Work with sensor Data
Role: Data Architect, Data Steward
Goal: You want to store sensor data (Things, Datastreams, measurements) in the Sensor Data Storage and publish it through the SensorThings API.
Required Permissions: update Dataset
Before you start
CIVITAS/CORE stores sensor data in a Sensor Data Storage based on the OGC SensorThings API (FROST). This guide explains how to configure the Sensor Data Storage node in a Pipeline, which fields your Mapping must write, and how to publish the stored data.
The SensorThings API organizes sensor data in entities:
- Thing — the station or device, e.g. a weather station
- Location — where the Thing is placed
- Datastream — one measuring point of a Thing, e.g. temperature, with its unit of measurement
- Sensor and ObservedProperty — what measures and what is measured
- Observation — a single measurement in a Datastream
Sensor data support is continuously evolving. The features described in this guide represent the current functionality of the Platform. Additional capabilities will be added in future releases.
Store sensor data
Configure the Sensor Data Storage
Navigate to: Datasets → Dataflow → Pipeline
Use the Sensor Data Storage node to write data into the SensorThings API backend.
Each Sensor Data Storage node requires a port. The port defines the write logic of the node:
- which entities it writes
- whether it creates, updates or appends them
- by which reference it finds an existing entity again
There is no default port. A new Sensor data storage node shows an error until you select a port, and a Dataset with an unconfigured node cannot be published. The node shows the selected port under its name, so the Pipeline shows at a glance what it writes.
Create and connect nodes, configure Data sources and Storage nodes, and define how data flows through a Pipeline. → Build Pipelines
Choose a port
| Port | What it writes | Typical use |
|---|---|---|
Things | One Thing for each record | Load or correct master data only, without measurements |
Observations | One measurement into a Datastream that already exists | Write measurements into Datastreams that another Pipeline created |
ThingTree | A Thing with its Location, Datastream, Sensor and ObservedProperty, plus one measurement — as one unit | Write master data and measurements together from one source |
A common setup uses two Pipelines: one with the Things or ThingTree port for master data, and one with the Observations port for continuous measurements, e.g. from MQTT.
After you select a port, the node panel shows the structure the port expects: the field names, their data types, the mandatory fields (marked with *) and the field the port uses as reference. Use this structure as the target of your Mapping.
Transform sensor data
Configure mappings
Navigate to: Datasets → Dataflow → Pipeline → Mapping Node
The Mapping directly before the Sensor Data Storage node transforms incoming data into the structure of the selected port.
Assign input and output Data structures, connect Attributes, and apply transformations where required. → Create Mappings
Map the references
Each entity is found again by a reference. The reference is your own key: the identifier the source system uses for the station, the measuring point or the measurement. The Platform stores it with the entity and looks for it on the next delivery. This keeps a second delivery from creating a second entity.
Write the reference into the free-attribute field of the entity. SensorThings calls this field properties on every entity, and parameters on an Observation.
| Port | Entity | Field | Meaning |
|---|---|---|---|
Things | Thing | properties.reference | The key of the station |
ThingTree | Thing | properties.reference | The key of the station |
ThingTree | Datastream | properties.reference | The key of the measuring point. It only has to be unique within the station, so temperature is enough — it does not have to be station-7-temperature |
ThingTree | Location | properties.reference | Optional. Without it, the Location takes the reference of its Thing |
Observations | Observation | parameters.thingReference | The key of the station the Datastream belongs to |
Observations | Observation | parameters.datastreamReference | The key of the measuring point |
Observations | Observation | parameters.reference | Optional. With it, a second delivery of the same measurement corrects the first one; without it, every delivery appends a new measurement |
A record without its mandatory reference is not written and goes to the error sink. A record without a key could not be found again, and every later record without a key would land on the same entity.
Map a Location
A Location (port ThingTree) needs the following fields:
| Field | Value |
|---|---|
name | Name of the Location, e.g. Station Prinzipalmarkt |
description | Description of the Location |
encodingType | application/geo+json |
location | The position as a GeoJSON geometry |
The SensorThings API expects the location field as a GeoJSON object, not as a text or as separate latitude and longitude values:
{
"type": "Point",
"coordinates": [
7.639286227368158,
51.95226057725515
]
}
Note the order of the coordinates: longitude first, then latitude (WGS 84).
This object must reach the Sensor Data Storage in exactly this form. There are two ways:
- The source already delivers GeoJSON: map the attribute directly to
location. - The source delivers separate values, e.g. latitude and longitude: build the object in the Mapping editor — set
typetoPointand fillcoordinateswith longitude and latitude in this order.
Conversion of geometries in WKT format (e.g. POINT(7.6392 51.9522)) into the GeoJSON representation is planned but not yet available for the Sensor Data Storage. Until then, convert WKT into GeoJSON in the source or build the GeoJSON object in the Mapping editor.
Handle failed records
Error sink
The Sensor Data Storage writes one record at a time, and the records of one delivery do not depend on each other. One defective record does not hold up the others: the correct records are written, and only the defective one reaches the error sink.
Each error entry names what failed and why:
Pipeline record dropped: entity=Datastream status=404
reason=the record names no Datastream that exists file=<record id>
entity— the SensorThings entity the write failed onstatus— the response of the Sensor Data Storagereason— the message it returned, or the reason the Platform rejected the record
Nothing of a failed record is written, so you can deliver it again after you have corrected the data.
When the Sensor Data Storage is temporarily unavailable — e.g. it is restarting or overloaded — the record is not treated as an error. The Pipeline retries and writes the record to the error sink only when all attempts have failed.
Publish sensor data
Create a SensorThings API
Navigate to: Dataset → Dataflow → APIs
A Dataset can expose its sensor data through a SensorThings API – Time-series Data.
Requirements:
- a valid Pipeline
- at least one Sensor Data Storage node
Add an API to a Dataset, choose the API type and share the API URL with consumers. → Define an API
Summary
CIVITAS/CORE supports sensor data workflows throughout the Platform:
- store sensor data using the Sensor Data Storage node
- select a port that defines what the node writes
- map references so entities are found again on the next delivery
- map Locations as GeoJSON geometries
- identify defective records in the error sink
- publish sensor data through the SensorThings API