"""Shared helpers for object-based validation.""" from __future__ import annotations from datetime import datetime from functools import lru_cache from math import ceil from pathlib import Path from typing import Any import numpy as np import yaml from scipy.ndimage import binary_dilation, distance_transform_edt DT_FORMAT = "%Y%m%d%H%M" def parse_cloud_id(cloud_id: str) -> tuple[datetime, int]: dt_str, number_str = cloud_id.split("_", 1) return datetime.strptime(dt_str, DT_FORMAT), int(number_str) def format_dt(dt: datetime) -> str: return dt.strftime(DT_FORMAT) def day_str(dt: datetime) -> str: return dt.strftime("%Y%m%d") def km_to_pixels(km: float | int | None, pixel_size_km: float) -> int: if km is None: return 0 km = float(km) if km <= 0: return 0 return int(ceil(km / float(pixel_size_km))) @lru_cache(maxsize=None) def circular_footprint(radius_pixels: int) -> np.ndarray: radius_pixels = int(radius_pixels) if radius_pixels <= 0: footprint = np.ones((1, 1), dtype=bool) footprint.flags.writeable = False return footprint y, x = np.ogrid[-radius_pixels : radius_pixels + 1, -radius_pixels : radius_pixels + 1] footprint = (x * x + y * y) <= radius_pixels * radius_pixels footprint.flags.writeable = False return footprint def normalize_buffer_backend(backend: str | None) -> str: backend = str(backend or "auto").lower() if backend not in {"auto", "binary", "edt"}: raise ValueError(f"unsupported buffer backend: {backend!r}") return backend def use_edt_backend(radius_pixels: int, backend: str | None = "auto") -> bool: backend = normalize_buffer_backend(backend) radius_pixels = int(radius_pixels) if radius_pixels <= 0: return False if backend == "edt": return True if backend == "binary": return False return radius_pixels > 4 def dilate_fast(mask: np.ndarray, radius_pixels: int, backend: str | None = "auto") -> np.ndarray: if radius_pixels <= 0: return mask.astype(bool, copy=True) mask = mask.astype(bool, copy=False) if not np.any(mask): return np.zeros_like(mask, dtype=bool) if use_edt_backend(radius_pixels, backend): return distance_transform_edt(~mask) <= int(radius_pixels) return binary_dilation(mask, structure=circular_footprint(radius_pixels)) def dilate(mask: np.ndarray, radius_pixels: int) -> np.ndarray: return dilate_fast(mask, radius_pixels, backend="binary") def load_yaml(path: str | Path) -> dict[str, Any]: with open(path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) return data or {} def write_yaml(data: dict[str, Any], path: str | Path) -> None: with open(path, "w", encoding="utf-8") as f: yaml.safe_dump(data, f, sort_keys=False, allow_unicode=True) def expand_modes(values: list[str] | tuple[str, ...] | str, both_values: tuple[str, str]) -> list[str]: if isinstance(values, str): values = [values] out: list[str] = [] for value in values: if value == "both": out.extend(both_values) else: out.append(value) deduped: list[str] = [] for value in out: if value not in deduped: deduped.append(value) return deduped