Spaces:
Sleeping
Sleeping
File size: 3,572 Bytes
1e732dd 696f787 1e732dd 696f787 1e732dd 696f787 1e732dd 696f787 1e732dd 696f787 1e732dd 696f787 1e732dd 696f787 1e732dd 9659593 1e732dd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | """
MediGuard AI — Biomarker Validation Service
Wraps the existing BiomarkerValidator as a production service with caching,
observability, and Pydantic-typed outputs.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from functools import lru_cache
from typing import Any
from src.biomarker_normalization import normalize_biomarker_name
from src.biomarker_validator import BiomarkerValidator
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class BiomarkerResult:
"""Validated result for a single biomarker."""
name: str
value: float
unit: str
status: str # NORMAL | HIGH | LOW | CRITICAL_HIGH | CRITICAL_LOW
reference_range: str
warning: str | None = None
@dataclass
class ValidationReport:
"""Complete biomarker validation report."""
results: list[BiomarkerResult] = field(default_factory=list)
safety_alerts: list[dict[str, Any]] = field(default_factory=list)
recognized_count: int = 0
unrecognized: list[str] = field(default_factory=list)
class BiomarkerService:
"""Production biomarker validation service."""
def __init__(self) -> None:
self._validator = BiomarkerValidator()
# --------------------------------------------------------------------- #
# Public API
# --------------------------------------------------------------------- #
def validate(
self,
biomarkers: dict[str, float],
gender: str | None = None,
) -> ValidationReport:
"""Validate a dict of biomarker name → value and return a report."""
report = ValidationReport()
for raw_name, value in biomarkers.items():
normalized = normalize_biomarker_name(raw_name)
flag = self._validator.validate_biomarker(normalized, value, gender=gender)
if flag is None:
report.unrecognized.append(raw_name)
continue
if flag.status == "UNKNOWN":
report.unrecognized.append(raw_name)
continue
report.recognized_count += 1
report.results.append(
BiomarkerResult(
name=flag.name,
value=flag.value,
unit=flag.unit,
status=flag.status,
reference_range=flag.reference_range,
warning=flag.warning,
)
)
if flag.status.startswith("CRITICAL"):
report.safety_alerts.append(
{
"severity": "CRITICAL",
"biomarker": normalized,
"message": flag.warning or f"{normalized} is critically out of range",
"action": "Seek immediate medical attention",
}
)
return report
def list_supported(self) -> list[dict[str, Any]]:
"""Return metadata for all supported biomarkers."""
result = []
for name, ref in self._validator.references.items():
result.append(
{
"name": name,
"unit": ref.get("unit", ""),
"normal_range": ref.get("normal_range", {}),
"critical_low": ref.get("critical_low"),
"critical_high": ref.get("critical_high"),
}
)
return result
@lru_cache(maxsize=1)
def make_biomarker_service() -> BiomarkerService:
return BiomarkerService()
|