Aumentando a Qualidade do Atendimento ao Paciente através de Stream Processing

Camada de Integração de Dados

A camada de integração normalizará esses fluxos de dados pois se conectará a várias fontes e dicionários de dados para tornar os dados brutos significativos. Os dados de saúde do paciente na forma de resultados de exames médicos precisam de referências, a camada de integração pode chamar serviços de dicionário e enriquecer a forma bruta antes de ser processada posteriormente. A camada de integração também é responsável pela criação de um modelo de unificação padrão. Diferentes fontes de dados podem ter diferentes formatos (JSON, XML, EDI) e diferentes vocabulários (dos fornecedores), e é importante criar um modelo de unificação padrão para facilitar o processamento. A camada de integração emprega uma boa quantidade de padrões de integração de dados corporativos, como transformação, construção de carga útil, filtragem de mensagens e coleta dispersa. Essa camada também incorpora o agendamento para extrair dados em intervalos e em um determinado horário (agendado). Finalmente, os dados normalizados são persistidos para um processamento posterior.

Camada de Stream Processing

A camada de stream processing é a camada de inteligência modelada. Essa camada normalmente forma uma arquitetura de tipo lambda. Ela processará dados de integridade normalizados que persistem e também correlacionará os fluxos de eventos em tempo real que são alimentados diretamente por meio de outros sistemas. A camada de stream processing é configurável por natureza, e as regras precisam ser alimentadas para o resultado esperado. Um trabalho típico dessa camada seria correlacionar a atividade diária de um paciente em um fluxo de dados do Fitbit com os indicadores de um monitor de pressão arterial e comparar/contrastar com os registros históricos de saúde pessoais. Essa visão aumentará a capacidade de tomada de decisão do médico, simplificará as recomendações de investigação e reduzirá os erros médicos devido à falta de uma visão holística. Como tal, essa camada é equipada para processar o processamento de eventos baseado em regras, fazendo o stream processing em lote, em tempo real e de maneira preditiva.

Camada de Personalização/Enriquecimento

Cada EMR e dispositivo podem fornecer apenas uma visualização localizada dos dados que possui. No entanto, uma visão geral de saúde centrada no paciente é uma agregação de muitos sistemas. Essa agregação acontece na camada de integração e uma grande quantidade de inteligência é adicionada na camada de stream processing. A camada de personalização adiciona nos dados o contexto centrado no paciente, incluindo as informações pessoais do paciente como um enriquecimento para que os provedores e os pacientes possam visualizar os dados como informações significativas. Essa camada também modela as políticas de autorização de dados. As regras necessárias para expor os dados do paciente a partes relevantes com base em suas necessidades e privilégios.

Camada de Assinatura

Dados enriquecidos e indicadores personalizados precisam ser entregues aos pacientes, médicos e outros profissionais de saúde. A camada de assinatura fornece uma plataforma múltipla e intermediária para os consumidores aceitarem enviar esses fluxos de dados. As aplicações do provedor podem consumir esses dados como APIs ou como assinaturas no pub/sub-mode. A camada de assinatura mantém o handshake de segurança com a plataforma do cliente, geralmente com protocolos de segurança padrão, como OAuth/mTLS, etc.

Implementação da solução

Pode haver diferentes opções de tecnologia para cada uma das várias camadas citadas nesta solução. Para tornar a solução mais genérica e amplamente acessível, nos voltamos para os softwares open source e para as escolhas de framework. Para a implementação das camadas de acesso e integração de dados, temos opções como o Apache Camel, o Apache Synapse, o WSO2 Data Services (Apache 2.0) e o Spring Integration. Para a camada de stream processing, podemos usar frameworks como o Siddhi (Apache 2.0), Apache Spark, Apache Storm ou Apache Flink. A personalização e o enriquecimento também são serviços providos pelas mesmas ferramentas que usamos para integração e normalização de dados. O Synapse e o Camel são os preferidos neste espaço. A camada de assinatura expõe os endpoints com segurança para consumo externo e integração. As ferramentas open source de gerenciamento de API, como o WSO2 API Manager (Apache 2.0), o Kong API Manager (Apache 2.0) e o Tyk (MPL) preenchem o restante.

Para acessar dados de wearables, dispositivos de assistência médica e plataformas de EMR específicas de fornecedor (Cerner, Epic, etc), a implementação pode usar extensões de conector em ferramentas diferentes. No Apache Camel são chamados de componentes Camel, no Apache Synapse são extensões de sinapse. Esses conectores permitem interagir com a funcionalidade e extrair dados de um produto ou plataforma de terceiros da camada de acesso a dados. Essas ferramentas têm uma ampla gama de conectores para diferentes plataformas que podem ser instaladas sob demanda, que permitirão o acesso a diferentes funções da plataforma (por exemplo, Cerner EMR, Epic EMR, Fitbit, etc.). Os conectores encapsulam o protocolo do fornecedor, o mecanismo de autenticação e o acesso aos dados, incluindo serialização e desserialização.

<propertydescription="ID"expression="//patientId"name="id"scope="default"type="STRING"/>
 <!-- INICIALIZANDO O CONECTOR EPIC -->
<sequencekey="EPIC_CONNECTOR_INITIALIZATION"/>
<!-- Search Observation EPIC-Connector Operation →
<!-- Pesquisando uma Observação através de uma operação com o EPIC-Connector -->
 <epic.searchObservation>
    <patient>{$ctx:id}</patient>
    <code>718-7</code>
 </epic.searchObservation>
   …

O Synapse e o Camel também desempenham o papel da camada de integração nesta solução. Depois que os dados são extraídos de fontes diferentes, por meio de conectores e links JDBC, os dados são canonicalizados para um formato comum com o auxílio de dicionários de dados de saúde e vocabulários específicos de fornecedores. Tanto o Camel quanto o Synapse podem atuar como um ESB nessa camada e executar vários padrões de integração corporativa como transformação de mensagem, coleta de dispersão, fábrica de carga, etc. Finalmente, o fluxo de dados normalizado é gravado em um armazenamento persistente ou em uma fila/fonte de eventos.

<!-- EPIC Error code 4101 = Resource request returns no results -->
<!-- Capture Issue from EPIC’s predefined Error Response  -->
<filter description="Compares EPIC error code with Response" regex="4101" source="get-property('uri.var.issue')">
    <then>
      <log level="custom">
        <property name="EPIC_ERROR_CODE" value="4101"/>
      </log>
      <drop/>
    </then>
    <else>
       <!-- This property will be used only when Reports are updated -->
       <property name="updatedReportCount" scope="default" type="DOUBLE" value="1.0"/>
       <!-- Use EPIC_REPORT_FILTER Sequence for Filter According to Date -->
      <filter xpath="$ctx:updatedReportCount >  0 ">
        <then>
        <log level="custom">
        <property expression="get-property('updatedReportCount')" name="Updated Report Count"/>

Os dados normalizados são então buscados pela camada de stream processing. Durante a implementação, escolhemos o "Siddhi", um framework open source de stream processing para essa tarefa. O Siddhi tomará os eventos normalizados e executará as regras definidas neles para gerar resumos, alertas e previsões. O tipo de consultas e regras que podem ser aplicadas aos dados é modelado e parametrizado. Com base nos cenários e no tipo de implantação (voltado para o provedor ou voltado para o paciente), esses modelos podem ser materializados para os livros de regras reais que o mecanismo do Siddhi aplicará nos dados. Os resumos podem então ser publicados em tópicos, para armazenamentos persistentes para processamento/otimização adicionais.

A implementação acima será a mesma com outros frameworks como o Apache Spark, Storm ou Flink. Todas essas ferramentas são ricas em recursos de stream processing, uso de memória, bem como em correlação com dados históricos, implementando padrões de tipo lambda.

-- Siddhi App for Analyzing Observation Records
@App:name("DiagnosticReportApp")
@App:description("Analyze Diagnostic Reports and alert generating Test App")

-- Fetch Observation Reports from Kafka Stream and publish them to Siddhi Stream
@info(name = 'inputStream') 
@source(
type='kafka', 
topic.list='hemoglobin-epic', 
group.id='test-consumer-group', 
threading.option='single.thread', 
bootstrap.servers='localhost:9092', 
partition.no.list = '0',
@map(type='json' , @attributes(entry="$.entry")))
define stream DiagnosticReportDataStream (entry string);

@info(name='Keeps splitted test results') 
define stream SplittedDiagnosticReportStream(entry string, jsonElement string);

@info(name='Keeps Final Analyzed Report values') 
define stream ReportStream(patientId string ,title string, value double, low double, high double, reportdate string,reportId string,unit string);

-- Intermediate Stream used for internal communication between apps  
@sink(type = 'inMemory' , topic = 'DIAGNOSTIC_ALERT',
	@map(type = 'passThrough'))
define stream AlertStream(patientId string , report string, result string,reportdate string,reportId string , highValue string , lowValue string , myValue string );

-- $.entry array contains multiple test Results. Split those into separate events for further processing
@info(name = 'split_entry_query')
from DiagnosticReportDataStream#json:tokenize(entry,'$')
select *
insert into SplittedDiagnosticReportStream;

-- Extract necessary fields for analyzing
@info(name='extract_report_fields_query')
from SplittedDiagnosticReportStream
select str:split(json:getString(jsonElement,'$.resource.subject.reference'), "/" , 8) as patientId , json:getString(jsonElement,'$.resource.code.text') as title , json:getDouble(jsonElement, '$.resource.valueQuantity.value') as value , json:getDouble(jsonElement, '$.resource.referenceRange[0].low.value') as low , json:getDouble(jsonElement, '$.resource.referenceRange[0].high.value') as high,json:getString(jsonElement, '$.resource.effectiveDateTime') as reportdate, json:getString(jsonElement, '$.resource.id') as reportId , json:getString(jsonElement,'$.resource.valueQuantity.unit') as unit
insert into ReportStream;


-- If the value is out of the reference range, alert the alert recipients.
@info(name = 'abnormal_filtering_query')
from ReportStream[value < low or value > high]
select patientId as patientId , title as report, 'ABNORMAL' as result , reportdate as reportdate , reportId as reportId , str:concat(high , unit) as highValue , str:concat(low , unit) as lowValue , str:concat(value , unit) as myValue 

Para a personalização e enriquecimento, podemos reutilizar o EI da pilha de integração do WSO2. O EI buscará resumos, informações de alerta, previsões e irá correlacionar esses dados com as informações de identificação pessoal do paciente, além de informações de identificação do provedor para fornecer o contexto e como preparação para o consumo do usuário final.

<!-- EPIC CONNECTOR INITIALIZATION -->
<sequence key="EPIC_CONNECTOR_INITIALIZATION"/>

<!--Get Patient Data From EPIC-->
<epic.searchPatient>
   <id>{$ctx:uri.var.patientid}</id>
</epic.searchPatient>

<!--Persist for alert enrichment-->
<property description="Patient Name" expression="json-eval($.name[0].text)" name="uri.var.patientName" scope="default" type="STRING"/>
<property description="Home Address Line" expression="json-eval($.address[0].line[0])" name="uri.var.patientAddress" scope="default" type="STRING"/>
...
<!--Get Physician Data from EPIC-->
<epic.readById>
   <type>Practitioner</type>
   <id>{$ctx:uri.var.prescriberId}</id>
</epic.readById>

<!--Persist for alert enrichment-->
<property description="Practitioner Name" expression="json-eval($.name.text)" name="uri.var.pratitionerName" scope="default" type="STRING"/>
...

Para a camada de assinaturas, usamos o produto de gerenciamento de API de software livre (API-M) do WSO’2 a partir de uma variedade de produtos gerenciadores de API open source. Os fluxos de eventos enriquecidos são então expostos como tópicos e APIs, e são listados no portal do desenvolvedor do API Manager. O portal do desenvolvedor mantém as assinaturas do cliente, a segurança com os clientes e a emissão de chaves. O gateway de API irá encapsular as solicitações para a camada de integração corporativa enquanto valida as políticas de segurança, limitação e gerenciamento de acesso..

Resumo

Este artigo propõe uma solução para problemas de fluxo de dados e eventos desagregados centralizados no provedor e em eventos na área da saúde. A solução fornece uma visão consolidada de dados centrados no paciente, utilizando diferentes fontes espalhadas pelo hospital ou outras organizações de cuidado à saúde. Também fornecemos uma implementação de referência com uma seleção de tecnologias que fornece opções de ferramentas que possuem recursos para ETL, integração orientada a mensagens, stream processing e gerenciamento de API.

Sobre os autores

Nuwan Bandara é arquiteto de soluções e trabalha em estreita colaboração com empresas globais da Fortune 1000 em projetos de integração e mensagens de software. É diretor da WSO2 e lidera times de engenharia de soluções na América do Norte. Trabalhou em projetos interessantes nos setores de saúde, governo e finanças, simplificando a integração de sistemas. É programador, palestrante e autor.

Himasha Guruge é engenheira de soluções focada em gerenciamento de APIs, gerenciamento de fluxos de trabalho e integração. É engenheira sênior na WSO2, trabalhando de perto em soluções corporativas na Europa.

Nadeeshan Gimhana é formando de Ciência da Computação e Engenharia na Universidade de Moratuwa, no Sri Lanka. É o desenvolvedor principal da implementação de referência da solução discutida neste artigo.