healthcare-expert

SKILL.md

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/