Digital Health Engineering: How to Build AI Triage and Health-Data Tools for African Clinics

# app/services/triage_service.py
from typing import List, Dict
import httpx
 
class TriageService:
    def __init__(self, ai_client):
        self.ai_client = ai_client
        self.severity_levels = ["low", "medium", "high", "critical"]
    
    async def assess_symptoms(
        self,
        symptoms: List[str],
        patient_age: int,
        patient_sex: str,
        vital_signs: Dict[str, float]
    ) -> Dict:
        """
        Assess patient symptoms and determine urgency
        """
        # Build context-aware prompt
        prompt = self._build_triage_prompt(
            symptoms, patient_age, patient_sex, vital_signs
        )
        
        # Get AI assessment
        assessment = await self.ai_client.chat_completion([
            {"role": "system", "content": self._get_system_prompt()},
            {"role": "user", "content": prompt}
        ])
        
        # Parse and validate response
        result = self._parse_assessment(assessment)
        
        # Apply safety rules
        result = self._apply_safety_rules(result, vital_signs)
        
        return result
    
    def _build_triage_prompt(
        self,
        symptoms: List[str],
        age: int,
        sex: str,
        vitals: Dict
    ) -> str:
        return f"""
Patient Information:
- Age: {age} years
- Sex: {sex}
- Symptoms: {', '.join(symptoms)}
- Vital Signs:
  - Temperature: {vitals.get('temperature', 'N/A')}°C
  - Blood Pressure: {vitals.get('bp_systolic', 'N/A')}/{vitals.get('bp_diastolic', 'N/A')} mmHg
  - Heart Rate: {vitals.get('heart_rate', 'N/A')} bpm
  - Respiratory Rate: {vitals.get('resp_rate', 'N/A')} breaths/min
 
Assess the urgency level and provide:
1. Severity (low/medium/high/critical)
2. Recommended action
3. Red flags to watch for
4. Estimated wait time
"""
    
    def _get_system_prompt(self) -> str:
        return """You are a medical triage AI assistant for a rural African clinic.
Your role is to assess symptom urgency and guide healthcare workers.
 
Guidelines:
- Be conservative: when in doubt, escalate
- Consider limited resources
- Account for tropical diseases (malaria, typhoid, etc.)
- Provide clear, actionable recommendations
- Use simple language (healthcare workers may have basic training)
 
CRITICAL: Always flag these as HIGH or CRITICAL:
- Chest pain
- Difficulty breathing
- Severe bleeding
- Altered consciousness
- Severe abdominal pain in pregnancy
- High fever in children under 5
"""
    
    def _apply_safety_rules(
        self,
        assessment: Dict,
        vitals: Dict
    ) -> Dict:
        """Override AI if vital signs indicate emergency"""
        
        # Critical vital signs = automatic escalation
        if vitals.get('temperature', 0) > 39.5:  # High fever
            assessment['severity'] = 'high'
        
        if vitals.get('bp_systolic', 120) > 180:  # Hypertensive crisis
            assessment['severity'] = 'critical'
        
        if vitals.get('heart_rate', 80) > 120:  # Tachycardia
            if assessment['severity'] == 'low':
                assessment['severity'] = 'medium'
        
        return assessment

// services/TriageService.ts
import NetInfo from '@react-native-community/netinfo';
import { db } from './database';
 
class TriageService {
  async submitTriage(triageData: TriageData): Promise<void> {
    // Save locally first
    await db.triage.add({
      ...triageData,
      synced: false,
      timestamp: Date.now()
    });
    
    // Try to sync if online
    const netInfo = await NetInfo.fetch();
    if (netInfo.isConnected) {
      await this.syncToServer(triageData);
    }
  }
  
  async syncToServer(triageData: TriageData): Promise<void> {
    try {
      const response = await fetch('https://api.clinic.com/triage', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(triageData),
        timeout: 10000 // 10 second timeout
      });
      
      if (response.ok) {
        // Mark as synced
        await db.triage.update(triageData.id, { synced: true });
      }
    } catch (error) {
      // Sync failed, will retry later
      console.log('Sync failed, will retry when online');
    }
  }
  
  // Background sync when connection restored
  async syncPendingRecords(): Promise<void> {
    const pending = await db.triage
      .where('synced')
      .equals(false)
      .toArray();
    
    for (const record of pending) {
      await this.syncToServer(record);
    }
  }
}

// database/schema.ts
import Dexie from 'dexie';
 
class ClinicDatabase extends Dexie {
  triage: Dexie.Table<TriageRecord, number>;
  patients: Dexie.Table<Patient, string>;
  
  constructor() {
    super('ClinicDB');
    
    this.version(1).stores({
      triage: '++id, patientId, timestamp, synced, severity',
      patients: 'id, name, phone, lastVisit',
      vitals: '++id, patientId, timestamp, synced'
    });
  }
}
 
interface TriageRecord {
  id?: number;
  patientId: string;
  symptoms: string[];
  vitals: VitalSigns;
  assessment: Assessment;
  timestamp: number;
  synced: boolean;
}

# app/services/malaria_screening.py
 
class MalariaScreening:
    """
    Malaria is the #1 cause of clinic visits in sub-Saharan Africa
    """
    
    MALARIA_SYMPTOMS = [
        "fever", "chills", "headache", "sweating",
        "fatigue", "nausea", "vomiting", "body aches"
    ]
    
    def assess_malaria_risk(
        self,
        symptoms: List[str],
        temperature: float,
        age: int
    ) -> Dict:
        """Calculate malaria probability"""
        
        # Count matching symptoms
        matches = sum(1 for s in symptoms if s.lower() in self.MALARIA_SYMPTOMS)
        
        # High fever is strong indicator
        fever_score = 0
        if temperature > 38.5:
            fever_score = 2
        elif temperature > 37.5:
            fever_score = 1
        
        # Children under 5 are high risk
        age_factor = 1.5 if age < 5 else 1.0
        
        # Calculate risk score
        risk_score = (matches + fever_score) * age_factor
        
        if risk_score >= 5:
            return {
                "risk": "high",
                "recommendation": "Immediate RDT test and treatment",
                "priority": "high"
            }
        elif risk_score >= 3:
            return {
                "risk": "medium",
                "recommendation": "RDT test recommended",
                "priority": "medium"
            }
        else:
            return {
                "risk": "low",
                "recommendation": "Monitor symptoms",
                "priority": "low"
            }

# app/services/maternal_health.py
 
class MaternalHealthService:
    """
    Pregnancy complications are a major concern in rural areas
    """
    
    DANGER_SIGNS = [
        "severe headache",
        "blurred vision",
        "severe abdominal pain",
        "vaginal bleeding",
        "reduced fetal movement",
        "severe swelling",
        "high blood pressure"
    ]
    
    async def assess_pregnancy_risk(
        self,
        gestational_age: int,  # weeks
        symptoms: List[str],
        vitals: Dict
    ) -> Dict:
        """Assess pregnancy-related risks"""
        
        # Check for danger signs
        danger_signs_present = [
            s for s in symptoms 
            if any(d in s.lower() for d in self.DANGER_SIGNS)
        ]
        
        # Check blood pressure (preeclampsia screening)
        bp_systolic = vitals.get('bp_systolic', 0)
        bp_diastolic = vitals.get('bp_diastolic', 0)
        
        if bp_systolic >= 140 or bp_diastolic >= 90:
            return {
                "risk": "critical",
                "condition": "Possible preeclampsia",
                "action": "IMMEDIATE doctor consultation",
                "transfer": "Consider referral to hospital"
            }
        
        if danger_signs_present:
            return {
                "risk": "high",
                "danger_signs": danger_signs_present,
                "action": "Urgent doctor consultation",
                "monitoring": "Close observation required"
            }
        
        # Routine pregnancy check
        return {
            "risk": "low",
            "action": "Routine antenatal care",
            "next_visit": self._calculate_next_visit(gestational_age)
        }

# app/services/sms_service.py
from twilio.rest import Client
 
class SMSTriageService:
    """
    SMS-based triage for zero-connectivity scenarios
    """
    
    def __init__(self):
        self.client = Client(account_sid, auth_token)
    
    async def process_sms_triage(self, from_number: str, message: str):
        """
        Process triage via SMS
        Format: "TRIAGE <age> <sex> <symptoms>"
        Example: "TRIAGE 25 M fever headache vomiting"
        """
        
        parts = message.upper().split()
        
        if parts[0] != "TRIAGE" or len(parts) < 4:
            await self.send_sms(
                from_number,
                "Format: TRIAGE <age> <sex> <symptoms>\nExample: TRIAGE 25 M fever headache"
            )
            return
        
        age = int(parts[1])
        sex = parts[2]
        symptoms = parts[3:]
        
        # Simple rule-based triage
        severity = self._assess_via_keywords(symptoms, age)
        
        response = self._format_sms_response(severity)
        await self.send_sms(from_number, response)
    
    def _assess_via_keywords(self, symptoms: List[str], age: int) -> str:
        """Simple keyword-based assessment"""
        
        critical_keywords = ["chest pain", "breathing", "bleeding", "unconscious"]
        high_keywords = ["severe", "high fever", "vomiting"]
        
        symptoms_text = " ".join(symptoms).lower()
        
        if any(k in symptoms_text for k in critical_keywords):
            return "critical"
        elif any(k in symptoms_text for k in high_keywords):
            return "high"
        elif age < 5 and "fever" in symptoms_text:
            return "high"  # Children with fever = high priority
        else:
            return "medium"
    
    def _format_sms_response(self, severity: str) -> str:
        """Format response for SMS (160 char limit)"""
        
        responses = {
            "critical": "URGENT: Go to clinic NOW. Critical symptoms detected.",
            "high": "HIGH PRIORITY: Visit clinic today. Bring this SMS.",
            "medium": "Visit clinic when possible. Monitor symptoms.",
            "low": "Low urgency. Rest and hydrate. Visit if worsens."
        }
        
        return responses.get(severity, responses["medium"])

# app/core/security.py
from cryptography.fernet import Fernet
import hashlib
 
class HealthDataEncryption:
    """Encrypt patient data at rest"""
    
    def __init__(self, encryption_key: str):
        self.cipher = Fernet(encryption_key.encode())
    
    def encrypt_patient_data(self, data: dict) -> str:
        """Encrypt sensitive patient information"""
        json_data = json.dumps(data)
        encrypted = self.cipher.encrypt(json_data.encode())
        return encrypted.decode()
    
    def decrypt_patient_data(self, encrypted_data: str) -> dict:
        """Decrypt patient information"""
        decrypted = self.cipher.decrypt(encrypted_data.encode())
        return json.loads(decrypted.decode())
    
    @staticmethod
    def anonymize_for_analytics(patient_data: dict) -> dict:
        """Remove PII for analytics"""
        return {
            "age_group": patient_data["age"] // 10 * 10,  # 25 → 20-29
            "sex": patient_data["sex"],
            "symptoms": patient_data["symptoms"],
            "severity": patient_data["severity"],
            # Remove: name, phone, address, ID numbers
        }

# app/services/analytics_service.py
 
class ClinicAnalytics:
    """Generate insights for clinic management"""
    
    async def get_daily_stats(self, clinic_id: str, date: str) -> Dict:
        """Daily clinic statistics"""
        
        stats = await db.query("""
            SELECT 
                COUNT(*) as total_patients,
                AVG(wait_time_minutes) as avg_wait_time,
                COUNT(CASE WHEN severity = 'critical' THEN 1 END) as critical_cases,
                COUNT(CASE WHEN diagnosis LIKE '%malaria%' THEN 1 END) as malaria_cases
            FROM triage_records
            WHERE clinic_id = ? AND DATE(timestamp) = ?
        """, [clinic_id, date])
        
        return stats
    
    async def get_disease_trends(self, clinic_id: str, days: int = 30) -> List[Dict]:
        """Track disease patterns over time"""
        
        trends = await db.query("""
            SELECT 
                DATE(timestamp) as date,
                diagnosis,
                COUNT(*) as cases
            FROM triage_records
            WHERE clinic_id = ? 
              AND timestamp >= DATE('now', '-{days} days')
            GROUP BY DATE(timestamp), diagnosis
            ORDER BY date DESC
        """, [clinic_id])
        
        return trends

# docker-compose.yml
version: '3.8'
 
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/clinic
      - REDIS_URL=redis://redis:6379
    restart: always
    deploy:
      resources:
        limits:
          cpus: '0.5'  # Low resource usage
          memory: 512M
  
  db:
    image: postgres:15-alpine  # Lightweight
    volumes:
      - postgres_data:/var/lib/postgresql/data
    environment:
      POSTGRES_DB: clinic
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
  
  redis:
    image: redis:7-alpine
    command: redis-server --maxmemory 128mb
 
volumes:
  postgres_data: