Pretty _repr_html_ of cube dimension metadata

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Richer (HTML based) representation of cube dimension metadata in Jupyter context ([#827](https://github.com/Open-EO/openeo-python-client/issues/827))
+
 ### Changed
 
 - Ranged downloads (e.g. of batch job results) no longer implement their own retry loop; transient failures are retried by the standard urllib3 retry configuration of the connection's session ([#934](https://github.com/Open-EO/openeo-python-client/issues/934))

openeo/dates.py

@@ -7,6 +7,8 @@
 
 from openeo.util import rfc3339
 
+# TODO #465 move this to openeo.util.datetime?
+
 
 def get_temporal_extent(
     *args,

openeo/metadata/_extents.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import typing
+
+
+class _NonConcreteExtent:
+    """
+    Special marker class for non-concrete spatial, temporal, or other extents.
+    There can be several reasons for an extent to be non-concrete at a given time:
+    being generally unknown, being undefined or unspecified by the user,
+    being (unintendedly) unavailable/unresolvable, being dynamic or parameterized, etc.
+
+    The essential aspects this class signals:
+    - it is known that there is an extent
+    - but one or more actual bounds are non-concrete in some sense (unknown, dynamic, ...)
+
+    Note that this is different from, so this class should not be used for:
+    - the absence of an extent (e.g. no limits) or partial open bounds
+    - an empty extent
+    """
+
+    def __repr__(self) -> str:
+        return "<_NonConcreteExtent>"
+
+    def __eq__(self, value: object, /) -> bool:
+        if isinstance(value, _NonConcreteExtent):
+            # Non-concrete extents are not equal to each other
+            # even not to themselves (because they are non-concrete).
+            return False
+        # Still let third-party equality tooling hook in (e.g. dirty_equals.IsInstance)
+        return NotImplemented
+
+
+class _EmptyExtent:
+    """
+    Special marker class for an empty extent (spatial, temporal, or other), like the empty set.
+    The essential aspects this class signals: there is a known, concrete extent,
+    but has no coverage: nothing matches or falls within it.
+
+    Typical result of successive filter operations that have no overlap.
+
+    Note that this is different from:
+    - the absence of an extent (e.g. no limits) or partial open bounds
+    - a non-concrete extent (which is known to have coverage, but the actual bounds are unknown, dynamic, etc.)
+    """
+
+    def __repr__(self) -> str:
+        return "<_EmptyExtent>"
+
+    def __eq__(self, other) -> bool:
+        if isinstance(other, _EmptyExtent):
+            return True
+        return NotImplemented
+
+
+class NumericalInterval(typing.NamedTuple):
+    """
+    Simple numerical interval defined by a start and end point
+    each of type float/int;
+    or None to indicate being open-ended in that direction (no limit).
+    """
+
+    start: float | int | None
+    end: float | int | None
+
+    @classmethod
+    def try_from(cls, x: typing.Any) -> NumericalInterval | None:
+        if isinstance(x, (list, tuple)) and len(x) == 2 and all(isinstance(v, (int, float)) or v is None for v in x):
+            return NumericalInterval(*x)
+        return None
+
+    def intersection(self, other: NumericalInterval) -> NumericalInterval | _EmptyExtent:
+        """
+        Compute the intersection of this interval with another interval.
+        If the intervals do not overlap, return an _EmptyExtent.
+        """
+        if self.start is None:
+            start = other.start
+        elif other.start is None:
+            start = self.start
+        else:
+            start = max(self.start, other.start)
+
+        if self.end is None:
+            end = other.end
+        elif other.end is None:
+            end = self.end
+        else:
+            end = min(self.end, other.end)
+
+        if start is not None and end is not None and end < start:
+            return _EmptyExtent()
+
+        return NumericalInterval(start, end)

openeo/metadata/_jupyter.py

@@ -0,0 +1,48 @@
+"""
+HTML rendering helpers for metadata models (e.g. ``CubeMetadata``), used for Jupyter's ``_repr_html_``.
+
+Kept in a separate module (instead of directly in ``openeo.metadata``) to keep the core metadata
+data model free of rendering/presentation concerns.
+"""
+
+from __future__ import annotations
+
+import html
+from typing import List
+
+from openeo.metadata import (
+    BandDimension,
+    Dimension,
+    SpatialDimension,
+    TemporalDimension,
+)
+
+
+def repr_html_dimensions(dimensions: List[Dimension]) -> str:
+    """
+    Render a compact HTML table with an overview of cube dimensions
+    (name, type, size, labels/extent), based on available (client-side) metadata.
+    """
+
+    header = ["Dimension", "Type", "Size", "Labels / Extent"]
+    rows = []
+    for dim in dimensions:
+        row = [dim.name, str(dim.type)]
+        if isinstance(dim, BandDimension):
+            labels = ", ".join(str(name) if name else "?" for name in dim.band_names)
+            row.extend([str(len(dim.bands)), "bands: " + labels])
+        elif isinstance(dim, SpatialDimension):
+            row.extend(["?", f"extent: {repr(dim.extent)}"])
+        elif isinstance(dim, TemporalDimension):
+            row.extend(["?", f"extent: {repr(dim.extent)}"])
+        else:
+            row.extend(["?", ""])
+        rows.append("<tr>" + "".join(f"<td>{html.escape(c)}</td>" for c in row) + "</tr>")
+
+    thead = "<tr>" + "".join(f"<th>{html.escape(c)}</th>" for c in header) + "</tr>"
+    tbody = "".join(rows)
+
+    return f"""<table>
+            <thead>{thead}</thead>
+            <tbody>{tbody}</tbody>
+        </table>"""

openeo/rest/_datacube.py

@@ -120,7 +120,8 @@ def _repr_html_(self):
             "explicit-zoom": True,
             "height": "400px",
         }
-        return render_component("model-builder", data=process, parameters=parameters)
+        graph_html = render_component("model-builder", data=process, parameters=parameters)
+        return graph_html
 
 
 class UDF:

openeo/rest/datacube.py

@@ -114,6 +114,19 @@ def __init__(
         super().__init__(pgnode=graph, connection=connection)
         self.metadata: Optional[CubeMetadata] = metadata
 
+    def _repr_html_(self) -> str:
+        # Assumption: upstream _repr_html_ gives process graph representation
+        html = graph_html = super()._repr_html_()
+
+        if isinstance(self.metadata, CubeMetadata):
+            metadata_html = self.metadata._repr_html_()
+            html = f"""
+                <details open><summary>Metadata</summary>{metadata_html}</details>
+                <details><summary>Process Graph</summary>{graph_html}</details>
+            """
+
+        return html
+
     def process(
         self,
         process_id: str,
@@ -243,6 +256,10 @@ def load_collection(
             if (connection and fetch_metadata)
             else None
         )
+        if metadata and spatial_extent:
+            metadata = metadata.filter_bbox(extent=spatial_extent)
+        if metadata and temporal_extent:
+            metadata = metadata.filter_temporal(extent=temporal_extent)
         if bands is not None:
             bands = cls._get_bands(bands, process_id="load_collection")
             if isinstance(bands, Parameter):
@@ -440,7 +457,7 @@ def load_stac(
         """
         arguments: dict = {"url": url}
         if spatial_extent:
-            arguments["spatial_extent"] = _get_geometry_argument(
+            spatial_extent = _get_geometry_argument(
                 argument=spatial_extent,
                 valid_geojson_types=["Polygon", "MultiPolygon", "Feature", "FeatureCollection"],
                 connection=connection,
@@ -450,9 +467,12 @@ def load_stac(
                 argument_name="spatial_extent",
                 process_id="load_stac",
             )
+            arguments["spatial_extent"] = spatial_extent
 
         if temporal_extent:
-            arguments["temporal_extent"] = DataCube._get_temporal_extent(extent=temporal_extent)
+            temporal_extent = cls._get_temporal_extent(extent=temporal_extent)
+            arguments["temporal_extent"] = temporal_extent
+
         bands = cls._get_bands(bands, process_id="load_stac")
         if bands is not None:
             arguments["bands"] = bands
@@ -492,6 +512,12 @@ def load_stac(
                 f"Failed to extract cube metadata from STAC URL {url!r}. Falling back on no-metadata mode. Exception: {e!r}"
             )
             metadata = None
+
+        if metadata and spatial_extent:
+            metadata = metadata.filter_bbox(extent=spatial_extent)
+        if metadata and temporal_extent:
+            metadata = metadata.filter_temporal(extent=temporal_extent)
+
         return cls(graph=graph, connection=connection, metadata=metadata)
 
     @classmethod
@@ -578,12 +604,11 @@ def filter_temporal(
                 f" If you want a half-unbounded interval, use something like filter_temporal({args[0]!r}, None) or use explicit keyword arguments."
                 f" If you want the full interval covering all of {args[0]!r}, use something like filter_temporal(extent={args[0]!r})."
             )
+        extent = self._get_temporal_extent(*args, start_date=start_date, end_date=end_date, extent=extent)
         return self.process(
-            process_id='filter_temporal',
-            arguments={
-                'data': THIS,
-                'extent': self._get_temporal_extent(*args, start_date=start_date, end_date=end_date, extent=extent)
-            }
+            process_id="filter_temporal",
+            arguments={"data": THIS, "extent": extent},
+            metadata=self.metadata.filter_temporal(extent=extent) if self.metadata else None,
         )
 
     @openeo_process
@@ -693,12 +718,14 @@ def filter_bbox(
             extent = {'west': west, 'east': east, 'north': north, 'south': south}
             extent.update(dict_no_none(crs=crs, base=base, height=height))
 
+        metadata = self.metadata.filter_bbox(extent=extent) if self.metadata else None
         return self.process(
-            process_id='filter_bbox',
+            process_id="filter_bbox",
             arguments={
-                'data': THIS,
-                'extent': extent
-            }
+                "data": THIS,
+                "extent": extent,
+            },
+            metadata=metadata,
         )
 
     @openeo_process
@@ -754,12 +781,14 @@ def filter_spatial(
             "FeatureCollection",
         ]
         geometries = _get_geometry_argument(geometries, valid_geojson_types=valid_geojson_types, connection=self.connection, crs=None)
+        metadata = self.metadata.filter_bbox(extent=geometries) if self.metadata else None
         return self.process(
-            process_id='filter_spatial',
+            process_id="filter_spatial",
             arguments={
-                'data': THIS,
-                'geometries': geometries
-            }
+                "data": THIS,
+                "geometries": geometries,
+            },
+            metadata=metadata,
         )
 
     @openeo_process

openeo/rest/vectorcube.py

@@ -52,6 +52,19 @@ def __init__(self, graph: PGNode, connection: Union[Connection, None], metadata:
         super().__init__(pgnode=graph, connection=connection)
         self.metadata = metadata
 
+    def _repr_html_(self) -> str:
+        # Assumption: upstream _repr_html_ gives process graph representation
+        html = graph_html = super()._repr_html_()
+
+        if isinstance(self.metadata, CubeMetadata):
+            metadata_html = self.metadata._repr_html_()
+            html = f"""
+                <details open><summary>Metadata</summary>{metadata_html}</details>
+                <details><summary>Process Graph</summary>{graph_html}</details>
+            """
+
+        return html
+
     @classmethod
     def _build_metadata(cls, add_properties: bool = False) -> CubeMetadata:
         """Helper to build a (minimal) `CubeMetadata` object."""

openeo/util.py

@@ -62,6 +62,9 @@ class Rfc3339:
 
     Also see https://tools.ietf.org/html/rfc3339#section-5.6
     """
+
+    # TODO #465 move this to openeo.util.datetime
+
     # TODO: currently we hard code timezone 'Z' for simplicity. Add real time zone support?
     _FMT_DATE = '%Y-%m-%d'
     _FMT_TIME = '%H:%M:%SZ'
@@ -206,6 +209,7 @@ def now_utc(self) -> str:
 
 
 # Default RFC3339 date-time formatter
+# TODO #465 move this to openeo.util.datetime
 rfc3339 = Rfc3339()
 
 
@@ -531,7 +535,15 @@ class BBoxDict(dict):
     .. versionadded:: 0.10.1
     """
 
-    def __init__(self, *, west: float, south: float, east: float, north: float, crs: Optional[Union[str, int]] = None):
+    def __init__(
+        self,
+        *,
+        west: float,
+        south: float,
+        east: float,
+        north: float,
+        crs: int | str | dict | None = None,
+    ):
         super().__init__(west=west, south=south, east=east, north=north)
         if crs is not None:
             self.update(crs=normalize_crs(crs))
@@ -555,6 +567,8 @@ def from_any(cls, x: Any, *, crs: Optional[str] = None) -> BBoxDict:
     @classmethod
     def from_dict(cls, data: dict) -> BBoxDict:
         """Build from dictionary with at least keys "west", "south", "east", and "north"."""
+        if not isinstance(data, dict):
+            raise InvalidBBoxException(f"Expected dict, but got {type(data).__name__}.")
         expected_fields = {"west", "south", "east", "north"}
         # TODO: also support upper case fields?
         # TODO: optional support for parameterized bbox fields?
@@ -566,6 +580,14 @@ def from_dict(cls, data: dict) -> BBoxDict:
             raise InvalidBBoxException(f"Non-numerical bbox fields {invalid}.")
         return cls(west=data["west"], south=data["south"], east=data["east"], north=data["north"], crs=data.get("crs"))
 
+    @classmethod
+    def try_from_dict(cls, data: dict | Any) -> BBoxDict | None:
+        """Try to build from dictionary with at least keys "west", "south", "east", and "north"."""
+        try:
+            return cls.from_dict(data)
+        except ValueError:
+            return None
+
     @classmethod
     def from_sequence(cls, seq: Union[list, tuple], crs: Optional[str] = None) -> BBoxDict:
         """Build from sequence of 4 bounds (west, south, east and north)."""