Skip to content

stream format

Jump to bottom
Lukas Sigrist edited this page Feb 19, 2023 · 2 revisions

RocketLogger Stream Interfaces

Starting with release 2.1.0, the RocketLogger software exposes two ZeroMQ based message interfaces for observing measurement status and receiving data blocks in real-time. The interfaces follows the publisher-subscriber pattern, where the RocketLogger measurement binary publishes a new message on each interface when a new data block has been recorded. Detailed configuration of the interfaces and their message data format is given below.

The interfaces were introduced with version 2.0, but not documented before release 2.1.0. They may be used with version 2.0 already, see the respective sections on compatibility.

Measurement Status Stream Interface

Status ZeroMQ Interface

The RocketLogger measurement binary publishes a new message to the measurement status interface every time a new measurement data block is processed, i.e. at the update rate configured via the --update CLI option.

The status interface binds the following endpoint a client can connect and subscribe to:

  • Transport: TCP
  • Address: 127.0.0.1 (i.e. accepting machine local connections only)
  • Port: 8276

Status Message Format

The status message contains the latest measurement status information, replacing any previously received status. The status information is encoded in JSON format with the following structure (showing sample values for illustration):

{
    "sampling": true,
    "error": false,
    "sample_count": 267200,
    "buffer_count": 334,
    "calibration_time": 1494315381, // unix timestamp, see calibration file format
    "calibration_file": "/home/rocketlogger/.config/rocketlogger/calibration.dat",
    "disk_free_bytes": 118452912128,
    "disk_free_permille": 945,
    "disk_use_rate": 288364,        // in bytes per second
    "sensor_count": 4,
    "sensor_names": [
        "TSL4531_left",
        "BME280_temp",
        "BME280_rh",
        "BME280_press"
    ],
    "config": {
        "ambient_enable": true,
        "background_enable": false,
        "interactive_enable": false,
        "calibration_ignore": false,
        "channel_enable": [
            "V1",
            "V2",
            "V3",
            "V4",
            "I1L",
            "I1H",
            "I2L",
            "I2H"
        ],
        "channel_force_range": [],
        "digital_enable": true,
        "file": {
            "comment": "Sampled using the RocketLogger",
            "filename": "/home/rocketlogger/data/data.rld",
            "format": "rld",
            "size": 1000000000
        },
        "sample_limit": 0,
        "sample_rate": 8000,    // in samples per second
        "update_rate": 10,      // in updates per second
        "web_enable": true
    }
}

If measurement data is not stored to file config.file is set to null.

Compatibility of Version 2.0

The measurement status interface exposed with 2.0.x releases is fully compatible with above listed specifications.

Measurement Data Stream Interface

Data ZeroMQ Interface

The RocketLogger measurement binary publishes a new multi-part message to the measurement data interface for every measurement data block, i.e. data messages are published at the update rate configured via the --update CLI option. Measurement data blocks are published only when data streaming is enabled (configured using the --web/--stream CLI option, by default enabled).

The data interface binds the following endpoint a client can connect and subscribe to:

  • Transport: TCP
  • Address: 127.0.0.1 (i.e. supporting machine local connections only)
  • Port: 8277

Data Message Format

The data message is a ZeroMQ multi-part message that consists of (a subset of) the following parts:

  1. JSON formatted metadata: measurement data rate and the sampled channel info
  2. data block timestamps: Linux real-time and raw monotonic clock timestamp for the data block
  3. channel data (separate message part per channel) the order is as specified in the metadata JSON of the first message part
  4. binary channel data (last message part): binary channel data aggregated into 32-bit values

The first two and last message parts are always available, the others depend on the active measurement channels: only sampled channels are available, all others are skipped. The binary data include the low current channel valid data when the corresponding current channel is measured.

Channel Metadata

Apart from the sample rate of the exposed measurement values, the metadata lists all channels with their specific properties:

  • channel name
  • measurement unit (with binary unit for binary values)
  • measurement scale (for non-binary values)
  • bit offset in the shared digital data values (for binary values)
  • an optional hidden boolean flag for data valid channels linked to measurement data of other channels

A sample metadata JSON is given below.

{
    "data_rate": 8000,  // in samples per second
    "channels": [
        {"name": "V1", "unit": "V", "scale": 1e-08},
        {"name": "V2", "unit": "V", "scale": 1e-08},
        {"name": "V3", "unit": "V", "scale": 1e-08},
        {"name": "V4", "unit": "V", "scale": 1e-08},
        {"name": "I1L", "unit": "A", "scale": 1e-11},
        {"name": "I1H", "unit": "A", "scale": 1e-09},
        {"name": "I2L", "unit": "A", "scale": 1e-11},
        {"name": "I2H", "unit": "A", "scale": 1e-09},
        {"name": "TSL4531_left", "unit": "lx", "scale": 1.0},
        {"name": "BME280_temp", "unit": "C", "scale": 0.001},
        {"name": "BME280_rh", "unit": "", "scale": 1e-06},
        {"name": "BME280_press", "unit": "P", "scale": 0.001},
        {"name": "DI1", "unit": "binary", "bit": 0},
        {"name": "DI2", "unit": "binary", "bit": 1},
        {"name": "DI3", "unit": "binary", "bit": 2},
        {"name": "DI4", "unit": "binary", "bit": 3},
        {"name": "DI5", "unit": "binary", "bit": 4},
        {"name": "DI6", "unit": "binary", "bit": 5},
        {"name": "I1L_valid", "unit": "binary", "bit": 6, "hidden": true},
        {"name": "I2L_valid", "unit": "binary", "bit": 7, "hidden": true}
    ]
}

Block Timestamp

Similar to the binary data file, the timestamp message part has a fixed size of 32 bytes that are to be interpreted as shown in the table below. Timestamp data is stored in little endian format.

Offset Size (byte) Field Content
0x00 8 Seconds Timestamp (real-time clock) of first sample in UNIX UTC seconds (integer)
0x08 8 Nanoseconds Nanoseconds of timestamp (real-time clock) of first sample (integer)
0x10 8 Seconds Timestamp (raw monotonic clock) of first sample in seconds (integer)
0x18 8 Nanoseconds Nanoseconds of timestamp (monotonic clock) of first sample (integer)

Data Channels

The data of the measurement channels are stored in separate message parts in the order the channels are listed in the metadata. The channel data of each message part represent an array of signed 32-bit integer values stored in little endian format.

If any binary channel is active, the binary channel message part data follow the same structure. All binary channels are merged into bit-stuffed 32-bit values stored in little endian format (same as for the binary data file. The bit value defined in the channel metadata specifies the bit offset of each binary channel. If not binary channel is active, the message part data is truncated to zero length.

Compatibility of Version 2.0

The measurement data interface exposed with 2.0.x releases uses the very same ZeroMQ interface. However, the metadata of binary channels may be interleaved with other channel metadata. Further, the binary channel message part always exposes data, i.e. is never truncated to zero length.

Stream Client Examples

Node.js Client

Separation of the RocketLogger measurement software and web interface by using an explicit message interface has been the main motivation to introduce the above two interfaces with RocketLogger version 2.0.

Therefore, the web interface client is the primary client of the data and status stream interfaces. For a complete example of a Node.js client implementation refer to the web interface server implementation, in particular at rl.data.js.

Python Client

Two minimal Python examples for a status and data stream client are given below as reference.

status_client.py: subscribe to, receive, decode and print measurement status updates

import json
import zmq

# ZeroMQ socket identifier for status publishing
STATUS_SOCKET = "tcp://127.0.0.1:8276"


context = zmq.Context()

print("subscribe to RocketLogger status interface")
socket = context.socket(zmq.SUB)
socket.connect(STATUS_SOCKET)
socket.subscribe("")

while True:
    message = socket.recv()
    status = json.loads(message)
    print(f"status received: {status}")

data_client.py: subscribe to, receive, decode and print measurement data

import json
import numpy as np
import zmq

# ZeroMQ socket identifier for data publishing
DATA_SOCKET = "tcp://127.0.0.1:8277"

# timestamp block data format
DT_TIMESTAMP = np.dtype(
    [
        ("realtime_sec", "<M8[s]"),
        ("realtime_ns", "<m8[ns]"),
        ("monotonic_sec", "<M8[s]"),
        ("monotonic_ns", "<m8[ns]"),
    ]
)


context = zmq.Context()

print("subscribe to RocketLogger data interface")
socket = context.socket(zmq.SUB)
socket.connect(DATA_SOCKET)
socket.subscribe("")

while True:
    message = socket.recv_multipart()

    # 1. JSON channel metadata
    meta = json.loads(message[0])
    print(f"data received: {meta}")

    # 2. data block timestamps
    time = np.frombuffer(message[1], dtype=DT_TIMESTAMP)
    print(f"time: {time}")

    channel_index = 2
    # 3. non-binary channels in order of channel metadata
    for channel in meta["channels"]:
        if channel["unit"] == "binary":
            continue

        data = np.frombuffer(message[channel_index], dtype="<i4")
        print(channel["name"], data)

        channel_index += 1

    # 4. binary channel bitmaps (if any listed in metadata)
    binary = np.frombuffer(message[channel_index], dtype="<u4")
    print("binary", binary)

Clone this wiki locally