API Reference¶

def registered_names(self) -> frozenset[str]:
    """Collect registered device/telemetry/command/periodic names."""
    all_regs = (
        self._devices,
        self._telemetry,
        self._commands,
        self._periodic,
        self._streams,
    )
    return frozenset(r.name for regs in all_regs for r in regs)

def include_router(
    self,
    router: Router,
    *,
    prefix: str | None = None,
    tags: list[str] | None = None,
    dependencies: list[Any] | None = None,
    adapters: dict[
        type,
        type
        | str
        | Callable[..., object]
        | tuple[
            type | str | Callable[..., object],
            type | str | Callable[..., object],
        ],
    ]
    | None = None,
) -> None:
    """Include a router's registrations in this application.

    Applies snapshot semantics: registrations are captured at call time.
    Later mutations to the router do not affect prior inclusions.
    Multiple inclusions with different prefixes are allowed.

    Args:
        router: Router instance to include.
        prefix: Optional single MQTT topic segment prepended to all
            router operation names.  Must not contain ``/``, ``+``,
            ``#``, or NUL.  Combined with router's own prefix.
        tags: Additional tags applied to all router operations.
            Accumulates in order: router constructor → include_router → operation.
        dependencies: Reserved for cos-ebc.  Must be None or empty.
        adapters: Adapter declarations merged into the app's registry.
            Same shape as ``App(adapters=...)``.  Conflicts (same port
            type already registered) raise ValueError at include time.

    Raises:
        ValueError: If *prefix* contains MQTT special characters.
        ValueError: If an adapter port type conflict is detected.
        NotImplementedError: If *dependencies* is not None or empty.

    See Also:
        ADR-044 — Public Router and composition API.

    Example::

        # sensors.py
        router = cosalette.Router(prefix="sensors")

        @router.telemetry("temperature", interval=30)
        async def read_temperature() -> dict:
            return {"celsius": 22.5}

        # main.py
        app = cosalette.App("bridge")
        app.include_router(router, tags=["production"])
        # → publishes to: bridge/sensors/temperature/state
    """
    if prefix is not None:
        validate_mqtt_name(prefix)

    if dependencies is not None and len(dependencies) > 0:
        msg = (
            "dependencies= is reserved for the cos-ebc epic "
            "and is not yet implemented. Pass None or omit the parameter."
        )
        raise NotImplementedError(msg)

    # Compute combined prefix
    combined_prefix = self._compute_combined_prefix(router._prefix, prefix)

    # Merge adapters first (fail fast on conflicts)
    self._merge_include_adapters(adapters)
    self._merge_router_adapters(router)

    # Include tags: accumulate router constructor → include_router → operation
    include_tags = list(tags) if tags is not None else []

    # Snapshot existing names for collision detection across all types
    existing_names: set[str] = set(self.registered_names())

    # Copy standard registrations with prefix/tag transformations
    self._copy_standard_registrations(
        router, combined_prefix, router._tags, include_tags, existing_names
    )

    # Handle periodic registrations: apply prefix, accumulate tags, check collisions
    for reg in router._periodic:
        accumulated_tags = self._accumulate_tags(
            router._tags, include_tags, reg.tags
        )
        new_name = self._apply_prefix(reg.name, combined_prefix)
        if new_name in existing_names:
            msg = f"Name {new_name!r} is already registered on the app"
            raise ValueError(msg)
        existing_names.add(new_name)
        new_reg = replace(reg, name=new_name, tags=tuple(accumulated_tags))
        self._periodic.append(new_reg)

    # Merge reactors with validation
    self._merge_reactors(router)

def adapter[T](self, port_type: type[T]) -> T:
    """Resolve an adapter by port type.

    Args:
        port_type: The Protocol type to look up.

    Returns:
        The adapter instance registered for that port type.

    Raises:
        LookupError: If no adapter is registered for the port type.
    """
    try:
        return cast(T, self._adapters[port_type])
    except KeyError:
        msg = f"No adapter registered for {port_type!r}"
        raise LookupError(msg) from None

def get_command_handler(
    self,
    sub_topic: str | None = None,
) -> CommandHandler | None:
    """Look up the command handler for a sub-topic (or root)."""
    return self._command_handlers.get(sub_topic)

async def publish_state(
    self,
    payload: dict[str, object],
    *,
    retain: bool = True,
) -> None:
    """Publish device state to ``{prefix}/{device}/state`` as JSON.

    For root devices (unnamed), publishes to ``{prefix}/state`` instead.

    This is the primary publication method for device telemetry.
    The payload dict is JSON-serialised automatically.

    Args:
        payload: Dict to serialise as JSON.
        retain: Whether the message should be retained (default True).
    """
    topic = f"{self._topic_base}/state"
    await self._mqtt.publish(topic, payload, retain=retain, qos=1)

async def publish(
    self,
    channel: str,
    payload: str,
    *,
    retain: bool = False,
    qos: int = 1,
) -> None:
    """Publish to an arbitrary sub-channel: ``{prefix}/{device}/{channel}``.

    For root devices (unnamed), publishes to ``{prefix}/{channel}`` instead.

    Escape hatch for non-standard topics. Prefer publish_state() for
    normal device state updates.
    """
    topic = f"{self._topic_base}/{channel}"
    await self._mqtt.publish(topic, payload, retain=retain, qos=qos)

async def sleep(self, seconds: float) -> None:
    """Shutdown-aware sleep.

    Returns early (without exception) if shutdown is requested during
    the sleep period. This enables the idiomatic pattern::

        while not ctx.shutdown_requested:
            await ctx.sleep(10)
            # ... do work ...
    """
    if self._shutdown_event.is_set():
        return

    sleep_task = asyncio.ensure_future(self._clock.sleep(seconds))
    shutdown_task = asyncio.ensure_future(self._shutdown_event.wait())

    done, pending = await asyncio.wait(
        {sleep_task, shutdown_task},
        return_when=asyncio.FIRST_COMPLETED,
    )

    for task in pending:
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task

async def sleep_until(
    self,
    target: datetime.time | Sequence[datetime.time],
    *,
    tz: datetime.tzinfo | None = None,
) -> None:
    """Shutdown-aware sleep until a wall-clock time.

    Sleeps until the next occurrence of *target* (or the nearest
    upcoming time if a sequence is given).  Uses local timezone
    when *tz* is ``None``.

    Returns early (without exception) if shutdown is requested
    during the sleep, via :meth:`sleep`.

    Example — poll twice daily at 06:00 and 18:00 local time::

        while not ctx.shutdown_requested:
            data = await read_data()
            await ctx.publish_state(data)
            await ctx.sleep_until([time(6, 0), time(18, 0)])

    Args:
        target: A :class:`datetime.time` or sequence of times to
            sleep until.
        tz: Timezone for interpreting *target*.  ``None`` (default)
            uses the system's local timezone.

    Raises:
        ValueError: If *target* is an empty sequence.

    See Also:
        ADR-032 — Wall-clock scheduling design.
    """
    seconds = _seconds_until(target, tz=tz)
    await self.sleep(seconds)

@contextlib.asynccontextmanager
async def sub_entity(self, name: str) -> AsyncIterator[SubEntityContext]:
    """Scoped sub-entity lifecycle with automatic availability.

    Publishes ``"online"`` on enter and ``"offline"`` on exit to
    ``{topic_base}/{name}/availability``.  Clears retained state
    on exit by publishing an empty payload to the state topic.

    Args:
        name: Sub-entity name (single MQTT topic level).

    Yields:
        A :class:`SubEntityContext` scoped to the sub-entity's topics.

    Raises:
        ValueError: If the name fails validation.

    See Also:
        ADR-031 — Sub-entity context manager.
    """
    from cosalette._context._sub_entity_context import SubEntityContext

    self._validate_sub_entity_name(name)
    self._active_sub_entities.add(name)
    sub = SubEntityContext(name=name, parent=self)
    avail_topic = f"{self._topic_base}/{name}/availability"
    try:
        await self._mqtt.publish(avail_topic, "online", retain=True, qos=1)
    except Exception:
        self._active_sub_entities.discard(name)
        raise
    try:
        yield sub
    finally:
        try:
            state_topic = f"{self._topic_base}/{name}/state"
            await self._mqtt.publish(state_topic, "", retain=True, qos=1)
            await self._mqtt.publish(avail_topic, "offline", retain=True, qos=1)
        finally:
            self._active_sub_entities.discard(name)

def on_command(
    self,
    handler_or_sub_topic: CommandHandler | str | None = None,
    /,
) -> CommandHandler | Callable[[CommandHandler], CommandHandler]:
    """Register a command handler for this device.

    Supports three call patterns:

    1. Decorator — root handler::

        @ctx.on_command
        async def handle(sub_topic: str | None, payload: str) -> None: ...

    2. Direct call — root handler::

        ctx.on_command(handle)

    3. Decorator factory — sub-topic handler::

        @ctx.on_command("calibrate")
        async def handle_cal(sub_topic: str | None, payload: str) -> None: ...

    Handlers may also accept a single :class:`Command` argument
    (new-style), detected automatically via type annotation::

        @ctx.on_command
        async def handle(cmd: Command) -> None: ...

    Raises:
        RuntimeError: If a handler is already registered for the same
            sub-topic, or if ``commands()`` is active and a root
            handler is being registered.
        ValueError: If the sub-topic string is empty or contains
            ``/``, ``+``, or ``#``.

    Returns:
        The handler unchanged when called with a callable, or a
        decorator function when called with a sub-topic string or None.
    """

    def _register(
        handler: CommandHandler,
        sub_topic: str | None,
    ) -> CommandHandler:
        if sub_topic is None and self._commands_consumed:
            msg = (
                f"Cannot register on_command — commands() iterator already "
                f"active for device '{self._name}'"
            )
            raise RuntimeError(msg)
        if sub_topic in self._command_handlers:
            label = f"sub-topic '{sub_topic}'" if sub_topic else "root"
            msg = (
                f"Command handler already registered for {label} "
                f"on device '{self._name}'"
            )
            raise RuntimeError(msg)
        self._command_handlers[sub_topic] = handler
        return handler

    # --- callable → register as root handler immediately ---
    if callable(handler_or_sub_topic):
        return _register(handler_or_sub_topic, None)  # ty: ignore[invalid-argument-type]

    # --- string → validate and return decorator for that sub-topic ---
    if isinstance(handler_or_sub_topic, str):
        if not handler_or_sub_topic:
            msg = "Sub-topic must not be empty"
            raise ValueError(msg)
        _invalid = set(handler_or_sub_topic) & {"/", "+", "#"}
        if _invalid:
            msg = (
                f"Sub-topic contains invalid MQTT characters "
                f"{_invalid}: '{handler_or_sub_topic}'"
            )
            raise ValueError(msg)
        sub = handler_or_sub_topic

        def _decorator(handler: CommandHandler) -> CommandHandler:
            return _register(handler, sub)

        return _decorator

    # --- None → return decorator for root handler ---
    def _root_decorator(handler: CommandHandler) -> CommandHandler:
        return _register(handler, None)

    return _root_decorator

def commands(
    self,
    timeout: float | None = None,
) -> AsyncIterator[Command | None]:
    """Async iterator that yields inbound commands from the internal queue.

    Provides a queue-backed async iterator for ``@app.device`` loops,
    eliminating the need for manual ``asyncio.Queue`` bridges.

    When *timeout* is provided, yields ``None`` on timeout expiry,
    enabling periodic-work patterns::

        async for cmd in ctx.commands(timeout=5):
            if cmd is None:
                await periodic_check()
            else:
                await process(cmd.payload)

    Without *timeout*, blocks until a command arrives or shutdown is
    requested. Shutdown is detected immediately via event racing.

    Args:
        timeout: Seconds to wait for a command before yielding None.
            When None (default), blocks indefinitely until a command
            arrives or shutdown is requested.

    Yields:
        Command when a command arrives, or None on timeout expiry.

    Raises:
        RuntimeError: If called more than once on the same context,
            or if a command handler is already registered via
            :meth:`on_command`.

    See Also:
        ADR-025 — Command channel and sub-topic routing.
    """
    if self._commands_consumed:
        msg = f"commands() already active for device '{self._name}'"
        raise RuntimeError(msg)
    if None in self._command_handlers:
        msg = (
            f"Cannot use commands() — on_command handler already "
            f"registered for device '{self._name}'"
        )
        raise RuntimeError(msg)
    self._commands_consumed = True

    async def _iter() -> AsyncIterator[Command | None]:
        while not self.shutdown_requested:
            result = await self._await_command(timeout)
            if result is not None:
                yield result
            elif self.shutdown_requested:
                break
            elif timeout is not None:
                yield None
        # Drain any commands that arrived before/during shutdown
        while not self._command_queue.empty():
            yield self._command_queue.get_nowait()

    return _iter()

def adapter[T](self, port_type: type[T]) -> T:
    """Resolve an adapter by port type.

    Args:
        port_type: The Protocol type to look up.

    Returns:
        The adapter instance registered for that port type.

    Raises:
        LookupError: If no adapter is registered for the port type.
    """
    try:
        return cast(T, self._adapters[port_type])
    except KeyError:
        msg = f"No adapter registered for {port_type!r}"
        raise LookupError(msg) from None

async def publish_state(
    self,
    payload: dict[str, object],
    *,
    retain: bool = True,
) -> None:
    """Publish sub-entity state to ``{device}/{name}/state`` as JSON.

    Args:
        payload: Dict to serialise as JSON.
        retain: Whether the message should be retained (default True).
    """
    topic = f"{self.parent._topic_base}/{self.name}/state"
    await self.parent._mqtt.publish(topic, payload, retain=retain, qos=1)

def on_command(
    self,
    handler: CommandHandler,
) -> CommandHandler:
    """Register a command handler for this sub-entity's sub-topic.

    Delegates to the parent device's :meth:`~DeviceContext.on_command`
    with this sub-entity's name as the sub-topic.

    Args:
        handler: Async callable to handle inbound commands.

    Returns:
        The handler, unchanged.
    """
    return self.parent.on_command(self.name)(handler)