Experiência 4: Integração e Interoperabilidade

Experiência 4: Integração e Interoperabilidade de Sistemas

Objetivo

Capacitar os estudantes na implementação de comunicação entre sistemas de saúde utilizando padrões de interoperabilidade modernos (HL7 FHIR) e clássicos (HL7v2). A experiência demonstra como integrar diferentes sistemas através de middleware especializado, garantindo troca segura e fidedigna de informação clínica.

Objetivos de Aprendizagem

Implementar operações CRUD completas em servidores HL7 FHIR
Configurar canais de integração com Mirth Connect
Transformar dados entre formatos HL7v2 ↔︎ HL7 FHIR
Garantir integridade e auditoria das trocas de dados
Aplicar conceitos de autenticação e autorização em APIs de saúde
Simular workflows hospitalares reais de troca de informação

Resultados de Aprendizagem

Ao completar esta experiência, os estudantes serão capazes de:

Realizar operações HL7 FHIR completas (CRUD + Search)
Configurar e monitorizar canais Mirth Connect
Implementar transformações bidirecionais HL7v2 ↔︎ HL7 FHIR
Configurar autenticação OAuth2/JWT para APIs
Simular integração entre sistemas hospitalares
Implementar logging e auditoria de transações

Exercício Prático Detalhado

Fase 1: Servidor HL7 FHIR e Operações Básicas (45 min)

Configuração do ambiente HAPI FHIR:

# Docker compose para HAPI FHIR Server
version: '3.7'
services:
  hapi-fhir:
    image: hapiproject/hapi:latest
    ports:
      - "8080:8080"
    environment:
      - spring.datasource.url=jdbc:h2:mem:testdb
      - hapi.fhir.fhir_version=R4
      - hapi.fhir.subscription.websocket_enabled=true
      - hapi.fhir.allow_external_references=true
    volumes:
      - ./hapi-config:/app/config

  # Base de dados PostgreSQL para produção
  postgres:
    image: postgres:13
    environment:
      POSTGRES_DB: hapi
      POSTGRES_USER: hapi
      POSTGRES_PASSWORD: HapiPassword123
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

Operações HL7 FHIR fundamentais:

import requests
import json
from datetime import datetime, timedelta

class FHIRClient:
    def __init__(self, base_url="http://localhost:8080/fhir"):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/fhir+json',
            'Accept': 'application/fhir+json'
        })

    def create_patient(self, patient_data):
        """Criar um novo paciente"""
        patient_resource = {
            "resourceType": "Patient",
            "identifier": [{
                "system": "http://hospital.example.org/patients",
                "value": patient_data['patient_id']
            }],
            "name": [{
                "family": patient_data['family_name'],
                "given": [patient_data['given_name']]
            }],
            "gender": patient_data.get('gender', 'unknown'),
            "birthDate": patient_data.get('birth_date'),
            "active": True
        }

        response = self.session.post(f"{self.base_url}/Patient",
                                   json=patient_resource)
        return response.json()

    def create_observation(self, patient_id, observation_data):
        """Criar observação de sinais vitais"""
        observation = {
            "resourceType": "Observation",
            "status": "final",
            "category": [{
                "coding": [{
                    "system": "http://terminology.hl7.org/CodeSystem/observation-category",
                    "code": "vital-signs",
                    "display": "Vital Signs"
                }]
            }],
            "code": {
                "coding": [{
                    "system": "http://loinc.org",
                    "code": observation_data['loinc_code'],
                    "display": observation_data['display']
                }]
            },
            "subject": {"reference": f"Patient/{patient_id}"},
            "effectiveDateTime": observation_data.get('datetime', datetime.now().isoformat()),
            "valueQuantity": {
                "value": observation_data['value'],
                "unit": observation_data['unit'],
                "system": "http://unitsofmeasure.org",
                "code": observation_data['ucum_code']
            }
        }

        response = self.session.post(f"{self.base_url}/Observation",
                                   json=observation)
        return response.json()

    def search_observations(self, patient_id, code=None, date_range=None):
        """Pesquisar observações com filtros"""
        params = {"subject": f"Patient/{patient_id}"}

        if code:
            params["code"] = code
        if date_range:
            params["date"] = f"ge{date_range['start']}&date=le{date_range['end']}"

        response = self.session.get(f"{self.base_url}/Observation", params=params)
        return response.json()

    def update_patient(self, patient_id, updated_data):
        """Atualizar dados do paciente"""
        # Primeiro buscar o recurso existente
        current = self.session.get(f"{self.base_url}/Patient/{patient_id}")
        patient = current.json()

        # Atualizar campos específicos
        if 'phone' in updated_data:
            patient.setdefault('telecom', []).append({
                "system": "phone",
                "value": updated_data['phone']
            })

        # PUT para atualização completa
        response = self.session.put(f"{self.base_url}/Patient/{patient_id}",
                                  json=patient)
        return response.json()

# Exemplo de utilização
fhir_client = FHIRClient()

# Criar paciente com dados da Experiência 3
patient_result = fhir_client.create_patient({
    'patient_id': 'PAT001',
    'family_name': 'Silva',
    'given_name': 'João',
    'gender': 'male',
    'birth_date': '1990-05-15'
})

patient_id = patient_result['id']

# Criar observações com dados das experiências anteriores
observations_data = [
    {
        'loinc_code': '8867-4',
        'display': 'Heart rate',
        'value': 72,
        'unit': 'beats/min',
        'ucum_code': '/min'
    },
    {
        'loinc_code': '8310-5',
        'display': 'Body temperature',
        'value': 36.7,
        'unit': 'degrees Celsius',
        'ucum_code': 'Cel'
    }
]

for obs_data in observations_data:
    fhir_client.create_observation(patient_id, obs_data)

Fase 2: Configuração Mirth Connect (60 min)

Instalação e configuração básica:

# Docker para Mirth Connect
version: '3.7'
services:
  mirth:
    image: nextgenhealthcare/mirth-connect:latest
    ports:
      - "8443:8443"  # Admin interface
      - "6661:6661"  # HL7 receiver
      - "6662:6662"  # HTTP sender
    environment:
      - DATABASE=postgres
      - DATABASE_URL=jdbc:postgresql://postgres:5432/mirthdb
      - DATABASE_USERNAME=mirth
      - DATABASE_PASSWORD=MirthPassword123
    volumes:
      - ./mirth-config:/opt/mirth-connect/appdata
    depends_on:
      - postgres

  postgres:
    image: postgres:13
    environment:
      POSTGRES_DB: mirthdb
      POSTGRES_USER: mirth
      POSTGRES_PASSWORD: MirthPassword123

Canal 1: HL7 → FHIR (Observações)

// Transform script em Mirth (JavaScript)
#### Canal 1: HL7v2 → HL7 FHIR (Observações)

```javascript
// Converter mensagem ORU^R01 para HL7 FHIR Observation

// Input: HL7 ORU^R01 message
var hl7Message = msg;

// Extrair dados do paciente (PID segment)
var pid = hl7Message['PID'];
var patientId = pid['PID.3']['PID.3.1'].toString();
var patientName = pid['PID.5']['PID.5.2'].toString() + " " + pid['PID.5']['PID.5.1'].toString();

// Extrair observações (OBX segments)
var obxSegments = hl7Message['OBX'];
var observations = [];

for (var i = 0; i < obxSegments.length(); i++) {
    var obx = obxSegments[i];

    var observation = {
        "resourceType": "Observation",
        "id": "obs-" + i + "-" + new Date().getTime(),
        "status": "final",
        "code": {
            "coding": [{
                "system": "http://loinc.org",
                "code": obx['OBX.3']['OBX.3.1'].toString(),
                "display": obx['OBX.3']['OBX.3.2'].toString()
            }]
        },
        "subject": {
            "reference": "Patient/" + patientId
        },
        "effectiveDateTime": formatHL7Date(obx['OBX.14'].toString()),
        "valueQuantity": {
            "value": parseFloat(obx['OBX.5'].toString()),
            "unit": obx['OBX.6']['OBX.6.1'].toString()
        }
    };

    observations.push(observation);
}

// Criar FHIR Bundle
var fhirBundle = {
    "resourceType": "Bundle",
    "type": "transaction",
    "entry": []
};

// Adicionar Patient se não existe
fhirBundle.entry.push({
    "request": {
        "method": "PUT",
        "url": "Patient/" + patientId
    },
    "resource": {
        "resourceType": "Patient",
        "id": patientId,
        "name": [{
            "text": patientName
        }]
    }
});

// Adicionar todas as observações
for (var j = 0; j < observations.length; j++) {
    fhirBundle.entry.push({
        "request": {
            "method": "POST",
            "url": "Observation"
        },
        "resource": observations[j]
    });
}

// Output: HL7 FHIR Bundle JSON
return JSON.stringify(fhirBundle, null, 2);

function formatHL7Date(hl7Date) {
    // Converter YYYYMMDDHHMMSS para ISO 8601
    if (hl7Date.length >= 8) {
        var year = hl7Date.substring(0, 4);
        var month = hl7Date.substring(4, 6);
        var day = hl7Date.substring(6, 8);
        var hour = hl7Date.length > 8 ? hl7Date.substring(8, 10) : "00";
        var minute = hl7Date.length > 10 ? hl7Date.substring(10, 12) : "00";

        return year + "-" + month + "-" + day + "T" + hour + ":" + minute + ":00Z";
    }
    return new Date().toISOString();
}

Canal 2: FHIR → HL7 (Sistema legacy)

#### Canal 2: HL7 FHIR → HL7v2 (Sistema legacy)

```javascript
// Transform HL7 FHIR Bundle para HL7v2 ADT^A08 (Patient Update)

var fhirBundle = JSON.parse(msg.toString());
var hl7Message = "";

// Processar cada entry no bundle
for (var i = 0; i < fhirBundle.entry.length; i++) {
    var resource = fhirBundle.entry[i].resource;

    if (resource.resourceType === "Patient") {
        // Criar mensagem HL7 ADT^A08
        var msh = "MSH|^~\\&|FHIR_SYSTEM|HOSPITAL|HIS|HOSPITAL|" +
                  getCurrentDateTime() + "||ADT^A08^ADT_A01|" +
                  generateControlId() + "|P|2.5||||\r";

        var pid = "PID|1||" + resource.id + "^^^HOSPITAL^MR||" +
                  (resource.name ? resource.name[0].family + "^" + resource.name[0].given[0] : "^^^") +
                  "||" + (resource.birthDate || "") + "|" +
                  (resource.gender ? resource.gender.charAt(0).toUpperCase() : "") + "|||||||||||||||||||||\r";

        hl7Message = msh + pid;

    } else if (resource.resourceType === "Observation") {
        // Adicionar segmento OBX para observações
        var obx = "OBX|" + (i+1) + "|NM|" +
                  (resource.code.coding[0].code || "") + "^" +
                  (resource.code.coding[0].display || "") + "^LN||" +
                  (resource.valueQuantity ? resource.valueQuantity.value : "") + "|" +
                  (resource.valueQuantity ? resource.valueQuantity.unit : "") + "|||||F|||" +
                  formatFHIRDate(resource.effectiveDateTime) + "\r";

        hl7Message += obx;
    }
}

return hl7Message;

function getCurrentDateTime() {
    var now = new Date();
    return now.getFullYear() +
           String(now.getMonth() + 1).padStart(2, '0') +
           String(now.getDate()).padStart(2, '0') +
           String(now.getHours()).padStart(2, '0') +
           String(now.getMinutes()).padStart(2, '0') +
           String(now.getSeconds()).padStart(2, '0');
}

function generateControlId() {
    return "MSG" + new Date().getTime();
}

function formatFHIRDate(isoDate) {
    if (!isoDate) return "";
    return isoDate.replace(/[-T:Z]/g, "").substring(0, 14);
}

Fase 3: Simulação de Workflow Hospitalar (45 min)

Cenário: Admissão de paciente com monitorização contínua

import time
import threading
from datetime import datetime, timedelta

class HospitalWorkflowSimulator:
    def __init__(self, fhir_client, mirth_url):
        self.fhir_client = fhir_client
        self.mirth_url = mirth_url

    def simulate_patient_admission(self, patient_data):
        """Simular admissão completa de paciente"""

        # 1. Criar registo no sistema FHIR
        print("Admitindo paciente...")
        patient = self.fhir_client.create_patient(patient_data)
        patient_id = patient['id']

        # 2. Enviar ADT^A01 (Admission) via HL7
        hl7_admission = self.create_hl7_admission(patient_data)
        self.send_hl7_message(hl7_admission)

        # 3. Iniciar monitorização contínua (dados das experiências anteriores)
        monitoring_thread = threading.Thread(
            target=self.continuous_monitoring,
            args=(patient_id,)
        )
        monitoring_thread.start()

        return patient_id

    def continuous_monitoring(self, patient_id):
        """Simular monitorização contínua de sinais vitais"""

        # Simular dados de 30 minutos de monitorização
        for minute in range(30):
            # Gerar sinais vitais simulados (baseados nos dados reais das experiências)
            vitals = self.generate_realistic_vitals(minute)

            # Criar observações FHIR
            for vital in vitals:
                self.fhir_client.create_observation(patient_id, vital)

            # Se valores críticos, enviar alerta via HL7
            if self.is_critical_value(vitals):
                alert_msg = self.create_hl7_alert(patient_id, vitals)
                self.send_hl7_message(alert_msg)
                print(f"Alerta crítico enviado para paciente {patient_id}")

            time.sleep(2)  # Simular 2 segundos = 1 minuto de tempo real

    def generate_realistic_vitals(self, minute):
        """Gerar sinais vitais realistas com tendências"""
        import random

        # Tendência de deterioração ao longo do tempo (simulação)
        base_hr = 70 + (minute * 0.5) + random.gauss(0, 5)
        base_temp = 36.5 + (minute * 0.02) + random.gauss(0, 0.3)

        return [
            {
                'loinc_code': '8867-4',
                'display': 'Heart rate',
                'value': max(50, min(150, base_hr)),
                'unit': 'beats/min',
                'ucum_code': '/min',
                'datetime': (datetime.now() - timedelta(minutes=30-minute)).isoformat()
            },
            {
                'loinc_code': '8310-5',
                'display': 'Body temperature',
                'value': round(max(35.0, min(42.0, base_temp)), 1),
                'unit': 'degrees Celsius',
                'ucum_code': 'Cel',
                'datetime': (datetime.now() - timedelta(minutes=30-minute)).isoformat()
            }
        ]

    def is_critical_value(self, vitals):
        """Detectar valores críticos"""
        for vital in vitals:
            if vital['loinc_code'] == '8867-4' and (vital['value'] < 50 or vital['value'] > 120):
                return True
            if vital['loinc_code'] == '8310-5' and (vital['value'] < 35.5 or vital['value'] > 38.5):
                return True
        return False

# Executar simulação
simulator = HospitalWorkflowSimulator(fhir_client, "http://localhost:6661")

patient_data = {
    'patient_id': 'SIM001',
    'family_name': 'Santos',
    'given_name': 'Maria',
    'gender': 'female',
    'birth_date': '1985-03-20'
}

patient_id = simulator.simulate_patient_admission(patient_data)
print(f"Simulação iniciada para paciente {patient_id}")

Fase 4: Auditoria e Monitorização (30 min)

Dashboard de monitorização de integrações:

class IntegrationMonitor:
    def __init__(self, fhir_url, mirth_stats_url):
        self.fhir_url = fhir_url
        self.mirth_stats_url = mirth_stats_url

    def generate_daily_report(self):
        """Gerar relatório diário de integrações"""

        report = {
            'date': datetime.now().strftime('%Y-%m-%d'),
            'fhir_stats': self.get_fhir_statistics(),
            'mirth_stats': self.get_mirth_statistics(),
            'errors': self.get_integration_errors(),
            'performance': self.get_performance_metrics()
        }

        return report

    def get_fhir_statistics(self):
        """Estatísticas do servidor FHIR"""
        stats = {}

        # Contar recursos por tipo
        resource_types = ['Patient', 'Observation', 'Device', 'DiagnosticReport']
        for resource_type in resource_types:
            response = requests.get(f"{self.fhir_url}/{resource_type}?_summary=count")
            if response.status_code == 200:
                stats[resource_type] = response.json().get('total', 0)

        # Atividade das últimas 24h
        yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
        recent_obs = requests.get(f"{self.fhir_url}/Observation?date=ge{yesterday}&_summary=count")
        stats['recent_observations'] = recent_obs.json().get('total', 0) if recent_obs.status_code == 200 else 0

        return stats

    def get_mirth_statistics(self):
        """Estatísticas dos canais Mirth"""
        # Esta seria implementada com a API REST do Mirth
        return {
            'channels_active': 2,
            'messages_processed_24h': 1250,
            'errors_24h': 3,
            'average_processing_time_ms': 45
        }

    def validate_data_integrity(self, sample_size=100):
        """Validar integridade dos dados entre sistemas"""

        # Selecionar amostra aleatória de observações
        observations = requests.get(f"{self.fhir_url}/Observation?_count={sample_size}")

        integrity_issues = []

        for obs in observations.json().get('entry', []):
            resource = obs['resource']

            # Validar estrutura FHIR
            if not self.validate_fhir_resource(resource):
                integrity_issues.append({
                    'type': 'invalid_fhir_structure',
                    'resource_id': resource.get('id'),
                    'issues': 'Missing required fields or invalid structure'
                })

            # Validar valores fisiológicos plausíveis
            if not self.validate_physiological_values(resource):
                integrity_issues.append({
                    'type': 'implausible_values',
                    'resource_id': resource.get('id'),
                    'issues': 'Physiologically implausible values detected'
                })

        return {
            'total_checked': sample_size,
            'integrity_issues': len(integrity_issues),
            'issues_detail': integrity_issues
        }

# Executar monitorização
monitor = IntegrationMonitor(
    fhir_url="http://localhost:8080/fhir",
    mirth_stats_url="http://localhost:8443/api/channels/statistics"
)

daily_report = monitor.generate_daily_report()
integrity_report = monitor.validate_data_integrity()

print("Relatório de Integração:")
print(json.dumps(daily_report, indent=2))
print("\nRelatório de Integridade:")
print(json.dumps(integrity_report, indent=2))

Recursos Necessários

Infraestrutura Base

Docker e Docker Compose
Servidor HAPI FHIR (contêinerizado)
Mirth Connect (contêinerizado)
PostgreSQL para persistência
Rede virtual para comunicação entre serviços

Software de Desenvolvimento

Python 3.8+ com bibliotecas: requests, fhir.resources, hl7apy
Postman/Insomnia para testes de API
HL7 Inspector para validação de mensagens
Git para versionamento de configurações

Ferramentas de Monitorização

Prometheus + Grafana para métricas
ELK Stack para logs centralizados
Mirth Administrator para gestão de canais
FHIR validation tools para testes de conformidade

Metodologias de Ensino

Sessão Teórica (45 min)

Fundamentos de interoperabilidade em saúde
Padrões HL7 v2 vs FHIR: quando usar cada um
Arquiteturas de integração: hub-and-spoke vs point-to-point
Segurança e auditoria em trocas de dados clínicos

Sessão Prática (135 min)

Setup do ambiente (30 min): Docker, serviços básicos
Operações FHIR (45 min): CRUD completo, pesquisas avançadas
Configuração Mirth (45 min): Canais, transformações, testes
Simulação completa (15 min): Workflow end-to-end

Trabalho Autónomo (90 min)

Implementar autenticação OAuth2 no FHIR
Criar canais Mirth adicionais para outros tipos de mensagem
Desenvolver dashboard de monitorização personalizado
Configurar alertas automáticos para falhas de integração

Avaliação

Critérios de Avaliação (100 pontos)

Operações FHIR (25 pts): Correção e eficiência das operações
Configuração Mirth (30 pts): Canais funcionais, transformações corretas
Simulação de workflow (20 pts): Cenário realista e completo
Monitorização e auditoria (15 pts): Implementação de controlos
Documentação técnica (10 pts): Procedimentos e troubleshooting

Entregáveis

Configuração completa do ambiente (Docker Compose)
Scripts Python para todas as operações FHIR
Canais Mirth exportados com documentação
Relatório de teste com validação de integrações
Dashboard de monitorização funcional

Extensões Avançadas

Segurança Avançada

mTLS para autenticação de serviços
JWT tokens com scopes específicos
SMART on FHIR para aplicações clínicas
Audit logging conforme IHE ATNA

Performance e Escalabilidade

Clustering de servidores FHIR
Load balancing para alta disponibilidade
Caching de recursos frequentemente acedidos
Async processing para grandes volumes

Integração com Sistemas Reais

Conectores para OpenEMR, GNU Health
APIs de dispositivos médicos
Integração com laboratórios (LIS)
Workflows de farmácia e prescrição

Integração com Pipeline

Input das Experiências Anteriores

Dados estruturados do REDCap (Experiência 3)
Resources FHIR válidos e testados
Metadados de qualidade e linhagem dos dados
APIs REST para acesso programático

Output para Experiências Seguintes

Dados integrados de múltiplas fontes
Workflows de troca padronizados
APIs seguras para consumo por aplicações
Audit trails completos para compliance

Preparação para Analytics

Data warehouse alimentado por integrações
ETL pipelines automáticos
Real-time streams para dashboards
Historical data para machine learning