"""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", ]