Healthcare Expert

Expert guidance for healthcare systems, medical informatics, regulatory compliance (HIPAA), and health data standards (HL7, FHIR).

Core Concepts

Healthcare IT

• Electronic Health Records (EHR)

• Health Information Exchange (HIE)

• Clinical Decision Support Systems

• Telemedicine platforms

• Medical imaging systems (PACS)

• Laboratory information systems

Standards and Protocols

• HL7 (Health Level 7)

• FHIR (Fast Healthcare Interoperability Resources)

• DICOM (Digital Imaging and Communications in Medicine)

• ICD-10 (diagnostic codes)

• CPT (procedure codes)

• SNOMED CT (clinical terminology)

Regulatory Compliance

• HIPAA (Health Insurance Portability and Accountability Act)

• HITECH Act

• GDPR for health data

• FDA regulations for medical devices

• 21 CFR Part 11 for electronic records

FHIR Resource Handling

from fhirclient import client from fhirclient.models import patient, observation, medication from datetime import datetime

FHIR Client setup

settings = { 'app_id': 'my_healthcare_app', 'api_base': 'https://fhir.example.com/r4' } smart = client.FHIRClient(settings=settings)

Patient resource

def create_patient(first_name, last_name, gender, birth_date): """Create FHIR Patient resource""" p = patient.Patient() p.name = [{ 'use': 'official', 'family': last_name, 'given': [first_name] }] p.gender = gender # 'male', 'female', 'other', 'unknown' p.birthDate = birth_date.isoformat()

return p.create(smart.server)

Observation resource (vital signs)

def create_vital_signs_observation(patient_id, code, value, unit): """Create vital signs observation""" obs = observation.Observation() obs.status = 'final' obs.category = [{ 'coding': [{ 'system': 'http://terminology.hl7.org/CodeSystem/observation-category', 'code': 'vital-signs', 'display': 'Vital Signs' }] }]

obs.code = {
    'coding': [{
        'system': 'http://loinc.org',
        'code': code,  # e.g., '8867-4' for heart rate
        'display': 'Heart rate'
    }]
}

obs.subject = {'reference': f'Patient/{patient_id}'}
obs.effectiveDateTime = datetime.now().isoformat()

obs.valueQuantity = {
    'value': value,
    'unit': unit,
    'system': 'http://unitsofmeasure.org',
    'code': unit
}

return obs.create(smart.server)

Search patients

def search_patients(family_name=None, given_name=None): """Search for patients by name""" search = patient.Patient.where(struct={})

if family_name:
    search = search.where(struct={'family': family_name})
if given_name:
    search = search.where(struct={'given': given_name})

return search.perform(smart.server)

Get patient observations

def get_patient_observations(patient_id, category=None): """Retrieve patient observations""" search = observation.Observation.where(struct={ 'patient': patient_id })

if category:
    search = search.where(struct={'category': category})

return search.perform(smart.server)

HL7 v2 Message Processing

import hl7

Parse HL7 message

def parse_hl7_message(message_text): """Parse HL7 v2 message""" h = hl7.parse(message_text)

# Extract message type
message_type = str(h.segment('MSH')[9])

# Extract patient information from PID segment
pid = h.segment('PID')
patient_info = {
    'patient_id': str(pid[3]),
    'name': str(pid[5]),
    'dob': str(pid[7]),
    'gender': str(pid[8])
}

return {
    'message_type': message_type,
    'patient': patient_info
}

Create ADT^A01 message (Patient Admission)

def create_admission_message(patient_id, patient_name, dob, gender): """Create HL7 ADT^A01 admission message""" message = hl7.Message( "MSH", [ "MSH", "|", "^~\&", "SENDING_APP", "SENDING_FACILITY", "RECEIVING_APP", "RECEIVING_FACILITY", datetime.now().strftime("%Y%m%d%H%M%S"), "", "ADT^A01", "MSG00001", "P", "2.5" ] )

# PID segment
message.append(hl7.Segment(
    "PID",
    [
        "PID", "", "", patient_id, "",
        patient_name, "", dob, gender
    ]
))

# PV1 segment (Patient Visit)
message.append(hl7.Segment(
    "PV1",
    [
        "PV1", "", "I", "ER", "", "", "",
        "", "", "", "", "", "", "",
        "", "", "", "", "", "", "", ""
    ]
))

return str(message)

Validate HL7 message

def validate_hl7_message(message_text): """Validate HL7 message structure""" try: h = hl7.parse(message_text)

    # Check required segments
    if not h.segment('MSH'):
        return False, "Missing MSH segment"

    # Verify message structure
    msh = h.segment('MSH')
    if len(msh) < 12:
        return False, "Invalid MSH segment"

    return True, "Valid HL7 message"
except Exception as e:
    return False, f"Parsing error: {str(e)}"

HIPAA Compliance Implementation

from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2 import hashlib import logging from datetime import datetime

class HIPAACompliantLogger: """HIPAA-compliant logging system"""

def __init__(self, log_file):
    self.logger = logging.getLogger('hipaa_audit')
    self.logger.setLevel(logging.INFO)

    handler = logging.FileHandler(log_file)
    formatter = logging.Formatter(
        '%(asctime)s - %(levelname)s - %(message)s'
    )
    handler.setFormatter(formatter)
    self.logger.addHandler(handler)

def log_access(self, user_id, patient_id, action, phi_accessed):
    """Log PHI access (HIPAA audit requirement)"""
    self.logger.info(
        f"USER:{user_id} | PATIENT:{patient_id} | "
        f"ACTION:{action} | PHI:{phi_accessed}"
    )

def log_modification(self, user_id, resource_type, resource_id, changes):
    """Log data modifications"""
    self.logger.info(
        f"USER:{user_id} | MODIFIED:{resource_type}/{resource_id} | "
        f"CHANGES:{changes}"
    )

def log_disclosure(self, user_id, patient_id, recipient, purpose):
    """Log PHI disclosure"""
    self.logger.info(
        f"DISCLOSURE | USER:{user_id} | PATIENT:{patient_id} | "
        f"TO:{recipient} | PURPOSE:{purpose}"
    )

class PHIEncryption: """Encryption for Protected Health Information"""

def __init__(self, master_key):
    self.fernet = Fernet(master_key)

def encrypt_phi(self, data):
    """Encrypt PHI data"""
    if isinstance(data, str):
        data = data.encode()
    return self.fernet.encrypt(data)

def decrypt_phi(self, encrypted_data):
    """Decrypt PHI data"""
    decrypted = self.fernet.decrypt(encrypted_data)
    return decrypted.decode()

@staticmethod
def hash_identifier(identifier):
    """Hash patient identifiers for de-identification"""
    return hashlib.sha256(identifier.encode()).hexdigest()

class HIPAAAccessControl: """Role-based access control for HIPAA compliance"""

ROLES = {
    'physician': ['read', 'write', 'prescribe'],
    'nurse': ['read', 'write'],
    'administrative': ['read'],
    'patient': ['read_own']
}

def __init__(self, user_role):
    self.role = user_role
    self.permissions = self.ROLES.get(user_role, [])

def can_access(self, action, patient_id, user_patient_id=None):
    """Check if user can perform action"""
    if action not in self.permissions:
        if action == 'read' and 'read_own' in self.permissions:
            return patient_id == user_patient_id
        return False

    return True

def require_permission(self, action):
    """Decorator for enforcing permissions"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            if action not in self.permissions:
                raise PermissionError(
                    f"Role '{self.role}' lacks permission: {action}"
                )
            return func(*args, **kwargs)
        return wrapper
    return decorator

Electronic Health Record System

from dataclasses import dataclass from typing import List, Optional from datetime import datetime

@dataclass class Patient: """Patient record""" patient_id: str mrn: str # Medical Record Number first_name: str last_name: str dob: datetime gender: str ssn: Optional[str] # Encrypted address: dict phone: str email: str emergency_contact: dict insurance: dict

@dataclass class Encounter: """Clinical encounter""" encounter_id: str patient_id: str encounter_date: datetime encounter_type: str # 'inpatient', 'outpatient', 'emergency' chief_complaint: str provider_id: str facility_id: str diagnosis_codes: List[str] # ICD-10 procedure_codes: List[str] # CPT notes: str

@dataclass class Medication: """Medication order""" medication_id: str patient_id: str drug_name: str dosage: str frequency: str route: str # 'oral', 'IV', etc. start_date: datetime end_date: Optional[datetime] prescriber_id: str pharmacy_notes: str

class EHRSystem: """Electronic Health Record system"""

def __init__(self, db, logger, access_control, encryption):
    self.db = db
    self.logger = logger
    self.access_control = access_control
    self.encryption = encryption

def get_patient_record(self, user_id, patient_id):
    """Retrieve patient record with audit logging"""
    # Check permissions
    if not self.access_control.can_access('read', patient_id):
        self.logger.log_access(
            user_id, patient_id, 'DENIED', 'patient_record'
        )
        raise PermissionError("Access denied")

    # Log access
    self.logger.log_access(
        user_id, patient_id, 'READ', 'patient_record'
    )

    # Retrieve and decrypt
    patient = self.db.get_patient(patient_id)
    if patient.ssn:
        patient.ssn = self.encryption.decrypt_phi(patient.ssn)

    return patient

def create_encounter(self, user_id, encounter: Encounter):
    """Create clinical encounter"""
    if not self.access_control.can_access('write', encounter.patient_id):
        raise PermissionError("Cannot create encounter")

    # Encrypt sensitive data
    if encounter.notes:
        encounter.notes = self.encryption.encrypt_phi(encounter.notes)

    # Save encounter
    self.db.save_encounter(encounter)

    # Log creation
    self.logger.log_modification(
        user_id, 'encounter', encounter.encounter_id, 'created'
    )

    return encounter

def get_patient_medications(self, user_id, patient_id):
    """Get active medications for patient"""
    if not self.access_control.can_access('read', patient_id):
        raise PermissionError("Access denied")

    self.logger.log_access(
        user_id, patient_id, 'READ', 'medications'
    )

    return self.db.get_active_medications(patient_id)

def prescribe_medication(self, user_id, medication: Medication):
    """Prescribe new medication"""
    if not self.access_control.can_access('prescribe', medication.patient_id):
        raise PermissionError("Cannot prescribe medication")

    # Drug interaction check
    active_meds = self.get_patient_medications(user_id, medication.patient_id)
    interactions = self.check_drug_interactions(medication, active_meds)

    if interactions:
        return {'status': 'warning', 'interactions': interactions}

    self.db.save_medication(medication)

    self.logger.log_modification(
        user_id, 'medication', medication.medication_id, 'prescribed'
    )

    return {'status': 'success', 'medication_id': medication.medication_id}

def check_drug_interactions(self, new_med, existing_meds):
    """Check for drug-drug interactions"""
    # This would integrate with a drug interaction database
    interactions = []
    # Implementation would check against drug interaction database
    return interactions

Best Practices

Security and Compliance

• Encrypt PHI at rest and in transit

• Implement comprehensive audit logging

• Use role-based access control

• Conduct regular security assessments

• Implement data backup and disaster recovery

• Train staff on HIPAA requirements

• Use de-identification for research data

Data Standards

• Use standard terminologies (SNOMED, LOINC)

• Implement FHIR for interoperability

• Support HL7 messaging

• Use ICD-10 for diagnoses

• Use CPT for procedures

• Validate data quality

System Design

• Design for high availability

• Implement redundancy

• Ensure data integrity

• Support audit trails

• Enable patient access portals

• Integrate with HIE networks

Anti-Patterns

❌ Storing PHI unencrypted ❌ No audit logging ❌ Inadequate access controls ❌ Using proprietary formats ❌ No data backup strategy ❌ Ignoring interoperability standards ❌ Weak authentication

Resources

• FHIR Specification: https://hl7.org/fhir/

• HL7 International: https://www.hl7.org/

• HIPAA Guidelines: https://www.hhs.gov/hipaa/

• ICD-10: https://www.who.int/classifications/icd/

• LOINC: https://loinc.org/

• SNOMED CT: https://www.snomed.org/