mcp_tox_calc/units.py

import re
from typing import Any, Dict


class UnitError(ValueError):
    pass


def _to_float(v: Any, field: str) -> float:
    try:
        x = float(v)
    except Exception as exc:
        raise UnitError(f"{field} must be numeric.") from exc
    return x


def _norm_unit(unit: Any) -> str:
    s = str(unit or "").strip().lower()
    s = s.replace(" ", "")
    s = s.replace("μ", "u")
    return s


def normalize_route(route: Any) -> str:
    r = str(route or "").strip().lower()
    if r in {"oral", "inhalation"}:
        return r
    raise UnitError("route must be oral or inhalation")


def normalize_oral_exposure(exposure_value: Any, exposure_unit: Any, body_weight_kg: Any) -> Dict[str, Any]:
    value = _to_float(exposure_value, "exposure_value")
    unit = _norm_unit(exposure_unit)
    bw = _to_float(body_weight_kg, "body_weight_kg") if body_weight_kg not in (None, "") else None

    conversions = []

    if unit in {"mg/kg-day", "mg/kg/d", "mg/kgday", "mgkgday"}:
        cdi = value
    elif unit in {"ug/kg-day", "ug/kg/d", "ug/kgday", "ugkgday"}:
        cdi = value / 1000.0
        conversions.append("exposure ug/kg-day -> mg/kg-day")
    elif unit in {"mg/day", "mg/d", "mgday"}:
        if bw is None or bw <= 0:
            raise UnitError("body_weight_kg is required for exposure unit mg/day")
        cdi = value / bw
        conversions.append("exposure mg/day -> mg/kg-day")
    elif unit in {"ug/day", "ug/d", "ugday"}:
        if bw is None or bw <= 0:
            raise UnitError("body_weight_kg is required for exposure unit ug/day")
        cdi = (value / 1000.0) / bw
        conversions.append("exposure ug/day -> mg/kg-day")
    else:
        raise UnitError(f"Unsupported oral exposure unit: {exposure_unit}")

    return {
        "value_mg_per_kg_day": cdi,
        "unit": "mg/kg-day",
        "conversions": conversions,
    }


def normalize_air_concentration(air_conc_value: Any, air_conc_unit: Any) -> Dict[str, Any]:
    value = _to_float(air_conc_value, "air_conc_value")
    unit = _norm_unit(air_conc_unit)
    conversions = []

    if unit in {"ug/m3", "ugm3", "ug/m^3"}:
        conc = value
    elif unit in {"mg/m3", "mgm3", "mg/m^3"}:
        conc = value * 1000.0
        conversions.append("air concentration mg/m3 -> ug/m3")
    elif unit in {"ng/m3", "ngm3", "ng/m^3"}:
        conc = value / 1000.0
        conversions.append("air concentration ng/m3 -> ug/m3")
    else:
        raise UnitError(f"Unsupported air concentration unit: {air_conc_unit}")

    return {
        "value_ug_per_m3": conc,
        "unit": "ug/m3",
        "conversions": conversions,
    }


def normalize_csf(csf_value: Any, csf_unit: Any) -> Dict[str, Any]:
    value = _to_float(csf_value, "csf_value")
    unit = _norm_unit(csf_unit)
    conversions = []

    if unit in {
        "(mg/kg-day)^-1",
        "1/(mg/kg-day)",
        "per(mg/kg-day)",
        "permg/kg-day",
        "(mgkgday)^-1",
        "1/mgkgday",
    }:
        out = value
    elif unit in {
        "(ug/kg-day)^-1",
        "1/(ug/kg-day)",
        "per(ug/kg-day)",
        "(ugkgday)^-1",
        "1/ugkgday",
    }:
        out = value * 1000.0
        conversions.append("CSF per (ug/kg-day) -> per (mg/kg-day)")
    elif unit in {"", "na", "n/a"}:
        out = value
    else:
        raise UnitError(f"Unsupported csf unit: {csf_unit}")

    return {
        "value_per_mg_per_kg_day": out,
        "unit": "(mg/kg-day)^-1",
        "conversions": conversions,
    }


def normalize_iur(iur_value: Any, iur_unit: Any) -> Dict[str, Any]:
    value = _to_float(iur_value, "iur_value")
    unit = _norm_unit(iur_unit)
    conversions = []

    if unit in {
        "(ug/m3)^-1",
        "1/(ug/m3)",
        "per(ug/m3)",
        "1/ugm3",
        "(ugm3)^-1",
    }:
        out = value
    elif unit in {
        "(mg/m3)^-1",
        "1/(mg/m3)",
        "per(mg/m3)",
        "1/mgm3",
        "(mgm3)^-1",
    }:
        out = value / 1000.0
        conversions.append("IUR per (mg/m3) -> per (ug/m3)")
    elif unit in {"", "na", "n/a"}:
        out = value
    else:
        raise UnitError(f"Unsupported iur unit: {iur_unit}")

    return {
        "value_per_ug_per_m3": out,
        "unit": "(ug/m3)^-1",
        "conversions": conversions,
    }