import math import torch from src.shared.common import cfg def _training_cfg(config: dict | None = None) -> dict: source = cfg if config is None else config if not isinstance(source, dict): return {} return source.get("training", {}) if isinstance(source.get("training", {}), dict) else {} def target_transform_mode(*, config: dict | None = None, mode: str | None = None) -> str: """Return the configured target transform mode.""" if mode is not None: return str(mode).strip().lower() return str(_training_cfg(config).get("target_transform", "none")).strip().lower() def transform_target_scalar( value: float, *, config: dict | None = None, mode: str | None = None, ) -> float: """Map a raw rainfall target into training space.""" mode = target_transform_mode(config=config, mode=mode) value = max(float(value), 0.0) if mode == "log1p": return math.log1p(value) return value def inverse_target_scalar( value: float, *, config: dict | None = None, mode: str | None = None, ) -> float: """Map a prediction from training space back into raw rainfall units.""" mode = target_transform_mode(config=config, mode=mode) value = float(value) if mode == "log1p": return max(math.expm1(value), 0.0) return value def transform_target_tensor( tensor: torch.Tensor, *, config: dict | None = None, mode: str | None = None, ) -> torch.Tensor: """Map a raw rainfall tensor into training space.""" mode = target_transform_mode(config=config, mode=mode) if mode == "log1p": return torch.log1p(torch.clamp(tensor, min=0.0)) return tensor def rain_threshold_mm(*, config: dict | None = None) -> float: """Rain/no-rain decision threshold in raw rainfall units (mm).""" return float(_training_cfg(config).get("rain_threshold_mm", 0.1)) def is_rain(value: float, *, threshold: float | None = None) -> bool: """Classify rainfall value (mm) as rain/dry using configured threshold.""" if threshold is None: threshold = rain_threshold_mm() return float(value) > float(threshold) def rain_probability_threshold(*, config: dict | None = None) -> float: """Probability threshold for deciding whether to emit non-zero rainfall prediction.""" return float(_training_cfg(config).get("rain_probability_threshold", 0.5))