Validate Protobuf Telemetry
Validate equipment telemetry before it reaches analytical calculations. This recipe checks Protobuf encoding, field presence, duplicate event IDs, and sensor ranges, then writes accepted readings and rejection reasons to separate Parquet files. Missing temperatures remain distinguishable from valid zero-degree readings.
Uses: generated Protobuf, native validation, Data SDK. Services: none. Difficulty: intermediate.
Define the Message
Save the schema and generate its native module in the same directory as the
program. optional makes presence observable independently of the scalar value.
syntax = "proto3";
package equipment;
message Reading {
int64 event_id = 1;
string machine_id = 2;
optional double celsius = 3;
}
exaloop protoc -I. --codon_out=. telemetry.proto
Validate Before Computing
The fixture includes two accepted readings and four rejection cases: a missing temperature, an out-of-range value, a duplicate event, and malformed bytes. Each bytes value contains one complete message supplied by the fixture. Raw Protobuf does not itself delimit concatenated messages; a real transport must provide those message boundaries.
from math import isfinite
from telemetry_pb import Reading
import exaloop.data as data
def reading(identifier: int, temperature: float) -> bytes:
message = Reading()
message.event_id = identifier
message.machine_id = "pump-7"
message.celsius = temperature
return message.to_bytes()
missing = Reading()
missing.event_id = 3
missing.machine_id = "pump-7"
frames = [reading(1, 0.0), reading(2, 92.5), missing.to_bytes(),
reading(4, 250.0), reading(2, 92.5), b'\xff']
seen: set[int] = set()
identifiers: list[int] = []
temperatures: list[float] = []
alerts: list[bool] = []
rejected: list[int] = []
reasons: list[str] = []
for position, frame in enumerate(frames):
try:
message = Reading.from_bytes(frame)
except ValueError:
rejected.append(position)
reasons.append("malformed")
continue
reason = ""
if message.event_id <= 0 or not message.machine_id:
reason = "invalid_identity"
elif not message.has("celsius"):
reason = "missing_temperature"
elif not isfinite(message.celsius) or not -40.0 <= message.celsius <= 150.0:
reason = "out_of_range"
elif message.event_id in seen:
reason = "duplicate"
if reason:
rejected.append(position)
reasons.append(reason)
continue
seen.add(message.event_id)
identifiers.append(message.event_id)
temperatures.append(message.celsius)
alerts.append(message.celsius >= 90.0)
assert temperatures == [0.0, 92.5] and alerts == [False, True]
assert reasons == ["missing_temperature", "out_of_range", "duplicate", "malformed"]
with data.table({"event_id": data.array(identifiers),
"celsius": data.array(temperatures),
"alert": data.array(alerts)}) as accepted:
accepted.write("accepted.parquet")
with data.table({"input_position": data.array(rejected),
"reason": data.array(reasons)}) as quarantine:
quarantine.write("rejected.parquet")
print("accepted", len(identifiers), "alerts", sum(int(value) for value in alerts),
"rejected", len(rejected))
exaloop --with data run -release telemetry.codon
accepted 2 alerts 1 rejected 4
The zero-degree reading is retained. The missing field is rejected rather than
silently treated as Protobuf's default 0.0. Parsing, generated field access,
and the validation loop execute natively; Parquet remains available to other
analytics tools.
Operational Choices
This example assumes globally unique event IDs. For device-local sequence
numbers, use a key such as (machine_id, boot_id, sequence) instead. The in-memory
set only deduplicates this run and grows with accepted events. A durable pipeline
needs a bounded deduplication window or an idempotent sink with persistent keys;
do not describe this local set as exactly-once processing.
Preserve the source topic/partition/offset or immutable object location in a real quarantine record. Raw frames may contain sensitive data: use access-controlled storage and retention limits, rather than logging all payloads. An alert is a classification here, not an external notification; retries need a separate delivery policy before sending email or opening maintenance tickets.
The simple threshold is intentionally inspectable. Replace it with calibrated rules or numerical code only after defining units, missing-value behavior, acceptable ranges, and the expected result for boundary cases. Changing those rules does not require changing the wire format.