feat: add backwards compatibility for the expression parameter

Author: hrodmn

tests/test_dependencies.py

@@ -4,7 +4,11 @@
 
 
 from titiler.cmr import dependencies
-from titiler.cmr.dependencies import InterpolatedXarrayParams
+from titiler.cmr.dependencies import (
+    CMRAssetsExprParams,
+    CMRXarrayExprParams,
+    InterpolatedXarrayParams,
+)
 from titiler.cmr.models import GranuleSearch
 
 
@@ -24,6 +28,40 @@ def test_granule_search_temporal_interval_unchanged():
     assert GranuleSearch(temporal=half_open).temporal == half_open
 
 
+def test_cmr_assets_expr_params_three_assets():
+    """Legacy expression with three assets: all detected and mapped in appearance order."""
+    params = CMRAssetsExprParams(expression="(NIR-RED)/(NIR+RED+BLUE)")
+    assert list(params.assets) == ["NIR", "RED", "BLUE"]
+    assert params.expression == "(b1-b2)/(b1+b2+b3)"
+
+
+def test_cmr_assets_expr_params_substring_asset_names():
+    """Asset names that are substrings of each other are distinguished by word boundaries."""
+    params = CMRAssetsExprParams(assets=["B4", "B4_mask"], expression="B4 * B4_mask")
+    assert params.expression == "b1 * b2"
+
+
+def test_cmr_assets_expr_params_math_functions():
+    """Math function names (sqrt, log) are not treated as asset names."""
+    params = CMRAssetsExprParams(expression="sqrt(NIR)/log(RED)")
+    assert list(params.assets) == ["NIR", "RED"]
+    assert params.expression == "sqrt(b1)/log(b2)"
+
+
+def test_cmr_assets_expr_params_extra_assets():
+    """Assets list with more entries than referenced in expression: order preserved, extras ignored."""
+    params = CMRAssetsExprParams(assets=["B03", "B04", "B05"], expression="B04-B05")
+    assert list(params.assets) == ["B03", "B04", "B05"]
+    assert params.expression == "b2-b3"
+
+
+def test_cmr_assets_expr_params_b_prefix_asset_names():
+    """Asset names starting with 'b' but not new-style (e.g. blue, band1) are treated as legacy."""
+    params = CMRAssetsExprParams(expression="(blue-red)/(blue+red)")
+    assert list(params.assets) == ["blue", "red"]
+    assert params.expression == "(b1-b2)/(b1+b2)"
+
+
 def test_interpolated_xarray_params_single_datetime():
     """Test InterpolatedXarrayParams with single datetime interpolation."""
     xarray_params = InterpolatedXarrayParams(
@@ -94,6 +132,72 @@ def test_interpolated_xarray_params_no_sel():
     assert result.variables == ["temperature"]
 
 
+def test_cmr_assets_expr_params_legacy_no_assets():
+    """Legacy expression with no assets: auto-detect assets and translate expression."""
+    params = CMRAssetsExprParams(expression="(B04-B05)/(B05+B04)")
+    assert list(params.assets) == ["B04", "B05"]
+    assert params.expression == "(b1-b2)/(b2+b1)"
+
+
+def test_cmr_assets_expr_params_legacy_with_assets():
+    """Legacy expression with assets provided: use assets order for mapping."""
+    params = CMRAssetsExprParams(
+        assets=["B05", "B04"], expression="(B04-B05)/(B04+B05)"
+    )
+    assert list(params.assets) == ["B05", "B04"]
+    assert params.expression == "(b2-b1)/(b2+b1)"
+
+
+def test_cmr_assets_expr_params_new_style_passthrough():
+    """New-style expression (b1, b2, ...) passes through unchanged."""
+    params = CMRAssetsExprParams(assets=["B04", "B05"], expression="(b1-b2)/(b1+b2)")
+    assert list(params.assets) == ["B04", "B05"]
+    assert params.expression == "(b1-b2)/(b1+b2)"
+
+
+def test_cmr_assets_expr_params_no_expression():
+    """No expression: assets unchanged, no error."""
+    params = CMRAssetsExprParams(assets=["B04"])
+    assert list(params.assets) == ["B04"]
+    assert params.expression is None
+
+
+def test_cmr_xarray_expr_params_legacy_variable_names():
+    """Legacy variable names are translated to positional bN format."""
+    params = CMRXarrayExprParams(
+        variables=["temperature", "pressure"], expression="temperature/pressure"
+    )
+    assert params.expression == "b1/b2"
+
+
+def test_cmr_xarray_expr_params_legacy_partial_match():
+    """Legacy NDVI-style expression with partial variable subset."""
+    params = CMRXarrayExprParams(
+        variables=["nir", "red", "green"], expression="(nir-red)/(nir+red)"
+    )
+    assert params.expression == "(b1-b2)/(b1+b2)"
+
+
+def test_cmr_xarray_expr_params_new_style_passthrough():
+    """New-style bN expressions pass through unchanged."""
+    params = CMRXarrayExprParams(variables=["nir", "red"], expression="(b1-b2)/(b1+b2)")
+    assert params.expression == "(b1-b2)/(b1+b2)"
+
+
+def test_cmr_xarray_expr_params_with_math_functions():
+    """Math function names are not treated as variable names."""
+    params = CMRXarrayExprParams(
+        variables=["nir", "red"], expression="log10(nir)/sqrt(red)"
+    )
+    assert params.expression == "log10(b1)/sqrt(b2)"
+
+
+def test_cmr_xarray_expr_params_no_expression():
+    """No expression: no error, variables unchanged."""
+    params = CMRXarrayExprParams(variables=["nir"], expression=None)
+    assert params.expression is None
+
+
 def test_interpolated_xarray_params_multiple_templates():
     """Test InterpolatedXarrayParams with multiple datetime templates."""
     xarray_params = InterpolatedXarrayParams(

titiler/cmr/dependencies.py

@@ -1,13 +1,20 @@
 """titiler.cmr FastAPI dependencies."""
 
+import re
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Annotated, List, Optional
 
 from fastapi import Depends, HTTPException, Query, Request
 from httpx import Client
-from titiler.core.dependencies import DefaultDependency, ExpressionParams
+from pydantic import AfterValidator
+from titiler.core.dependencies import (
+    AssetsExprParams,
+    DefaultDependency,
+    ExpressionParams,
+    _parse_asset,
+)
 from titiler.xarray.dependencies import SelDimStr, XarrayIOParams
 
 from titiler.cmr.models import (
@@ -58,7 +65,11 @@ def GranuleSearchParams(
 
 @dataclass(init=False)
 class BackendParams(DefaultDependency):
-    """backend parameters."""
+    """Reader backend parameters sourced from application state.
+
+    Reads the HTTP client, Earthdata auth token, S3 access flag, and S3
+    credential provider from the FastAPI app state on each request.
+    """
 
     client: Client = field(init=False)
     auth_token: str | None = field(init=False)
@@ -76,7 +87,7 @@ def __init__(self, request: Request):
 
 @dataclass
 class GranuleSearchBackendParams(DefaultDependency):
-    """PgSTAC parameters."""
+    """Backend parameters controlling granule search coverage behaviour."""
 
     items_limit: Annotated[
         int | None,
@@ -118,6 +129,72 @@ def __post_init__(self):
         self.bands_regex = None
 
 
+def _translate_legacy_expr(expression: str, names: list[str]) -> str:
+    """Translate legacy name-based expression to positional bN format.
+
+    If the expression contains identifiers from `names` (not already bN-style,
+    not function calls), replaces them with b1, b2, ... based on their position
+    in `names`.
+    """
+    identifiers = re.findall(r"\b([a-zA-Z_]\w*)\b(?!\s*\()", expression)
+    new_style = re.compile(r"^b[1-9][0-9]*$", re.IGNORECASE)
+    legacy = list(dict.fromkeys(n for n in identifiers if not new_style.match(n)))
+    if not legacy:
+        return expression
+    mapping = {name: f"b{i + 1}" for i, name in enumerate(names)}
+    expr = expression
+    for name, band_ref in mapping.items():
+        expr = re.sub(r"\b" + re.escape(name) + r"\b", band_ref, expr)
+    return expr
+
+
+@dataclass
+class CMRAssetsExprParams(AssetsExprParams):
+    """AssetsExprParams with backwards-compatible legacy expression translation.
+
+    Detects legacy expressions that reference asset names directly (e.g. B04, NIR)
+    and translates them to the new rio-tiler 9.0 positional band format (b1, b2, ...).
+    """
+
+    assets: Annotated[
+        list[str] | None,
+        AfterValidator(_parse_asset),
+        Query(
+            title="Asset names",
+            description="Asset's names.",
+        ),
+    ] = None
+
+    def __post_init__(self):
+        """Translate legacy asset-name expressions to positional bN format.
+
+        If the expression already uses bN references (e.g. b1-b2), it is left
+        unchanged. Otherwise, identifiers are matched against the provided or
+        auto-detected asset list and substituted with b1, b2, ... in order.
+        """
+        if not self.expression:
+            return
+
+        identifiers = re.findall(r"\b([a-zA-Z_]\w*)\b(?!\s*\()", self.expression)
+        new_style_pattern = re.compile(r"^b[1-9][0-9]*$", re.IGNORECASE)
+        asset_names = list(
+            dict.fromkeys(
+                name for name in identifiers if not new_style_pattern.match(name)
+            )
+        )
+
+        if not asset_names:
+            return
+
+        if self.assets:
+            ordered_assets = list(self.assets)
+        else:
+            ordered_assets = asset_names
+            self.assets = ordered_assets
+
+        self.expression = _translate_legacy_expr(self.expression, ordered_assets)
+
+
 @dataclass
 class XarrayDsParams(DefaultDependency):
     """Xarray Dataset Options."""
@@ -152,12 +229,29 @@ class InterpolatedXarrayParams(XarrayParams):
     ] = None
 
 
+@dataclass
+class CMRXarrayExprParams(InterpolatedXarrayParams):
+    """InterpolatedXarrayParams with legacy variable-name expression translation.
+
+    Translates expressions like `temperature/pressure` to `b1/b2` based on the
+    order of `variables`.
+    """
+
+    def __post_init__(self):
+        """Translate legacy variable-name expressions to positional bN format.
+
+        Skipped when expression is already new-style (contains only bN refs) or
+        when no expression is provided. Safe to call more than once — already-
+        translated expressions are returned unchanged.
+        """
+        if self.expression and self.variables:
+            self.expression = _translate_legacy_expr(self.expression, self.variables)
+
+
 def interpolated_xarray_ds_params(
-    xarray_params: Annotated[
-        InterpolatedXarrayParams, Depends(InterpolatedXarrayParams)
-    ],
+    xarray_params: Annotated[CMRXarrayExprParams, Depends(CMRXarrayExprParams)],
     granule_search: Annotated[GranuleSearch, Depends(GranuleSearchParams)],
-) -> InterpolatedXarrayParams:
+) -> CMRXarrayExprParams:
     """
     Xarray parameters with string interpolation support for the sel parameter.
 
@@ -184,9 +278,10 @@ def interpolated_xarray_ds_params(
         else:
             interpolated_sel.append(sel_item)
 
-    return InterpolatedXarrayParams(
+    return CMRXarrayExprParams(
         variables=xarray_params.variables,
         group=xarray_params.group,
         sel=interpolated_sel,
         decode_times=xarray_params.decode_times,
+        expression=xarray_params.expression,
     )

titiler/cmr/expression.py

@@ -0,0 +1,80 @@
+"""Expression evaluation utilities for xarray band math."""
+
+from typing import Any
+
+import numpy as np
+import xarray as xr
+from rio_tiler.expression import get_expression_blocks
+
+from titiler.cmr.logger import logger
+
+_MATH_FUNCTIONS: dict[str, Any] = {
+    "abs": np.abs,
+    "ceil": np.ceil,
+    "floor": np.floor,
+    "round": np.round,
+    "trunc": np.trunc,
+    "sign": np.sign,
+    "sqrt": np.sqrt,
+    "exp": np.exp,
+    "expm1": np.expm1,
+    "log": np.log,
+    "log1p": np.log1p,
+    "log10": np.log10,
+    "log2": np.log2,
+    "sin": np.sin,
+    "cos": np.cos,
+    "tan": np.tan,
+    "arcsin": np.arcsin,
+    "arccos": np.arccos,
+    "arctan": np.arctan,
+    "arctan2": np.arctan2,
+    "sinh": np.sinh,
+    "cosh": np.cosh,
+    "tanh": np.tanh,
+    "arcsinh": np.arcsinh,
+    "arccosh": np.arccosh,
+    "arctanh": np.arctanh,
+    "isnan": np.isnan,
+    "isfinite": np.isfinite,
+    "isinf": np.isinf,
+    "signbit": np.signbit,
+    "fmod": np.fmod,
+    "hypot": np.hypot,
+    "maximum": np.maximum,
+    "minimum": np.minimum,
+    "where": np.where,
+}
+
+
+def apply_expression(
+    da: xr.DataArray,
+    expression: str,
+) -> xr.DataArray:
+    """Evaluate a band-math expression against a DataArray.
+
+    The DataArray must have a "band" dimension. Each band is exposed as b1, b2, ...
+    in the expression namespace, along with numexpr-compatible math functions and
+    the full `np` and `xr` namespaces for backwards compatibility.
+
+    Args:
+        da: Input DataArray with a "band" dimension.
+        expression: Band-math expression string (e.g. "log10(b1)/sqrt(b2)").
+
+    Returns:
+        Result DataArray, preserving the CRS if present.
+    """
+    logger.info(f"applying expression: {expression}")
+    pre_expression_crs = da.rio.crs
+    expression_blocks = get_expression_blocks(expression)
+    band_vars = {
+        f"b{i + 1}": da.isel(band=i, drop=True) for i in range(da.sizes["band"])
+    }
+    namespace = {"np": np, "xr": xr, **_MATH_FUNCTIONS, **band_vars}
+    results = [
+        eval(block, {"__builtins__": {}}, namespace) for block in expression_blocks
+    ]
+    result = results[0] if len(results) == 1 else xr.concat(results, dim="band")
+    if pre_expression_crs is not None:
+        result = result.rio.write_crs(pre_expression_crs)
+    return result