Automate Patient Intake 🚀

import boto3
import re
from typing import Dict, List

class PHIRedactor:
    def __init__(self):
        self.comprehend = boto3.client('comprehendmedical')
    
    def redact_phi(self, text: str) -> Dict[str, any]:
        """Redact PHI using AWS Comprehend Medical"""
        # Detect PHI entities
        result = self.comprehend.detect_phi(Text=text)
        
        # Sort entities by position (reverse order to maintain indices)
        entities = sorted(result['Entities'], key=lambda x: x['BeginOffset'], reverse=True)
        
        redacted_text = text
        phi_map = {}  # Store original values for re-identification later
        
        for entity in entities:
            original = entity['Text']
            entity_type = entity['Type']
            begin = entity['BeginOffset']
            end = entity['EndOffset']
            
            # Generate replacement token
            replacement = f"[{entity_type}_{len(phi_map)}]"
            phi_map[replacement] = original
            
            # Replace in text
            redacted_text = redacted_text[:begin] + replacement + redacted_text[end:]
        
        return {
            'redacted_text': redacted_text,
            'phi_map': phi_map,
            'entities_found': len(entities)
        }
    
    def restore_phi(self, text: str, phi_map: Dict[str, str]) -> str:
        """Restore original PHI after processing"""
        restored = text
        for token, original in phi_map.items():
            restored = restored.replace(token, original)
        return restored

# Usage in intake automation
redactor = PHIRedactor()

def safe_automate_intake(patient_text: str) -> Dict:
    # Step 1: Redact PHI
    redaction = redactor.redact_phi(patient_text)
    safe_text = redaction['redacted_text']
    
    # Step 2: Process with LLM (now HIPAA compliant)
    result = automate_patient_intake(safe_text)
    
    # Step 3: Restore PHI in final output
    result['extracted']['patient_name'] = redactor.restore_phi(
        result['extracted']['patient_name'],
        redaction['phi_map']
    )
    
    return result

import logging
import json
from datetime import datetime
import boto3
from typing import Dict

class HIPAALogger:
    def __init__(self, log_group: str, log_stream: str):
        self.cloudwatch = boto3.client('logs')
        self.log_group = log_group
        self.log_stream = log_stream
        
        # Create log stream if it doesn't exist
        try:
            self.cloudwatch.create_log_stream(
                logGroupName=log_group,
                logStreamName=log_stream
            )
        except self.cloudwatch.exceptions.ResourceAlreadyExistsException:
            pass
    
    def log_phi_access(self, 
                       user_id: str,
                       patient_id: str,
                       action: str,
                       reason: str,
                       ip_address: str,
                       additional_data: Dict = None) -> None:
        """Log PHI access event"""
        event = {
            'timestamp': datetime.now().isoformat(),
            'user_id': user_id,
            'patient_id': patient_id,  # Use patient_id, NOT patient_name
            'action': action,  # 'extract', 'validate', 'generate_questions'
            'reason': reason,  # 'automated_intake', 'manual_review'
            'ip_address': ip_address,
            'system': 'intake_automation',
            'version': '1.0',
            'additional_data': additional_data or {}
        }
        
        # Send to CloudWatch
        self.cloudwatch.put_log_events(
            logGroupName=self.log_group,
            logStreamName=self.log_stream,
            logEvents=[{
                'timestamp': int(datetime.now().timestamp() * 1000),
                'message': json.dumps(event)
            }]
        )

# Usage in intake automation
hipaa_logger = HIPAALogger(
    log_group='/healthcare/phi-access',
    log_stream='intake-automation'
)

def logged_automate_intake(patient_text: str, 
                           user_id: str,
                           patient_id: str,
                           ip_address: str) -> Dict:
    # Log extraction
    hipaa_logger.log_phi_access(
        user_id=user_id,
        patient_id=patient_id,
        action='extract',
        reason='automated_intake',
        ip_address=ip_address
    )
    
    result = automate_patient_intake(patient_text)
    
    # Log validation
    hipaa_logger.log_phi_access(
        user_id=user_id,
        patient_id=patient_id,
        action='validate',
        reason='automated_intake',
        ip_address=ip_address,
        additional_data={'is_complete': result['is_complete']}
    )
    
    # Log question generation if needed
    if not result['is_complete']:
        hipaa_logger.log_phi_access(
            user_id=user_id,
            patient_id=patient_id,
            action='generate_questions',
            reason='automated_intake',
            ip_address=ip_address,
            additional_data={'question_count': len(result['questions'])}
        )
    
    return result

import re
from typing import Dict

class MedicalAbbreviationExpander:
    # Common medical abbreviations
    ABBREVS = {
        'SOB': 'shortness of breath',
        'CP': 'chest pain',
        'N/V': 'nausea and vomiting',
        'HA': 'headache',
        'HTN': 'hypertension',
        'DM': 'diabetes mellitus',
        'COPD': 'chronic obstructive pulmonary disease',
        'CHF': 'congestive heart failure',
        'MI': 'myocardial infarction',
        'CVA': 'cerebrovascular accident',
        'Pt': 'patient',
        'Hx': 'history',
        'Sx': 'symptoms',
        'Tx': 'treatment',
        'Dx': 'diagnosis',
        'Rx': 'prescription',
        'c/o': 'complains of',
        'w/': 'with',
        'w/o': 'without',
        'prn': 'as needed',
        'bid': 'twice daily',
        'tid': 'three times daily',
        'qid': 'four times daily',
        'qd': 'once daily',
        'qhs': 'at bedtime',
        'NPO': 'nothing by mouth',
        'STAT': 'immediately'
    }
    
    def expand(self, text: str) -> Dict[str, any]:
        """Expand medical abbreviations in text"""
        expanded_text = text
        expansions_made = []
        
        for abbrev, full in self.ABBREVS.items():
            # Use word boundaries to avoid partial matches
            pattern = r'\b' + re.escape(abbrev) + r'\b'
            
            if re.search(pattern, expanded_text, re.IGNORECASE):
                expansions_made.append({
                    'abbreviation': abbrev,
                    'expanded_to': full
                })
                expanded_text = re.sub(
                    pattern, 
                    full, 
                    expanded_text, 
                    flags=re.IGNORECASE
                )
        
        return {
            'original_text': text,
            'expanded_text': expanded_text,
            'expansions': expansions_made
        }

# Usage in intake automation
expander = MedicalAbbreviationExpander()

def preprocess_patient_text(patient_text: str) -> str:
    """Preprocess text before sending to LLM"""
    # Expand abbreviations
    expansion = expander.expand(patient_text)
    
    # Log what was expanded for audit trail
    if expansion['expansions']:
        print(f"Expanded {len(expansion['expansions'])} abbreviations:")
        for exp in expansion['expansions']:
            print(f"  {exp['abbreviation']} → {exp['expanded_to']}")
    
    return expansion['expanded_text']

# Example
patient_text = "Pt is 54yo F c/o severe HA x 2 weeks. CP 7/10. Hx of HTN, on lisinopril 10mg qd."
expanded = preprocess_patient_text(patient_text)
print(f"\nExpanded: {expanded}")
# Output: "patient is 54yo F complains of severe headache x 2 weeks. chest pain 7/10. history of hypertension, on lisinopril 10mg once daily."

from fhirclient import client
from fhirclient.models import patient, humanname, condition, codeableconcept, reference
from fhirclient.models import medicationstatement, allergyintolerance
from typing import Dict
import os

class EHRIntegration:
    def __init__(self, ehr_type: str = 'epic'):
        if ehr_type == 'epic':
            settings = {
                'app_id': os.getenv('EPIC_APP_ID'),
                'api_base': os.getenv('EPIC_FHIR_BASE'),  # https://fhir.epic.com/interconnect-fhir-oauth/api/FHIR/R4/
                'redirect_uri': os.getenv('EPIC_REDIRECT_URI')
            }
        elif ehr_type == 'cerner':
            settings = {
                'app_id': os.getenv('CERNER_APP_ID'),
                'api_base': os.getenv('CERNER_FHIR_BASE'),
                'redirect_uri': os.getenv('CERNER_REDIRECT_URI')
            }
        else:
            raise ValueError(f"Unsupported EHR type: {ehr_type}")
        
        self.smart = client.FHIRClient(settings=settings)
    
    def write_intake_to_ehr(self, extracted_data: Dict, patient_id: str = None) -> Dict:
        """Write extracted intake data to EHR as FHIR resources"""
        results = {}
        
        # 1. Create or update Patient resource
        if not patient_id:
            pat = patient.Patient()
            name = humanname.HumanName()
            name.given = [extracted_data.get('patient_name', '').split()[0]]
            name.family = ' '.join(extracted_data.get('patient_name', '').split()[1:])
            pat.name = [name]
            pat.birthDate = self._calculate_birth_date(extracted_data.get('age'))
            pat.gender = extracted_data.get('gender', 'unknown')
            
            pat.create(self.smart.server)
            patient_id = pat.id
            results['patient_id'] = patient_id
        
        # 2. Create Condition resource for chief complaint
        cond = condition.Condition()
        cond.code = codeableconcept.CodeableConcept()
        cond.code.text = extracted_data.get('chief_complaint')
        cond.subject = reference.Reference()
        cond.subject.reference = f'Patient/{patient_id}'
        cond.severity = self._map_severity(extracted_data.get('severity', 5))
        
        cond.create(self.smart.server)
        results['condition_id'] = cond.id
        
        # 3. Create MedicationStatement resources
        medication_ids = []
        for med in extracted_data.get('medications', []):
            med_stmt = medicationstatement.MedicationStatement()
            med_stmt.medicationCodeableConcept = codeableconcept.CodeableConcept()
            med_stmt.medicationCodeableConcept.text = f"{med.get('name')} {med.get('dose')}"
            med_stmt.subject = reference.Reference()
            med_stmt.subject.reference = f'Patient/{patient_id}'
            med_stmt.status = 'active'
            
            med_stmt.create(self.smart.server)
            medication_ids.append(med_stmt.id)
        
        results['medication_ids'] = medication_ids
        
        # 4. Create AllergyIntolerance resources
        allergy_ids = []
        for allergy in extracted_data.get('allergies', []):
            allergy_int = allergyintolerance.AllergyIntolerance()
            allergy_int.code = codeableconcept.CodeableConcept()
            allergy_int.code.text = allergy
            allergy_int.patient = reference.Reference()
            allergy_int.patient.reference = f'Patient/{patient_id}'
            allergy_int.clinicalStatus = codeableconcept.CodeableConcept()
            allergy_int.clinicalStatus.text = 'active'
            
            allergy_int.create(self.smart.server)
            allergy_ids.append(allergy_int.id)
        
        results['allergy_ids'] = allergy_ids
        
        return results
    
    def _calculate_birth_date(self, age: int) -> str:
        """Calculate approximate birth date from age"""
        from datetime import datetime, timedelta
        birth_year = datetime.now().year - age
        return f"{birth_year}-01-01"
    
    def _map_severity(self, severity: int) -> codeableconcept.CodeableConcept:
        """Map 1-10 severity to FHIR severity code"""
        severity_code = codeableconcept.CodeableConcept()
        if severity <= 3:
            severity_code.text = 'mild'
        elif severity <= 6:
            severity_code.text = 'moderate'
        else:
            severity_code.text = 'severe'
        return severity_code

# Usage after extraction
ehr = EHRIntegration(ehr_type='epic')

def automate_with_ehr_write(patient_text: str) -> Dict:
    # Extract data
    result = automate_patient_intake(patient_text)
    
    # Write to EHR if extraction was successful
    if result['extracted']:
        try:
            ehr_result = ehr.write_intake_to_ehr(
                extracted_data=result['extracted']
            )
            result['ehr_write'] = {
                'success': True,
                'patient_id': ehr_result['patient_id'],
                'resources_created': len(ehr_result)
            }
        except Exception as e:
            result['ehr_write'] = {
                'success': False,
                'error': str(e)
            }
    
    return result

import { createClient } from '@supabase/supabase-js';

interface ConsentRecord {
  patient_id: string;
  consent_type: 'ai_processing' | 'data_sharing' | 'research';
  granted: boolean;
  granted_date: string;
  expiration_date?: string;
  revoked_date?: string;
}

class ConsentManager {
  private db: any;

  constructor() {
    this.db = createClient(
      process.env.SUPABASE_URL!,
      process.env.SUPABASE_KEY!
    );
  }

  async checkAIConsent(patientId: string): Promise<{
    hasConsent: boolean;
    reason?: string;
    consentDate?: string;
  }> {
    try {
      const { data, error } = await this.db
        .from('patient_consent')
        .select('*')
        .eq('patient_id', patientId)
        .eq('consent_type', 'ai_processing')
        .single();

      if (error) {
        return {
          hasConsent: false,
          reason: 'no_consent_record',
        };
      }

      const consent = data as ConsentRecord;

      // Check if consent was revoked
      if (consent.revoked_date) {
        return {
          hasConsent: false,
          reason: 'consent_revoked',
        };
      }

      // Check if consent expired
      if (consent.expiration_date) {
        const expirationDate = new Date(consent.expiration_date);
        if (expirationDate < new Date()) {
          return {
            hasConsent: false,
            reason: 'consent_expired',
          };
        }
      }

      // Check if consent was granted
      if (!consent.granted) {
        return {
          hasConsent: false,
          reason: 'consent_denied',
        };
      }

      return {
        hasConsent: true,
        consentDate: consent.granted_date,
      };
    } catch (error) {
      console.error('Consent check failed:', error);
      // Fail closed - if we can't verify consent, don't process
      return {
        hasConsent: false,
        reason: 'consent_check_error',
      };
    }
  }

  async logConsentCheck(
    patientId: string,
    hasConsent: boolean,
    reason: string
  ): Promise<void> {
    await this.db.from('consent_audit_log').insert({
      patient_id: patientId,
      check_timestamp: new Date().toISOString(),
      has_consent: hasConsent,
      reason: reason,
      checked_by: 'intake_automation',
    });
  }
}

const consentManager = new ConsentManager();

async function automateWithConsentCheck(
  patientText: string,
  patientId: string
): Promise<any> {
  // Check consent before processing
  const consentCheck = await consentManager.checkAIConsent(patientId);

  // Log the consent check
  await consentManager.logConsentCheck(
    patientId,
    consentCheck.hasConsent,
    consentCheck.reason || 'consent_granted'
  );

  if (!consentCheck.hasConsent) {
    // Route to manual intake workflow
    return {
      status: 'manual_review_required',
      reason: consentCheck.reason,
      message: 'Patient has not consented to AI processing',
      manual_workflow_url: `/manual-intake/${patientId}`,
    };
  }

  // Proceed with automated intake
  const result = await automateIntake(patientText);

  return {
    status: 'automated_processing_complete',
    consent_verified: true,
    consent_date: consentCheck.consentDate,
    ...result,
  };
}

// Usage
const result = await automateWithConsentCheck(patientText, 'patient-12345');

if (result.status === 'manual_review_required') {
  console.log(`Routing to manual workflow: ${result.reason}`);
} else {
  console.log(`Automated processing complete: ${result.is_complete}`);
}