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())