MA Streaming Open Data

!!! info New in v3.28.0

What this is

The MA Streaming Open Data v1 protocol is a Protobuf-based telemetry wire format published by Motion Applied (Apache 2.0). Every Kafka message is framed as an outer Packet envelope that carries a type discriminator string and an content field — raw bytes of a separate inner-payload protobuf message whose schema depends on that type. ATLAS-based data systems emit this format.

MAStreamingDeserializer handles both decoding stages in one call: it unwraps the outer Packet, dispatches the content bytes to the matching inner-payload class, and — for the three sample-bearing packet types (PeriodicData, SynchroData, RowData) — fans the samples out into flat row dicts ready for time-series sinks. The generic ProtobufDeserializer handles only the outer envelope; per-type dispatch, timestamp reconstruction, and row projection would still be your responsibility.

Prerequisites

Install Quix Streams with the Protobuf extra:

pip install quixstreams[protobuf]

No additional build toolchain is required. The vendored open_data_pb2.py is committed alongside the library.

Import the deserializer from quixstreams.models.serializers.ma_streaming_open_data. import quixstreams on its own does not load the protobuf bindings.

Quickstart

The smallest working pipeline reads a topic carrying MA Streaming Open Data packets and prints every flat row:

from quixstreams import Application
from quixstreams.models.serializers.ma_streaming_open_data import MAStreamingDeserializer

app = Application(
    broker_address="localhost:9092",
    consumer_group="my-group",
)

topic = app.topic(
    name="telemetry-input",
    value_deserializer=MAStreamingDeserializer(),
    key_deserializer="str",
)

sdf = app.dataframe(topic)
sdf = sdf.update(print)

app.run()

With default settings, each Kafka message carrying a PeriodicData, SynchroData, or RowData packet becomes one row per sample: the deserializer returns a list of flat dicts, and Quix Streams processes each item as its own row. A timestamp_extractor, if you set one, is called once per row.

The default configuration (no filter) drops every non-sample packet. Only PeriodicData, SynchroData and RowData produce output; session management, configuration and event packets produce no rows. To keep them, add a filter (see Filtering).

If your producer sends a numeric data_format_identifier instead of parameter names (MA's managed producers do), also set definitions_topic and consumer; see Resolving parameter names.

Default output shape

A single PeriodicData Kafka message with two signals, each recorded at two timestamps, produces four rows:

[
    {"packet_type": "PeriodicData", "timestamp": 1715000000, "name": "vCar",    "value": 312.4,   "validity": "DATA_STATUS_VALID"},
    {"packet_type": "PeriodicData", "timestamp": 1715000010, "name": "vCar",    "value": 313.1,   "validity": "DATA_STATUS_VALID"},
    {"packet_type": "PeriodicData", "timestamp": 1715000000, "name": "nEngine", "value": 11200.0, "validity": "DATA_STATUS_VALID"},
    {"packet_type": "PeriodicData", "timestamp": 1715000010, "name": "nEngine", "value": 11350.0, "validity": "DATA_STATUS_VALID"},
]

Understanding the output shapes

The deserializer uses two different shapes depending on packet type. This asymmetry is intentional: engineering time-series data benefits from a flat schema that fits any column store; non-engineering payloads vary too much across types to flatten uniformly.

Sample-bearing packets — flat rows

PeriodicData, SynchroData, and RowData carry time-series sample columns. The deserializer fans each sample into one flat dict. The default projection emits five columns:

Column	Source	Description
`packet_type`	`Packet.type`	Discriminator. Useful for `partitionBy("packet_type")` in lakehouse sinks.
`timestamp`	Reconstructed per sample	Absolute timestamp (see below).
`name`	Inline `data_format.parameter_identifiers.parameter_identifiers[i]`, or the definition for `data_format_identifier` (see Resolving parameter names)	Signal name (parameter identifier).
`value`	Sample value field	Measured value; type varies by sample list (double, int32, bool, string).
`validity`	`DataStatus` enum name	Status string such as `DATA_STATUS_VALID`, `DATA_STATUS_MISSING` or `DATA_STATUS_ERROR`; the integer with `use_integers_for_enums=True`.

Timestamps are reconstructed from the wire format. PeriodicData stores samples as start_time + i × interval; SynchroData uses a cumulative sum of the intervals field; RowData carries per-row timestamps. You never reconstruct timestamps manually.

Non-sample-bearing packets — tagged-union (when filter is active)

When a filter is set and a non-sample-bearing packet type survives the type allowlist, the deserializer emits it as a tagged-union record: a single-key dict whose key is the Packet.type string and whose value is the packet envelope without type. The envelope holds session_key, is_essential, content (the decoded inner payload) and id. 64-bit integers such as id are rendered as strings.

# A NewSession packet inside a filtered pipeline:
[
    {
        "NewSession": {
            "session_key": "run-42",
            "is_essential": True,
            "content": {"data_source": "Car1", "topic_partition_offsets": {}},
            "id": "0",
        }
    }
]

# A Configuration packet:
[
    {
        "Configuration": {
            "session_key": "run-42",
            "is_essential": True,
            "content": {
                "config_id": "cfg-1",
                "parameter_definitions": [...],
                "event_definitions": [],
                "group_definitions": [],
                "partial_packet": False,
                "packet_number": 0,
                "total_packet_number": 0,
            },
            "id": "0",
        }
    }
]

Without a filter, non-sample-bearing packets return [] and do not appear in the stream at all.

One record per packet (build_table=None)

Pass build_table=None to disable row projection. The deserializer returns one tagged-union record per packet, sample packets included, with no per-sample fan-out. Useful for diagnostic pipelines that need the full unmodified payload.

MAStreamingDeserializer(build_table=None)

On this path parameter names are not resolved: content.data_format stays as on the wire, for example {"data_format_identifier": "7"}.

Filtering

filter accepts a dict with two optional keys: "types" and "signals". Both are pure allowlists — anything not listed is dropped.

Filter by packet type

MAStreamingDeserializer(
    filter={"types": ["PeriodicData", "SynchroData", "NewSession"]},
)

All other packet types return []. Non-sample-bearing types that survive (e.g. NewSession above) are returned as tagged-union records.

Filter by signal name

MAStreamingDeserializer(
    filter={"signals": ["vCar", "nEngine"]},
)

Signal filtering applies inside sample-bearing packets only. Rows whose name is not in the allowlist are skipped before any dict is constructed. A packet whose columns contain none of the listed signals produces [].

Combining both axes

MAStreamingDeserializer(
    filter={
        "types": ["PeriodicData", "RowData", "NewSession"],
        "signals": ["vCar", "nEngine"],
    },
)

Type filtering runs first. Signal filtering runs inside surviving sample-bearing packets. Non-sample-bearing survivors (NewSession) return as tagged-union records; signal filtering does not apply to them.

Output shape summary

Condition	Per-message output
No filter — sample-bearing packet	`[{flat row}, ...]`
No filter — non-sample-bearing packet	`[]`
Filter active — sample-bearing, signals match	`[{flat row}, ...]`
Filter active — sample-bearing, signals don't match	`[]`
Filter active — non-sample-bearing, type in allowlist	`[{"NewSession": {envelope}}]`
Filter active — any packet, type not in allowlist	`[]`
`build_table=None` — any packet whose type passes the filter	`[{"PeriodicData": {envelope}}]`

Quix Streams processes each list item as its own row. Read the single root key of a tagged-union record to route by type:

def route(record: dict):
    if "packet_type" in record:
        handle_sample(record)                        # flat row from default projection
    else:
        (packet_type, inner), = record.items()       # tagged-union
        handle_non_sample(packet_type, inner)

Validation errors raised at construction

The filter argument is validated when you construct MAStreamingDeserializer(...), not at message-processing time:

Error	Cause
`TypeError: filter must be a Mapping`	Passed a string, list, or other non-dict value.
`TypeError: filter["types"] must be a sequence of strings, got str`	Passed a bare string instead of a list, e.g. `"types": "PeriodicData"`.
`ValueError: filter["types"] is empty`	Passed an empty list. Omit the key to disable that axis.
`ValueError: filter has unknown keys: ['signal']`	Typo — the key is `"signals"`, not `"signal"`.
`ValueError: filter["types"] contains unknown packet types: ['Periodicdata']`	Packet-type strings are case-sensitive.
`ValueError: filter["signals"] is set but build_table is disabled`	`build_table=None` and `filter["signals"]` are incompatible.
`ValueError: filter["signals"] is set but decode_content=False`	Signal filtering requires the inner payload to be decoded.

Driving the filter from environment variables

For containerised deployments, store each axis as a comma-separated environment variable:

import os
from quixstreams.models.serializers.ma_streaming_open_data import MAStreamingDeserializer

def csv_env(name: str) -> list[str]:
    raw = os.environ.get(name, "")
    return [item.strip() for item in raw.split(",") if item.strip()]

filter_spec: dict = {}
types = csv_env("FILTER_TYPES")
if types:
    filter_spec["types"] = types
signals = csv_env("FILTER_SIGNALS")
if signals:
    filter_spec["signals"] = signals

deserializer = MAStreamingDeserializer(filter=filter_spec or None)

Leaving an env var empty or unset omits that axis. The library rejects an explicit empty list with ValueError, so the if ...: guard converts "empty env var" into "axis not present" before the validator sees it. If both axes end up empty, filter_spec is {}, which the library accepts as a no-op alias for filter=None.

Custom projections with build_table

The build_table parameter controls which columns appear in the flat rows produced for sample-bearing packets. The default projection emits five columns. You can subset or rename them.

Supported columns

Only these seven source field names are recognised by the projection engine:

Column	Wire source	Notes
`parameter_identifier`	Inline `parameter_identifiers[i]`, or the definition for `data_format_identifier` (see Resolving parameter names)	Signal name. Appears as `name` in the default projection.
`timestamp`	Reconstructed per sample	Absolute timestamp.
`value`	Sample value	Type varies by sample list kind.
`status`	`DataStatus` enum name	Appears as `validity` in the default projection. The integer with `use_integers_for_enums=True`.
`session_key`	`Packet.session_key`	Envelope field; same value for every row from the same packet.
`packet_type`	`Packet.type`	Envelope field; discriminator.
`is_essential`	`Packet.is_essential`	Envelope boolean.

Note that the default projection's output columns (name, validity) are different from the source field names (parameter_identifier, status). When you pass build_table as a list you must use the source field names from the table above, not the default output names.

List form — subset columns, keep source names

Pass a list of source field names. The output column names match the source field names:

MAStreamingDeserializer(
    build_table=["parameter_identifier", "timestamp", "value"],
)
# Row: {"parameter_identifier": "vCar", "timestamp": 1715000000, "value": 312.4}

Dict form — rename output columns

Pass a dict where keys are the output column names and values are lists of source field aliases. The first alias in the list that matches a supported column is used as the source:

MAStreamingDeserializer(
    build_table={
        "signal":    ["parameter_identifier"],
        "ts_ns":     ["timestamp"],
        "raw_value": ["value"],
        "source":    ["packet_type"],
    },
)
# Row: {"signal": "vCar", "ts_ns": 1715000000, "raw_value": 312.4, "source": "PeriodicData"}

The dict-form alias list is evaluated at construction time to resolve the canonical source field; it is not used again per message.

What happens with an unsupported column name

Passing an unrecognised name raises ValueError at construction, not at runtime:

# Raises: ValueError: build_table contains unsupported column: 'signal_name'
MAStreamingDeserializer(build_table=["signal_name", "timestamp", "value"])

Supported packet types

Every built-in Packet.type string, its inner protobuf message class, and the output it produces under the default configuration:

`Packet.type`	Inner class	Sample-bearing	Default output
`PeriodicData`	`PeriodicDataPacket`	Yes	Flat rows (list)
`SynchroData`	`SynchroDataPacket`	Yes	Flat rows (list)
`RowData`	`RowDataPacket`	Yes	Flat rows (list)
`NewSession`	`NewSessionPacket`	No	`[]` (tagged-union with filter)
`EndOfSession`	`EndOfSessionPacket`	No	`[]` (tagged-union with filter)
`SessionInfo`	`SessionInfoPacket`	No	`[]` (tagged-union with filter)
`StreamStarted`	`StreamStartedPacket`	No	`[]` (tagged-union with filter)
`StreamStopped`	`StreamStoppedPacket`	No	`[]` (tagged-union with filter)
`Configuration`	`ConfigurationPacket`	No	`[]` (tagged-union with filter)
`Event`	`EventPacket`	No	`[]` (tagged-union with filter)
`Marker`	`MarkerPacket`	No	`[]` (tagged-union with filter)
`Error`	`ErrorPacket`	No	`[]` (tagged-union with filter)
`Metadata`	`MetadataPacket`	No	`[]` (tagged-union with filter)
`RawCANData`	`RawCANDataPacket`	No	`[]` (tagged-union with filter)
`AxisData`	`AxisDataPacket`	No	`[]` (tagged-union with filter)
`MapData`	`MapDataPacket`	No	`[]` (tagged-union with filter)
`DataFormatConfiguration`	`DataFormatConfigurationPacket`	No	`[]` (tagged-union with filter)
`DataFormatDefinition`	`DataFormatDefinitionPacket`	No	`[]` (tagged-union with filter)
`CoverageCursorInfo`	`CoverageCursorInfoPacket`	No	`[]` (tagged-union with filter)
`SystemStatus`	`SystemStatusMessage`	No	`[]` (tagged-union with filter)

A type not in this table is dropped under the default configuration. With build_table=None, or with a filter that has no types, it arrives as a tagged-union record whose content is the base64 string of the raw bytes. No warning is logged.

Recipes

1 — Filter to specific signals and write to a topic

Subscribe to two signals from periodic and synchro packets only, then forward the flat rows to a downstream topic for lakehouse ingestion.

from quixstreams import Application
from quixstreams.models.serializers.ma_streaming_open_data import MAStreamingDeserializer

app = Application(
    broker_address="localhost:9092",
    consumer_group="signal-sink",
)

topic = app.topic(
    name="telemetry-input",
    value_deserializer=MAStreamingDeserializer(
        filter={
            "types": ["PeriodicData", "SynchroData"],
            "signals": ["vCar", "nEngine"],
        },
        build_table=[
            "parameter_identifier",
            "timestamp",
            "value",
            "status",
            "session_key",
        ],
    ),
    key_deserializer="str",
)

out = app.topic("engineering-rows", value_serializer="json")

sdf = app.dataframe(topic)
# Each row: {"parameter_identifier": ..., "timestamp": ..., "value": ...,
#            "status": ..., "session_key": ...}
sdf.to_topic(out)

app.run()

2 — Route engineering rows and session events to separate topics

Use sdf.filter() to branch on whether a record is a flat engineering row or a tagged-union session event, then send each branch to its own output topic.

from quixstreams import Application
from quixstreams.models.serializers.ma_streaming_open_data import MAStreamingDeserializer

app = Application(
    broker_address="localhost:9092",
    consumer_group="router",
)

topic = app.topic(
    name="telemetry-input",
    value_deserializer=MAStreamingDeserializer(
        filter={
            "types": [
                "PeriodicData", "SynchroData", "RowData",
                "NewSession", "EndOfSession",
            ],
        },
    ),
    key_deserializer="str",
)

engineering_out = app.topic("engineering-rows", value_serializer="json")
session_out = app.topic("session-events", value_serializer="json")

sdf = app.dataframe(topic)

sdf.filter(lambda r: "packet_type" in r).to_topic(engineering_out)
sdf.filter(lambda r: "packet_type" not in r).to_topic(session_out)

app.run()

Engineering rows carry a packet_type key (flat projection). Tagged-union session records do not — their single root key is the packet type string ("NewSession", "EndOfSession").

3 — Resolving parameter names

Sample packets name their parameters in one of two wire forms:

Identifier form. data_format carries only a numeric data_format_identifier. MA's managed producers send this form.
Inline form. data_format.parameter_identifiers lists the names. This form also occurs on the wire and needs no setup.

The names for an identifier live in a DataFormatDefinition packet. The MA Stream API server writes it once, when the parameter list is first registered: to the data source's <prefix>Essentials.<datasource> topic and to partition 0 of <prefix>Data.<datasource>, never to per-stream topics. A consumer of the data topic cannot rely on seeing it, so identifier-form feeds require definitions_topic.

Subscribe only the data topic, and pass the deserializer the essentials topic name and a consumer to read it with:

from quixstreams import Application
from quixstreams.models.serializers.ma_streaming_open_data import MAStreamingDeserializer

app = Application(broker_address="localhost:9092", consumer_group="ma-telemetry")

with app.get_consumer(auto_commit_enable=False) as definitions_consumer:
    topic = app.topic(
        "Data.Car1",
        value_deserializer=MAStreamingDeserializer(
            definitions_topic="Essentials.Car1",
            consumer=definitions_consumer,
        ),
        key_deserializer="str",
    )
    sdf = app.dataframe(topic)
    sdf.update(print)
    app.run()

On a miss, the deserializer reads the definitions topic from where its previous read stopped until the identifier appears or no message arrives for 5 seconds, then looks the identifier up again. DataFormatDefinition packets that arrive on the data topic are cached too.
definitions_topic is the full broker-side topic name, including any prefix. Do not pass it through app.topic().
Identifiers are per data source. Use one deserializer per data source, each with that data source's essentials topic.
A KeyError after seeding means the definitions topic is wrong. The error goes to on_consumer_error, and by default the application stops.
Replicas and restarts need no extra settings. Each process reads the definitions topic on its first miss and commits no offsets for it.
build_table=None does not resolve names. Its records keep content.data_format as on the wire.

4 — Register a custom (OEM extension) packet type

If your feed carries private extension packets not listed in the built-in type table, register them with extra_content_types. Built-in types take precedence on key collision; to override a built-in, subclass MAStreamingDeserializer and rebind _content_types after super().__init__().

from quixstreams.models.serializers.ma_streaming_open_data import MAStreamingDeserializer
from my_oem_proto_pb2 import MyOemPacket

deserializer = MAStreamingDeserializer(
    extra_content_types={"MyOemType": MyOemPacket},
    filter={"types": ["PeriodicData", "MyOemType"]},
)

Options reference

Parameter	Type	Default	Description
`decode_content`	`bool`	`True`	When `True`, decode the inner payload bytes into a typed dict (Stage 2). When `False`, leave `content` as the Stage-1 base64 string and return one tagged-union record per packet — useful for envelope-only diagnostics.
`extra_content_types`	`dict[str, type[Message]]`	`None`	Additional `Packet.type → protobuf class` mappings for OEM extension packets. Built-in types win on key collision.
`use_integers_for_enums`	`bool`	`False`	When `True`, enum fields are emitted as integers instead of string names (e.g. `1` instead of `"DATA_STATUS_VALID"`), both in tagged-union records and in the projected `status` column.
`build_table`	list or dict or `None`	`DEFAULT_BUILD_TABLE`	Column projection for sample-bearing packets. Pass a list of source field names or a dict of output name → source alias list. Pass `None` to disable projection and get one tagged-union record per packet.
`filter`	`dict` or `None`	`None`	Packet-type and signal allowlists. Two optional keys: `"types"` (list of `Packet.type` strings) and `"signals"` (list of parameter identifier strings). `{}` is accepted as a no-op alias for `None`.
`definitions_topic`	`str`	`None`	Full broker-side name of the data source's essentials topic (`<prefix>Essentials.<datasource>`). Read on an unknown `data_format_identifier`. Requires `consumer`.
`consumer`	`Consumer`	`None`	A consumer from `app.get_consumer(auto_commit_enable=False)`, used to read `definitions_topic`. The deserializer only assigns and polls it; it never commits or closes it.

Maintenance

Regenerating the protobuf bindings

The vendored open_data_pb2.py is generated from proto/open_data.proto. End users never need to run this step. Library maintainers should regenerate after bumping the bundled schema to a new protocol revision:

pip install grpcio-tools==1.66.2
python quixstreams/models/serializers/ma_streaming_open_data/regenerate.py

Run it in a virtual environment that holds only grpcio-tools, and run the script by file path: python -m imports the package, which loads the committed bindings under the environment's protobuf runtime. The script uses the grpc_tools.protoc compiler bundled with grpcio-tools — no system-level protoc binary is required. The generated file is committed alongside the library so that pip install quixstreams[protobuf] works without a build toolchain.

After generation, the # Protobuf Python Version: comment at the top of open_data_pb2.py must read 5.27.2, the floor of the protobuf extra. Bindings generated for a newer version fail to import on older protobuf runtimes.

Protocol attribution

The open_data.proto schema is sourced from the MA.DataPlatforms.Protocol repository published by Motion Applied, licensed under the Apache License 2.0. MAStreamingDeserializer.protocol_version is "v1", mirroring the ma.streaming.open_data.v1 package name in the schema.

Limitations

Importing MAStreamingDeserializer and your own generated ma.streaming.open_data.v1 bindings in one process makes protobuf raise a duplicate symbol error, because both register the same message names; use one or the other.

Troubleshooting

Non-sample packets produce empty lists and never reach downstream steps

Symptom: You consume a topic carrying NewSession or Configuration packets. Nothing flows through the pipeline; no errors appear.

Cause: Without a filter, non-sample-bearing packets always return []. The default projection is optimised for time-series data; session and configuration events are intentionally excluded.

Fix: Add a filter that includes the packet types you need. They will appear as tagged-union records:

MAStreamingDeserializer(
    filter={"types": ["PeriodicData", "NewSession", "Configuration"]},
)

A signal I expect is missing from the output

Symptom: You set filter={"signals": ["vCar"]} but rows for that signal never appear.

Cause: Signal names are matched exactly and are case-sensitive. If the parameter_identifier on the wire is "v_car" or "VCAR", the filter drops it.

Fix: Temporarily remove the signals axis, run the pipeline, and print the name column from a few rows to see the exact wire name. Correct the name in your filter and restore it.

`ValueError` at construction when combining `build_table=None` and `filter["signals"]`

Symptom: MAStreamingDeserializer(build_table=None, filter={"signals": ["vCar"]}) raises immediately with a ValueError.

Cause: Signal filtering is implemented inside the row-projection fast path. Without a build_table, the deserializer has no mechanism to compare sample names against the allowlist.

Fix: Keep the default build_table (omit the parameter) or pass a custom list or dict. If you need build_table=None, use a "types"-only filter.

`KeyError: <n>` from the deserializer

Symptom: The application stops with KeyError: 7 (or another number) raised while deserializing a sample packet.

Cause: The packet carries data_format_identifier 7 and no definition for it is known. If definitions_topic is set, the deserializer already read that topic and the definition was not there, so definitions_topic names the wrong topic, or another data source's. If definitions_topic is not set, the definition was never seen: it is written only once, so identifier-form feeds need definitions_topic.

Fix: Set definitions_topic to the data source's full topic name, together with consumer; see Resolving parameter names.

`SerializationError` from the deserializer

Symptom: The application stops with a SerializationError raised while deserializing a message.

Cause: The message bytes, or the payload inside them, are not a valid MA Streaming Open Data Packet.

Fix: Check that the topic carries MA Streaming Open Data. To skip such messages instead of stopping, handle the error in on_consumer_error.

Message keys arrive as bytes instead of strings

Symptom: Keys appear as b"run-42" instead of "run-42", or the pipeline raises a deserialization error on message keys.

Cause: The default key_deserializer is "bytes". MA Streaming Open Data topics typically use string session identifiers as keys.

Fix: Pass key_deserializer="str" to app.topic():

topic = app.topic(
    name="telemetry-input",
    value_deserializer=MAStreamingDeserializer(),
    key_deserializer="str",
)