SensorReadings — The Data Currency of the Edge

What Is a SensorReading?

Every piece of data that flows through a Pyvorin Edge pipeline is wrapped in a SensorReading. It is the atomic unit of the edge runtime: a single observation captured at a specific moment from a named sensor. Understanding how to create, validate, transform, and batch these objects is the first skill every edge developer must master.

The SensorReading dataclass lives in pyvorin_edge.sensors and is intentionally minimal. It contains exactly the fields required to describe a sensor observation without imposing unnecessary structure on your metadata.

Field Reference

Creating Readings

The simplest way to instantiate a reading is with positional or keyword arguments. Because SensorReading is a dataclass, you also get free __repr__, __eq__, and hashing behavior (if all fields are hashable).

import time
from pyvorin_edge.sensors import SensorReading

reading = SensorReading(
    sensor_name="boiler_room_temp",
    timestamp=time.time(),
    value=67.3,
    unit="°C",
    metadata={"calibrated": True, "probe_id": "T-1024"},
)

print(reading)
# SensorReading(sensor_name='boiler_room_temp', timestamp=1717152000.0,
#               value=67.3, unit='°C', metadata={'calibrated': True, 'probe_id': 'T-1024'})
  

Serialization and Deserialization

Edge pipelines frequently need to persist readings to local SQLite stores, forward them over MQTT, or buffer them for cloud upload. The built-in to_dict() method produces a JSON-safe dictionary. For the reverse path, you can reconstruct a reading manually or via helper utilities in your own adapter layer.

import json
from pyvorin_edge.sensors import SensorReading

reading = SensorReading(
    sensor_name="chiller_vibration",
    timestamp=time.time(),
    value=4.2,
    unit="mm/s",
    metadata={"axis": "z"},
)

# To JSON payload
payload = json.dumps(reading.to_dict())
print(payload)
# {"sensor_name": "chiller_vibration", "timestamp": 1717152000.0,
#  "value": 4.2, "unit": "mm/s", "metadata": {"axis": "z"}}

# From JSON payload back to object
data = json.loads(payload)
restored = SensorReading(**data)
assert restored == reading
  

Unit Conversion

The SDK does not enforce unit consistency at the reading level—your pipeline or adapter layer is responsible for normalisation. However, a common pattern is to attach a conversion helper directly to your ingestion code. Below is a drop-in Celsius-to-Fahrenheit converter that preserves metadata and produces a new SensorReading instance.

from pyvorin_edge.sensors import SensorReading

def celsius_to_fahrenheit(r: SensorReading) -> SensorReading:
    if r.unit != "°C":
        raise ValueError(f"Expected unit °C, got {r.unit}")
    return SensorReading(
        sensor_name=r.sensor_name,
        timestamp=r.timestamp,
        value=r.value * 9 / 5 + 32,
        unit="°F",
        metadata={**r.metadata, "converted_from": "°C"},
    )

original = SensorReading("lab_temp", 1717152000.0, 22.0, "°C", {})
converted = celsius_to_fahrenheit(original)
print(converted.value)   # 71.6
print(converted.unit)    # °F
  

Working with Batches

Real-world edge devices rarely emit readings one at a time. A Modbus poll, an MQTT bulk message, or a CSV log ingestion all produce batches. The SDK treats batches as plain Python lists, which means you can leverage list comprehensions, map, and standard library tools without learning new APIs.

import time
from pyvorin_edge.sensors import SensorReading

# Simulate a batch of 100 temperature readings
base_time = time.time()
batch = [
    SensorReading(
        sensor_name="warehouse_temp",
        timestamp=base_time + i * 60.0,
        value=18.0 + (i % 5),
        unit="°C",
        metadata={"batch_index": i},
    )
    for i in range(100)
]

# Filter anomalies (values > 21 °C)
anomalies = [r for r in batch if r.value > 21.0]
print(f"Anomalies detected: {len(anomalies)}")

# Compute average value manually
avg = sum(r.value for r in batch) / len(batch)
print(f"Average temperature: {avg:.2f} °C")

# Extract a list of dicts for JSON serialization
json_batch = [r.to_dict() for r in batch]
  

Validation with SensorRegistry

When you register a Sensor definition with a Pipeline, the runtime creates a SensorRegistry that can validate incoming SensorReading objects. This catches naming mismatches and unit errors before they pollute your windows or trigger false rule evaluations.

from pyvorin_edge.sensors import Sensor, SensorType, SensorReading, SensorRegistry

registry = SensorRegistry()
registry.register(Sensor(
    name="cold_store_temp",
    sensor_type=SensorType.TEMPERATURE,
    unit="°C",
    sampling_interval_seconds=30.0,
    normal_range=(-25.0, -18.0),
    alert_threshold=-15.0,
    location="Building-C",
))

# Valid reading
valid = SensorReading("cold_store_temp", 1717152000.0, -20.0, "°C")
print(registry.validate_reading(valid))   # True
print(registry.check_alert(valid))        # False

# Wrong unit — validation fails
invalid = SensorReading("cold_store_temp", 1717152000.0, -4.0, "°F")
print(registry.validate_reading(invalid)) # False

# Above threshold — alert fires
alert = SensorReading("cold_store_temp", 1717152000.0, -14.0, "°C")
print(registry.check_alert(alert))        # True
  

Best Practices

• Use epoch floats for timestamps — they compress well, sort without parsing, and are timezone-agnostic. Convert to ISO 8601 only at presentation time.
• Keep metadata flat — deeply nested dictionaries complicate SQLite serialization and cloud policy redaction. Prefer string keys and scalar values.
• Never mutate a reading in-place — treat SensorReading objects as immutable. Create new instances when converting units or enriching metadata.
• Name sensors hierarchically — building/zone/sensor_type/index (e.g., "hq_floor2_co2_01") makes wildcard subscriptions and log analysis far easier.