Betegellátási folyamat adatsablonja

// AIE Transformation Script (Pseudo-code for Config)
// Objective: Parse HL7 message and append line to CSV
// Output Columns: PatientEpisodeId, ActivityName, EventDateTime, PatientMrn, AttendingProvider, DepartmentName, PrimaryDiagnosisCode

// 1. Initialize Variables
var msgType = HL7.GetField("MSH-9-1") + "^" + HL7.GetField("MSH-9-2");
var visitNumber = HL7.GetField("PV1-19");
var mrn = HL7.GetField("PID-3");
var eventTime = HL7.GetField("MSH-7"); // Or EVN-2
var provider = HL7.GetField("PV1-7");
var dept = HL7.GetField("PV1-3");
var diagnosis = HL7.GetField("DG1-3"); // For Diagnosis events
var orderStatus = HL7.GetField("ORC-5");
var resultStatus = HL7.GetField("OBR-25");
var activityName = "";

// 2. Map Message Types to Activities
// Registration & Admission
if (msgType == "ADT^A04" || msgType == "ADT^A01") {
    activityName = "Patient Registered";
}
else if (msgType == "ADT^A02") {
    activityName = "Transfer to Department";
}
else if (msgType == "ADT^A03") {
    activityName = "Patient Discharged";
}
// Scheduling
else if (msgType == "SIU^S12") {
    activityName = "Follow-up Scheduled";
}
// Orders (Diagnostics, Care Plans, Discharge Planning)
else if (msgType == "ORM^O01") {
    var orderControl = HL7.GetField("ORC-1");
    var orderType = HL7.GetField("OBR-4"); // Check for specific service codes
    
    if (orderControl == "NW") { // New Order
        if (orderType.Contains("LAB") || orderType.Contains("RAD")) {
            activityName = "Diagnostic Test Ordered";
        }
        else if (orderType.Contains("DISCHARGE")) {
            activityName = "Discharge Planning Initiated";
        }
        else if (orderType.Contains("CARE_PLAN")) {
            activityName = "Treatment Plan Created";
        }
    }
}
// Results & Vitals
else if (msgType == "ORU^R01") {
    // Check if it is Vitals or Lab Result based on OBX codes or sending app
    var observationId = HL7.GetField("OBX-3"); 
    if (observationId.Contains("VITALS")) {
        activityName = "Vitals Recorded";
    } else {
        if (resultStatus == "F" || resultStatus == "C") {
            activityName = "Diagnostic Test Completed";
        }
    }
    
    // Initial Assessment Check (often a specific doc type in ORU or MDM)
    if (observationId.Contains("INITIAL_ASSESS")) {
        activityName = "Initial Assessment Completed";
    }
}
// Medication Administration (RAS)
else if (msgType == "RAS^O17") {
    activityName = "Medication Administered";
}
// Billing/Procedures (DFT)
else if (msgType == "DFT^P03") {
    activityName = "Procedure Performed";
}
// Diagnosis Updates (Often in ADT or DFT, logic depends on config)
else if (msgType == "ADT^A08" && diagnosis != "") {
    // Check if diagnosis segment is new/updated
    activityName = "Diagnosis Confirmed";
}

// 3. Write to File if Activity Identified
if (activityName != "" && visitNumber != "") {
    // Format Date to ISO
    var formattedDate = FormatDate(eventTime, "yyyy-MM-dd HH:mm:ss");
    
    // Construct CSV Line
    var csvLine = visitNumber + "," + 
                  activityName + "," + 
                  formattedDate + "," + 
                  mrn + "," + 
                  provider + "," + 
                  dept + "," + 
                  diagnosis;
                  
    File.WriteLine("[Configured_Output_Path]\\PatientJourneyLog.csv", csvLine);
}

import requests
import pandas as pd
import json
from datetime import datetime, timedelta

class=class="hl-string">"hl-comment"># === CONFIGURATION ===
API_URL = class="hl-string">"https://unity.veradigm.com/[YourEnvironment]/unity/unityservice.svc/json"
APP_NAME = class="hl-string">"[YourAppName]"
USER_ID = class="hl-string">"[YourUserID]"
PASSWORD = class="hl-string">"[YourPassword]"
START_DATE = class="hl-string">"2023-01-01"
END_DATE = class="hl-string">"2023-01-31"
OUTPUT_FILE = class="hl-string">"veradigm_patient_journey_log.csv"

class=class="hl-string">"hl-comment"># Global Session Token
TOKEN = None

def get_security_token():
    class="hl-string">"""Authenticate and retrieve security token."""
    payload = {
        class="hl-string">"Action": class="hl-string">"GetToken",
        class="hl-string">"Appname": APP_NAME,
        class="hl-string">"Username": USER_ID,
        class="hl-string">"Password": PASSWORD
    }
    try:
        response = requests.post(API_URL, json=payload)
        response.raise_for_status()
        data = response.json()
        class=class="hl-string">"hl-comment"># Depending on Unity version, token might be in body or header. Assuming body here.
        return data[0][class="hl-string">'Token'] if isinstance(data, list) else data.get(class="hl-string">'Token')
    except Exception as e:
        print(fclass="hl-string">"Authentication failed: {e}")
        return None

def make_unity_request(action, params):
    class="hl-string">"""Wrapper for Unity API calls."""
    payload = {
        class="hl-string">"Action": action,
        class="hl-string">"Appname": APP_NAME,
        class="hl-string">"Token": TOKEN
    }
    payload.update(params)
    
    try:
        response = requests.post(API_URL, json=payload, timeout=30)
        response.raise_for_status()
        class=class="hl-string">"hl-comment"># Unity often returns a list of dictionaries
        return response.json()
    except Exception as e:
        print(fclass="hl-string">"API Call Error ({action}): {e}")
        return []

def extract_patient_journey():
    global TOKEN
    TOKEN = get_security_token()
    if not TOKEN:
        return

    print(class="hl-string">"Token acquired. Starting extraction...")
    
    event_log = []

    class=class="hl-string">"hl-comment"># 1. Get Encounters within Date Range
    class=class="hl-string">"hl-comment"># Using GetEncounterInfo or similar search action
    encounters = make_unity_request(class="hl-string">"GetEncounterInfo", {
        class="hl-string">"StartEncounterDate": START_DATE,
        class="hl-string">"EndEncounterDate": END_DATE
    })

    print(fclass="hl-string">"Found {len(encounters)} encounters. Processing details...")

    for enc in encounters:
        try:
            class=class="hl-string">"hl-comment"># Basic Encounter Data
            episode_id = str(enc.get(class="hl-string">'EncounterID'))
            patient_id = enc.get(class="hl-string">'PatientID')
            mrn = enc.get(class="hl-string">'MRN', class="hl-string">'')
            admit_time = enc.get(class="hl-string">'EncounterDate') class=class="hl-string">"hl-comment"># ISO string expected
            discharge_time = enc.get(class="hl-string">'DischargeDate')
            dept_name = enc.get(class="hl-string">'Department', class="hl-string">'General')
            attending = enc.get(class="hl-string">'ProviderName', class="hl-string">'Unknown')
            
            class=class="hl-string">"hl-comment"># --- Activity: Patient Registered ---
            if admit_time:
                event_log.append({
                    class="hl-string">"PatientEpisodeId": episode_id,
                    class="hl-string">"ActivityName": class="hl-string">"Patient Registered",
                    class="hl-string">"EventDateTime": admit_time,
                    class="hl-string">"PatientMrn": mrn,
                    class="hl-string">"AttendingProvider": attending,
                    class="hl-string">"DepartmentName": dept_name,
                    class="hl-string">"AdmissionDate": admit_time,
                    class="hl-string">"DischargeDate": discharge_time
                })

            class=class="hl-string">"hl-comment"># --- Activity: Patient Discharged ---
            if discharge_time:
                 event_log.append({
                    class="hl-string">"PatientEpisodeId": episode_id,
                    class="hl-string">"ActivityName": class="hl-string">"Patient Discharged",
                    class="hl-string">"EventDateTime": discharge_time,
                    class="hl-string">"PatientMrn": mrn,
                    class="hl-string">"AttendingProvider": attending,
                    class="hl-string">"DepartmentName": dept_name,
                    class="hl-string">"AdmissionDate": admit_time,
                    class="hl-string">"DischargeDate": discharge_time
                })

            class=class="hl-string">"hl-comment"># 2. Get Orders (Labs, Procedures, Consults)
            orders = make_unity_request(class="hl-string">"GetOrders", {class="hl-string">"PatientID": patient_id, class="hl-string">"EncounterID": episode_id})
            for order in orders:
                order_time = order.get(class="hl-string">'OrderDate')
                order_type = order.get(class="hl-string">'OrderType', class="hl-string">'').lower()
                order_name = order.get(class="hl-string">'OrderName', class="hl-string">'Unknown Order')
                
                class=class="hl-string">"hl-comment"># --- Activity: Diagnostic Test Ordered ---
                if class="hl-string">'lab' in order_type or class="hl-string">'imaging' in order_type or class="hl-string">'radiology' in order_type:
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Diagnostic Test Ordered",
                        class="hl-string">"EventDateTime": order_time,
                        class="hl-string">"PatientMrn": mrn,
                        class="hl-string">"Attribute_OrderName": order_name
                    })

                class=class="hl-string">"hl-comment"># --- Activity: Procedure Performed ---
                class=class="hl-string">"hl-comment"># Often procedures are orders with status completed or specific types
                if class="hl-string">'procedure' in order_type and order.get(class="hl-string">'Status') == class="hl-string">'Completed':
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Procedure Performed",
                        class="hl-string">"EventDateTime": order.get(class="hl-string">'CompletionDate', order_time),
                        class="hl-string">"PatientMrn": mrn,
                        class="hl-string">"Attribute_ProcedureName": order_name
                    })
                    
                class=class="hl-string">"hl-comment"># --- Activity: Discharge Planning Initiated ---
                if class="hl-string">'discharge' in order_name.lower():
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Discharge Planning Initiated",
                        class="hl-string">"EventDateTime": order_time,
                        class="hl-string">"PatientMrn": mrn
                    })

            class=class="hl-string">"hl-comment"># 3. Get Results (Lab completion)
            results = make_unity_request(class="hl-string">"GetResults", {class="hl-string">"PatientID": patient_id, class="hl-string">"EncounterID": episode_id})
            for res in results:
                result_time = res.get(class="hl-string">'ResultDate')
                if result_time:
                    class=class="hl-string">"hl-comment"># --- Activity: Diagnostic Test Completed ---
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Diagnostic Test Completed",
                        class="hl-string">"EventDateTime": result_time,
                        class="hl-string">"PatientMrn": mrn,
                        class="hl-string">"Attribute_ResultName": res.get(class="hl-string">'TestName', class="hl-string">'Unknown Test')
                    })

            class=class="hl-string">"hl-comment"># 4. Get Problems (Diagnosis)
            problems = make_unity_request(class="hl-string">"GetPatientProblems", {class="hl-string">"PatientID": patient_id, class="hl-string">"EncounterID": episode_id})
            for prob in problems:
                onset_date = prob.get(class="hl-string">'OnsetDate') or prob.get(class="hl-string">'DateEntered')
                if onset_date:
                     class=class="hl-string">"hl-comment"># --- Activity: Diagnosis Confirmed ---
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Diagnosis Confirmed",
                        class="hl-string">"EventDateTime": onset_date,
                        class="hl-string">"PatientMrn": mrn,
                        class="hl-string">"PrimaryDiagnosisCode": prob.get(class="hl-string">'ICD10', class="hl-string">''),
                        class="hl-string">"Attribute_DiagnosisName": prob.get(class="hl-string">'ProblemDescription', class="hl-string">'')
                    })

            class=class="hl-string">"hl-comment"># 5. Get Vitals
            vitals = make_unity_request(class="hl-string">"GetVitals", {class="hl-string">"PatientID": patient_id, class="hl-string">"EncounterID": episode_id})
            for vit in vitals:
                taken_time = vit.get(class="hl-string">'DateTaken')
                if taken_time:
                    class=class="hl-string">"hl-comment"># --- Activity: Vitals Recorded ---
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Vitals Recorded",
                        class="hl-string">"EventDateTime": taken_time,
                        class="hl-string">"PatientMrn": mrn
                    })

            class=class="hl-string">"hl-comment"># 6. Get Clinical Notes (Assessments, Plans)
            class=class="hl-string">"hl-comment"># Sometimes accessed via GetNoteInfo or GetChart
            notes = make_unity_request(class="hl-string">"GetNoteInfo", {class="hl-string">"PatientID": patient_id, class="hl-string">"EncounterID": episode_id})
            for note in notes:
                note_date = note.get(class="hl-string">'NoteDate')
                note_title = note.get(class="hl-string">'DocumentType', class="hl-string">'').lower()
                
                class=class="hl-string">"hl-comment"># --- Activity: Initial Assessment Completed ---
                if class="hl-string">'initial' in note_title or class="hl-string">'assessment' in note_title:
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Initial Assessment Completed",
                        class="hl-string">"EventDateTime": note_date,
                        class="hl-string">"PatientMrn": mrn
                    })
                
                class=class="hl-string">"hl-comment"># --- Activity: Treatment Plan Created ---
                if class="hl-string">'plan' in note_title or class="hl-string">'care' in note_title:
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Treatment Plan Created",
                        class="hl-string">"EventDateTime": note_date,
                        class="hl-string">"PatientMrn": mrn
                    })

            class=class="hl-string">"hl-comment"># 7. Med Administration (GetMedicationAdministration or inferred from Med History)
            meds = make_unity_request(class="hl-string">"GetMedicationAdministration", {class="hl-string">"PatientID": patient_id, class="hl-string">"EncounterID": episode_id})
            for med in meds:
                admin_time = med.get(class="hl-string">'AdministrationDate')
                if admin_time:
                    class=class="hl-string">"hl-comment"># --- Activity: Medication Administered ---
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Medication Administered",
                        class="hl-string">"EventDateTime": admin_time,
                        class="hl-string">"PatientMrn": mrn,
                        class="hl-string">"Attribute_MedicationName": med.get(class="hl-string">'MedicationName', class="hl-string">'')
                    })

            class=class="hl-string">"hl-comment"># 8. Transfer Events (Infer from ADT or Audit log)
            class=class="hl-string">"hl-comment"># Simulating transfer extraction via AuditLog for Location changes
            audit_logs = make_unity_request(class="hl-string">"GetAuditLog", {class="hl-string">"PatientID": patient_id, class="hl-string">"EncounterID": episode_id})
            for log in audit_logs:
                if class="hl-string">"Transfer" in log.get(class="hl-string">'EventName', class="hl-string">'') or class="hl-string">"Location Change" in log.get(class="hl-string">'EventName', class="hl-string">''):
                     class=class="hl-string">"hl-comment"># --- Activity: Transfer to Department ---
                     event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Transfer to Department",
                        class="hl-string">"EventDateTime": log.get(class="hl-string">'EventDate'),
                        class="hl-string">"PatientMrn": mrn,
                        class="hl-string">"DepartmentName": log.get(class="hl-string">'NewValue', class="hl-string">'Unknown Dept')
                     })

            class=class="hl-string">"hl-comment"># 9. Follow-up Scheduled
            if discharge_time:
                future_appts = make_unity_request(class="hl-string">"GetSchedule", {class="hl-string">"PatientID": patient_id, class="hl-string">"StartDate": discharge_time})
                for appt in future_appts:
                    class=class="hl-string">"hl-comment"># --- Activity: Follow-up Scheduled ---
                    class=class="hl-string">"hl-comment"># Note: We want the time the appt was BOOKED, but Unity usually gives ApptDate.
                    class=class="hl-string">"hl-comment"># We will use the creation date of the appt if available, otherwise ApptDate as proxy
                    event_log.append({
                        class="hl-string">"PatientEpisodeId": episode_id,
                        class="hl-string">"ActivityName": class="hl-string">"Follow-up Scheduled",
                        class="hl-string">"EventDateTime": appt.get(class="hl-string">'DateBooked', appt.get(class="hl-string">'ApptDate')), 
                        class="hl-string">"PatientMrn": mrn
                    })

        except Exception as e:
            print(fclass="hl-string">"Error processing encounter {enc.get('EncounterID')}: {e}")
            continue

    class=class="hl-string">"hl-comment"># Post-Processing: Create DataFrame
    df = pd.DataFrame(event_log)
    
    class=class="hl-string">"hl-comment"># Helper: Convert strings to datetime
    df[class="hl-string">'EventDateTime'] = pd.to_datetime(df[class="hl-string">'EventDateTime'], errors=class="hl-string">'coerce')
    df = df.dropna(subset=[class="hl-string">'EventDateTime', class="hl-string">'PatientEpisodeId'])
    
    class=class="hl-string">"hl-comment"># Calculate LOS and Readmission
    df[class="hl-string">'AdmissionDate'] = pd.to_datetime(df[class="hl-string">'AdmissionDate'], errors=class="hl-string">'coerce')
    df[class="hl-string">'DischargeDate'] = pd.to_datetime(df[class="hl-string">'DischargeDate'], errors=class="hl-string">'coerce')
    
    class=class="hl-string">"hl-comment"># LOS
    df[class="hl-string">'LengthOfStay'] = (df[class="hl-string">'DischargeDate'] - df[class="hl-string">'AdmissionDate']).dt.days

    class=class="hl-string">"hl-comment"># IsReadmission (Simplified: check if patient had previous discharge within 30 days)
    class=class="hl-string">"hl-comment"># Requires sorting by patient and admission time first
    class=class="hl-string">"hl-comment"># This is complex in a flat event log context, simplified here as placeholder logic
    df[class="hl-string">'IsReadmission'] = False 

    print(fclass="hl-string">"Extraction complete. Rows: {len(df)}")
    df.to_csv(OUTPUT_FILE, index=False)
    print(fclass="hl-string">"Data saved to {OUTPUT_FILE}")

if __name__ == class="hl-string">"__main__":
    extract_patient_journey()