Skip to content

Telemetry Devices

@app.telemetry is the recommended decorator for devices that read a sensor on a recurring schedule and publish the result to MQTT. It owns the polling loop, error isolation, and publication strategy — your handler just reads and returns a dict.

Handler anatomy

The simplest telemetry handler takes zero arguments:

@app.telemetry("temperature", interval=60)  # (1)!
async def temperature() -> dict[str, object]:
    reading = await read_i2c_sensor()  # (2)!
    return {"celsius": reading.temp, "humidity": reading.rh}  # (3)!
  1. Framework calls this function every 60 seconds.
  2. Your code reads the hardware (or adapter).
  3. The returned dict is JSON-serialised and published to {prefix}/temperature/state as a retained QoS 1 message.

When you need infrastructure access (adapters, settings, MQTT publishing), declare a ctx: DeviceContext parameter and the framework injects it:

@app.telemetry("temperature", interval=60)
async def temperature(ctx: cosalette.DeviceContext) -> dict[str, object]:
    sensor = ctx.adapter(SensorPort)
    return {"celsius": sensor.read_temp()}

Telemetry devices are normally poll-only, but triggerable= gives them a trigger source: "mqtt" (== True) responds to inbound MQTT on {prefix}/{device}/set, "local" is woken in-process by an injected EntityNotifier when hardware pushes, and "both" accepts either — see the Triggerable Telemetry guide.

@app.device takes triggerable="local" too. Because the device owns its own loop, the framework injects a DeviceTrigger the handler awaits itself instead of racing the wake against an interval — see Local Triggers on @app.device.

A chatty trigger source can wake an entity far more often than is useful. min_interval=<seconds> bounds the spacing between trigger-initiated runs: the first wake after a quiet period runs immediately, wakes arriving inside the window coalesce into exactly one run carrying the last payload, and the interval= heartbeat keeps its own cadence throughout. It is opt-in — the default is off — see Throttling a Trigger Storm.

How the framework runs telemetry

Under the hood, @app.telemetry is syntactic sugar for a polling loop inside the framework:

# Simplified TelemetryRunner.run_telemetry (see _telemetry_runner.py)
async def run_telemetry(self, reg, ctx, error_publisher):
    last_published = None
    last_error_type = None
    while not ctx.shutdown_requested:
        try:
            result = await reg.func(ctx)
            if result is None:
                await ctx.sleep(reg.interval)
                continue
            strategy = reg.publish_strategy
            should_publish = (
                last_published is None          # First → always
                or strategy is None             # No strategy → always
                or strategy.should_publish(result, last_published)
            )
            if should_publish:
                await ctx.publish_state(result)
                last_published = result
                if strategy is not None:
                    strategy.on_published()
            if last_error_type is not None:
                last_error_type = None  # Recovery
        except asyncio.CancelledError:
            raise  # Let shutdown cancellation propagate
        except Exception as exc:
            if type(exc) is not last_error_type:
                await error_publisher.publish(exc, device=reg.name)
            last_error_type = type(exc)
        await ctx.sleep(reg.interval)

The framework wraps each telemetry call in error isolation with state-transition deduplication — the first error of each type is published, but repeated same-type errors are suppressed to prevent flooding. When the sensor recovers, the framework logs recovery and restores the device health status.

Publish strategies

By default, the framework publishes every probe result. Publish strategies decouple the probing frequency from the publishing frequency — probe often, publish selectively:

from cosalette import Every, OnChange

@app.telemetry("temperature", interval=10, publish=Every(seconds=300))
async def temperature() -> dict[str, object]:
    return {"celsius": await read_sensor()}

Here, interval=10 means the sensor is probed every 10 seconds, but Every(seconds=300) ensures state is published at most once every 5 minutes. This is useful when you want responsive readings locally (e.g. for EWMA smoothing) but don't need to flood MQTT.

For threshold modes, composition operators, and the full strategy reference, see Publish Strategies.

Coalescing groups

When multiple telemetry handlers share a physical resource — such as a serial bus, SPI interface, or rate-limited API — they can be grouped into a shared execution window using the group= parameter:

@app.telemetry("outdoor", interval=300, group="optolink")
async def outdoor(port: OptolinkPort) -> dict[str, object]:
    return await port.read_signals(["outdoor_temp"])

@app.telemetry("hotwater", interval=300, group="optolink")
async def hotwater(port: OptolinkPort) -> dict[str, object]:
    return await port.read_signals(["hot_water_temp"])

Handlers in the same group are managed by a shared tick-aligned scheduler. At t=0 all grouped handlers fire together; at subsequent ticks only those whose interval divides evenly into the elapsed time fire. This reduces resource sessions from N (one per handler) down to 1 per coinciding tick, eliminates timing drift, and enables adapter session sharing.

Each handler retains its own publish strategy, error isolation, persistence policy, and init function — group= is purely an execution scheduling hint.

A group member may also declare triggerable=. The wake is per member: arming one member runs that member alone, members armed at the same moment share one batch and one adapter session, and a member with no new input is never invoked. Its interval= heartbeat stays anchored to the group's shared epoch, so an out-of-cycle run cannot drift it out of alignment with its siblings.

See ADR-018 for the full design rationale and ADR-067 for the wake semantics.

Deferred registration

Sometimes a device should only be registered when the app's settings dictate it. All three decorator forms (@app.telemetry, @app.device, @app.command) accept enabled= as a callable that receives the resolved Settings instance and returns a bool:

@app.telemetry(
    "magnetometer",
    interval=lambda s: s.poll_interval,
    enabled=lambda s: s.enable_debug_device,  # resolved at bootstrap
)
async def magnetometer(mag: MagnetometerPort) -> dict[str, object]:
    reading = mag.read()
    return {"bx": reading.bx, "by": reading.by, "bz": reading.bz}

When the callable returns False, the device is silently dropped from the registry before MQTT wiring begins. Both interval= and enabled= support deferred resolution — all callables receive the same Settings instance at bootstrap time.

This preserves the fully-declarative main.py style: every device is visible at module level, and no @app.on_configure boilerplate is needed just to conditionally register one device.

Imperative add_*() methods

app.add_telemetry(), app.add_device(), and app.add_command() only accept enabled: bool. Inside @app.on_configure, settings are already available, so a callable is unnecessary.

See ADR-038 and ADR-020 for the design records.

See also