Spaces:
Sleeping
Sleeping
| """Method A: Rule-Based Clinical Analysis Engine — Phase 3. | |
| Evaluates evidence-based clinical guidelines and sourced reference ranges | |
| for 5 target domains: Blood, Sugar, Liver, Lungs, and Intestine. | |
| Strict Safety Rules: | |
| - All thresholds cite established medical guidelines (ADA 2024, WHO, AASLD, ATS/GOLD, ACG). | |
| - Non-diagnostic research terminology only (e.g. ELEVATED_GLYCEMIC_INDICATOR, ANEMIA_INDICATOR). | |
| - Preserves assertion status (negated, historical, or suspected findings are not marked active). | |
| - Explicit trace generation for full explainability. | |
| """ | |
| from __future__ import annotations | |
| import time | |
| from typing import Any | |
| from ehr_intel.analysis.features import ( | |
| ClinicalFeatureVector, | |
| MissingnessStatus, | |
| ) | |
| from ehr_intel.analysis.models import ( | |
| AnalysisMethod, | |
| ClinicalAnalysisResult, | |
| DomainIndicator, | |
| IndicatorStatus, | |
| ) | |
| from ehr_intel.clinical.canonical import AssertionStatus | |
| from ehr_intel.models import ConfidenceTier, ReviewStatus | |
| class RuleBasedClinicalAnalyzer: | |
| """Deterministic, rule-based clinical indicator analyzer.""" | |
| def __init__(self) -> None: | |
| pass | |
| def analyze(self, features: ClinicalFeatureVector) -> ClinicalAnalysisResult: | |
| """Evaluate clinical feature vector against evidence-based rules.""" | |
| start_time = time.perf_counter() | |
| indicators: list[DomainIndicator] = [] | |
| feature_trace: list[dict[str, Any]] = [] | |
| uncertainties: list[str] = [] | |
| conflicts: list[dict[str, Any]] = [] | |
| domain_statuses: dict[str, IndicatorStatus] = { | |
| "blood": IndicatorStatus.INSUFFICIENT_DATA, | |
| "sugar": IndicatorStatus.INSUFFICIENT_DATA, | |
| "liver": IndicatorStatus.INSUFFICIENT_DATA, | |
| "lungs": IndicatorStatus.INSUFFICIENT_DATA, | |
| "intestine": IndicatorStatus.INSUFFICIENT_DATA, | |
| } | |
| # 1. Sugar Domain Analysis | |
| sugar_inds, sugar_trace, sugar_status, sugar_conflicts = self._analyze_sugar(features.sugar) | |
| indicators.extend(sugar_inds) | |
| feature_trace.extend(sugar_trace) | |
| domain_statuses["sugar"] = sugar_status | |
| conflicts.extend(sugar_conflicts) | |
| # 2. Blood Domain Analysis | |
| blood_inds, blood_trace, blood_status = self._analyze_blood(features.blood) | |
| indicators.extend(blood_inds) | |
| feature_trace.extend(blood_trace) | |
| domain_statuses["blood"] = blood_status | |
| # 3. Liver Domain Analysis | |
| liver_inds, liver_trace, liver_status = self._analyze_liver(features.liver) | |
| indicators.extend(liver_inds) | |
| feature_trace.extend(liver_trace) | |
| domain_statuses["liver"] = liver_status | |
| # 4. Lung Domain Analysis | |
| lung_inds, lung_trace, lung_status = self._analyze_lungs(features.lungs) | |
| indicators.extend(lung_inds) | |
| feature_trace.extend(lung_trace) | |
| domain_statuses["lungs"] = lung_status | |
| # 5. Intestine Domain Analysis | |
| intestine_inds, intestine_trace, intestine_status = self._analyze_intestine(features.intestine) | |
| indicators.extend(intestine_inds) | |
| feature_trace.extend(intestine_trace) | |
| domain_statuses["intestine"] = intestine_status | |
| # Overall review status | |
| has_review_req = any(ind.review_required for ind in indicators) or len(conflicts) > 0 | |
| overall_status = ReviewStatus.REVIEW_REQUIRED if has_review_req else ReviewStatus.ACCEPTED | |
| elapsed = time.perf_counter() - start_time | |
| return ClinicalAnalysisResult( | |
| patient_id=features.patient_id, | |
| document_id=features.document_id, | |
| method=AnalysisMethod.RULE_BASED, | |
| overall_review_status=overall_status, | |
| indicators=indicators, | |
| domain_statuses=domain_statuses, | |
| feature_trace=feature_trace, | |
| conflicts=conflicts, | |
| uncertainties=uncertainties, | |
| provenance="RuleBasedClinicalAnalyzer (ADA 2024, WHO, AASLD, ATS, ACG guidelines)", | |
| latency_seconds=round(elapsed, 4), | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Domain Rule Implementations | |
| # --------------------------------------------------------------------------- | |
| def _analyze_sugar( | |
| self, sugar: Any | |
| ) -> tuple[list[DomainIndicator], list[dict[str, Any]], IndicatorStatus, list[dict[str, Any]]]: | |
| inds: list[DomainIndicator] = [] | |
| trace: list[dict[str, Any]] = [] | |
| conflicts: list[dict[str, Any]] = [] | |
| if sugar.is_empty: | |
| return [ | |
| DomainIndicator( | |
| domain="sugar", | |
| indicator_name="ELEVATED_GLYCEMIC_INDICATOR", | |
| status=IndicatorStatus.INSUFFICIENT_DATA, | |
| severity=None, | |
| evidence="No glycemic parameters (HbA1c, Fasting Glucose, Random Glucose) reported in document.", | |
| confidence=ConfidenceTier.UNKNOWN, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ], trace, IndicatorStatus.INSUFFICIENT_DATA, conflicts | |
| hba1c = sugar.hba1c | |
| fbg = sugar.fasting_glucose | |
| rbg = sugar.random_glucose | |
| # Check assertion safety: if explicitly negated or historical, skip active indicator | |
| if hba1c.status == AssertionStatus.ABSENT_NEGATED or fbg.status == AssertionStatus.ABSENT_NEGATED: | |
| trace.append({"parameter": "sugar", "note": "Sugar condition explicitly negated in source text"}) | |
| supported_reasons: list[str] = [] | |
| supporting_params: list[str] = [] | |
| severity: str = "MODERATE" | |
| confidence = ConfidenceTier.HIGH | |
| # Rule 1: HbA1c | |
| if hba1c.missingness == MissingnessStatus.REPORTED and hba1c.value_numeric is not None: | |
| val = hba1c.value_numeric | |
| trace.append({"param": "hba1c", "value": val, "unit": "%", "threshold": ">= 6.5 (ADA 2024)"}) | |
| if val >= 6.5: | |
| supported_reasons.append(f"HbA1c = {val}% (threshold >= 6.5%, ADA 2024)") | |
| supporting_params.append("hba1c") | |
| if val >= 9.0: | |
| severity = "SEVERE" | |
| elif val >= 5.7: | |
| supported_reasons.append(f"HbA1c = {val}% in prediabetes range [5.7 - 6.4%] (ADA 2024)") | |
| supporting_params.append("hba1c") | |
| severity = "MILD" | |
| # Rule 2: Fasting Glucose | |
| if fbg.missingness == MissingnessStatus.REPORTED and fbg.value_numeric is not None: | |
| val = fbg.value_numeric | |
| trace.append({"param": "fasting_glucose", "value": val, "unit": "mg/dL", "threshold": ">= 126 (ADA 2024)"}) | |
| if val >= 126.0: | |
| supported_reasons.append(f"Fasting Blood Glucose = {val} mg/dL (threshold >= 126 mg/dL, ADA 2024)") | |
| supporting_params.append("fasting_glucose") | |
| if val >= 250.0: | |
| severity = "SEVERE" | |
| elif val >= 100.0: | |
| supported_reasons.append(f"Fasting Blood Glucose = {val} mg/dL in impaired fasting glucose range [100 - 125 mg/dL] (ADA 2024)") | |
| supporting_params.append("fasting_glucose") | |
| if severity != "SEVERE": | |
| severity = "MILD" | |
| # Rule 3: Random Glucose | |
| if rbg.missingness == MissingnessStatus.REPORTED and rbg.value_numeric is not None: | |
| val = rbg.value_numeric | |
| trace.append({"param": "random_glucose", "value": val, "unit": "mg/dL", "threshold": ">= 200 (ADA 2024)"}) | |
| if val >= 200.0: | |
| supported_reasons.append(f"Random Blood Glucose = {val} mg/dL (threshold >= 200 mg/dL, ADA 2024)") | |
| supporting_params.append("random_glucose") | |
| severity = "SEVERE" | |
| # Check for intra-domain conflicts (e.g. normal fasting glucose < 100 but HbA1c >= 7.0) | |
| if ( | |
| fbg.missingness == MissingnessStatus.REPORTED | |
| and fbg.value_numeric is not None | |
| and fbg.value_numeric < 100.0 | |
| and hba1c.missingness == MissingnessStatus.REPORTED | |
| and hba1c.value_numeric is not None | |
| and hba1c.value_numeric >= 7.0 | |
| ): | |
| conflicts.append({ | |
| "domain": "sugar", | |
| "type": "LAB_DISCREPANCY", | |
| "detail": f"Normal fasting glucose ({fbg.value_numeric} mg/dL) conflicts with elevated HbA1c ({hba1c.value_numeric}%)", | |
| "review_required": True, | |
| }) | |
| if supported_reasons: | |
| inds.append( | |
| DomainIndicator( | |
| domain="sugar", | |
| indicator_name="ELEVATED_GLYCEMIC_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity=severity, | |
| evidence="; ".join(supported_reasons), | |
| confidence=confidence, | |
| supporting_parameters=supporting_params, | |
| review_required=len(conflicts) > 0 or severity == "SEVERE", | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| domain_status = IndicatorStatus.SUPPORTED | |
| else: | |
| inds.append( | |
| DomainIndicator( | |
| domain="sugar", | |
| indicator_name="ELEVATED_GLYCEMIC_INDICATOR", | |
| status=IndicatorStatus.NOT_SUPPORTED, | |
| severity=None, | |
| evidence="Glycemic parameters within normal sourced reference bounds (ADA 2024).", | |
| confidence=ConfidenceTier.HIGH, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| domain_status = IndicatorStatus.NOT_SUPPORTED | |
| return inds, trace, domain_status, conflicts | |
| def _analyze_blood(self, blood: Any) -> tuple[list[DomainIndicator], list[dict[str, Any]], IndicatorStatus]: | |
| inds: list[DomainIndicator] = [] | |
| trace: list[dict[str, Any]] = [] | |
| if blood.is_empty: | |
| return [ | |
| DomainIndicator( | |
| domain="blood", | |
| indicator_name="HEMATOLOGIC_ABNORMALITY_INDICATOR", | |
| status=IndicatorStatus.INSUFFICIENT_DATA, | |
| evidence="No hematologic parameters (Hemoglobin, Hematocrit, WBC, Platelets) reported.", | |
| confidence=ConfidenceTier.UNKNOWN, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ], trace, IndicatorStatus.INSUFFICIENT_DATA | |
| hb = blood.hemoglobin | |
| wbc = blood.wbc | |
| plt = blood.platelets | |
| any_supported = False | |
| # Hemoglobin (Anemia) | |
| if hb.missingness == MissingnessStatus.REPORTED and hb.value_numeric is not None: | |
| val = hb.value_numeric | |
| trace.append({"param": "hemoglobin", "value": val, "unit": "g/dL", "threshold": "< 12.0 (WHO)"}) | |
| if val < 12.0: | |
| any_supported = True | |
| sev = "SEVERE" if val < 8.0 else ("MODERATE" if val < 11.0 else "MILD") | |
| inds.append( | |
| DomainIndicator( | |
| domain="blood", | |
| indicator_name="ANEMIA_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity=sev, | |
| evidence=f"Hemoglobin = {val} g/dL (below standard WHO reference threshold of 12.0 g/dL)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["hemoglobin"], | |
| review_required=(sev == "SEVERE"), | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| # WBC (Leukocytosis / Leukopenia) | |
| if wbc.missingness == MissingnessStatus.REPORTED and wbc.value_numeric is not None: | |
| val = wbc.value_numeric | |
| trace.append({"param": "wbc", "value": val, "unit": "x10^3/uL", "normal_range": "4.5 - 11.0"}) | |
| if val > 11.0: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="blood", | |
| indicator_name="LEUKOCYTOSIS_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="SEVERE" if val > 20.0 else "MODERATE", | |
| evidence=f"WBC = {val} x10^3/uL (exceeds upper limit 11.0 x10^3/uL, Harrison's/Mayo)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["wbc"], | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| elif val < 4.0: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="blood", | |
| indicator_name="LEUKOPENIA_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="SEVERE" if val < 2.0 else "MODERATE", | |
| evidence=f"WBC = {val} x10^3/uL (below lower limit 4.0 x10^3/uL)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["wbc"], | |
| review_required=True, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| # Platelets (Thrombocytopenia / Thrombocytosis) | |
| if plt.missingness == MissingnessStatus.REPORTED and plt.value_numeric is not None: | |
| val = plt.value_numeric | |
| trace.append({"param": "platelets", "value": val, "unit": "x10^3/uL", "normal_range": "150 - 450"}) | |
| if val < 150.0: | |
| any_supported = True | |
| sev = "SEVERE" if val < 50.0 else "MODERATE" | |
| inds.append( | |
| DomainIndicator( | |
| domain="blood", | |
| indicator_name="THROMBOCYTOPENIA_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity=sev, | |
| evidence=f"Platelet count = {val} x10^3/uL (below reference threshold 150 x10^3/uL)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["platelets"], | |
| review_required=(sev == "SEVERE"), | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| elif val > 450.0: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="blood", | |
| indicator_name="THROMBOCYTOSIS_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="MODERATE", | |
| evidence=f"Platelet count = {val} x10^3/uL (exceeds reference threshold 450 x10^3/uL)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["platelets"], | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| if not any_supported: | |
| inds.append( | |
| DomainIndicator( | |
| domain="blood", | |
| indicator_name="HEMATOLOGIC_ABNORMALITY_INDICATOR", | |
| status=IndicatorStatus.NOT_SUPPORTED, | |
| evidence="Reported hematologic parameters are within normal reference ranges.", | |
| confidence=ConfidenceTier.HIGH, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| return inds, trace, IndicatorStatus.NOT_SUPPORTED | |
| return inds, trace, IndicatorStatus.SUPPORTED | |
| def _analyze_liver(self, liver: Any) -> tuple[list[DomainIndicator], list[dict[str, Any]], IndicatorStatus]: | |
| inds: list[DomainIndicator] = [] | |
| trace: list[dict[str, Any]] = [] | |
| if liver.is_empty: | |
| return [ | |
| DomainIndicator( | |
| domain="liver", | |
| indicator_name="HEPATIC_INJURY_INDICATOR", | |
| status=IndicatorStatus.INSUFFICIENT_DATA, | |
| evidence="No hepatic panel parameters (ALT, AST, ALP, Bilirubin, Albumin) reported.", | |
| confidence=ConfidenceTier.UNKNOWN, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ], trace, IndicatorStatus.INSUFFICIENT_DATA | |
| alt = liver.alt | |
| ast = liver.ast | |
| alp = liver.alp | |
| bili = liver.total_bilirubin | |
| alb = liver.albumin | |
| any_supported = False | |
| # Hepatocellular injury (ALT / AST) | |
| transaminase_elevated = False | |
| ev_trans: list[str] = [] | |
| supp_trans: list[str] = [] | |
| if alt.missingness == MissingnessStatus.REPORTED and alt.value_numeric is not None: | |
| trace.append({"param": "alt", "value": alt.value_numeric, "unit": "U/L", "upper_limit": 50.0}) | |
| if alt.value_numeric > 50.0: | |
| transaminase_elevated = True | |
| ev_trans.append(f"ALT = {alt.value_numeric} U/L (ULN 50 U/L, AASLD)") | |
| supp_trans.append("alt") | |
| if ast.missingness == MissingnessStatus.REPORTED and ast.value_numeric is not None: | |
| trace.append({"param": "ast", "value": ast.value_numeric, "unit": "U/L", "upper_limit": 48.0}) | |
| if ast.value_numeric > 48.0: | |
| transaminase_elevated = True | |
| ev_trans.append(f"AST = {ast.value_numeric} U/L (ULN 48 U/L, AASLD)") | |
| supp_trans.append("ast") | |
| if transaminase_elevated: | |
| any_supported = True | |
| is_marked = (alt.value_numeric and alt.value_numeric > 150.0) or (ast.value_numeric and ast.value_numeric > 150.0) | |
| inds.append( | |
| DomainIndicator( | |
| domain="liver", | |
| indicator_name="HEPATOCELLULAR_INJURY_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="SEVERE" if is_marked else "MODERATE", | |
| evidence="; ".join(ev_trans), | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=supp_trans, | |
| review_required=is_marked or False, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| # Cholestatic injury (ALP / Bilirubin) | |
| cholestatic_elevated = False | |
| ev_chol: list[str] = [] | |
| supp_chol: list[str] = [] | |
| if alp.missingness == MissingnessStatus.REPORTED and alp.value_numeric is not None: | |
| trace.append({"param": "alp", "value": alp.value_numeric, "unit": "U/L", "upper_limit": 129.0}) | |
| if alp.value_numeric > 129.0: | |
| cholestatic_elevated = True | |
| ev_chol.append(f"ALP = {alp.value_numeric} U/L (ULN 129 U/L)") | |
| supp_chol.append("alp") | |
| if bili.missingness == MissingnessStatus.REPORTED and bili.value_numeric is not None: | |
| trace.append({"param": "total_bilirubin", "value": bili.value_numeric, "unit": "mg/dL", "upper_limit": 1.2}) | |
| if bili.value_numeric > 1.2: | |
| cholestatic_elevated = True | |
| ev_chol.append(f"Total Bilirubin = {bili.value_numeric} mg/dL (ULN 1.2 mg/dL)") | |
| supp_chol.append("total_bilirubin") | |
| if cholestatic_elevated: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="liver", | |
| indicator_name="CHOLESTATIC_LIVER_INJURY_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="MODERATE", | |
| evidence="; ".join(ev_chol), | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=supp_chol, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| # Synthetic dysfunction (Albumin) | |
| if alb.missingness == MissingnessStatus.REPORTED and alb.value_numeric is not None: | |
| trace.append({"param": "albumin", "value": alb.value_numeric, "unit": "g/dL", "lower_limit": 3.5}) | |
| if alb.value_numeric < 3.5: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="liver", | |
| indicator_name="HEPATIC_SYNTHETIC_DYSFUNCTION_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="SEVERE" if alb.value_numeric < 2.5 else "MODERATE", | |
| evidence=f"Serum Albumin = {alb.value_numeric} g/dL (lower limit 3.5 g/dL)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["albumin"], | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| if not any_supported: | |
| inds.append( | |
| DomainIndicator( | |
| domain="liver", | |
| indicator_name="HEPATIC_INJURY_INDICATOR", | |
| status=IndicatorStatus.NOT_SUPPORTED, | |
| evidence="Reported liver enzymes and hepatic markers are within normal reference ranges.", | |
| confidence=ConfidenceTier.HIGH, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| return inds, trace, IndicatorStatus.NOT_SUPPORTED | |
| return inds, trace, IndicatorStatus.SUPPORTED | |
| def _analyze_lungs(self, lungs: Any) -> tuple[list[DomainIndicator], list[dict[str, Any]], IndicatorStatus]: | |
| inds: list[DomainIndicator] = [] | |
| trace: list[dict[str, Any]] = [] | |
| if lungs.is_empty: | |
| return [ | |
| DomainIndicator( | |
| domain="lungs", | |
| indicator_name="PULMONARY_IMPAIRMENT_INDICATOR", | |
| status=IndicatorStatus.INSUFFICIENT_DATA, | |
| evidence="No pulmonary vitals (SpO2, Respiratory Rate) or symptoms reported.", | |
| confidence=ConfidenceTier.UNKNOWN, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ], trace, IndicatorStatus.INSUFFICIENT_DATA | |
| spo2 = lungs.spo2 | |
| rr = lungs.respiratory_rate | |
| any_supported = False | |
| if spo2.missingness == MissingnessStatus.REPORTED and spo2.value_numeric is not None: | |
| val = spo2.value_numeric | |
| trace.append({"param": "spo2", "value": val, "unit": "%", "threshold": "< 95% (WHO)"}) | |
| if val < 90.0: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="lungs", | |
| indicator_name="HYPOXEMIA_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="SEVERE", | |
| evidence=f"SpO2 = {val}% (below critical hypoxemia threshold 90%, WHO/ATS)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["spo2"], | |
| review_required=True, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| elif val < 95.0: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="lungs", | |
| indicator_name="HYPOXEMIA_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="MILD", | |
| evidence=f"SpO2 = {val}% (mild hypoxemia range 90-94%, WHO)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["spo2"], | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| if rr.missingness == MissingnessStatus.REPORTED and rr.value_numeric is not None: | |
| val = rr.value_numeric | |
| trace.append({"param": "respiratory_rate", "value": val, "unit": "breaths/min", "normal_range": "12 - 20"}) | |
| if val > 20.0: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="lungs", | |
| indicator_name="TACHYPNEA_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="MODERATE", | |
| evidence=f"Respiratory Rate = {val} breaths/min (exceeds normal upper limit of 20 breaths/min)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["respiratory_rate"], | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| if not any_supported: | |
| inds.append( | |
| DomainIndicator( | |
| domain="lungs", | |
| indicator_name="PULMONARY_IMPAIRMENT_INDICATOR", | |
| status=IndicatorStatus.NOT_SUPPORTED, | |
| evidence="Reported respiratory vitals (SpO2, Respiratory Rate) are within normal reference limits.", | |
| confidence=ConfidenceTier.HIGH, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| return inds, trace, IndicatorStatus.NOT_SUPPORTED | |
| return inds, trace, IndicatorStatus.SUPPORTED | |
| def _analyze_intestine(self, intestine: Any) -> tuple[list[DomainIndicator], list[dict[str, Any]], IndicatorStatus]: | |
| inds: list[DomainIndicator] = [] | |
| trace: list[dict[str, Any]] = [] | |
| if intestine.is_empty: | |
| return [ | |
| DomainIndicator( | |
| domain="intestine", | |
| indicator_name="INTESTINAL_INFLAMMATION_INDICATOR", | |
| status=IndicatorStatus.INSUFFICIENT_DATA, | |
| evidence="No gastrointestinal or enteric markers (Calprotectin, CRP, ESR, Occult Blood) reported.", | |
| confidence=ConfidenceTier.UNKNOWN, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ], trace, IndicatorStatus.INSUFFICIENT_DATA | |
| calprot = intestine.fecal_calprotectin | |
| crp = intestine.crp | |
| esr = intestine.esr | |
| fobt = intestine.stool_occult_blood | |
| any_supported = False | |
| # Fecal Calprotectin | |
| if calprot.missingness == MissingnessStatus.REPORTED and calprot.value_numeric is not None: | |
| val = calprot.value_numeric | |
| trace.append({"param": "fecal_calprotectin", "value": val, "unit": "ug/g", "threshold": "> 50 (ACG)"}) | |
| if val > 120.0: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="intestine", | |
| indicator_name="INTESTINAL_INFLAMMATION_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="SEVERE", | |
| evidence=f"Fecal Calprotectin = {val} ug/g (threshold > 120 ug/g active inflammation, ACG guidelines)", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["fecal_calprotectin"], | |
| review_required=True, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| elif val > 50.0: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="intestine", | |
| indicator_name="INTESTINAL_INFLAMMATION_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="MILD", | |
| evidence=f"Fecal Calprotectin = {val} ug/g (borderline range 51 - 120 ug/g, ACG guidelines)", | |
| confidence=ConfidenceTier.MEDIUM, | |
| supporting_parameters=["fecal_calprotectin"], | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| # Stool Occult Blood | |
| if fobt.missingness == MissingnessStatus.REPORTED and fobt.raw_value: | |
| raw_upper = fobt.raw_value.strip().upper() | |
| trace.append({"param": "stool_occult_blood", "value": raw_upper}) | |
| if any(pos in raw_upper for pos in ("POS", "POSITIVE", "DETECTED", "REACTIVE", "ABNORMAL")): | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="intestine", | |
| indicator_name="GI_BLEEDING_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="MODERATE", | |
| evidence=f"Stool Occult Blood test reported POSITIVE ({fobt.raw_value})", | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=["stool_occult_blood"], | |
| review_required=True, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| # Systemic Inflammatory Markers (CRP / ESR) | |
| syst_inflam: list[str] = [] | |
| supp_inflam: list[str] = [] | |
| if crp.missingness == MissingnessStatus.REPORTED and crp.value_numeric is not None: | |
| trace.append({"param": "crp", "value": crp.value_numeric, "unit": "mg/L", "upper_limit": 5.0}) | |
| if crp.value_numeric > 5.0: | |
| syst_inflam.append(f"CRP = {crp.value_numeric} mg/L (ULN 5.0 mg/L)") | |
| supp_inflam.append("crp") | |
| if esr.missingness == MissingnessStatus.REPORTED and esr.value_numeric is not None: | |
| trace.append({"param": "esr", "value": esr.value_numeric, "unit": "mm/hr", "upper_limit": 20.0}) | |
| if esr.value_numeric > 20.0: | |
| syst_inflam.append(f"ESR = {esr.value_numeric} mm/hr (ULN 20 mm/hr)") | |
| supp_inflam.append("esr") | |
| if syst_inflam: | |
| any_supported = True | |
| inds.append( | |
| DomainIndicator( | |
| domain="intestine", | |
| indicator_name="SYSTEMIC_INFLAMMATION_INDICATOR", | |
| status=IndicatorStatus.SUPPORTED, | |
| severity="MODERATE", | |
| evidence="; ".join(syst_inflam), | |
| confidence=ConfidenceTier.HIGH, | |
| supporting_parameters=supp_inflam, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| if not any_supported: | |
| inds.append( | |
| DomainIndicator( | |
| domain="intestine", | |
| indicator_name="INTESTINAL_INFLAMMATION_INDICATOR", | |
| status=IndicatorStatus.NOT_SUPPORTED, | |
| evidence="Reported gastrointestinal markers (Calprotectin, CRP, ESR) are within normal reference ranges.", | |
| confidence=ConfidenceTier.HIGH, | |
| source_method=AnalysisMethod.RULE_BASED, | |
| ) | |
| ) | |
| return inds, trace, IndicatorStatus.NOT_SUPPORTED | |
| return inds, trace, IndicatorStatus.SUPPORTED | |
| def get_rule_analyzer() -> RuleBasedClinicalAnalyzer: | |
| """Return singleton rule analyzer instance.""" | |
| return RuleBasedClinicalAnalyzer() | |
| __all__ = [ | |
| "RuleBasedClinicalAnalyzer", | |
| "get_rule_analyzer", | |
| ] | |