| """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 |
|
|
