Functions

def convert_units(
    magnitude: float | int | str, from_unit: str, to_unit: str, system: UnitSystem | None = None
) -> float:
    """
    Convert a quantity between units.

    >>> convert_units(1, "m", "cm")
    100.0
    >>> convert_units(1, "m", "cm", system=UnitSystem.UCUM)
    100.0
    >>> convert_units(1, "m", "cm", system=UnitSystem.SI)
    100.0
    >>> convert_units(1.0, "hectare", "m^2", system=UnitSystem.SI)
    10000.0
    >>> convert_units(1.0, "hectare", "m ** 2", system=UnitSystem.SI)
    10000.0
    >>> convert_units(1.0, "hectare", "m ** 2", system=None)
    10000.0
    >>> convert_units(1.0, "km2", "m2", system=UnitSystem.UCUM)
    1000000.0
    >>> convert_units("100", "m", "cm")
    10000.0
    >>> convert_units("3.14", "m", "cm")
    314.0

    :param magnitude:
    :param from_unit:
    :param to_unit:
    :return: converted magnitude
    """
    magnitude = float(magnitude)
    ureg: pint.UnitRegistry = get_unit_registry(system)
    from_unit = normalize_unit(from_unit, system)
    to_unit = normalize_unit(to_unit, system)
    try:
        from_unit_q = ureg.parse_units(from_unit)
    except lark.exceptions.UnexpectedCharacters as err:
        msg = f"Cannot parse unit: {from_unit}"
        raise UndefinedUnitError(msg) from err
    except pint.errors.UndefinedUnitError as err:
        msg = f"Unknown source unit: {from_unit}"
        raise UndefinedUnitError(msg) from err
    quantity = magnitude * from_unit_q
    try:
        return quantity.to(to_unit).magnitude
    except pint.errors.UndefinedUnitError as err:
        msg = f"Unknown target unit: {from_unit}"
        raise UndefinedUnitError(msg) from err
    except pint.errors.DimensionalityError as err:
        msg = f"Cannot convert from {from_unit} to {to_unit}"
        raise DimensionalityError(msg) from err

@lru_cache
def get_unit_registry(
    system: UnitSystem | None = None,
) -> pint.UnitRegistry | PintUcumRegistry:
    """
    Get a unit registry.

    >>> ureg = get_unit_registry(UnitSystem.UCUM)
    >>> ureg.from_ucum("m/s2.kg")
    <Quantity(1.0, 'meter * kilogram / second ** 2')>
    >>> str(_)
    '1.0 kilogram * meter / second ** 2'
    >>> ureg.from_ucum("m[H2O]{35Cel}")  # annotated UCUM code; ucumvert <0.3 names it m_H2O, >=0.3 meter_H2O
    <Quantity(1, '..._H2O')>
    >>> _.to("mbar")
    <Quantity(98.0665, 'millibar')>
    >>> ureg("degC")   # a standard pint unit
    <Quantity(1, 'degree_Celsius')>
    >>> ureg.from_ucum("g/m2")
    <Quantity(1.0, 'gram / meter ** 2')>
    >>> _.to(ureg.from_ucum("kg/m2"))
    <Quantity(0.001, 'kilogram / meter ** 2')>
    >>> ureg.from_ucum("nmol/mmol{Cre}")
    <Quantity(1.0, 'nanomole / millimole')>
    >>> sireg = get_unit_registry(UnitSystem.SI)
    >>> sireg("degC")
    <Quantity(1, 'degree_Celsius')>
    >>> sireg("ha")
    <Quantity(1, 'hectare')>

    :param system: currently only supported values are None or UnitSystem.UCUM
    :return:
    """
    import pint

    ureg = pint.UnitRegistry()
    if not system:
        return ureg
    if system in REGISTRIES:
        return REGISTRIES[system]
    if system.value in dir(ureg.sys):
        ureg.default_system = system.value
        return ureg
    msg = f"Unknown unit system: {system}"
    raise NotImplementedError(msg)

def normalize_unit(unit: str, system: UnitSystem | None = None) -> str:
    """Normalize the unit to UnitSystem.UCUM, if possible."""
    if system is None or system != UnitSystem.UCUM:
        return unit

    # this is UnitSystem.UCUM
    try:
        return str(get_unit_registry(system).from_ucum(unit))
    except pint.errors.UndefinedUnitError as err:
        msg = f"Unknown unit: {unit}"
        raise UndefinedUnitError(msg) from err
    except lark.exceptions.UnexpectedCharacters as err:
        msg = f"Cannot parse unit: {unit}"
        raise UndefinedUnitError(msg) from err

def perform_unit_conversion(
    slot_derivation: SlotDerivation,
    source_obj: dict[str, Any],
    sv: SchemaView,
    source_type: str,
) -> float | dict | None:
    """Convert a slot's value between units, per its ``unit_conversion`` config.

    Takes the source row, schemaview and source type explicitly rather than a
    ``DerivationContext``: that type lives in ``object_transformer``, which imports
    this module, so depending on it here would be circular.

    :param slot_derivation: the derivation carrying the ``unit_conversion`` block
    :param source_obj: the source row
    :param sv: source schema view, for resolving the slot's declared unit
    :param source_type: source class name
    :return: the converted magnitude, a structured value, or None
    :rtype: float | dict | None
    """
    uc = slot_derivation.unit_conversion
    curr_v = source_obj.get(slot_derivation.populated_from, None)

    if curr_v is None:
        logger.debug(f"No value found for slot '{slot_derivation.populated_from}'; skipping conversion")
        return None

    slot = sv.induced_slot(slot_derivation.populated_from, source_type)
    schema_unit = None
    from_unit = None
    system = UnitSystem.UCUM

    if slot.unit:
        if slot.unit.ucum_code:
            schema_unit = slot.unit.ucum_code
        elif slot.unit.iec61360code:
            schema_unit = slot.unit.iec61360code
            system = UnitSystem.IEC61360
        elif slot.unit.symbol:
            schema_unit = slot.unit.symbol
            system = None
        elif slot.unit.abbreviation:
            schema_unit = slot.unit.abbreviation
            system = None
        elif slot.unit.descriptive_name:
            schema_unit = slot.unit.descriptive_name
            system = None
        else:
            raise NotImplementedError(f"Cannot determine unit system for slot '{slot.name}' — all unit fields are None")

    spec_unit = uc.source_unit if uc.source_unit else None

    if schema_unit and spec_unit:
        if schema_unit != spec_unit:
            raise ValueError(
                f"Mismatch in source units for slot '{slot_derivation.populated_from}': "
                f"schema unit '{schema_unit}' vs. transformation spec '{spec_unit}'"
            )
        from_unit = schema_unit
    elif schema_unit:
        from_unit = schema_unit
    elif spec_unit:
        from_unit = spec_unit
    else:
        if uc.source_unit_slot:
            from_unit = None
        else:
            slot_name = slot_derivation.populated_from
            raise ValueError(f"No source unit provided in schema or transformation spec for slot '{slot_name}'")

    if uc.source_unit_slot:
        # Structured input, e.g., {"value": 120, "unit": "cm"}
        from_unit_val = curr_v.get(uc.source_unit_slot)
        if from_unit_val:
            if from_unit and from_unit_val != from_unit:
                slot_name = slot_derivation.populated_from
                raise ValueError(
                    f"Value unit '{from_unit_val}' does not match expected '{from_unit}' for slot '{slot_name}'"
                )
            from_unit = from_unit_val
        else:
            raise ValueError(f"Missing unit in structured value for slot '{slot_derivation.populated_from}': {curr_v}")

        magnitude = curr_v.get(uc.source_magnitude_slot)
        if magnitude is None:
            raise ValueError(
                f"Missing magnitude in structured value for slot '{slot_derivation.populated_from}': {curr_v}"
            )
    else:
        magnitude = curr_v

    try:
        magnitude = float(magnitude)
    except (TypeError, ValueError):
        if uc.none_if_non_numeric:
            return None
        raise

    to_unit = uc.target_unit or from_unit
    if from_unit == to_unit:
        result = magnitude
    else:
        result = convert_units(magnitude, from_unit=from_unit, to_unit=to_unit, system=system)

    if uc.target_magnitude_slot:
        return {uc.target_magnitude_slot: result, uc.target_unit_slot: to_unit}
    return result