Devices¶

def make_gas_counter(
    settings: Gas2MqttSettings,
    store: DeviceStore,
    logger: logging.Logger,
) -> GasCounterState:
    """Factory — creates stateful domain objects.

    Called once by the @app.state provider before the telemetry loop.
    The returned GasCounterState is injected into gas_counter() on every
    scheduled and triggered run.
    """
    store.load()
    trigger = SchmittTrigger(settings.trigger_level, settings.trigger_hysteresis)
    counter = _restore_counter(store, logger)
    consumption = _restore_consumption(store, settings, logger)
    return GasCounterState(trigger, counter, consumption, store)

async def gas_counter(
    state: GasCounterState,
    magnetometer: MagnetometerPort,
    logger: logging.Logger,
) -> dict[str, object] | None:
    """Gas counter telemetry handler (read-only).

    Called periodically to read magnetometer and count ticks.
    Returns dict to publish, or None to skip.
    """
    return _poll(state, magnetometer, logger)

async def update_consumption(
    payload: str,
    state: GasCounterState,
    logger: logging.Logger,
) -> dict[str, object] | None:
    """Handle inbound MQTT command to set consumption value.

    Expects JSON payload: {"consumption_m3": <float>}
    Publishes updated state on success; returns None if disabled or invalid.
    """
    if state.consumption is None:
        logger.warning("Consumption command received but tracking is disabled")
        return None

    if not payload.strip():
        return None

    try:
        data = json.loads(payload)
    except json.JSONDecodeError:
        logger.warning("Ignoring invalid JSON in consumption command")
        return None

    if not isinstance(data, dict):
        logger.warning("Ignoring non-object JSON in consumption command")
        return None

    if "consumption_m3" in data:
        consumption_m3 = _parse_consumption_m3(data["consumption_m3"], logger)
        if consumption_m3 is None:
            return None
        state.consumption.set_consumption(consumption_m3)
        logger.info("Consumption set to %.3f m³", state.consumption.consumption_m3)
        state.stage_state()
        return state.build_state()
    return None

def make_pt1(settings: Gas2MqttSettings) -> Pt1Filter:
    """Create PT1 filter from settings for temperature smoothing.

    This is the ``init=`` factory: cosalette calls it once before the
    handler loop and injects the returned ``Pt1Filter`` by type.
    """
    return Pt1Filter(tau=settings.smoothing_tau, dt=settings.temperature_interval)

async def temperature(
    magnetometer: MagnetometerPort,
    settings: Gas2MqttSettings,
    pt1: Pt1Filter,
) -> dict[str, object]:
    """Read temperature, calibrate, filter, and return state dict.

    Args:
        magnetometer: Sensor adapter (injected by cosalette DI).
        settings: Application settings with calibration coefficients.
        pt1: PT1 filter instance (created by :func:`make_pt1`).

    Returns:
        ``{"temperature": <rounded float>}`` — published to MQTT by
        the framework when the ``OnChange`` strategy fires.
    """
    reading = magnetometer.read()
    raw_celsius = settings.temp_scale * reading.temperature_raw + settings.temp_offset
    return {"temperature": round(pt1.update(raw_celsius), 1)}

async def magnetometer(
    magnetometer: MagnetometerPort,
) -> dict[str, object]:
    """Read and return raw magnetic field values.

    Args:
        magnetometer: Sensor adapter (injected by cosalette DI).

    Returns:
        ``{"bx": int, "by": int, "bz": int}`` — published to MQTT
        on every poll cycle.
    """
    reading = magnetometer.read()
    return {"bx": reading.bx, "by": reading.by, "bz": reading.bz}