File size: 3,335 Bytes
7da2ecb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """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
|