"""Clinical Unit Normalizer and Numeric Validator — Phase 3. Safely converts clinical measurements to expected canonical units: - Glucose: mmol/L -> mg/dL (x 18.0182) - Hemoglobin: g/L -> g/dL (x 0.1) - Bilirubin: umol/L -> mg/dL (x 0.05847) - WBC / Platelets: /uL -> x10^3/uL (x 0.001) - CRP: mg/dL -> mg/L (x 10.0) - HbA1c: mmol/mol -> % ((val * 0.09148) + 2.152) Safety rules: - Preserves raw_value, raw_unit, normalized_value, normalized_unit, conversion_applied. - Rejects malformed values (e.g. '140/90' is recognized as compound BP, not scalar lab). - Rejects out-of-bounds physiological values as invalid/malformed. """ from __future__ import annotations import re from dataclasses import dataclass from typing import Any from ehr_intel.analysis.dictionary import ( ClinicalParameterDefinition, get_parameter_dictionary, ) @dataclass class NormalizedParameterValue: """Validated and unit-normalized clinical measurement.""" param_key: str canonical_name: str raw_value: str raw_numeric: float | None raw_unit: str | None normalized_numeric: float | None normalized_unit: str | None conversion_applied: bool is_valid: bool validation_error: str | None = None domain: str | None = None class ClinicalUnitNormalizer: """Safe clinical unit converter and numeric parameter validator.""" def __init__(self) -> None: self.dict_engine = get_parameter_dictionary() def normalize( self, name: str, value: str | float | int, unit: str | None = None, ) -> NormalizedParameterValue: """Validate and normalize a raw clinical parameter and its unit.""" param_key, defn = self.dict_engine.lookup(name) if not defn or not param_key: # Parameter not in 5 target domains return NormalizedParameterValue( param_key="unknown", canonical_name=name, raw_value=str(value), raw_numeric=self._try_parse_float(value), raw_unit=unit, normalized_numeric=self._try_parse_float(value), normalized_unit=unit, conversion_applied=False, is_valid=True, ) raw_str = str(value).strip() # Reject compound values like '140/90' for scalar laboratory tests if "/" in raw_str and defn.category == "laboratory": return NormalizedParameterValue( param_key=param_key, canonical_name=defn.canonical_name, raw_value=raw_str, raw_numeric=None, raw_unit=unit, normalized_numeric=None, normalized_unit=defn.expected_unit, conversion_applied=False, is_valid=False, validation_error="Malformed scalar value: compound ratio or blood pressure string rejected for lab", domain=defn.domain, ) parsed_num = self._try_parse_float(raw_str) if parsed_num is None: return NormalizedParameterValue( param_key=param_key, canonical_name=defn.canonical_name, raw_value=raw_str, raw_numeric=None, raw_unit=unit, normalized_numeric=None, normalized_unit=defn.expected_unit, conversion_applied=False, is_valid=False, validation_error="Unable to parse scalar numeric value", domain=defn.domain, ) # Check valid numeric range if defn.valid_numeric_range: min_val, max_val = defn.valid_numeric_range # Check range in raw or normalized # If unit is already expected: clean_unit = self._clean_unit_str(unit) expected_clean = self._clean_unit_str(defn.expected_unit) if clean_unit == expected_clean and (parsed_num < min_val * 0.5 or parsed_num > max_val * 2.0): return NormalizedParameterValue( param_key=param_key, canonical_name=defn.canonical_name, raw_value=raw_str, raw_numeric=parsed_num, raw_unit=unit, normalized_numeric=None, normalized_unit=defn.expected_unit, conversion_applied=False, is_valid=False, validation_error=f"Value {parsed_num} outside physiologically plausible bounds [{min_val}, {max_val}]", domain=defn.domain, ) # Apply Unit Conversion norm_val, norm_unit, converted, err = self._convert_unit(parsed_num, unit, defn) if err: return NormalizedParameterValue( param_key=param_key, canonical_name=defn.canonical_name, raw_value=raw_str, raw_numeric=parsed_num, raw_unit=unit, normalized_numeric=None, normalized_unit=defn.expected_unit, conversion_applied=False, is_valid=False, validation_error=err, domain=defn.domain, ) return NormalizedParameterValue( param_key=param_key, canonical_name=defn.canonical_name, raw_value=raw_str, raw_numeric=parsed_num, raw_unit=unit, normalized_numeric=round(norm_val, 4), normalized_unit=norm_unit, conversion_applied=converted, is_valid=True, domain=defn.domain, ) def _convert_unit( self, value: float, unit: str | None, defn: ClinicalParameterDefinition, ) -> tuple[float, str, bool, str | None]: """Convert scalar value to defn.expected_unit using defn.convertible_units.""" expected = defn.expected_unit or "" if not unit or not unit.strip(): # No unit provided — assume expected unit only if value is plausible return value, expected, False, None c_unit = self._clean_unit_str(unit) c_expected = self._clean_unit_str(expected) if c_unit == c_expected: return value, expected, False, None # Check convertible units for conv_u, factor in defn.convertible_units.items(): if self._clean_unit_str(conv_u) == c_unit: if isinstance(factor, (int, float)): converted = value * float(factor) return converted, expected, True, None if factor == "IFCC_TO_DCCT": # (mmol/mol * 0.09148) + 2.152 converted = (value * 0.09148) + 2.152 return converted, expected, True, None # If unit mismatch cannot be safely converted, return error rather than guessing return value, unit, False, f"Incompatible unit '{unit}' cannot be safely converted to '{expected}'" def _clean_unit_str(self, u: str | None) -> str: if not u: return "" return re.sub(r"[\s\.\-_]", "", u.lower()) def _try_parse_float(self, v: Any) -> float | None: if isinstance(v, (int, float)): return float(v) try: cleaned = str(v).replace("%", "").strip() return float(cleaned) except (ValueError, TypeError): return None _global_normalizer: ClinicalUnitNormalizer | None = None def get_unit_normalizer() -> ClinicalUnitNormalizer: global _global_normalizer if _global_normalizer is None: _global_normalizer = ClinicalUnitNormalizer() return _global_normalizer