"""This module provides functions to retrieve and process map visualization parameters."""
import json
import logging
import warnings
from typing import Mapping, Optional, Sequence, Union
import ee
from pysepal import color as scolors
from pysepal.frontend import styles as ss
from pysepal.scripts.gee_interface import GEEInterface
from pysepal.scripts.warning import SepalWarning
log = logging.getLogger("sepalui.mapping.visualization")
VIZ_LIST_KEYS = ("bands", "palette", "labels")
VIZ_NUMERIC_LIST_KEYS = ("min", "max", "values")
VIZ_BOOL_LIST_KEYS = ("inverted",)
VIZ_SCALAR_KEYS = ("name", "type")
VIZ_ALL_KEYS = (
*VIZ_LIST_KEYS,
*VIZ_NUMERIC_LIST_KEYS,
*VIZ_BOOL_LIST_KEYS,
*VIZ_SCALAR_KEYS,
)
def _strtobool(val: str) -> bool:
"""Convert a string representation of truth to True or False.
This function replaces the deprecated distutils.util.strtobool.
Args:
val: A string representation of a boolean value. Accepts:
- "true", "false" (case insensitive)
- "yes", "no" (case insensitive)
- "1", "0"
- "on", "off" (case insensitive)
- Empty string defaults to False
Returns:
The boolean value represented by the string.
Raises:
ValueError: If the string cannot be converted to a boolean.
"""
if not val: # Handle empty strings
return False
val = val.lower()
if val in ("y", "yes", "t", "true", "on", "1"):
return True
elif val in ("n", "no", "f", "false", "off", "0"):
return False
else:
raise ValueError(f"invalid truth value {val!r}")
PREFIX = "visualization"
"""the constant prefix for SEPAL visualization parameters"""
[docs]
async def get_viz_params_async(
ee_object: ee.ComputedObject,
gee_interface: Optional[GEEInterface] = None,
) -> tuple:
"""Asynchronously retrieve and process visualization parameters for a given Earth Engine object.
Args:
ee_object: The Earth Engine object (Image, FeatureCollection, etc.) to process.
gee_interface: The GEE interface to use for fetching properties. If not provided, a new instance will be created.
Returns:
A tuple containing the processed image, object, and visualization parameters.
Raises:
AttributeError: If the provided Earth Engine object is not of a recognized type.
"""
validate_ee_object(ee_object)
props = {}
if gee_interface is None:
gee_interface = GEEInterface()
raw_prop_list = await get_props_list_async(gee_interface, ee_object)
return process_props(raw_prop_list, props)
[docs]
def get_viz_params(
ee_object: ee.ComputedObject,
gee_interface: Optional[GEEInterface] = None,
) -> tuple:
"""Retrieve and process visualization parameters for a given Earth Engine object.
Args:
ee_object: The Earth Engine object (Image, FeatureCollection, etc.) to process.
gee_interface: The GEE interface to use for fetching properties. If not provided, a new instance will be created.
Returns:
A tuple containing the processed image, object, and visualization parameters.
Raises:
AttributeError: If the provided Earth Engine object is not of a recognized type.
"""
validate_ee_object(ee_object)
props = {}
if gee_interface is None:
gee_interface = GEEInterface()
raw_prop_list = get_props_list(gee_interface, ee_object)
return process_props(raw_prop_list, props)
[docs]
def get_props_list(gee_interface: GEEInterface, ee_object: ee.ComputedObject) -> list:
"""Retrieve the list of visualization properties from an Earth Engine object.
Args:
gee_interface: The GEE interface to use for fetching properties.
ee_object: The Earth Engine object (Image, FeatureCollection, etc.) to process.
Returns:
A list of visualization properties.
Raises:
AttributeError: If the provided Earth Engine object is not of a recognized type.
"""
if not isinstance(ee_object, ee.Image):
return []
prop_names = ee_object.propertyNames()
viz_props = prop_names.filter(ee.Filter.stringStartsWith("item", PREFIX))
# Check if there are any visualization properties
if gee_interface.get_info(viz_props.size()) == 0:
log.warning("Image has no visualization properties, returning empty viz params")
return []
viz_dict = ee_object.toDictionary(viz_props)
raw_prop_list = gee_interface.get_info(viz_dict)
return raw_prop_list
[docs]
async def get_props_list_async(gee_interface: GEEInterface, ee_object: ee.ComputedObject) -> list:
"""Asynchronously retrieve the list of visualization properties from an Earth Engine object.
Args:
gee_interface: The GEE interface to use for fetching properties.
ee_object: The Earth Engine object (Image, FeatureCollection, etc.) to process.
Returns:
A list of visualization properties.
Raises:
AttributeError: If the provided Earth Engine object is not of a recognized type.
"""
if not isinstance(ee_object, ee.Image):
return []
prop_names = ee_object.propertyNames()
viz_props = prop_names.filter(ee.Filter.stringStartsWith("item", PREFIX))
# Check if there are any visualization properties
if await gee_interface.get_info_async(viz_props.size()) == 0:
log.warning("Image has no visualization properties, returning empty viz params")
return []
# Get only the visualization properties as a dictionary
viz_dict = ee_object.toDictionary(viz_props)
raw_prop_list = await gee_interface.get_info_async(viz_dict)
return raw_prop_list
[docs]
def process_props(raw_prop_list: Optional[list], props: dict = {}) -> dict:
"""Process the raw visualization properties into a structured dictionary.
Args:
raw_prop_list: The raw list of properties retrieved from the Earth Engine object.
props: An optional dictionary to store processed properties.
Returns:
A dictionary containing processed visualization properties.
"""
if not raw_prop_list:
return props
# decompose each property by its number
# and gather the properties in a sub dictionary
for p, val in raw_prop_list.items():
# extract the number and create the sub-dict
_, number, name = p.split("_")
props.setdefault(number, {})
# modify the values according to prop key
if isinstance(val, str):
if name in ["bands", "palette", "labels"]:
val = val.split(",")
elif name in ["max", "min", "values"]:
val = [float(i) for i in val.split(",")]
elif name in ["inverted"]:
val = [bool(_strtobool(i)) for i in val.split(",")]
# set the value
props[number][name] = val
for i in props.keys():
if "type" in props[i]:
# categorical values need to be cast to int
if props[i]["type"] == "categorical":
props[i]["values"] = [int(val) for val in props[i]["values"]]
else:
# if no "type" is provided guess it from the different parameters gathered
if len(props[i]["bands"]) == 1:
props[i]["type"] = "continuous"
elif len(props[i]["bands"]) == 3:
props[i]["type"] = "rgb"
else:
warnings.warn(
"the embed viz properties are incomplete or badly set, "
"please review our documentation",
SepalWarning,
)
props = {}
return props
[docs]
def process_vis_params(
ee_object: ee.ComputedObject,
vis_params: dict,
viz: dict,
use_map_vis: bool,
viz_name: str = None,
) -> tuple:
"""Process the visualization parameters for a given Earth Engine object.
Args:
ee_object: The Earth Engine object (Image, FeatureCollection, etc.) to process.
vis_params: The visualization parameters to apply.
viz: The visualization properties dictionary.
use_map_vis: Whether to use map visualization parameters.
viz_name: Name of the visualization parameters to retrieve.
Returns:
A tuple containing the processed image, object, and visualization parameters.
Raises:
ValueError: If the provided viz_name cannot be found in the image metadata.
"""
# apply it to vis_params
if not vis_params and viz and use_map_vis:
# find the viz params in the list
try:
vis_params = next(i for p, i in viz.items() if i["name"] == viz_name)
except StopIteration:
raise ValueError(
f"the provided viz_name ({viz_name}) cannot be found in the image metadata"
)
# invert the bands if needed
inverted = vis_params.pop("inverted", None)
if inverted is not None:
# get the index of the bands that need to be inverted
index_list = [i for i, v in enumerate(inverted) if v is True]
# multiply everything by -1
for i in index_list:
min_ = vis_params["min"][i]
max_ = vis_params["max"][i]
vis_params["min"][i] = max_
vis_params["max"][i] = min_
# specific case of categorical images
# Pad the palette when using non-consecutive values
# instead of remapping or using sldStyle
# to preserve the class values in the image, for inspection
if vis_params["type"] == "categorical":
colors = vis_params["palette"]
values = vis_params["values"]
min_ = min(values)
max_ = max(values)
# set up a black palette of correct length
palette = ["#000000"] * (max_ - min_ + 1)
# replace the values within the palette
for i, val in enumerate(values):
palette[val - min_] = colors[i]
# adapt the vizparams
vis_params["palette"] = palette
vis_params["min"] = min_
vis_params["max"] = max_
# specific case of hsv
elif vis_params["type"] == "hsv":
# set to_min to 0 and to_max to 1
# in the original expression:
# 'to_min + (v - from_min) * (to_max - to_min) / (from_max - from_min)'
expression = "{band} = (b('{band}') - {from_min}) / ({from_max} - {from_min})"
# get the maxs and mins
# removing them from the parameter
mins = vis_params.pop("min")
maxs = vis_params.pop("max")
# create the rgb bands
asset = ee_object
for i, band in enumerate(vis_params["bands"]):
# adapt the expression
exp = expression.format(from_min=mins[i], from_max=maxs[i], band=band)
asset = asset.addBands(asset.expression(exp), [band], True)
# set the arguments
ee_object = asset.select(vis_params["bands"]).hsvToRgb()
vis_params["bands"] = ["red", "green", "blue"]
# force cast to featureCollection if needed
if isinstance(
ee_object,
(
ee.geometry.Geometry,
ee.feature.Feature,
ee.featurecollection.FeatureCollection,
),
):
default_vis = json.loads((ss.JSON_DIR / "layer.json").read_text())["ee_layer"]
default_vis.update(color=scolors.primary)
# We want to get all the default styles and only change those whose are
# in the provided visualization.
default_vis.update(vis_params)
vis_params = default_vis
features = ee.FeatureCollection(ee_object)
const_image = ee.Image.constant(0.5)
try:
image_fill = features.style(**vis_params).updateMask(const_image)
image_outline = features.style(**vis_params)
except AttributeError:
# Raise a more understandable error
raise AttributeError(
"You can only use the following styles: 'color', 'pointSize', "
"'pointShape', 'width', 'fillColor', 'styleProperty', "
"'neighborhood', 'lineType'"
)
image = image_fill.blend(image_outline)
obj = features
# use directly the ee object if Image
elif isinstance(ee_object, ee.image.Image):
image = obj = ee_object
# use mosaicing if the ee_object is a ImageCollection
elif isinstance(ee_object, ee.imagecollection.ImageCollection):
image = obj = ee_object.mosaic()
log.debug(f"Returning image with vis_params: {vis_params}")
return image, obj, vis_params
[docs]
def validate_ee_object(ee_object: ee.ComputedObject) -> None:
"""Validate the type of the ee object.
Args:
ee_object: the ee object to validate
Raises:
AttributeError: if the ee_object is not an instance of ee.Image, ee.Geometry, ee.Feature or ee.FeatureCollection
"""
# check the type of the ee object and raise an error if it's not recognized
if not isinstance(
ee_object,
(
ee.Image,
ee.ImageCollection,
ee.FeatureCollection,
ee.Feature,
ee.Geometry,
),
):
raise AttributeError(
"\n\nThe image argument in 'addLayer' function must be an instance of "
"one of ee.Image, ee.Geometry, ee.Feature or ee.FeatureCollection."
)
def _serialize_viz_value(key: str, value: object) -> str:
"""Serialize a single visualization parameter value to the SEPAL property format.
SEPAL stores visualization parameters as flat image properties prefixed with
``visualization_<index>_<key>``. List-valued keys (``bands``, ``palette``,
``labels``, ``min``, ``max``, ``values``, ``inverted``) are stored as
comma-separated strings; scalar keys (``name``, ``type``) as plain strings.
"""
if key in VIZ_SCALAR_KEYS:
return str(value)
if key in VIZ_BOOL_LIST_KEYS:
sequence = value if isinstance(value, (list, tuple)) else [value]
return ",".join("true" if bool(item) else "false" for item in sequence)
if isinstance(value, str):
# Allow callers to pass a pre-serialized comma-separated string.
return value
if isinstance(value, (list, tuple)):
return ",".join(str(item) for item in value)
return str(value)
[docs]
def set_viz_params(
image: ee.Image,
*,
name: str = "default",
type: Optional[str] = None,
bands: Optional[Sequence[str]] = None,
min: Union[float, Sequence[float], None] = None,
max: Union[float, Sequence[float], None] = None,
palette: Union[str, Sequence[str], None] = None,
values: Optional[Sequence[Union[int, float]]] = None,
labels: Optional[Sequence[str]] = None,
inverted: Optional[Sequence[bool]] = None,
index: int = 0,
) -> ee.Image:
"""Embed SEPAL-convention visualization parameters as image properties.
This is the inverse of :func:`get_viz_params`. SEPAL stores per-image
visualization as flat properties ``visualization_<index>_<key>``; SepalMap
reads them on display, and Earth Engine asset exports preserve them, so
callers downstream (other SEPAL recipes, the Code Editor) see the styling
automatically.
Args:
image: The :class:`ee.Image` to annotate. A new image is returned;
the original is not mutated.
name: Logical name of the visualization (e.g. ``"default"``). Stored
as ``visualization_<index>_name``.
type: Visualization type — one of ``"continuous"``, ``"categorical"``,
``"rgb"``, ``"hsv"``. Inferred by SepalMap when omitted, based on
the number of bands.
bands: Band names targeted by the visualization. Single-band for
``continuous``/``categorical``, three bands for ``rgb``/``hsv``.
min: Per-band minimum value(s). Scalar or sequence; both stored as
a comma-separated string.
max: Per-band maximum value(s). Same rules as ``min``.
palette: Hex color palette. Sequence of ``#RRGGBB`` or a
pre-comma-joined string.
values: For ``categorical`` only — the discrete pixel values that the
palette maps to. Preserves non-consecutive class codes.
labels: Per-value human-readable labels (used in legends).
inverted: Per-band inversion flags.
index: Slot index in the property name. Use distinct indices when
attaching multiple named visualizations to the same image.
Returns:
A new :class:`ee.Image` carrying the visualization properties.
Example:
>>> styled = set_viz_params(
... classified,
... name="loss_year",
... type="categorical",
... bands=["classification"],
... palette=["#ffff00", "#8b0000", "#d3d3d3"],
... values=[1, 25, 30],
... labels=["loss 2001", "loss 2025", "non forest"],
... )
"""
if not isinstance(image, ee.Image):
actual = image.__class__.__name__ if image is not None else "None"
raise TypeError(f"set_viz_params requires an ee.Image; got {actual}")
fields: dict[str, object] = {
"name": name,
"type": type,
"bands": bands,
"min": min,
"max": max,
"palette": palette,
"values": values,
"labels": labels,
"inverted": inverted,
}
properties: dict[str, str] = {}
for key, value in fields.items():
if value is None:
continue
properties[f"visualization_{index}_{key}"] = _serialize_viz_value(key, value)
if not properties:
return image
return image.set(properties)
[docs]
def merge_viz_params(*params_list: Mapping[str, object]) -> dict:
"""Merge several visualization parameter dicts, later entries taking precedence.
Useful when an app exposes a base palette (e.g. a shared common module)
and a per-instance override (e.g. user-tuned min/max).
"""
merged: dict = {}
for params in params_list:
if not params:
continue
for key, value in params.items():
if value is None:
continue
merged[key] = value
return merged
