Spaces:
Sleeping
Sleeping
| 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()) | |