Issue #460 use typing.NamedTuple for Band container

and some usage cleanups on the side

openeo/local/connection.py

@@ -236,7 +236,7 @@ def load_stac(
                 TemporalDimension(name=xarray_cube.openeo.temporal_dims[0], extent=[]),
                 BandDimension(
                     name=xarray_cube.openeo.band_dims[0],
-                    bands=[Band(x) for x in xarray_cube[xarray_cube.openeo.band_dims[0]].values],
+                    bands=[Band(name=x) for x in xarray_cube[xarray_cube.openeo.band_dims[0]].values],
                 ),
             ],
         )

openeo/metadata.py

@@ -1,11 +1,9 @@
 import logging
 import warnings
-from collections import namedtuple
-from typing import List, Union, Tuple, Callable, Any
+from typing import Any, Callable, List, NamedTuple, Optional, Tuple, Union
 
-from openeo.util import deep_get
 from openeo.internal.jupyter import render_component
-
+from openeo.util import deep_get
 
 _log = logging.getLogger(__name__)
 
@@ -82,9 +80,19 @@ def rename(self, name) -> 'Dimension':
         return TemporalDimension(name=name, extent=self.extent)
 
 
-# Simple container class for band metadata (incl. wavelength in micrometer)
-Band = namedtuple("Band", ["name", "common_name", "wavelength_um", "aliases", "gsd"])
-Band.__new__.__defaults__ = (None, None, None, None,)
+class Band(NamedTuple):
+    """
+    Simple container class for band metadata.
+    Based on https://github.com/stac-extensions/eo#band-object
+    """
+
+    name: str
+    common_name: Optional[str] = None
+    # wavelength in micrometer
+    wavelength_um: Optional[float] = None
+    aliases: Optional[List[str]] = None
+    # "openeo:gsd" field (https://github.com/Open-EO/openeo-stac-extensions#GSD-Object)
+    gsd: Optional[dict] = None
 
 
 class BandDimension(Dimension):
@@ -175,10 +183,15 @@ def rename_labels(self, target, source) -> 'Dimension':
             for old_name, new_name in zip(source, target):
                 band_index = self.band_index(old_name)
                 the_band = new_bands[band_index]
-                new_bands[band_index] = Band(new_name, the_band.common_name, the_band.wavelength_um, the_band.aliases,
-                                             the_band.gsd)
+                new_bands[band_index] = Band(
+                    name=new_name,
+                    common_name=the_band.common_name,
+                    wavelength_um=the_band.wavelength_um,
+                    aliases=the_band.aliases,
+                    gsd=the_band.gsd,
+                )
         else:
-            new_bands = [Band(name=n, common_name=None, wavelength_um=None) for n in target]
+            new_bands = [Band(name=n) for n in target]
         return BandDimension(name=self.name, bands=new_bands)
 
 
@@ -273,7 +286,7 @@ def _parse_dimensions(cls, spec: dict, complain: Callable[[str], None] = warning
             elif dim_type == "temporal":
                 dimensions.append(TemporalDimension(name=name, extent=info.get("extent")))
             elif dim_type == "bands":
-                bands = [Band(b, None, None) for b in info.get("values", [])]
+                bands = [Band(name=b) for b in info.get("values", [])]
                 if not bands:
                     complain("No band names in dimension {d!r}".format(d=name))
                 dimensions.append(BandDimension(name=name, bands=bands))
@@ -289,8 +302,16 @@ def _parse_dimensions(cls, spec: dict, complain: Callable[[str], None] = warning
         )
         if eo_bands:
             # center_wavelength is in micrometer according to spec
-            bands_detailed = [Band(b['name'], b.get('common_name'), b.get('center_wavelength'), b.get('aliases'),
-                                   b.get('openeo:gsd')) for b in eo_bands]
+            bands_detailed = [
+                Band(
+                    name=b["name"],
+                    common_name=b.get("common_name"),
+                    wavelength_um=b.get("center_wavelength"),
+                    aliases=b.get("aliases"),
+                    gsd=b.get("openeo:gsd"),
+                )
+                for b in eo_bands
+            ]
             # Update band dimension with more detailed info
             band_dimensions = [d for d in dimensions if d.type == "bands"]
             if len(band_dimensions) == 1:
@@ -439,7 +460,7 @@ def add_dimension(self, name: str, label: Union[str, float], type: str = None) -
         if any(d.name == name for d in self._dimensions):
             raise DimensionAlreadyExistsException(f"Dimension with name {name!r} already exists")
         if type == "bands":
-            dim = BandDimension(name=name, bands=[Band(label, None, None)])
+            dim = BandDimension(name=name, bands=[Band(name=label)])
         elif type == "spatial":
             dim = SpatialDimension(name=name, extent=[label, label])
         elif type == "temporal":

openeo/rest/connection.py

@@ -1160,12 +1160,15 @@ def load_result(
         :return: a :py:class:`DataCube`
         """
         # TODO: add check that back-end supports `load_result` process?
-        metadata = CollectionMetadata({}, dimensions=[
-            SpatialDimension(name="x", extent=[]),
-            SpatialDimension(name="y", extent=[]),
-            TemporalDimension(name='t', extent=[]),
-            BandDimension(name="bands", bands=[Band("unknown")]),
-        ])
+        metadata = CollectionMetadata(
+            {},
+            dimensions=[
+                SpatialDimension(name="x", extent=[]),
+                SpatialDimension(name="y", extent=[]),
+                TemporalDimension(name="t", extent=[]),
+                BandDimension(name="bands", bands=[Band(name="unknown")]),
+            ],
+        )
         cube = self.datacube_from_process(
             process_id="load_result",
             id=id,
@@ -1289,7 +1292,7 @@ def load_stac(
                 SpatialDimension(name="x", extent=[]),
                 SpatialDimension(name="y", extent=[]),
                 TemporalDimension(name="t", extent=[]),
-                BandDimension(name="bands", bands=[Band("unknown")]),
+                BandDimension(name="bands", bands=[Band(name="unknown")]),
             ],
         )
         arguments = {"url": url}

openeo/rest/datacube.py

@@ -149,7 +149,7 @@ def load_collection(
                 metadata = metadata.filter_bands(bands)
             else:
                 # Ensure minimal metadata with best effort band dimension guess (based on `bands` argument).
-                band_dimension = BandDimension("bands", bands=[Band(b, None, None) for b in bands])
+                band_dimension = BandDimension("bands", bands=[Band(name=b) for b in bands])
                 metadata = CollectionMetadata({}, dimensions=[band_dimension])
             arguments['bands'] = bands
         if max_cloud_cover:
@@ -201,12 +201,15 @@ def load_disk_collection(cls, connection: 'openeo.Connection', file_format: str,
             }
         )
 
-        metadata = CollectionMetadata({}, dimensions=[
-            SpatialDimension(name="x", extent=[]),
-            SpatialDimension(name="y", extent=[]),
-            TemporalDimension(name='t', extent=[]),
-            BandDimension(name="bands", bands=[Band("unknown")]),
-        ])
+        metadata = CollectionMetadata(
+            {},
+            dimensions=[
+                SpatialDimension(name="x", extent=[]),
+                SpatialDimension(name="y", extent=[]),
+                TemporalDimension(name="t", extent=[]),
+                BandDimension(name="bands", bands=[Band(name="unknown")]),
+            ],
+        )
         return cls(graph=pg, connection=connection, metadata=metadata)
 
     @classmethod
@@ -1472,7 +1475,7 @@ def ndvi(self, nir: str = None, red: str = None, target_band: str = None) -> 'Da
         if target_band is None:
             metadata = self.metadata.reduce_dimension(self.metadata.band_dimension.name)
         else:
-            metadata = self.metadata.append_band(Band(target_band, "ndvi", None))
+            metadata = self.metadata.append_band(Band(name=target_band, common_name="ndvi"))
         return self.process(
             process_id="ndvi",
             arguments=dict_no_none(

tests/test_metadata.py

@@ -27,6 +27,11 @@ def test_metadata_extent():
     assert metadata.extent == {"spatial": {"xmin": 4, "xmax": 10}}
 
 
+def test_band_minimal():
+    band = Band("red")
+    assert band.name == "red"
+
+
 def test_band_dimension():
     bdim = BandDimension(name="spectral", bands=[
         Band("B02", "blue", 0.490),