Build a Telemetry Device¶

Telemetry devices are the most simple archetype in cosalette. They poll a sensor at a fixed interval and publish a JSON state message — the framework handles the timing loop, serialisation, and error isolation for you.

How Telemetry Works¶

The @app.telemetry decorator registers a function that:

1. Optionally receives a DeviceContext or other injectable parameters.
2. Returns a dict — the framework JSON-serialises it and publishes to {prefix}/{name}/state.
3. Runs on a fixed interval — the framework calls await ctx.sleep(interval) between invocations under the hood. This is the probing frequency.
4. Optionally uses a publish strategy (publish=) to control which probe results are actually published — decoupling probing from publishing.
5. Can return None to suppress a single cycle.
6. Is error-isolated — if one poll raises an exception, the framework logs the error, publishes it to the error topic, and continues the loop. A single bad reading never stops the daemon.

This is the return-dict contract: your function produces data, the framework handles delivery. Compare this to @app.device where you own the main loop and call ctx.publish_state() manually (see Command & Control Device).

strategy = ...  # from the publish= parameter, or None
last_published = None
last_error_type = None
while not ctx.shutdown_requested:
    try:
        result = await your_function(ctx)
        if result is None:
            await ctx.sleep(interval)
            continue
        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:
            log_recovery()
            last_error_type = None
    except Exception as exc:
        if type(exc) is not last_error_type:
            log_and_publish_error(exc)
        last_error_type = type(exc)
    await ctx.sleep(interval)

A Minimal Telemetry Device¶

The simplest telemetry handler takes zero arguments — just return a dict:

import cosalette

app = cosalette.App(name="gas2mqtt", version="1.0.0")


@app.telemetry("counter", interval=60)  # (1)!
async def counter() -> dict[str, object]:  # (2)!
    """Read the gas meter impulse count."""
    return {"impulses": 42}  # (3)!


app.run()

1. "counter" is the device name — it determines the MQTT topic: gas2mqtt/counter/state. interval=60 means polling every 60 seconds.
2. Zero-arg handlers are valid. The framework injects nothing — your function just returns data. You can also request ctx: DeviceContext if needed.
3. The returned dict is published as {"impulses": 42} to gas2mqtt/counter/state with retain=True and qos=1.

When you run this, the framework:

• Connects to the MQTT broker.
• Calls counter() every 60 seconds.
• Publishes the returned dict as JSON to gas2mqtt/counter/state.
• Keeps running until SIGTERM or SIGINT.

Single-Device Apps (Root Device)¶

When your app has only one device, you can omit the device name entirely. The framework publishes directly to root-level topics — no /{device}/ segment:

import cosalette

app = cosalette.App(name="weather2mqtt", version="1.0.0")


@app.telemetry(interval=30)  # (1)!
async def read_sensor() -> dict[str, object]:
    """Read weather station sensors."""
    return {"temperature": 21.5, "humidity": 58.0}


app.run()

1. No device name — the function name read_sensor is used internally for logging. The MQTT topic is weather2mqtt/state (no device segment).

Topic layout:

Using DeviceContext¶

When your handler needs infrastructure access, declare a ctx: DeviceContext parameter — the framework injects it automatically:

@app.telemetry("counter", interval=60)
async def counter(ctx: cosalette.DeviceContext) -> dict[str, object]:
    settings = ctx.settings          # (1)!
    device_name = ctx.name           # (2)!
    clock_value = ctx.clock.now()    # (3)!

    return {"impulses": 42, "read_at": clock_value}

1. Access the application Settings instance (or your custom subclass).
2. The device name as registered — "counter" in this case.
3. The monotonic clock port — useful for timing calculations. In tests, this is a FakeClock you control directly.

Resolving Adapters¶

When your telemetry device needs hardware access, use the adapter pattern:

from gas2mqtt.ports import GasMeterPort

@app.telemetry("counter", interval=60)
async def counter(ctx: cosalette.DeviceContext) -> dict[str, object]:
    meter = ctx.adapter(GasMeterPort)  # (1)!
    reading = meter.read_impulses()
    return {"impulses": reading}

1. Resolves the adapter registered for GasMeterPort. Raises LookupError if no adapter is registered. See Hardware Adapters for registration.

Multiple Sensors in One App¶

A single app can register multiple telemetry devices, each with its own interval:

import cosalette
from gas2mqtt.ports import GasMeterPort

app = cosalette.App(name="gas2mqtt", version="1.0.0")


@app.telemetry("counter", interval=60)
async def counter(ctx: cosalette.DeviceContext) -> dict[str, object]:
    """Read impulse count every 60 seconds."""
    meter = ctx.adapter(GasMeterPort)
    return {"impulses": meter.read_impulses()}


@app.telemetry("temperature", interval=30)
async def temperature(ctx: cosalette.DeviceContext) -> dict[str, object]:
    """Read the meter's temperature sensor every 30 seconds."""
    meter = ctx.adapter(GasMeterPort)
    return {"celsius": meter.read_temperature()}


app.run()

Each telemetry device runs as an independent asyncio task. They share the same MQTT connection and adapter instances, but their polling loops are completely independent. If temperature fails, counter keeps running.

Topic layout:

Imperative Registration¶

The @app.telemetry decorator works great when the handler is defined in the same module as the App. When the handler lives in a separate module — a sensor library, a shared utility, or a generated function — the decorator forces you to write a pass-through wrapper:

from my_sensors import read_temperature

@app.telemetry("temperature", interval=30)
async def temperature(ctx: cosalette.DeviceContext) -> dict[str, object]:
    return await read_temperature(ctx)  # just forwarding

The app.add_telemetry() method eliminates the wrapper — register the imported function directly:

from my_sensors import read_temperature

app.add_telemetry("temperature", read_temperature, interval=30)

Full Signature¶

app.add_telemetry(
    name,                    # device name (always required — no root device)
    func,                    # async callable returning dict | None
    *,
    interval=0.0,            # polling interval in seconds (required unless schedule=)
    schedule=None,           # cron string or CronSchedule (mutually exclusive with interval=)
    schedule_spec=None,      # per-device callable CronSpec — (per_device_config) -> str | CronSchedule (use with name=callable)
    publish=None,            # optional PublishStrategy
    persist=None,            # optional PersistPolicy
    init=None,               # optional synchronous factory
    enabled=True,            # False to skip registration entirely
    group=None,              # coalescing group name (requires interval=, not schedule=)
    triggerable=False,       # trigger source: "mqtt" (== True), "local", or "both"
    min_interval=None,       # storm throttle in seconds for triggered runs (requires triggerable=)
    retry=0,                 # max retry attempts (0 = disabled)
    retry_on=None,           # exception types to retry on
    backoff=None,            # BackoffStrategy (default: ExponentialBackoff)
    circuit_breaker=None,    # optional CircuitBreaker
    # timeout omitted → auto-defaults to interval; timeout=None → disabled
)

All keyword parameters behave identically to the decorator form.

Using init=¶

init= works the same way as the decorator — pass a synchronous factory whose return value is injected by type:

from my_sensors import read_temperature
from cosalette.filters import Pt1Filter


def make_filter() -> Pt1Filter:
    return Pt1Filter(tau=5.0, dt=10.0)


app.add_telemetry(
    "temperature",
    read_temperature,
    interval=10,
    publish=cosalette.OnChange(threshold=0.5),
    init=make_filter,
)

Choosing Between Decorator and Imperative¶

/// admonition | Identical validation type: info

Both paths run the same registration logic — signature validation, init= type-collision checks, and duplicate-name detection happen identically whether you use the decorator or add_telemetry(). ///

/// admonition | Named devices only type: warning

add_telemetry() always requires a device name — root (unnamed) devices can only be created via the decorator. ///

Conditional Registration¶

Use enabled= to defer the decision until settings are available, without an import-time if block:

@app.telemetry(
    "temperature",
    interval=30,
    enabled=lambda settings: settings.enable_temperature,
)
async def temperature() -> dict[str, object]:
    return {"celsius": read_temp()}

If you prefer imperative registration, do the branch inside @app.on_configure after settings resolution:

@app.on_configure
def register_optional(settings: MySettings) -> None:
    if settings.enable_temperature:
        app.add_telemetry("temperature", temperature, interval=30)

Both patterns defer the decision until bootstrap. Avoid settings = app.settings at import time here; that bypasses deferred enabled= and can break --help, --env-file, and other settings-resolution flows.

/// admonition | Disabled devices are invisible type: info

When enabled=False, the device is not registered at all — it won't appear in MQTT topics, won't reserve a name slot, and won't consume resources at runtime. ///

Publish Strategies¶

By default, every probe result is published to MQTT. Publish strategies let you decouple the probing frequency from the publishing frequency — the handler runs on interval, but only selected results are actually sent.

Basic Usage¶

from cosalette import Every, OnChange

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

Without publish=, the behaviour is exactly as before — every result is published.

Available Strategies¶

Composing Strategies¶

Combine strategies with | (OR) and & (AND):

# Publish on change OR every 5 minutes (heartbeat guarantee)
@app.telemetry("temp", interval=10, publish=OnChange() | Every(seconds=300))
async def temp() -> dict[str, object]:
    return {"celsius": await read_sensor()}

# Publish only when changed AND at least 30s have passed (debounce)
@app.telemetry("temp", interval=10, publish=OnChange() & Every(seconds=30))
async def temp() -> dict[str, object]:
    return {"celsius": await read_sensor()}

• | (OR): publish if any strategy says yes — useful for change detection with a periodic heartbeat fallback.
• & (AND): publish only if all strategies agree — useful for debouncing rapid changes.

For threshold modes, comparison semantics, edge cases, and composition details, see Publish Strategies.

Returning None¶

Handlers can return None to suppress a single cycle, independently of any strategy. The strategy is not consulted for None returns, and the "last published" value is not updated.

@app.telemetry("counter", interval=5, publish=OnChange())
async def counter(ctx: cosalette.DeviceContext) -> dict[str, object] | None:
    meter = ctx.adapter(GasMeterPort)
    if not meter.is_ready():
        return None  # skips this cycle entirely
    return {"impulses": meter.read_impulses()}

Initialisation Callbacks (init=)¶

When a telemetry handler needs per-device state — such as a filter instance, a calibration table, or a connection pool — the init= parameter provides a clean way to create it once and inject it into every poll cycle.

Without init=, you'd resort to module-level globals or closures. init= keeps state creation explicit, co-located with the decorator, and testable in isolation.

Basic Usage¶

class SmoothingFilter:
    """Moving-average filter for noisy sensor readings."""

    def __init__(self, window: int = 5) -> None:
        self.readings: list[float] = []
        self.window = window

    def update(self, value: float) -> float:
        self.readings.append(value)
        if len(self.readings) > self.window:
            self.readings.pop(0)
        return sum(self.readings) / len(self.readings)


def make_filter() -> SmoothingFilter:  # (1)!
    return SmoothingFilter(window=10)


@app.telemetry("temperature", interval=30, init=make_filter)  # (2)!
async def temperature(smoother: SmoothingFilter) -> dict[str, object]:  # (3)!
    raw = read_sensor()
    return {"celsius": smoother.update(raw)}

1. The factory is a plain synchronous callable. It runs once before the first poll cycle — not on every interval.
2. init=make_filter tells the framework to call make_filter() and inject the result into the handler.
3. The handler declares smoother: SmoothingFilter — the framework matches the return type of the init callback to this parameter automatically.

How It Works¶

1. The framework calls init() once before the handler's polling loop starts.
2. The return value is added to the dependency-injection provider map, keyed by its type.
3. Any handler parameter whose type annotation matches the init result type receives the same instance on every invocation.
4. The init callback can itself receive injected parameters (e.g. Settings) — the same DI machinery used for handler parameters.

Combining with Filters and Strategies¶

init= pairs naturally with the framework's built-in filters and publish strategies. Use init= to create the filter instance, and publish= to control when results are sent:

from cosalette import OnChange
from cosalette.filters import Pt1Filter


def make_pt1() -> Pt1Filter:
    return Pt1Filter(tau=5.0, dt=10.0)


@app.telemetry(
    "temperature",
    interval=10,
    publish=OnChange(threshold=0.5),
    init=make_pt1,
)
async def temperature(pt1: Pt1Filter) -> dict[str, object]:
    raw = await read_sensor()
    return {"celsius": round(pt1.update(raw), 1)}

Compare this to a module-level pt1 = Pt1Filter(...) pattern — init= achieves the same result but scopes the filter to the device registration, making it explicit which device owns the state.

Rules and Constraints¶

• Synchronous only — async def init callbacks raise TypeError at decoration time. The callback runs during bootstrap, before the async event loop processes device tasks.
• Type collision guard — if the init callback returns a type the framework already provides (Settings, DeviceContext, Logger, ClockPort, Event), a TypeError is raised immediately. Use a wrapper class if you need to inject something with a colliding type.
• Fail-fast validation — bad signatures (e.g. un-annotated parameters) are caught at decoration time, not at runtime.

Signal Filters¶

Filters are handler-level data transformations that smooth or clean sensor readings before they reach publish strategies. Unlike strategies that control when to publish, filters control what is published. They implement the Filter protocol (update(value) -> float) and compose naturally.

Available Filters¶

cosalette ships three filter implementations in cosalette.filters:

Example: PT1 Filter with init=¶

from cosalette import OnChange
from cosalette.filters import Pt1Filter


def make_pt1() -> Pt1Filter:
    return Pt1Filter(tau=5.0, dt=10.0)


@app.telemetry(
    "temperature",
    interval=10,
    publish=OnChange(threshold=0.5),
    init=make_pt1,
)
async def temperature(pt1: Pt1Filter) -> dict[str, object]:
    raw = await read_sensor()
    return {"celsius": round(pt1.update(raw), 1)}

For algorithm details, parameter tuning, and the decision table, see Signal Filters.

Persistence¶

Telemetry devices can persist state across restarts using the store= and persist= parameters.

Basic Usage¶

Pass a Store backend to the app, then declare store: DeviceStore in your handler:

import cosalette
from cosalette import JsonFileStore, DeviceStore, SaveOnPublish

app = cosalette.App(
    "myapp", "1.0.0",
    store=JsonFileStore("./data/state.json"),
)

@app.telemetry("counter", interval=30, persist=SaveOnPublish())
async def counter(store: DeviceStore) -> dict[str, object]:
    store["total"] = store.get("total", 0) + 1
    return {"total": store["total"]}

Available Save Policies¶

Policies compose with | (OR) and & (AND):

persist = SaveOnPublish() | SaveOnChange()  # save on either condition

Combining with Other Features¶

Persistence works seamlessly with publish strategies, filters, and init callbacks:

from cosalette import DeviceStore, OnChange, Pt1Filter, SaveOnPublish

@app.telemetry(
    "sensor",
    interval=10,
    publish=OnChange(threshold=0.5),
    persist=SaveOnPublish(),
    init=lambda: Pt1Filter(tau=2.0, dt=10.0),
)
async def sensor(
    store: DeviceStore,
    lpf: Pt1Filter,
) -> dict[str, object]:
    raw = 21.5  # e.g. from an adapter
    filtered = lpf.update(raw)
    store["last_value"] = filtered
    return {"value": filtered}

For full details, see the Persistence concept.

Typed Telemetry Returns¶

Instead of returning a raw dict, you can return a Pydantic model. The framework serializes it via Pydantic TypeAdapter (JSON-mode serialization) before publishing. This supports BaseModel, stdlib dataclasses, TypedDict, and primitives — not just BaseModel. state_model= on the decorator drives normalization when declared, otherwise the return annotation does:

from pydantic import BaseModel

class SensorReading(BaseModel):
    celsius: float
    humidity: float

@app.telemetry("climate", interval=60, state_model=SensorReading)
async def climate(ctx: cosalette.DeviceContext) -> SensorReading:
    return SensorReading(celsius=21.5, humidity=58.0)

Primitive / list returns are wrapped as {"value": ...} automatically. Return None to suppress a cycle as usual.

Contract Metadata¶

Every @app.telemetry registration can carry contract metadata — optional fields that describe what the device produces and how it behaves. The metadata is surfaced by cosalette manifest and the MCP server; it has no runtime effect.

from pydantic import BaseModel
import cosalette

class CounterReading(BaseModel):
    impulses: int
    temperature_celsius: float

@app.telemetry(
    "counter",
    interval=cosalette.setting_ref("poll_interval"),  # (1)!
    triggerable=True,
    summary="Gas meter impulse count and ambient temperature",  # (2)!
    state_model=CounterReading,                                  # (3)!
    behavior=["reads serial port", "applies outlier rejection"],  # (4)!
    effects=["updates Home Assistant energy dashboard"],          # (5)!
)
async def counter(ctx: cosalette.DeviceContext):  # (6)!
    meter = ctx.adapter(GasMeterPort)
    return {"impulses": meter.read_impulses(), "temperature_celsius": meter.read_temperature()}

1. setting_ref("poll_interval") exposes the field name in the manifest. A raw lambda s: s.poll_interval works identically at runtime but shows "<deferred>" in manifest output.
2. summary is a one-line human-readable description of what this device reports.
3. state_model declares the shape of the published JSON. Since 0.9.0 it is a runtime contract, not documentation: the return value is validated against it before publishing.
4. behavior lists ordered operational steps; each string is one bullet in the manifest.
5. effects lists side effects and downstream mutations.
6. No return annotation: state_model= outranks it anyway, and a loose -> dict[str, object] alongside state_model= warns at registration since 0.9.0.

Use payload_model on triggerable telemetry to document the JSON shape accepted on the /set topic:

class RefreshRequest(BaseModel):
    days: int = 7

@app.telemetry(
    "counter",
    interval=cosalette.setting_ref("poll_interval"),
    triggerable=True,
    state_model=CounterReading,
    payload_model=RefreshRequest,   # shape accepted on /set
)
async def counter(): ...

For the full contract-first workflow — including the read/write split pattern, setting_ref inspectability, and viewing the manifest — see the Contract-First Route Design guide.

Practical Example: Gas Meter Impulse Counter¶

Here's a complete, realistic telemetry device for a gas meter with a reed switch impulse sensor:

"""gas2mqtt — Gas meter impulse counter bridge."""

from __future__ import annotations

from typing import Protocol, runtime_checkable

import cosalette
from pydantic import Field
from pydantic_settings import SettingsConfigDict


# --- Port (Protocol) for hardware abstraction ---

@runtime_checkable
class GasMeterPort(Protocol):
    """Hardware abstraction for gas meter impulse sensors."""

    def read_impulses(self) -> int: ...
    def read_temperature(self) -> float: ...


# --- Settings ---

class Gas2MqttSettings(cosalette.Settings):
    model_config = SettingsConfigDict(
        env_prefix="GAS2MQTT_",
        env_nested_delimiter="__",
        env_file=".env",
        env_file_encoding="utf-8",
    )
    serial_port: str = Field(default="/dev/ttyUSB0")
    poll_interval: int = Field(default=60, ge=1)


# --- App ---

app = cosalette.App(
    name="gas2mqtt",
    version="1.0.0",
    settings_class=Gas2MqttSettings,
)


# --- Telemetry device ---

@app.telemetry("counter", interval=app.settings.poll_interval)  # (1)!
async def counter(ctx: cosalette.DeviceContext) -> dict[str, object]:
    """Read gas meter impulses and publish state."""
    meter = ctx.adapter(GasMeterPort)
    impulses = meter.read_impulses()
    temp = meter.read_temperature()

    return {
        "impulses": impulses,
        "temperature_celsius": temp,
    }


app.run()

1. app.settings is available at decoration time because App.__init__ eagerly instantiates the settings class. The poll_interval value here reflects environment variables and .env files — no hardcoded constants needed.

Error Behaviour¶

When a telemetry function raises an exception, the framework applies state-transition deduplication:

1. First error — caught, logged at ERROR level, and published to gas2mqtt/error and gas2mqtt/counter/error. The device health status in the heartbeat is set to "error".
2. Repeated same-type errors — suppressed. No additional MQTT publishes until the error type changes. This prevents flooding the broker when a sensor is persistently broken.
3. Different error type — treated as a new error: published and logged.
4. Recovery — when the next poll succeeds after a failure, recovery is logged at INFO level and the device health status is restored to "ok" in the heartbeat.
5. Continues the polling loop — the next interval always runs.

This means transient failures (sensor timeouts, I/O glitches) are self-healing. The daemon stays up and retries on the next cycle. Persistent failures produce a single error event instead of flooding MQTT with identical messages every interval.

@app.telemetry("counter", interval=60)
async def counter(ctx: cosalette.DeviceContext) -> dict[str, object]:
    meter = ctx.adapter(GasMeterPort)
    reading = meter.read_impulses()  # (1)!
    if reading < 0:
        raise ValueError(f"Invalid impulse count: {reading}")  # (2)!
    return {"impulses": reading}

1. If read_impulses() raises OSError, the framework catches it and publishes an error payload. The loop continues.
2. You can also raise explicitly — the framework treats it the same way.

Advanced Topics¶

For coalescing groups, cron scheduling, retry/backoff resilience, and triggerable on-demand reads, see Advanced Telemetry Techniques.

For lookup tables — interval guidelines, cron syntax, backoff strategy comparison, and the timeout backstop semantics — see Telemetry Reference.

Using Telemetry with Router¶

For multi-module applications, define telemetry devices on a Router instead of directly on App:

import cosalette

router = cosalette.Router(prefix="sensors", tags=["environment"])


@router.telemetry("temperature", interval=30)
async def read_temperature() -> dict[str, object]:
    """Read I2C temperature sensor."""
    return {"celsius": 22.5}


@router.telemetry("humidity", interval=30)
async def read_humidity() -> dict[str, object]:
    """Read I2C humidity sensor."""
    return {"percent": 55.0}

import cosalette
from sensors import router as sensors_router

app = cosalette.App(name="home2mqtt", version="1.0.0")
app.include_router(sensors_router)

if __name__ == "__main__":
    app.run()

MQTT topics:

• home2mqtt/sensors/temperature/state
• home2mqtt/sensors/humidity/state

All telemetry features (publish strategies, filters, retry, triggerable, typed returns) work identically on routers. See Router Composition for full multi-module patterns.

See Also¶

• Advanced Telemetry Techniques — coalescing groups, cron scheduling, retry/backoff, and triggerable telemetry
• Telemetry Reference — interval guidelines, cron syntax, backoff strategy tables, timeout backstop
• Router Composition — organize telemetry devices in multi-module apps
• Device Archetypes — telemetry vs command archetypes
• MQTT Topics — the {prefix}/{device}/state topic layout
• Architecture — how devices fit into the framework
• Publish Strategies — publishing control concepts
• Signal Filters — handler-level data transformations
• ADR-010 — the decision behind device archetypes
• ADR-013 — the decision behind publish strategies
• ADR-014 — the decision behind signal filters
• ADR-018 — the decision behind coalescing groups
• ADR-024 — the decision behind retry/backoff
• ADR-032 — the decision behind cron scheduling and wall-clock sleep