환자 여정 데이터 템플릿

import pandas as pd
import numpy as np
from datetime import datetime

class=class="hl-string">"hl-comment"># Configuration: Input and Output filenames
INPUT_FILE = class="hl-string">'athena_raw_export.csv'
OUTPUT_FILE = class="hl-string">'patient_journey_event_log.csv'

class=class="hl-string">"hl-comment"># Load the raw wide-format report from athenaNet
class=class="hl-string">"hl-comment"># Assumes standard CSV export. Adjust encoding if necessary (e.g., class="hl-string">'utf-8-sig' or class="hl-string">'latin1')
df = pd.read_csv(INPUT_FILE, dtype=str)

class=class="hl-string">"hl-comment"># Define the mapping between Activity Names and the columns in your athenaNet report
class=class="hl-string">"hl-comment"># You must ensure these column names match your Report Builder export headers exactly
activity_mapping = {
    class="hl-string">'Appointment Scheduled': class="hl-string">'APPT_CREATED_TIMESTAMP',
    class="hl-string">'Patient Checked In': class="hl-string">'CHECKIN_TIMESTAMP',
    class="hl-string">'Initial Assessment Completed': class="hl-string">'INTAKE_COMPLETED_TIMESTAMP',
    class="hl-string">'Diagnostic Test Ordered': class="hl-string">'ORDER_ENTRY_TIMESTAMP',
    class="hl-string">'Specimen Collected': class="hl-string">'LAB_COLLECTION_TIMESTAMP',
    class="hl-string">'Test Results Received': class="hl-string">'LAB_RESULT_TIMESTAMP',
    class="hl-string">'Diagnosis Confirmed': class="hl-string">'DIAGNOSIS_ENTRY_TIMESTAMP',
    class="hl-string">'Treatment Plan Developed': class="hl-string">'PLAN_NOTE_TIMESTAMP',
    class="hl-string">'Medication Administered': class="hl-string">'MED_ADMIN_TIMESTAMP',
    class="hl-string">'Procedure Performed': class="hl-string">'PROCEDURE_END_TIMESTAMP',
    class="hl-string">'Patient Transferred': class="hl-string">'TRANSFER_TIMESTAMP',
    class="hl-string">'Discharge Order Written': class="hl-string">'DISCHARGE_ORDER_TIMESTAMP',
    class="hl-string">'Patient Discharged': class="hl-string">'DISCHARGE_TIMESTAMP',
    class="hl-string">'Follow-Up Appointment Scheduled': class="hl-string">'NEXT_APPT_CREATED_TIMESTAMP',
    class="hl-string">'Claim Submitted': class="hl-string">'CLAIM_SUBMISSION_TIMESTAMP'
}

class=class="hl-string">"hl-comment"># Define attribute columns to carry over to every event row
attribute_columns = {
    class="hl-string">'ENCOUNTER_ID': class="hl-string">'PatientEpisodeId',
    class="hl-string">'PATIENT_ID': class="hl-string">'PatientId',
    class="hl-string">'PROVIDER_NAME': class="hl-string">'ProviderName',
    class="hl-string">'DEPARTMENT_NAME': class="hl-string">'DepartmentName',
    class="hl-string">'PRIMARY_DIAGNOSIS_CODE': class="hl-string">'PrimaryDiagnosisCode',
    class="hl-string">'PATIENT_AGE': class="hl-string">'PatientAgeGroup',
    class="hl-string">'ENCOUNTER_TYPE': class="hl-string">'EncounterType',
    class="hl-string">'DISCHARGE_DISPOSITION': class="hl-string">'DischargeDisposition',
    class="hl-string">'IS_READMISSION': class="hl-string">'IsReadmission'
}

class=class="hl-string">"hl-comment"># Initialize list to hold event rows
event_log_data = []

class=class="hl-string">"hl-comment"># Transformation Loop
for index, row in df.iterrows():
    episode_id = row.get(class="hl-string">'ENCOUNTER_ID')
    
    class=class="hl-string">"hl-comment"># Skip rows without a valid identifier
    if pd.isna(episode_id) or str(episode_id).strip() == class="hl-string">'':
        continue

    class=class="hl-string">"hl-comment"># Iterate through each defined activity
    for activity_name, col_name in activity_mapping.items():
        timestamp_val = row.get(col_name)
        
        class=class="hl-string">"hl-comment"># Only create an event if a valid timestamp exists
        if pd.notna(timestamp_val) and str(timestamp_val).strip() != class="hl-string">'':
            
            class=class="hl-string">"hl-comment"># Create base event object
            event = {
                class="hl-string">'PatientEpisodeId': episode_id,
                class="hl-string">'ActivityName': activity_name,
                class="hl-string">'EventTimestamp': timestamp_val,
                class="hl-string">'SourceSystem': class="hl-string">'athenahealth',
                class="hl-string">'LastDataUpdate': datetime.now().strftime(class="hl-string">'%Y-%m-%d %H:%M:%S')
            }
            
            class=class="hl-string">"hl-comment"># Add attributes
            for csv_col, target_col in attribute_columns.items():
                event[target_col] = row.get(csv_col, None)
            
            event_log_data.append(event)

class=class="hl-string">"hl-comment"># Create DataFrame from the list of events
event_log_df = pd.DataFrame(event_log_data)

class=class="hl-string">"hl-comment"># Standardize Date Format (ISO 8601 Recommended)
class=class="hl-string">"hl-comment"># Adjust the format string matches the raw CSV format (e.g., class="hl-string">'%m/%d/%Y %H:%M')
event_log_df[class="hl-string">'EventTimestamp'] = pd.to_datetime(event_log_df[class="hl-string">'EventTimestamp'], errors=class="hl-string">'coerce')

class=class="hl-string">"hl-comment"># Remove rows where timestamp parsing failed
event_log_df = event_log_df.dropna(subset=[class="hl-string">'EventTimestamp'])

class=class="hl-string">"hl-comment"># Sort by Episode and Time
event_log_df = event_log_df.sort_values(by=[class="hl-string">'PatientEpisodeId', class="hl-string">'EventTimestamp'])

class=class="hl-string">"hl-comment"># Output to CSV
event_log_df.to_csv(OUTPUT_FILE, index=False)

print(fclass="hl-string">"Transformation complete. Processed {len(df)} encounters into {len(event_log_df)} events.")
print(fclass="hl-string">"File saved to: {OUTPUT_FILE}")

import requests
import pandas as pd
import time
from datetime import datetime, timedelta
import base64

class=class="hl-string">"hl-comment"># Configuration
API_KEY = class="hl-string">'YOUR_CLIENT_ID'
API_SECRET = class="hl-string">'YOUR_CLIENT_SECRET'
PRACTICE_ID = class="hl-string">'195900'  class=class="hl-string">"hl-comment"># Replace with your Practice ID
BASE_URL = class="hl-string">'https://api.athenahealth.com/preview1'  class=class="hl-string">"hl-comment"># Use production URL for live data
START_DATE = class="hl-string">'01/01/2023'
END_DATE = class="hl-string">'01/31/2023'
OUTPUT_FILE = class="hl-string">'athena_patient_journey.csv'

def get_access_token():
    url = fclass="hl-string">'{BASE_URL}/oauth2/v1/token'
    auth_str = fclass="hl-string">'{API_KEY}:{API_SECRET}'
    b64_auth = base64.b64encode(auth_str.encode()).decode()
    headers = {
        class="hl-string">'Authorization': fclass="hl-string">'Basic {b64_auth}',
        class="hl-string">'Content-Type': class="hl-string">'application/x-www-form-urlencoded'
    }
    data = {class="hl-string">'grant_type': class="hl-string">'client_credentials', class="hl-string">'scope': class="hl-string">'athena/service/Athenanet.MDP.*'}
    response = requests.post(url, headers=headers, data=data)
    if response.status_code == 200:
        return response.json()[class="hl-string">'access_token']
    else:
        raise Exception(fclass="hl-string">'Auth failed: {response.text}')

def fetch_paginated(url, headers, params):
    results = []
    while True:
        response = requests.get(url, headers=headers, params=params)
        if response.status_code == 429:
            time.sleep(2)
            continue
        if response.status_code != 200:
            print(fclass="hl-string">'Error fetching {url}: {response.status_code}')
            break
        data = response.json()
        if class="hl-string">'totalcount' in data and data[class="hl-string">'totalcount'] == 0:
            break
        class=class="hl-string">"hl-comment"># Handle different response structures (key might be appointments, orders, etc.)
        keys = [k for k in data.keys() if k not in [class="hl-string">'totalcount', class="hl-string">'next', class="hl-string">'previous']]
        if keys:
            results.extend(data[keys[0]])
        
        if class="hl-string">'next' in data and data[class="hl-string">'next']:
             class=class="hl-string">"hl-comment"># Athena pagination often uses offset or a link
             class=class="hl-string">"hl-comment"># Adjust based on specific endpoint logic if next is not a full URL
             params[class="hl-string">'offset'] = params.get(class="hl-string">'offset', 0) + len(data[keys[0]])
        else:
            break
    return results

def extract_events():
    token = get_access_token()
    headers = {class="hl-string">'Authorization': fclass="hl-string">'Bearer {token}'}
    events = []

    class=class="hl-string">"hl-comment"># 1. Fetch Appointments (Patient Episodes)
    class=class="hl-string">"hl-comment"># Using /appointments/booked for the range
    print(class="hl-string">'Fetching appointments...')
    appt_params = {
        class="hl-string">'startdate': START_DATE,
        class="hl-string">'enddate': END_DATE,
        class="hl-string">'showclaimdetail': class="hl-string">'true',  class=class="hl-string">"hl-comment"># optimize to get claim info early
        class="hl-string">'ignorepartners': class="hl-string">'true'
    }
    appointments = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/appointments/booked', headers, appt_params)

    for appt in appointments:
        episode_id = str(appt.get(class="hl-string">'appointmentid'))
        patient_id = appt.get(class="hl-string">'patientid')
        dept_id = appt.get(class="hl-string">'departmentid')
        provider_id = appt.get(class="hl-string">'providerid')
        
        class=class="hl-string">"hl-comment"># Base Activity: Appointment Scheduled
        if appt.get(class="hl-string">'date'): class=class="hl-string">"hl-comment"># capturing creation/scheduled date requires parsing history or booking info
            class=class="hl-string">"hl-comment"># Fallback to date if created date unavailable, but typically we want class="hl-string">'scheduleddatetime'
            sched_time = appt.get(class="hl-string">'scheduleddatetime', appt.get(class="hl-string">'date') + class="hl-string">' ' + appt.get(class="hl-string">'starttime'))
            events.append({
                class="hl-string">'PatientEpisodeId': episode_id,
                class="hl-string">'ActivityName': class="hl-string">'Appointment Scheduled',
                class="hl-string">'EventTimestamp': sched_time,
                class="hl-string">'SourceSystem': class="hl-string">'athenaOne',
                class="hl-string">'PatientId': patient_id,
                class="hl-string">'DepartmentName': dept_id
            })

        class=class="hl-string">"hl-comment"># Activity: Patient Checked In
        if appt.get(class="hl-string">'startcheckin'):
            events.append({
                class="hl-string">'PatientEpisodeId': episode_id,
                class="hl-string">'ActivityName': class="hl-string">'Patient Checked In',
                class="hl-string">'EventTimestamp': appt.get(class="hl-string">'date') + class="hl-string">' ' + appt.get(class="hl-string">'startcheckin'), class=class="hl-string">"hl-comment"># combining date and time
                class="hl-string">'SourceSystem': class="hl-string">'athenaOne',
                class="hl-string">'PatientId': patient_id
            })
            
        class=class="hl-string">"hl-comment"># Activity: Patient Discharged (Ambulatory Checkout)
        if appt.get(class="hl-string">'stopcheckin'):
            events.append({
                class="hl-string">'PatientEpisodeId': episode_id,
                class="hl-string">'ActivityName': class="hl-string">'Patient Discharged',
                class="hl-string">'EventTimestamp': appt.get(class="hl-string">'date') + class="hl-string">' ' + appt.get(class="hl-string">'stopcheckin'),
                class="hl-string">'SourceSystem': class="hl-string">'athenaOne',
                class="hl-string">'PatientId': patient_id,
                class="hl-string">'DischargeDisposition': class="hl-string">'Home' class=class="hl-string">"hl-comment"># Default for ambulatory
            })

        class=class="hl-string">"hl-comment"># Activity: Claim Submitted
        if class="hl-string">'claimid' in appt:
             class=class="hl-string">"hl-comment"># Ideally query /claims/{claimId} for exact submission timestamp
             class=class="hl-string">"hl-comment"># Using appointment date as proxy if detailed claim fetch skipped
             class=class="hl-string">"hl-comment"># Here we assume we fetch claim details:
             claim_res = requests.get(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/claims/{appt["claimid"]}', headers=headers)
             if claim_res.status_code == 200:
                 claim_data = claim_res.json()
                 if claim_data.get(class="hl-string">'billedservicedate'):
                     events.append({
                        class="hl-string">'PatientEpisodeId': episode_id,
                        class="hl-string">'ActivityName': class="hl-string">'Claim Submitted',
                        class="hl-string">'EventTimestamp': claim_data.get(class="hl-string">'createddate', claim_data.get(class="hl-string">'billedservicedate')) + class="hl-string">' 12:00:00',
                        class="hl-string">'SourceSystem': class="hl-string">'athenaCollector',
                        class="hl-string">'PatientId': patient_id
                     })

        class=class="hl-string">"hl-comment"># Clinical Data Extraction (Requires Encounter Context)
        encounter_id = appt.get(class="hl-string">'encounterid')
        if encounter_id:
            class=class="hl-string">"hl-comment"># Activity: Initial Assessment (Vitals)
            vitals = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/vitals', headers, {})
            if vitals:
                class=class="hl-string">"hl-comment"># Take the first vital sign entry time
                first_vital = sorted(vitals, key=lambda x: x.get(class="hl-string">'date') or class="hl-string">'')[0]
                events.append({
                    class="hl-string">'PatientEpisodeId': episode_id,
                    class="hl-string">'ActivityName': class="hl-string">'Initial Assessment Completed',
                    class="hl-string">'EventTimestamp': first_vital.get(class="hl-string">'date') + class="hl-string">' ' + (first_vital.get(class="hl-string">'time') or class="hl-string">'00:00'),
                    class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals',
                    class="hl-string">'PatientId': patient_id
                })

            class=class="hl-string">"hl-comment"># Activity: Diagnoses
            diagnoses = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/diagnoses', headers, {})
            for diag in diagnoses:
                events.append({
                    class="hl-string">'PatientEpisodeId': episode_id,
                    class="hl-string">'ActivityName': class="hl-string">'Diagnosis Confirmed',
                    class="hl-string">'EventTimestamp': diag.get(class="hl-string">'lastmodifieddatetime', appt.get(class="hl-string">'date') + class="hl-string">' 12:00:00'),
                    class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals',
                    class="hl-string">'PrimaryDiagnosisCode': diag.get(class="hl-string">'diagnosiscode')
                })

            class=class="hl-string">"hl-comment"># Activity: Orders (Labs, Procedures, Discharge)
            orders = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/orders', headers, {})
            for order in orders:
                order_type = order.get(class="hl-string">'documentationtype', class="hl-string">'')
                created_time = order.get(class="hl-string">'createddatetime', appt.get(class="hl-string">'date') + class="hl-string">' 12:00:00')
                
                if class="hl-string">'LAB' in order_type.upper() or class="hl-string">'IMAGING' in order_type.upper():
                    events.append({
                        class="hl-string">'PatientEpisodeId': episode_id,
                        class="hl-string">'ActivityName': class="hl-string">'Diagnostic Test Ordered',
                        class="hl-string">'EventTimestamp': created_time,
                        class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                    })
                    class=class="hl-string">"hl-comment"># Activity: Specimen Collected (Inferred from status or specific fields if available)
                    if order.get(class="hl-string">'status') in [class="hl-string">'COMPLETED', class="hl-string">'REVIEW']:
                         events.append({
                            class="hl-string">'PatientEpisodeId': episode_id,
                            class="hl-string">'ActivityName': class="hl-string">'Specimen Collected',
                            class="hl-string">'EventTimestamp': order.get(class="hl-string">'lastmodifieddatetime', created_time), class=class="hl-string">"hl-comment"># Approximation
                            class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                        })
                
                if class="hl-string">'SURGERY' in order_type.upper() or class="hl-string">'PROCEDURE' in order_type.upper():
                     events.append({
                        class="hl-string">'PatientEpisodeId': episode_id,
                        class="hl-string">'ActivityName': class="hl-string">'Procedure Performed',
                        class="hl-string">'EventTimestamp': order.get(class="hl-string">'performdate', created_time),
                        class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                    })

                if class="hl-string">'DISCHARGE' in order_type.upper():
                    events.append({
                        class="hl-string">'PatientEpisodeId': episode_id,
                        class="hl-string">'ActivityName': class="hl-string">'Discharge Order Written',
                        class="hl-string">'EventTimestamp': created_time,
                        class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                    })

            class=class="hl-string">"hl-comment"># Activity: Test Results Received
            results = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/analyteresults', headers, {})
            for res in results:
                events.append({
                    class="hl-string">'PatientEpisodeId': episode_id,
                    class="hl-string">'ActivityName': class="hl-string">'Test Results Received',
                    class="hl-string">'EventTimestamp': res.get(class="hl-string">'analytedatetime', appt.get(class="hl-string">'date') + class="hl-string">' 12:00:00'),
                    class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                })

            class=class="hl-string">"hl-comment"># Activity: Treatment Plan
            class=class="hl-string">"hl-comment"># Check for Care Plan document or assessment note
            notes = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/progressnotes', headers, {})
            if notes:
                 events.append({
                    class="hl-string">'PatientEpisodeId': episode_id,
                    class="hl-string">'ActivityName': class="hl-string">'Treatment Plan Developed',
                    class="hl-string">'EventTimestamp': notes[0].get(class="hl-string">'lastmodifieddatetime', appt.get(class="hl-string">'date') + class="hl-string">' 12:00:00'),
                    class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                })

            class=class="hl-string">"hl-comment"># Activity: Medication Administered
            class=class="hl-string">"hl-comment"># Look for administration records
            meds = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/medications', headers, {})
            for med in meds:
                 if med.get(class="hl-string">'events'): class=class="hl-string">"hl-comment"># Check if administration events exist
                     for event in med[class="hl-string">'events']:
                         if event.get(class="hl-string">'type') == class="hl-string">'ADMINISTERED':
                             events.append({
                                class="hl-string">'PatientEpisodeId': episode_id,
                                class="hl-string">'ActivityName': class="hl-string">'Medication Administered',
                                class="hl-string">'EventTimestamp': event.get(class="hl-string">'eventdate'),
                                class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                             })

            class=class="hl-string">"hl-comment"># Activity: Patient Transferred (Inferred from Hospital Stay/ADT data if available)
            class=class="hl-string">"hl-comment"># Mocking logic: If department ID changes in encounter history (requires fetching history)
            class=class="hl-string">"hl-comment"># For standard API, we check if class="hl-string">'admissionstatus' indicates transfer
            pass class=class="hl-string">"hl-comment"># Complexity high for standard extraction without Hospital scope

            class=class="hl-string">"hl-comment"># Activity: Follow-Up Appointment Scheduled
            class=class="hl-string">"hl-comment"># We check if this patient has a FUTURE appointment created around this time
            class=class="hl-string">"hl-comment"># This requires a separate query or processing the full appointment list post-loop
            class=class="hl-string">"hl-comment"># Simplified: Logic placeholder
            pass 

    class=class="hl-string">"hl-comment"># Post-processing for Follow-Ups (simplified)
    class=class="hl-string">"hl-comment"># Sort all appts by patient and date, check next appt creation date
    
    class=class="hl-string">"hl-comment"># Convert to DataFrame
    df = pd.DataFrame(events)
    df[class="hl-string">'LastDataUpdate'] = datetime.now().isoformat()
    
    class=class="hl-string">"hl-comment"># Final Validation and Export
    if not df.empty:
        df[class="hl-string">'EventTimestamp'] = pd.to_datetime(df[class="hl-string">'EventTimestamp'])
        df.sort_values(by=[class="hl-string">'PatientEpisodeId', class="hl-string">'EventTimestamp'], inplace=True)
        df.to_csv(OUTPUT_FILE, index=False)
        print(fclass="hl-string">'Successfully exported {len(df)} rows to {OUTPUT_FILE}')
    else:
        print(class="hl-string">'No events extracted.')

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