Medical Device Data Ingestion and File Upload in the LifeOmic Platform
Introduction
In the rapidly evolving landscape of healthcare technology, medical platforms have become integral for collecting and analyzing data from various devices. A critical aspect of these platforms is the seamless ingestion of data and efficient upload of files. In this article, we will delve into the intricacies of the code responsible for enabling data ingestion and file upload in the LifeOmic Platform. By gaining a comprehensive understanding of the underlying mechanisms, we can appreciate how this code streamlines the flow of information, ensuring accurate diagnostics and efficient data management.
We will go through an example project that demonstrates the ability to handle button events on a medical device, simulating real-world scenarios while recording relevant results. We will discuss how these observations are structured, published to specific topics using MQTT, and seamlessly transmitted to the LifeOmic Platform for further analysis.
Another critical aspect we will cover is the code’s capability to facilitate efficient file upload. We will examine how the code requests upload URLs via MQTT in the event of error detection, allowing diagnostic log files to be securely transmitted to the server. We will explore the process of establishing an HTTP connection, performing PUT requests, and ensuring successful file uploads.
We’ll also go over what can be accomplished within our Platform’s web application once device data is available, from creating a table that lets us visualize all observations recorded on a device, to creating a graph that plots device data over time.
By delving into the code implementation, you’ll have a comprehensive understanding of the data ingestion and file upload processes within the LifeOmic Platform. Armed with this knowledge, we hope that you can optimize your own solutions, ensuring accurate data capture, efficient diagnostics, and enhanced patient care, all within a single secure and easy-to-use service.
Prerequisites
In order for devices to be able to communicate with the LifeOmic Platform, they must be correctly provisioned. The process for provisioning devices is explained in detailed in this publication. For the purposes of this article, we’ll assume that our device has already been successfully provisioned. Ultimately, this means that the device has access to its device certificate, which is used to securely communicate with the LifeOmic Platform.
Overview
Before diving into the implementation code, let’s look at a high-level overview of the data ingestion and file upload flow. The sequence diagram below illustrates the interactions between the device, the LifeOmic MQTT broker, and the LifeOmic API.
At the start, the device establishes a connection with the LifeOmic MQTT broker, enabling communication between the device and the platform. Once connected, the device subscribes to its unique FHIR ingest and File Upload topics, ensuring it receives relevant messages from the broker. Both the FHIR ingest and File Upload topics have /accepted and /rejected sub-topics so devices can listen for success and error responses as messages are published to them.
As in our past articles, we’ll be using the M5Stack Core 2. The device is equipped with hardware buttons that we’ll use to trigger events. One button triggers ingesting data and the other will simulate an error for uploading device diagnostics.
In the case of data ingestion, the device publishes the calculated results to the FHIRIngest topic. The LifeOmic MQTT broker receives the result and acknowledges its successful delivery by publishing to the ${DEVICE_ID}/FHIRIngest/accepted topic.
However, if an error is simulated by pressing on the error button, the device requests diagnostics upload by publishing a message to the CreateFileUploadLinktopic. The LifeOmic MQTT broker responds by publishing a signed file upload URL to the ${DEVICE_ID}/CreateFileUploadLink/accepted topic.
With the upload URL in hand, the device initiates an HTTP PUT request to the LifeOmic API, transmitting the diagnostic log file for upload. The LifeOmic API processes the request and returns a result indicating the success or failure of the file upload.
This overview provides a high-level understanding of the data ingestion and file upload flow within the medical device platform. In the subsequent sections, we will explore the code implementation that enables these functionalities, examining the steps involved in establishing connections, publishing messages, and handling file uploads.
Code Overview
Without further ado, let’s take a look at the code. For brevity, we’ll only show the most relevant parts of the code, and use // ... to indicate where code was omitted. A repository with all of the code can be found here.
Establishing Connectivity
At its core, the code focuses on establishing connectivity, ensuring seamless communication between the medical device and the LifeOmic Platform. It begins by initializing an authenticated WiFi connection by using the device certificate and device private key.
// ...
void setupWifi();
// ...
void setup()
{
// ...
setupWifi();
// ...
}
void loop()
{
// ...
if (WiFi.status() != WL_CONNECTED)
{
M5.Lcd.println("error, wifi not connected");
}
// ...
}
void setupWifi()
{
M5.Lcd.print("Connecting to WiFi");
WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
while (WiFi.status() != WL_CONNECTED)
{
delay(500);
M5.Lcd.print(".");
}
M5.Lcd.println("");
M5.Lcd.println("Wifi Connected");
wifiClient.setCACert(AWS_CERT_CA);
wifiClient.setCertificate(LO_DEVICE_CERTIFICATE);
wifiClient.setPrivateKey(LO_DEVICE_PRIVATE_KEY);
mqttConnect();
delay(2000);
// ...
}Additionally, the code configures the MQTT connection by subscribing to the device topic and adding a handler for dealing with “file upload” messages that the broker publishes to the device.
// in "Config.h"
std::string LO_IOT_ENDPOINT = "data.iot.us.lifeomic.com";
std::string AWS_RULES_TOPIC = "$aws/rules";
std::string LO_FHIR_INGEST_TOPIC_NAME = "FHIRIngest";
std::string LO_FILE_UPLOAD_TOPIC_NAME = "CreateFileUploadLink";
std::string LO_FHIR_INGEST_RULES_TOPIC = AWS_RULES_TOPIC + "/" + LO_FHIR_INGEST_TOPIC_NAME;
std::string LO_FILE_UPLOAD_RULES_TOPIC = AWS_RULES_TOPIC + "/" + LO_FILE_UPLOAD_TOPIC_NAME;
// ...
// in "main.cpp"
// ...
std::string fhirIngestAcceptedTopic = DEVICE_ID + "/" + LO_FHIR_INGEST_TOPIC_NAME + "/accepted";
std::string fhirIngestRejectedTopic = DEVICE_ID + "/" + LO_FHIR_INGEST_TOPIC_NAME + "/rejected";
std::string fileUploadAcceptedTopic = DEVICE_ID + "/" + LO_FILE_UPLOAD_TOPIC_NAME + "/accepted";
std::string fileUploadRejectedTopic = DEVICE_ID + "/" + LO_FILE_UPLOAD_TOPIC_NAME + "/rejected";
// ...
void setupWifi();
void mqttConnect();
// ...
void handleMessage(String &topic, String &payload);
// ...
void setup()
{
// ...
setupWifi();
// ...
}
void loop()
{
mqttConnect();
mqttClient.loop();
// ...
if (WiFi.status() != WL_CONNECTED)
{
M5.Lcd.println("error, wifi not connected");
}
// ...
}
void setupWifi()
{
M5.Lcd.print("Connecting to WiFi");
WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
while (WiFi.status() != WL_CONNECTED)
{
delay(500);
M5.Lcd.print(".");
}
M5.Lcd.println("");
M5.Lcd.println("Wifi Connected");
wifiClient.setCACert(AWS_CERT_CA);
wifiClient.setCertificate(LO_DEVICE_CERTIFICATE);
wifiClient.setPrivateKey(LO_DEVICE_PRIVATE_KEY);
mqttConnect();
delay(2000);
}
void mqttConnect()
{
if (mqttClient.connected())
{
return;
}
Serial.print("Last MQTT Error: ");
Serial.println(mqttClient.lastError());
Serial.println("Connecting to MQTT broker");
mqttClient.begin(LO_IOT_ENDPOINT.c_str(), 8883, wifiClient);
mqttClient.setCleanSession(false);
while (!mqttClient.connect(DEVICE_ID))
{
Serial.print(".");
delay(100);
}
if (!mqttClient.connected())
{
Serial.println("");
Serial.println("Broker timeout!");
}
else
{
Serial.println("Connected to MQTT broker");
Serial.printf("Subscribing to topic %s\n", fhirIngestAcceptedTopic.c_str());
mqttClient.subscribe(fhirIngestAcceptedTopic.c_str());
Serial.printf("Subscribing to topic %s\n", fhirIngestRejectedTopic.c_str());
mqttClient.subscribe(fhirIngestRejectedTopic.c_str());
Serial.printf("Subscribing to topic %s\n", fileUploadAcceptedTopic.c_str());
mqttClient.subscribe(fileUploadAcceptedTopic.c_str());
Serial.printf("Subscribing to topic %s\n", fileUploadRejectedTopic.c_str());
mqttClient.subscribe(fileUploadRejectedTopic.c_str());
// We'll take a look at the implementation of "handleMessage" in the
// following sections.
mqttClient.onMessage(handleMessage);
}
}
// ...Button Event Handling
The code is set up to publish data when either of the two side buttons on the device are pressed. The code that gets triggered when pressing those buttons simulates result calculations and publishes them to the data ingestion topic. At the same time, the device diagnostics also get updated by recording the last triggered action.
// ...
String diagFileBuffer = "";
// ...
void handleButton(Button btn);
void recordObservation(String val, String code, String system, String display);
void updateDiagnostic(String val);
// ...
void loop()
{
// ...
handleButton(m5.BtnA);
handleButton(m5.BtnC);
// ...
}
// ...
void handleButton(Button btn)
{
if (btn.wasReleased() || btn.pressedFor(1000, 200))
{
M5.Lcd.print("Calculating Results...");
delay(500);
String result = "success";
// LO_RESULT_CODE and LO_CODE_SYSTEM are arbitrary here, and
// can either match the FHIR spec or your own ingestion
// structure. See more in the FHIR docs:
// http://hl7.org/fhir/STU3/terminologies-systems.html
recordObservation(result, LO_RESULT_CODE, LO_CODE_SYSTEM, "");
M5.Lcd.printf("%s: %s\n", btn.label(), result);
updateDiagnostic("Btn=" + String(btn.label()) + ";result=" + result);
}
}
// ...
void recordObservation(String val, String code, String system, String display)
{
StaticJsonDocument<200> doc;
doc["value"] = val;
doc["unit"] = "status";
StaticJsonDocument<200> codeObj;
codeObj["code"] = code;
codeObj["system"] = system;
// fn argument display variable results in null.
codeObj["display"] = "btn_press_event";
JsonArray coding = doc.createNestedArray("coding");
coding.add(codeObj);
char jsonBuffer[1024];
serializeJson(doc, jsonBuffer);
// retained and qos are required.
bool pubResp = mqttClient.publish(LO_FHIR_INGEST, jsonBuffer, false, 1);
Serial.print("pub response ");
Serial.println(pubResp);
}
void updateDiagnostic(String val)
{
diagFileBuffer += val + "\n";
}
// ...The data that gets published to the ingestion topic is a JSON buffer that has a specific structure. First, it includes the value and the unit of the data point being ingested. It also includes a coding property, which is itself another JSON object which includes code, system, and display properties.
In the FHIR specification, a code represents a concept or a value that has a defined meaning within a particular code system. It serves as a standardized representation of clinical or administrative data. A code system, on the other hand, is a collection of codes and their associated definitions. It provides a controlled vocabulary or terminology that facilitates the consistent and interoperable exchange of healthcare information.
FHIR supports the use of various existing code systems to ensure consistency and interoperability across different healthcare systems. Some common systems include LOINC, SNOMED CT, and ICD-10. When ingesting data with the LifeOmic Platform, you can choose an existing code system or decide to roll your own.
File Upload
In addition to data ingestion, the code facilitates efficient file upload for diagnostic logs. An error is simulated when the middle button on the device is pressed. The press event triggers a function that initiates the file upload process by requesting an upload URL via the file upload MQTT topic. Upon receiving the request, the broker generates a unique signed file upload URL and sends it back via the device topic. Since the device is listening for messages on the device topic, it receives the event with the new upload URL. It then performs a PUT request to the upload URL, transferring the diagnostic log file securely to the LifeOmic API.
// ...
void handleErrorButton(Button btn);
void requestDeviceDiagnosticUpload();
void uploadDeviceDiagnostic(String &topic, String &payload);
// ...
void loop()
{
// ...
handleErrorButton(m5.BtnB);
// ...
}
// ...
void handleErrorButton(Button btn)
{
if (btn.wasReleased() || btn.pressedFor(1000, 200))
{
M5.Lcd.print("Calculating Results...");
delay(500);
String result = "fail";
M5.Lcd.println("error detected. Uploading diagnostic log file");
startFileUpload();
recordObservation(result, LO_RESULT_CODE, LO_CODE_SYSTEM, "");
updateDiagnostic("Btn=" + String(btn.label()) + ";result=" + result);
}
}
// ...
void startFileUpload()
{
StaticJsonDocument<200> doc;
doc["fileName"] = String(DEVICE_ID) + "_" + String(millis()) + "_device_diagnostic.txt";
doc["contentType"] = "text/plain";
char jsonBuffer[1024];
serializeJson(doc, jsonBuffer);
Serial.println(jsonBuffer);
bool pubResp = mqttClient.publish(LO_FILE_UPLOAD_RULES_TOPIC.c_str(), jsonBuffer, false, 1);
Serial.print("pub response ");
Serial.println(pubResp);
}
// We previously registered this function to the "onMessage" MQTT handler
// on the device.
void handleMessage(String &topic, String &payload)
{
Serial.print("Message received on topic: ");
// In a production environment, we'd use the topic to know which type of
// payload we're getting:
// - success if the topic ends with "CreateFileUploadLink/accepted"
// - error if the topic ends with "CreateFileUploadLink/rejected"
Serial.println(topic);
Serial.print("Payload: ");
Serial.println(payload);
// unmarshall payload, extract uploadUrl
DynamicJsonDocument doc(2048);
DeserializationError err = deserializeJson(doc, payload);
if (err)
{
Serial.print("Deserialization Error: ");
Serial.println(err.c_str());
}
if (!doc["uploadUrl"].isNull())
{
HTTPClient http;
http.begin(wifiClient, doc["uploadUrl"]);
http.addHeader("Content-Type", "text/plain");
http.addHeader("Content-Length", String(diagFileBuffer.length()));
int response = http.PUT(diagFileBuffer);
if (response >= 200 && response < 300)
{
diagFileBuffer = "";
}
else
{
Serial.printf("Error uploading diagnostic file: %d\n", response);
}
}
}Keep in Mind
As you implement and adapt this code to your specific medical device platform, consider incorporating additional features such as robust error handling, data encryption, and optimization techniques.
Data Analysis Inside the LifeOmic Platform
Now that we have taken a closer look at the data ingestion and file upload process, let’s dive into how we can leverage our device data once it has made its way to the LifeOmic Platform.
Let’s start with checking out some of the button event data that we’ve been sending so far. Once we are in the LifeOmic Platform, we can navigate to the Medical Devices tab on the left menu. Once there, we can search for the device we’d like to inspect. In this scenario, we’ve chosen one of our test devices, which has been sending the events shown in the previous code.
We have selected a time range to get an overview of all the events that were recorded during a certain time period.
We’ve configured our table to show the event type, the event value, and the time when it was recorded. Tables are configurable and can adapt to many different types of values, as seen below:
The other type of visualization we can reach for is a graph of data points over time. In our example, we have configured a graph to display the different button event values over time as dots on the graph. When hovering over each dot, we can see all of the event’s details.
As with the table visualization, the graph is also configurable. Most importantly, we can choose which type of FHIR data we want to plot, and then filter the data by specific codes. In our example, we have chosen to trace Observations matching codes with the btn_press_event value.
On top of the specific data events we ingested, we were also able to upload a device diagnostics file when we simulated an error on the device. The LifeOmic Platform lets us inspect that file by navigating to the Files tab on the left navigation menu.
We can use the search bar to find and open our diagnostics file. In our example, once we open it, we’ll be able to see a timeline of all the actions the device recorded before it encountered the error.
Closing Thoughts
The capabilities offered by the LifeOmic Platform provide us with powerful tools to gain valuable insights into the behavior of a fleet of devices. The visualizations and data analytics within the Platform allow us to unlock a deeper understanding of device performance and utilization. Armed with this knowledge, we are empowered to make data-driven decisions that have the potential to enhance the overall experience for practitioners and patients alike.
The ability to monitor and analyze device data within the LifeOmic Platform not only improves operational efficiency but also enables us to proactively identify patterns, detect anomalies, and optimize device performance. This proactive approach translates into better outcomes, reduced downtime, and improved patient care.
Additionally, the LifeOmic Platform ensures the secure storage and management of all the essential data required for these insights. With a robust and reliable infrastructure in place, we can trust that our sensitive data is protected, maintaining the highest standards of privacy and security.
To find out more about our medical device solutions, check out our website. We’ve also published another article that goes deeper into device failure analysis, which pairs well with the topics we have just discussed.
