Data Pipeline¶

Overview¶

Physical Sensors
      |
      v
/-----------------------------\
|        Data Source          |
|  /----------\ /----------\  |
|  | channel  | | channel  |  |  *N
|  \----------/ \----------/  |
\--------+---------+-----------/
         | data    | status
         \----+----/
              |
              v
/-----------------------------------------------------\
|                      NATS                            |
|                                                     |
|  data        --------------------  recorder         |
|              --------------------  aggregator       |
|                                          |          |
|  data_status <------ (metadata only) ---/          |
|              -----------------------------------\   |
|  status      -----------------------------------+--> trigger
|  event2      -----------------------------------/   |
|                                                     |
|  cmd         --------------------  any component    |
\-----------------------------------------------------/

Flow¶

Data Sources manage sensors and their Data Channels. Each configured channel processes and publishes batched samples to NATS (data context).
Aggregator subscribes to data and publishes per-batch column statistics (min, max, sum, count) to data_status.
Trigger subscribes to data_status, status, and event2, evaluates expressions, and sends measurement events to flower when conditions are met.
Flower tracks which channels are inside an active measurement window and instructs sinks to record.
Sinks subscribe to NATS and write received data (for example to parquet).
Recorder writes the Parquet recordings, owns the catalog, and runs export workers that copy completed recordings to a configured destination - s3upload to a bucket, usbexport to a removable medium.

Data Sources¶

A Data Source is a module component (moduleId.datasource.id) that reads from one or more physical or logical sensors and manages a set of Data Channels. It:

Receives raw DataFrames (columnar batches with timestamp and typed columns) from its sensor driver
Handles Data Channel configurations - channels are created, updated, and removed at runtime via the config store
Dispatches each incoming frame to every configured channel's processor chain
Publishes encoded data to NATS (data context) only for channels that have an active configuration - unconfigured channels produce no NATS output

Examples:

Module	Component Ref	Description
`retx`	`retx.datasource.adc0_ain`	8-channel ADC sensor hub
`retx`	`retx.datasource.acs0_acc_lg`	3-axis accelerometer (large range)
`gps`	`gps.datasource.gps`	GPS module

Data Channels¶

A Data Channel is a configured output of a Data Source, identified by a component reference (moduleId.channel.id). Each channel runs the raw DataFrame through a processor chain before encoding and publishing to NATS.

Examples:

ChannelRef	Source	Type	Columns
`retx.channel.ain0`	`retx.datasource.adc0_ain`	`Scalar`	`ain0` (uint32)
`retx.channel.ain2`	`retx.datasource.adc0_ain`	`Scalar`	`ain2` (uint32)
`gps.channel.position`	`gps.datasource.gps`	`GeographicCoordinate`	`lat`, `lon`, `alt`
`gps.channel.direction`	`gps.datasource.gps`	`GPSVelocity`	`speed`, `track`, `climb`

Processor Chain¶

Each Data Channel has an ordered processor chain of 0-n processors. Every processor receives the output of the previous one and may redefine the batch size, sampling frequency, or columns of the data flowing through it.

Processor	Purpose	Can redefine
`frameValidator`	Validates frame integrity; drops malformed frames	-
`resampler`	Resamples to a target frequency (`addRemove` or `linearInterpolation` strategy)	frequency
`columnSelector`	Selects, reorders, and renames columns; sets the channel type	columns, channel type
`windower`	Accumulates frames up to a fixed window size before publishing	window size

A columnSelector is required in every channel.

Wire Format¶

All channels publish data as a columnar, batched protobuf table (pbtable_v1) with per-column statistics (min/max/sum/count). Each NATS message carries the headers X-Flow-Timestamp (RFC3339Nano) and X-Flow-Type: pbtable_v1.

Measurements¶

A Measurement is a time-bounded recording window managed by the flower module:

Has a start timestamp and a deadline timestamp
At most one measurement is active per flow at any time
The deadline can be extended by new trigger events while the measurement is active
A measurement cannot be aborted - it runs until its deadline

The recorder persists data that falls within an active measurement window, one Parquet file per (device, measurementId, channel, part).

Aggregator¶

The aggregator module derives per-batch column statistics from the data stream and publishes them on data_status under the source channel's component ref. That is what trigger expressions read with data_status_val.