Blame for codeblue_context.py

import json
import re
import sys
from typing import Any, Dict, List, Optional, Tuple


def _to_list(value: Any) -> List[Any]:
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]


def _first_non_empty(*values: Any) -> str:
    for value in values:
        if isinstance(value, str) and value.strip():
            return value.strip()
    return ""


def _extract_text(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, dict):
        direct = _first_non_empty(
            value.get("name"),
            value.get("display"),
            value.get("text"),
            value.get("description"),
            value.get("title"),
            value.get("condition"),
            value.get("diagnosis"),
            value.get("substance"),
            value.get("allergen"),
            value.get("medication"),
            value.get("drug"),
            value.get("note"),
        )
        if direct:
            return direct

        code = value.get("code")
        if isinstance(code, dict):
            code_text = _first_non_empty(code.get("text"), code.get("display"))
            if code_text:
                return code_text
            coding = _to_list(code.get("coding"))
            for c in coding:
                if isinstance(c, dict):
                    coding_text = _first_non_empty(c.get("display"), c.get("code"))
                    if coding_text:
                        return coding_text
        elif isinstance(code, str) and code.strip():
            return code.strip()

    return ""


def _parse_float(value: Any) -> Optional[float]:
    if isinstance(value, (int, float)):
        return float(value)
    if isinstance(value, str):
        cleaned = value.strip().replace(",", "")
        if not cleaned:
            return None
        match = re.search(r"-?\d+(\.\d+)?", cleaned)
        if match:
            try:
                return float(match.group(0))
            except ValueError:
                return None
    return None


def _extract_reference_range(lab: Dict[str, Any]) -> Tuple[Optional[float], Optional[float]]:
    rr = lab.get("reference_range")
    if isinstance(rr, dict):
        low = _parse_float(rr.get("low"))
        high = _parse_float(rr.get("high"))
        return low, high
    if isinstance(rr, list) and rr:
        first = rr[0]
        if isinstance(first, dict):
            low = _parse_float(first.get("low"))
            high = _parse_float(first.get("high"))
            return low, high
    return None, None


def _normalize_vitals(vitals: Any) -> Dict[str, Any]:
    normalized: Dict[str, Any] = {}
    if isinstance(vitals, dict):
        for k, v in vitals.items():
            normalized[str(k).lower().strip()] = v
    elif isinstance(vitals, list):
        for item in vitals:
            if not isinstance(item, dict):
                continue
            key = _extract_text(item.get("name") or item.get("type") or item.get("code"))
            if not key:
                continue
            normalized[key.lower().strip()] = item.get("value")
    return normalized


def _dedupe_keep_order(items: List[str]) -> List[str]:
    seen = set()
    result = []
    for item in items:
        text = item.strip()
        if not text:
            continue
        if text not in seen:
            seen.add(text)
            result.append(text)
    return result


def _has_documented_data(value: Any) -> bool:
    if value is None:
        return False
    if isinstance(value, str):
        return bool(value.strip())
    if isinstance(value, (list, tuple, set, dict)):
        return len(value) > 0
    return True


def _is_placeholder_entry(text: str) -> bool:
    t = text.strip().lower()
    return (
        t == "unknown"
        or t == "not documented"
        or t == "none reported"
        or t.startswith("none identified")
        or t.startswith("no critical abnormalities identified")
        or t.startswith("no immediate life-threatening alert identified")
    )


def build_emergency_snapshot(record: Dict[str, Any]) -> Tuple[str, List[Dict[str, str]], List[str], Dict[str, Any]]:
    if not isinstance(record, dict):
        record = {}

    reasoning: List[Dict[str, str]] = []
    def add_reasoning(explanation: str, source_text: str = ""):
        reasoning.append({"explanation": explanation, "source_text": source_text})

    add_reasoning("Initiating CodeBlue emergency analysis.")

    patient = record.get("patient", {})
    age = patient.get("age") if isinstance(patient, dict) else None
    gender = patient.get("gender") if isinstance(patient, dict) else None
    add_reasoning(f"Extracted patient demographics: Age {age or 'Unknown'}, Gender {gender or 'Unknown'}.")

    raw_allergies = _to_list(record.get("allergies"))
    raw_medications = _to_list(record.get("medications"))
    raw_conditions = _to_list(record.get("conditions"))
    raw_labs = _to_list(record.get("labs"))
    raw_history = _to_list(record.get("history"))
    raw_clinical_notes = _to_list(record.get("clinical_notes", record.get("clinical_notes", "")))
    raw_notes = _to_list(record.get("notes"))
    raw_vitals = record.get("vitals")
    vitals = _normalize_vitals(raw_vitals)
    add_reasoning("Normalized and loaded all available clinical data sources.")

    # Combine all text sources for narrative analysis
    all_text_sources = [
        str(item) for item in 
        raw_allergies + raw_medications + raw_conditions + raw_labs + 
        raw_history + raw_clinical_notes + raw_notes
    ]
    full_narrative = " ".join(all_text_sources).lower()


    has_labs_documented = _has_documented_data(record.get("labs"))
    has_vitals_documented = _has_documented_data(raw_vitals)
    has_medications_documented = _has_documented_data(record.get("medications"))

    critical_alerts: List[str] = []
    allergies_out: List[str] = []
    high_risk_medications: List[str] = []
    chronic_conditions: List[str] = []
    recent_abnormalities: List[str] = []
    emergency_risks: List[str] = []
    extracted_vitals: Dict[str, Any] = {}

    # Allergies
    severe_allergy = False
    for allergy in raw_allergies:
        allergy_text = _extract_text(allergy)
        if not allergy_text and isinstance(allergy, dict):
            allergy_text = _first_non_empty(
                _extract_text(allergy.get("substance")),
                _extract_text(allergy.get("allergen")),
                _extract_text(allergy.get("code")),
            )
        if allergy_text:
            allergies_out.append(allergy_text)
            add_reasoning(f"Found allergy: {allergy_text}", source_text=allergy_text)


    # Conditions
    conditions = [_extract_text(c) for c in raw_conditions]
    conditions = [c for c in conditions if c]
    chronic_keywords = (
        "diabetes",
        "hypertension",
        "heart",
        "coronary",
        "copd",
        "asthma",
        "ckd",
        "kidney disease",
        "renal",
        "stroke",
        "epilepsy",
        "cancer",
        "chf",
        "heart failure",
        "afib",
        "atrial fibrillation",
    )
    for condition in conditions:
        if any(keyword in condition.lower() for keyword in chronic_keywords):
            chronic_conditions.append(condition)
            add_reasoning(f"Identified emergency-relevant chronic condition: {condition}", source_text=condition)

    # Medications
    meds = [_extract_text(m) for m in raw_medications]
    meds = [m for m in meds if m]

    anticoagulants = (
        "warfarin",
        "heparin",
        "enoxaparin",
        "apixaban",
        "rivaroxaban",
        "dabigatran",
        "edoxaban",
    )
    opioids = (
        "morphine",
        "oxycodone",
        "hydromorphone",
        "fentanyl",
        "codeine",
        "tramadol",
        "methadone",
    )
    cardiac_meds = (
        "digoxin",
        "amiodarone",
        "metoprolol",
        "atenolol",
        "carvedilol",
        "diltiazem",
        "verapamil",
        "nitroglycerin",
    )
    has_anticoagulant = False
    has_insulin = False
    has_opioid = False
    has_cardiac_med = False

    for med in meds:
        med_l = med.lower()
        if any(name in med_l for name in anticoagulants):
            has_anticoagulant = True
            high_risk_medications.append(f"{med} (anticoagulant)")
            add_reasoning(f"Detected high-risk medication (anticoagulant): {med}", source_text=med)
        if "insulin" in med_l:
            has_insulin = True
            high_risk_medications.append(f"{med} (insulin)")
            add_reasoning(f"Detected high-risk medication (insulin): {med}", source_text=med)
        if any(name in med_l for name in opioids):
            has_opioid = True
            high_risk_medications.append(f"{med} (opioid)")
            add_reasoning(f"Detected high-risk medication (opioid): {med}", source_text=med)
        if any(name in med_l for name in cardiac_meds):
            has_cardiac_med = True
            high_risk_medications.append(f"{med} (cardiac medication)")
            add_reasoning(f"Detected high-risk medication (cardiac): {med}", source_text=med)

    # Labs
    for lab in raw_labs:
        if isinstance(lab, str):
            text = lab.strip()
            if text:
                text_l = text.lower()
                if any(word in text_l for word in ("critical", "abnormal", "high", "low")):
                    recent_abnormalities.append(text)
                    if "critical" in text_l:
                        critical_alerts.append(f"Critical lab finding: {text}.")
            continue

        if not isinstance(lab, dict):
            continue

        name = _extract_text(lab.get("name") or lab.get("test") or lab.get("code")) or "Lab result"
        value_raw = lab.get("value")
        if value_raw is None and isinstance(lab.get("result"), dict):
            value_raw = lab["result"].get("value")
        elif value_raw is None:
            value_raw = lab.get("result")

        value_num = _parse_float(value_raw)
        unit = _extract_text(lab.get("unit"))
        flag = _extract_text(lab.get("flag")).lower()
        low, high = _extract_reference_range(lab)

        abnormal = False
        critical = False
        if flag and any(token in flag for token in ("abnormal", "high", "low", "critical", "panic")):
            abnormal = True
        if flag and any(token in flag for token in ("critical", "panic", "severe")):
            critical = True
        if value_num is not None:
            if low is not None and value_num < low:
                abnormal = True
            if high is not None and value_num > high:
                abnormal = True

        if abnormal:
            value_text = str(value_raw) if value_raw is not None else "reported"
            detail = f"{name}: {value_text}{(' ' + unit) if unit else ''}"
            if flag:
                detail += f" ({flag})"
            recent_abnormalities.append(detail)
            add_reasoning(f"Found abnormal lab: {detail}", source_text=str(lab))
            if critical:
                critical_alerts.append(f"Critical lab abnormality: {detail}.")
                add_reasoning(f"Flagged CRITICAL lab: {detail}", source_text=str(lab))

    # Vitals Extraction and Analysis
    def add_vital_abnormality(label: str, value: Any, is_critical: bool, source: str) -> None:
        detail = f"{label}: {value}"
        recent_abnormalities.append(detail)
        if is_critical:
            critical_alerts.append(f"Critical vital sign: {detail}.")
            add_reasoning(f"Flagged CRITICAL vital: {detail}", source_text=source)
        else:
            add_reasoning(f"Noted abnormal vital: {detail}", source_text=source)

    # Regex patterns for vitals
    vital_patterns = {
        "Blood Pressure": r"bp\s*:?\s*(\d{2,3})\s*/\s*(\d{2,3})",
        "Heart Rate": r"hr\s*:?\s*(\d{2,3})",
        "SpO2": r"spo2\s*:?\s*(\d{2,3})%?",
        "Respiratory Rate": r"rr\s*:?\s*(\d{1,2})",
        "Temperature": r"temp\s*:?\s*(\d{2}(\.\d)?)",
    }

    for label, pattern in vital_patterns.items():
        match = re.search(pattern, full_narrative, re.IGNORECASE)
        if match:
            if label == "Blood Pressure":
                systolic, diastolic = map(int, match.groups())
                value_str = f"{systolic}/{diastolic}"
                extracted_vitals[label] = value_str
                add_reasoning(f"Extracted {label}: {value_str}", source_text=match.group(0))
                if systolic > 180 or diastolic > 120 or systolic < 90:
                     add_vital_abnormality(label, value_str, True, match.group(0))
            else:
                value = float(match.group(1))
                extracted_vitals[label] = value
                add_reasoning(f"Extracted {label}: {value}", source_text=match.group(0))
                # Add abnormality checks here based on value
                if label == "Heart Rate" and (value > 130 or value < 40):
                    add_vital_abnormality(label, f"{value} bpm", True, match.group(0))
                elif label == "SpO2" and value < 90:
                    add_vital_abnormality(label, f"{value}%", True, match.group(0))


    # History / notes signals
    narrative_text = " ".join(
        [_extract_text(item) for item in raw_history + raw_clinical_notes + raw_notes]
    ).lower()

    if "icu" in narrative_text:
        critical_alerts.append("ICU history documented.")
    if "anaphyl" in narrative_text and not severe_allergy:
        critical_alerts.append("History suggests severe allergic reaction.")
        severe_allergy = True

    # Emergency risks
    has_diabetes = any("diabetes" in c.lower() for c in conditions + chronic_conditions)
    has_cardiac_condition = any(
        token in c.lower()
        for c in conditions + chronic_conditions
        for token in ("heart", "coronary", "chf", "afib", "atrial fibrillation", "myocardial infarction")
    )
    trauma_or_fall_risk = any(token in narrative_text for token in ("fall", "trauma", "injury", "post-op", "surgery"))

    if has_anticoagulant:
        emergency_risks.append("Bleeding Risk (High)")
        add_reasoning("Synthesized risk: Anticoagulant exposure elevates bleeding risk.")
    if has_insulin or has_diabetes:
        emergency_risks.append("Hypoglycemia Risk (Moderate)")
        add_reasoning("Synthesized risk: Diabetes/insulin factors increase hypoglycemia risk.")
    if has_opioid:
        emergency_risks.append("Respiratory Depression Risk (High)")
        add_reasoning("Synthesized risk: Opioid exposure elevates respiratory depression risk.")
    if has_cardiac_condition or has_cardiac_med:
        emergency_risks.append("Cardiac Event Risk (Moderate)")
        add_reasoning("Synthesized risk: Cardiac history/meds increase risk of cardiac events.")
    if severe_allergy:
        emergency_risks.append("Anaphylaxis Risk (High)")
        add_reasoning("Synthesized risk: Severe allergy history poses high anaphylaxis risk.")


    # Final Snapshot Generation
    final_lines = []
    if critical_alerts:
        final_lines.extend(critical_alerts)

    summary_line = f"{age or 'Unknown'}-y/o {gender or 'Unknown'} with history of {', '.join(chronic_conditions[:2]) if chronic_conditions else 'undisclosed conditions'}."
    final_lines.append(summary_line)

    risk_summary_line = f"Primary risks: {', '.join(emergency_risks[:3]) if emergency_risks else 'none identified'}."
    final_lines.append(risk_summary_line)

    if not final_lines:
        final_lines.append("No critical issues identified from available data.")

    add_reasoning("Finalizing emergency snapshot based on synthesized risks and alerts.")
    final_snapshot = "\n".join(_dedupe_keep_order(final_lines)[:6])

    return final_snapshot, reasoning, emergency_risks, extracted_vitals


def main() -> int:
    raw = ""
    if len(sys.argv) > 1:
        path = sys.argv[1]
        with open(path, "r", encoding="utf-8") as f:
            raw = f.read()
    else:
        raw = sys.stdin.read()

    if not raw.strip():
        record: Dict[str, Any] = {}
    else:
        try:
            parsed = json.loads(raw)
            record = parsed if isinstance(parsed, dict) else {}
        except json.JSONDecodeError:
            record = {}

    snapshot, reasoning, risks, vitals = build_emergency_snapshot(record)
    result = {
        "snapshot": snapshot,
        "reasoning": reasoning,
        "risks": risks,
        "vitals": vitals,
    }
    print(json.dumps(result, indent=2, ensure_ascii=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())