Spaces:
Sleeping
Sleeping
| """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, | |
| ) | |
| 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 | |