Taking Action on Real-Time Data with FME

Introduction

Integrating real-time data into your workflows can be made easy with the use of the FME Platform. Real-time data provides us with the opportunity to act quickly on sudden changes. This article is a high-level overview of how to process and take action on real-time data using FME Server by looking at the demo of a fully-automated flood monitoring system.

This demo uses data from the U.S. Geological Survey to produce a live-updating ArcGIS Online Dashboard and an email notification service, all made possible by FME. With FME, we’re able to connect real-time data to these applications seamlessly. Whether it’s a third-party source or your own sensors, FME can manipulate and process data in real-time.
 

What is Real-Time Data?

Real-time data is data that is generated and delivered for immediate processing. The source of this data is often sensors, mobile devices, social media, and applications. An example of this could be river flood gauges, equipment temperature sensors, the location of your phone, or the status of your web server. 

Real-time data can be classified as event data, which is generated when defined conditions arise, or as stream data, which is continuous and has no discrete beginning or end. This demo focuses on stream data by looking at the example of a real-time flood monitoring system.
 

Creating a Real-Time Flood Monitoring System with FME

This section of the article provides an overview of how FME was used to create a real-time flood monitoring system. The goal was to create an email notification system that automatically reacts to incoming real-time water level data, as well as an ArcGIS Online Dashboard that visualizes the data on a public web map.

Workspaces

The demo has been split into three key workspaces. One workspace gathers the data and sends it out in an appropriate format for other workspaces to receive, and two workspaces process the data and take action on it.

1: Retrieving the Real-Time Data

The first workspace gathers the data from the source and processes it for sending out to other workspaces that will take action on the real-time data. This first workspace acts as an intermediary between the source and our applications.

For the purpose of demonstration, we used a pre-recorded dataset that streams out data at 30-second intervals to the WebSocketSender. We used simulated real-time data because the demo needs to be always available on Safe.com and the U.S. Geological Survey could be unreliable or have downtime.

Normally, this is not the case. Instead, you can connect directly to the source through a web connection and send the data out each time new data is received through a WebSocketSender.
 

Retrieving Data From the Source Using a REST API

The first step in our real-time data simulation workspace was retrieving data from the source API. We made one API call for each water gauge station, reading them from a simple list of station IDs.

@Value(site_num) is the Station ID and is used in the Request URL to get the current water levels for each station.

We then removed unneeded attributes using an AttributeManager. We used a Timestamper so we could later sort the data in chronological order, ensuring it was streamed in the correct order. Finally, we packed the data into JSON string and stored 24 hours of recorded data in an Excel sheet. 
 

Sending Out Pre-Recorded Data

We created a second workspace that sends this data out in 30-second intervals through a WebSocketSender. We assigned the WebSocketSender a unique ID called “StationInfo”. We used this same unique ID in later workspaces to ensure that we connect to the correct WebSocket stream. 

There are many more Connectors that are able to run in stream mode.
 

2: Creating an Email Notification Service

Rising water levels due to rainfall, dam release, or snowmelt often mean danger. With the knowledge of what is considered a dangerous water level at each of the stations, we can easily create an automated email notification service. This workspace runs in the background, constantly monitoring water levels, ready to alert in the case of a flood. Not only does this improve response time, but it means that as long as the system is up and running, you don’t need to worry about checking the current water levels yourself. All of this is made possible with FME’s TestFilter transformer and Emailer transformer.
 

Retrieving Values From the WebSocketReceiver

We used a WebSocketReceiver to receive a JSON message from the WebSocketSender in the first workspace. The JSONFlattener then extracts the WebSocket message into attributes and values.
 

Limiting the Frequency of Emails

Emails in this demo are limited to a frequency of 30 minutes so that the receiving email address doesn’t get spammed. In a regular deployment, emails could be configured to send with every new flooding event. Values are sorted by station, and a sampler releases the last value received for each station when the time window ends.

Testing Current Water Levels for a Flood

We then set up a TestFilter to check water levels against known flood levels at each station. If a flood is detected, an email is sent with a list of the stations affected, the current water level, and the known flood height. The email body was customized using the HTMLReportGenerator. The HTML content generated by the HTMLReportGenerator is linked to an Email External Action in FME Server using the Automations Writer. Below is a sample email.
 

Using the Group By Parameter

It is important to note the use of the Group By parameter here. We grouped water level values by their time of measurement so that when features reach the HTMLReportGenerator, they are bulked into one table and one email as opposed to creating individual emails for each feature received.
 

3: Updating an ArcGIS Online Feature Layer with Real-Time Data

Real-time data can also be used to create visuals such as maps and dashboards, but to do this, we need a way to connect the real-time data to the map. To do this, we need to continuously upload the real-time data coming from our source to an ArcGIS Online feature layer. 
 

Retrieving the Real-Time Data

In this step, the JSONFlattener connects to a “WebSocket message received” trigger in FME Server and receives a JSON message from the WebSocketSender in the first workspace. We then extracted the JSON message into attributes and values.
 

Creating Geometry

Next, geometry is added to the data, allowing us to view the data spatially on a map. We used a VertexCreator and a CoordinateSystemSetter to create a point object from the longitude and latitude attributes.
 

Filtering Attributes and Updating an ArcGIS Online Feature Layer

When updating an ArcGIS Online feature layer, the fields in the data you upload must match the fields and types in the ArcGIS Online feature layer. We filtered and formatted the attributes so that they match the fields in the ArcGIS Online feature layer. We then used the ArcGIS Online Feature Service Writer to update an ArcGIS Online feature layer. 

With the feature layer being updated in real-time, we were able to create a dashboard in ArcGIS Online showing which stations are currently experiencing flood levels. You can view the dashboard here .

Tips and Tricks for the ArcGIS Online Feature Service Writer

When using the ArcGIS Online Feature Service Writer, it is important to distinguish some important parameters before heading forward. Real-time data is a unique challenge, so hopefully, this clarifies any difficulties you run into.

Tip #1: Set the writer to read 1 feature per request

In the Navigator > AGOL Feature Service Writer > Parameters > Advanced, there is an option to set Features Per Request. Set this to 1. This ensures that when the ArcGIS Online Feature Service Writer receives a feature, it will write it out immediately. If this value is greater than 1, the writer will not update the data in real-time.

Tip #2: Make sure you have OBJECTIDs attached to your data

ArcGIS Online uses OBJECTIDs to identify unique features when it is used in the UPDATE mode. If no OBJECTIDs are attached to the data we are sending out, the writer will not successfully update the data. We can add OBJECTIDs by merging a table with each feature and its OBJECTID to the data we are reading in through the WebSocketReceiver. This is accomplished with a FeatureMerger.

 

Enabling the Workspaces in FME Server

Finally, we used FME Server to automate the workspaces and run them indefinitely.
We ran the first workspace that simulates real-time data using the Run Workspace function. Because we want to run the workspace in a loop, the Run Until Cancelled option was enabled. Run Until Cancelled restarts the workspace once it has reached the end of the pre-recorded dataset. We then ran the other two workspaces in separate automations.
 

Email Notification Service Workspace

We built an automation that connects the output of the automation writers to an external email action. A manual trigger was used to trigger the workspace as if using the Run Workspace function in FME Server. This allows the workspace to collect data from the WebSocketReceiver to send out bulk emails. With the email action connected, an email containing a list of stations that have triggered a flood alert is sent.

ArcGIS Online Feature Layer Updater Workspace

We built an automation that allows the ArcGIS Online feature layer updater workspace to access the WebSocket message sent out by the first workspace. The result is the ArcGIS Online feature layer is constantly updated with real-time water level data. 

 

Connecting it all With WebSockets

As mentioned earlier in the article, WebSockets allow for the free flow of data between servers and applications. To enable the WebSocket connection, we created a publication with a matching Stream ID to the one used in the WebSocketSender and WebSocketReceiver in our workspaces.

Conclusion

This is just one of the many practical uses of real-time data. When looking at the case of a flood monitoring system, there are far more complex applications that can be integrated into this use-case. For example, we could create a work order if water levels breach a designated flood height, dispatching a unit to deliver sandbags and other flood mitigation strategies (see an example workflow here ). Simple tasks like these that used to require human intervention can now be automated with peace-of-mind using Streams in FME Server.