csc8114 / code / src / shared / targets.py
targets.py
Raw
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))