Merge pull request #128 from ocean-uhh/103-feat-new-layer-conversiontransformation-for-internal-pipeline

feat: New layer conversion/transformation for internal pipeline

Author: ysorge

seasenselib/cli/commands/data_commands.py

@@ -144,18 +144,14 @@ def _build_stage_kwargs(args):
 
     # Check for --pipeline-skip-stages argument
     elif getattr(args, 'pipeline_skip_stages', None):
-        # Build a list of all steps except the skipped ones
         skip = [s.strip() for s in args.pipeline_skip_stages.split(',')]
         try:
-            from ...pipeline.registry import StageRegistry
             from ...pipeline.config import PipelineConfig
-            registry = StageRegistry.get_instance()
-            all_steps = registry.list_stages()
-            selected_steps = [s for s in all_steps if s not in skip]
-            # Build pipeline config with selected steps
-            config = PipelineConfig()
-            for step_name in selected_steps:
-                config.add_stage(step_name)
+            config = PipelineConfig.from_resource("default")
+            config.pipeline = [
+                stage for stage in config.pipeline
+                if stage.name not in skip
+            ]
             stage_kwargs['pipeline_config'] = config
         except Exception:
             # If registry fails, just pass the skip info and let read() handle it
@@ -272,6 +268,7 @@ def _write_processing_protocol(
         protocol["handlers_applied"] = metadata.get("handlers_applied")
         protocol["variable_mappings"] = metadata.get("variable_mappings")
         protocol["derived_parameters"] = metadata.get("derived_parameters")
+        protocol["transformations"] = metadata.get("transformations")
         unit_conversions = _format_unit_conversions(metadata.get("unit_conversions"))
         if unit_conversions is not None:
             protocol["unit_conversions"] = unit_conversions
@@ -296,6 +293,112 @@ def _write_processing_protocol(
         json.dump(protocol, f, indent=2, sort_keys=True)
 
 
+def _format_example_value(value) -> str:
+    """Format one scalar preview value without verbose dtype wrappers."""
+    try:
+        import numpy as np
+    except Exception:
+        np = None
+
+    if np is not None:
+        if isinstance(value, np.datetime64):
+            return np.datetime_as_string(value)
+        if isinstance(value, np.timedelta64):
+            return str(value)
+        if isinstance(value, np.generic):
+            value = value.item()
+
+    if isinstance(value, bytes):
+        try:
+            return value.decode("utf-8")
+        except UnicodeDecodeError:
+            return value.hex()
+    if isinstance(value, float):
+        return f"{value:.6g}"
+    return str(value)
+
+
+def _example_selector(array, max_values: int) -> tuple[dict, str, int]:
+    """Return a small indexer and sampled dimension for an array preview."""
+    if not array.dims:
+        return {}, "", 1
+
+    sample_dim = "time" if "time" in array.dims else array.dims[-1]
+    sample_count = min(int(array.sizes[sample_dim]), max_values)
+    indexer = {}
+    for dim in array.dims:
+        if dim == sample_dim:
+            indexer[dim] = slice(0, sample_count)
+        else:
+            indexer[dim] = 0
+    return indexer, sample_dim, sample_count
+
+
+def _format_example_selector(array, indexer: dict, sample_dim: str) -> str:
+    """Describe the small indexer used for an example preview."""
+    if not array.dims:
+        return ""
+
+    parts = []
+    for dim in array.dims:
+        selector = indexer[dim]
+        if dim == sample_dim:
+            parts.append(f"{dim}=0:{selector.stop}")
+            continue
+
+        label = "0"
+        if dim in array.coords and array.coords[dim].size:
+            try:
+                label = _format_example_value(array.coords[dim].isel({dim: 0}).values)
+            except Exception:
+                label = "0"
+        parts.append(f"{dim}={label}")
+    return ", ".join(parts)
+
+
+def _format_array_example(name: str, array, max_values: int = 5) -> str:
+    """Format a bounded example line for one xarray coordinate or variable."""
+    indexer, sample_dim, _sample_count = _example_selector(array, max_values)
+    subset = array.isel(indexer) if indexer else array
+
+    try:
+        values = subset.values
+    except Exception as exc:
+        return f"  {name}: <failed to read preview: {exc}>"
+
+    try:
+        import numpy as np
+        flat = np.asarray(values).reshape(-1)
+    except Exception:
+        flat = [values]
+
+    preview = ", ".join(
+        _format_example_value(value)
+        for value in list(flat)[:max_values]
+    )
+    selector_text = _format_example_selector(array, indexer, sample_dim)
+    selector_text = f" [{selector_text}]" if selector_text else ""
+    return (
+        f"  {name}{selector_text}: dims={array.dims}, "
+        f"shape={tuple(array.shape)}, values=[{preview}]"
+    )
+
+
+def _print_dataset_example(dataset, max_values: int = 5) -> None:
+    """Print bounded examples without materializing the whole Dataset."""
+    print(f"Example values (up to {max_values} values per variable):")
+
+    if dataset.coords:
+        print("\nCoordinates:")
+        for name, array in dataset.coords.items():
+            print(_format_array_example(name, array, max_values=max_values))
+
+    if dataset.data_vars:
+        print("\nData variables:")
+        for name, array in dataset.data_vars.items():
+            print(_format_array_example(name, array, max_values=max_values))
+
+
 class ConvertCommand(BaseCommand):
     """Handle file conversion with lazy loading."""
 
@@ -428,8 +531,7 @@ def execute(self, args: argparse.Namespace) -> CommandResult:
             elif args.schema == 'info':
                 data.info()
             elif args.schema == 'example':
-                df = data.to_dataframe()
-                print(df.head())
+                _print_dataset_example(data)
 
             # Write processing protocol if requested
             if want_protocol:

seasenselib/config/pipeline/default.json

@@ -20,6 +20,14 @@
         ]
       }
     },
+    {
+      "name": "transformation",
+      "config": {
+        "handlers": [
+          "reader"
+        ]
+      }
+    },
     {
       "name": "derivation",
       "config": {

seasenselib/config/pipeline/full.json

@@ -21,6 +21,14 @@
         ]
       }
     },
+    {
+      "name": "transformation",
+      "config": {
+        "handlers": [
+          "reader"
+        ]
+      }
+    },
     {
       "name": "derivation",
       "config": {

seasenselib/pipeline/__init__.py

@@ -8,6 +8,8 @@
 from .pipeline import Pipeline
 from .config import PipelineConfig, StageConfig
 from .registry import StageRegistry
+from .interfaces import ITransformation, TransformationRecord
+from .transformation import TransformationStage
 from .factory import (
     default_pipeline,
     minimal_pipeline,
@@ -22,6 +24,9 @@
     "PipelineConfig",
     "StageConfig",
     "StageRegistry",
+    "ITransformation",
+    "TransformationRecord",
+    "TransformationStage",
     "default_pipeline",
     "minimal_pipeline",
     "create_pipeline",

seasenselib/pipeline/factory.py

@@ -19,11 +19,12 @@ def default_pipeline() -> Pipeline:
     The default pipeline includes (in order):
     1. Mapping
     2. Unit Handling
-    3. Derivation
-    4. Metadata Extraction
-    5. Metadata Enrichment
-    6. Validation
-    7. Finalization
+    3. Transformation
+    4. Derivation
+    5. Metadata Extraction
+    6. Metadata Enrichment
+    7. Validation
+    8. Finalization
     """
     config = PipelineConfig.from_resource("default")
     return create_pipeline(config=config)

seasenselib/pipeline/finalization/handlers/global_attributes.py

@@ -149,5 +149,9 @@ def _build_history_entry(self, timestamp: str, context: StageContext) -> str:
         if 'derived_parameters' in context.metadata:
             derived = ', '.join(context.metadata['derived_parameters'])
             parts.append(f"Derived: {derived}")
+
+        if 'transformations' in context.metadata:
+            count = len(context.metadata['transformations'])
+            parts.append(f"Transformed: {count} step(s)")
 
         return '; '.join(parts)

seasenselib/pipeline/finalization/handlers/processor_metadata.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Dict, Any
+import json
 import logging
 import platform
 
@@ -72,6 +73,14 @@ def set_attr(key: str, value: Any) -> None:
         if self.include_os:
             set_attr("processor_os", f"{platform.system()} {platform.release()}")
 
+        transformations = meta.get("transformations")
+        if transformations:
+            set_attr(
+                "processor_transformations",
+                json.dumps(transformations, ensure_ascii=False, default=str),
+            )
+            set_attr("processor_transformations_count", len(transformations))
+
         context.dataset = ds
         logger.debug("Added processor metadata attributes")
         return context

seasenselib/pipeline/handler_catalog.py

@@ -19,13 +19,15 @@
     HANDLER_GROUP_DERIVATIONS,
     HANDLER_GROUP_METADATA_EXTRACTORS,
     HANDLER_GROUP_CONVENTIONS,
+    HANDLER_GROUP_TRANSFORMATIONS,
     HANDLER_GROUP_VALIDATORS,
 )
 from .interfaces import (
     IMappingStrategy,
     IDerivation,
     IMetadataExtractor,
     IConvention,
+    ITransformation,
     IValidator,
 )
 
@@ -56,6 +58,9 @@
         "normalize": "seasenselib.pipeline.unit_handling.handlers.unit_normalizer.UnitNormalizer",
         "convert": "seasenselib.pipeline.unit_handling.handlers.unit_converter.UnitConverter",
     },
+    "transformation": {
+        "reader": "seasenselib.pipeline.transformation.handlers.reader_transformations.ReaderTransformations",
+    },
     "metadata_enrichment": {
         "cf": "seasenselib.pipeline.metadata_enrichment.handlers.cf_convention.CFConvention",
         "acdd": "seasenselib.pipeline.metadata_enrichment.handlers.acdd_convention.ACDDConvention",
@@ -81,6 +86,7 @@
     "derivation": (HANDLER_GROUP_DERIVATIONS, IDerivation),
     "metadata_extraction": (HANDLER_GROUP_METADATA_EXTRACTORS, IMetadataExtractor),
     "metadata_enrichment": (HANDLER_GROUP_CONVENTIONS, IConvention),
+    "transformation": (HANDLER_GROUP_TRANSFORMATIONS, ITransformation),
     "validation": (HANDLER_GROUP_VALIDATORS, IValidator),
 }
 

seasenselib/pipeline/handler_registry.py

@@ -17,6 +17,7 @@
 HANDLER_GROUP_DERIVATIONS = "seasenselib.pipeline.derivations"
 HANDLER_GROUP_METADATA_EXTRACTORS = "seasenselib.pipeline.metadata_extractors"
 HANDLER_GROUP_CONVENTIONS = "seasenselib.pipeline.conventions"
+HANDLER_GROUP_TRANSFORMATIONS = "seasenselib.pipeline.transformations"
 HANDLER_GROUP_VALIDATORS = "seasenselib.pipeline.validators"
 
 
@@ -85,5 +86,6 @@ def get(cls, group: str, base_class: Type[T]) -> Dict[str, Type[T]]:
     "HANDLER_GROUP_DERIVATIONS",
     "HANDLER_GROUP_METADATA_EXTRACTORS",
     "HANDLER_GROUP_CONVENTIONS",
+    "HANDLER_GROUP_TRANSFORMATIONS",
     "HANDLER_GROUP_VALIDATORS",
 ]