-   [](https://platform.exaloop.io/)
-   [Cookbook](https://platform.exaloop.io/cookbook/)
-   Validate Protobuf Telemetry

# 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.

```protobuf title="telemetry.proto"
syntax = "proto3";
package equipment;

message Reading {
  int64 event_id = 1;
  string machine_id = 2;
  optional double celsius = 3;
}
```

```sh title="Terminal"
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.

```python title="telemetry.codon"
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))
```

```sh title="Terminal"
exaloop --with data run -release telemetry.codon
```

```text title="Output"
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.

## Related Guides

-   [Generated message presence](https://platform.exaloop.io/data/protobuf/#mutation-and-presence)
-   [Wire compatibility](https://platform.exaloop.io/data/protobuf-api/#wire-compatibility)
-   [Kafka delivery and commits](https://platform.exaloop.io/streaming/kafka/#consume-and-commit)

Source: https://platform.exaloop.io/cookbook/telemetry/