Skip to content

Reference

Complete lookup reference — settings, MQTT topics, commands, CLI options, error types, and API documentation.


Settings Reference

All settings are read from environment variables prefixed with JEELINK2MQTT_. A .env file in the working directory is loaded automatically. Nested settings use __ as delimiter (e.g. JEELINK2MQTT_MQTT__HOST).

Application Settings

Setting Env Variable Type Default Description
serial_port JEELINK2MQTT_SERIAL_PORT str /dev/ttyUSB0 Serial port path (must start with /dev/)
baud_rate JEELINK2MQTT_BAUD_RATE int 57600 Serial baud rate
sensors JEELINK2MQTT_SENSORS list[object] [] Sensor definitions (JSON array)
staleness_timeout_seconds JEELINK2MQTT_STALENESS_TIMEOUT_SECONDS float 600.0 Global staleness timeout in seconds (min: 60)
median_filter_window JEELINK2MQTT_MEDIAN_FILTER_WINDOW int 7 Median filter window size (3–21, must be odd)
heartbeat_interval_seconds JEELINK2MQTT_HEARTBEAT_INTERVAL_SECONDS float 180.0 Heartbeat re-publish interval in seconds (min: 10)

Inherited cosalette Settings

Setting Env Variable Type Default Description
mqtt.host JEELINK2MQTT_MQTT__HOST str localhost MQTT broker hostname
mqtt.port JEELINK2MQTT_MQTT__PORT int 1883 MQTT broker port
mqtt.username JEELINK2MQTT_MQTT__USERNAME str "" MQTT username
mqtt.password JEELINK2MQTT_MQTT__PASSWORD str "" MQTT password

Validators

Field Constraint
serial_port Must start with /dev/
median_filter_window Must be odd, between 3 and 21
staleness_timeout_seconds Minimum 60.0
heartbeat_interval_seconds Minimum 10.0

Sensor Configuration Fields

Each entry in the JEELINK2MQTT_SENSORS JSON array supports:

Field Type Required Default Description
name str Yes Logical sensor name (e.g. "office", "outdoor")
temp_offset float No 0.0 Calibration offset added to temperature (°C)
humidity_offset float No 0.0 Calibration offset added to humidity (percentage points)
staleness_timeout float \| null No null Per-sensor staleness override in seconds (null = use global)

Example:

JEELINK2MQTT_SENSORS='[
  {"name": "office", "temp_offset": -0.5, "humidity_offset": 2.0},
  {"name": "outdoor", "staleness_timeout": 900},
  {"name": "bedroom"}
]'

MQTT Topic Map

Topic Direction Retained Payload
jeelink2mqtt/{sensor}/state Out Yes {temperature, humidity, low_battery, timestamp}
jeelink2mqtt/{sensor}/availability Out Yes "online" or "offline"
jeelink2mqtt/raw/state Out No {sensor_id, temperature, humidity, low_battery, timestamp}
jeelink2mqtt/mapping/state Out Yes {sensor_name: {sensor_id, mapped_at, last_seen}}
jeelink2mqtt/mapping/event Out No {event_type, sensor_name, old_sensor_id, new_sensor_id, timestamp, reason}
jeelink2mqtt/mapping/set In No {command, ...params}

Sensor State Payload

{
  "temperature": 21.3,
  "humidity": 52,
  "low_battery": false,
  "timestamp": "2026-03-04T10:15:00+00:00"
}

Mapping Event Payload

{
  "event_type": "auto_adopt",
  "sensor_name": "office",
  "old_sensor_id": null,
  "new_sensor_id": 42,
  "timestamp": "2026-03-04T10:15:00+00:00",
  "reason": "Auto-adopted sensor ID 42 for 'office'"
}

Event types: auto_adopt, manual_assign, manual_reset, reset_all.


Command Reference

Commands are sent as JSON to jeelink2mqtt/mapping/set. Responses are published to jeelink2mqtt/mapping/state.

Command Parameters Description
assign sensor_name (str), sensor_id (int) Manually assign an ephemeral ID to a logical sensor name
reset sensor_name (str) Remove the mapping for a single sensor
reset_all (none) Clear all sensor mappings
list_unknown (none) List recently-seen unmapped sensor IDs

assign Response

{
  "status": "ok",
  "event": {
    "event_type": "manual_assign",
    "sensor_name": "office",
    "old_sensor_id": null,
    "new_sensor_id": 42,
    "reason": "Manually assigned sensor ID 42 to 'office'"
  }
}

reset Response

{
  "status": "ok",
  "event": {
    "event_type": "manual_reset",
    "sensor_name": "office",
    "old_sensor_id": 42
  }
}

reset_all Response

{
  "status": "ok",
  "cleared": 2,
  "sensors": ["office", "outdoor"]
}

list_unknown Response

{
  "status": "ok",
  "unknown_sensors": {
    "42": {
      "temperature": 21.3,
      "humidity": 55,
      "low_battery": false,
      "timestamp": "2026-03-04T10:15:00+00:00"
    }
  }
}

Error Responses

{"error": "Invalid JSON payload"}
{"error": "Unknown command: foo"}
{"error": "assign requires 'sensor_name' and 'sensor_id'"}
{"error": "Sensor ID 42 is already mapped to 'outdoor', cannot assign to 'office'"}
{"error": "Unknown sensor name 'foo' — must be one of the configured sensors"}

CLI Options

jeelink2mqtt uses the cosalette CLI framework (Typer-based). Available flags:

Flag Description
--dry-run Use fake adapter — no hardware required
--version Print version and exit
--log-level Set logging verbosity: DEBUG, INFO, WARNING, ERROR
--env-file Path to .env file (default: .env in working directory)
jeelink2mqtt --dry-run --log-level DEBUG --env-file /etc/jeelink2mqtt/.env

Error Types

Domain exceptions are mapped to MQTT error type strings for structured error reporting:

Exception MQTT Error Type Description
SerialConnectionError serial_connection JeeLink serial port unavailable or disconnected
FrameParseError frame_parse Received data doesn't match LaCrosse frame format
MappingConflictError mapping_conflict ID already assigned to another sensor
StalenessTimeoutError staleness_timeout Sensor hasn't sent readings within the staleness window
UnknownSensorError unknown_sensor Reading from an unrecognised/unmapped sensor ID

API Reference

Auto-generated from source docstrings.

Models

jeelink2mqtt.models

Domain models for jeelink2mqtt.

Immutable value objects representing sensor readings, configuration, mapping state, and mapping lifecycle events. All models use frozen dataclasses with __slots__ for memory efficiency and immutability guarantees.

SensorReading dataclass
SensorReading(
    sensor_id: int,
    temperature: float,
    humidity: int,
    low_battery: bool,
    timestamp: datetime = (lambda: now(UTC))(),
)

Raw reading received from the JeeLink USB receiver.

Represents a single LaCrosse frame decoded into typed fields. The sensor_id is ephemeral — it changes on every battery swap, so higher-level code must resolve it to a logical name via the sensor registry (see ADR-002).

sensor_id instance-attribute
sensor_id: int

Ephemeral LaCrosse sensor ID (changes on battery swap).

temperature instance-attribute
temperature: float

Temperature in degrees Celsius.

humidity instance-attribute
humidity: int

Relative humidity percentage (0–100).

low_battery instance-attribute
low_battery: bool

Battery warning flag from the sensor frame.

timestamp class-attribute instance-attribute
timestamp: datetime = field(
    default_factory=lambda: now(UTC)
)

When the reading was received (defaults to now in UTC).

SensorConfig dataclass
SensorConfig(
    name: str,
    temp_offset: float = 0.0,
    humidity_offset: float = 0.0,
    staleness_timeout: float | None = None,
)

Per-sensor configuration loaded from application settings.

Each configured sensor gets a logical name (e.g. "office", "outdoor") that remains stable across battery swaps and ID changes. Calibration offsets allow compensating for individual sensor inaccuracies.

name instance-attribute
name: str

Logical sensor name (e.g. "office", "outdoor").

temp_offset class-attribute instance-attribute
temp_offset: float = 0.0

Calibration offset added to temperature readings (°C).

humidity_offset class-attribute instance-attribute
humidity_offset: float = 0.0

Calibration offset added to humidity readings (percentage points).

staleness_timeout class-attribute instance-attribute
staleness_timeout: float | None = None

Per-sensor staleness override in seconds (None = use global).

SensorMapping dataclass
SensorMapping(
    sensor_id: int,
    sensor_name: str,
    mapped_at: datetime,
    last_seen: datetime,
)

Runtime mapping state for one sensor.

Tracks which ephemeral LaCrosse ID is currently associated with a logical sensor name. Updated when a battery swap causes a new ID to appear (auto-adopt) or via manual assignment.

sensor_id instance-attribute
sensor_id: int

Currently mapped LaCrosse ID.

sensor_name instance-attribute
sensor_name: str

Logical sensor name this ID is mapped to.

mapped_at instance-attribute
mapped_at: datetime

When this mapping was created (e.g. battery swap timestamp).

last_seen instance-attribute
last_seen: datetime

Last time a reading was received for this mapping.

MappingEvent dataclass
MappingEvent(
    event_type: str,
    sensor_name: str,
    old_sensor_id: int | None,
    new_sensor_id: int | None,
    timestamp: datetime,
    reason: str,
)

Immutable event recording a mapping change.

Produced by the sensor registry whenever a mapping is created, changed, or reset. Can be published to MQTT for observability or persisted for audit trails.

event_type instance-attribute
event_type: str

One of "auto_adopt", "manual_assign", "manual_reset", "reset_all".

sensor_name instance-attribute
sensor_name: str

Affected logical sensor name.

old_sensor_id instance-attribute
old_sensor_id: int | None

Previous LaCrosse ID (None if first assignment).

new_sensor_id instance-attribute
new_sensor_id: int | None

New LaCrosse ID (None if reset).

timestamp instance-attribute
timestamp: datetime

When the event occurred.

reason instance-attribute
reason: str

Human-readable explanation of the mapping change.

Settings

jeelink2mqtt.settings

Application configuration for jeelink2mqtt.

Extends cosalette's Settings base with JeeLink-specific fields for serial communication, sensor definitions, and signal-processing parameters. Loaded from environment variables prefixed with JEELINK2MQTT_ and/or a .env file.

SensorConfigSettings

Bases: BaseModel

Pydantic model for a single sensor definition in the config.

Mirrors :class:~jeelink2mqtt.models.SensorConfig but lives in the settings layer so pydantic-settings can deserialise it from environment variables or config files.

name instance-attribute
name: str

Logical sensor name (e.g. "office", "outdoor").

temp_offset class-attribute instance-attribute
temp_offset: float = 0.0

Calibration offset for temperature (°C).

humidity_offset class-attribute instance-attribute
humidity_offset: float = 0.0

Calibration offset for humidity (percentage points).

staleness_timeout class-attribute instance-attribute
staleness_timeout: float | None = None

Per-sensor staleness override in seconds (None = use global).

Jeelink2MqttSettings

Bases: Settings

Root settings for the jeelink2mqtt application.

Inherits MQTT and logging settings from cosalette and adds hardware, sensor, and signal-processing configuration.

Environment variable examples::

JEELINK2MQTT_SERIAL_PORT=/dev/ttyUSB0
JEELINK2MQTT_BAUD_RATE=57600
JEELINK2MQTT_STALENESS_TIMEOUT_SECONDS=600
JEELINK2MQTT_MQTT__HOST=broker.local

Errors

jeelink2mqtt.errors

Domain exceptions for jeelink2mqtt.

Each exception maps to a structured MQTT error type via the error_type_map dict, used by cosalette's error handling system.

error_type_map module-attribute
error_type_map: dict[type[Exception], str] = {
    SerialConnectionError: "serial_connection",
    FrameParseError: "frame_parse",
    MappingConflictError: "mapping_conflict",
    StalenessTimeoutError: "staleness_timeout",
    UnknownSensorError: "unknown_sensor",
}

Mapping from exception types to MQTT error-topic string identifiers.

Used by cosalette's error publisher to route structured error payloads to the correct MQTT sub-topic.

SerialConnectionError

Bases: Exception

JeeLink serial port unavailable or disconnected.

FrameParseError

Bases: Exception

Received data doesn't match the expected LaCrosse frame format.

MappingConflictError

Bases: Exception

Attempted to map an ID that's already assigned to another sensor.

StalenessTimeoutError

Bases: Exception

Sensor hasn't sent readings within the staleness window.

UnknownSensorError

Bases: Exception

Received a reading from an unrecognised / unmapped sensor ID.

Ports

jeelink2mqtt.ports

Protocol ports for jeelink2mqtt hardware abstraction.

Follows hexagonal architecture (ADR-003): device code depends on Protocol ports, never on concrete implementations.

JeeLinkPort

Bases: Protocol

Hardware abstraction for the JeeLink USB receiver.

Concrete implementations wrap the serial connection and frame parsing. Application code depends only on this protocol, making it straightforward to substitute a mock or simulator for testing.

open
open() -> None

Open the serial connection to the JeeLink receiver.

Source code in packages/src/jeelink2mqtt/ports.py
def open(self) -> None:
    """Open the serial connection to the JeeLink receiver."""
    ...
close
close() -> None

Close the serial connection.

Source code in packages/src/jeelink2mqtt/ports.py
def close(self) -> None:
    """Close the serial connection."""
    ...
start_scan
start_scan() -> None

Start scanning for incoming LaCrosse sensor frames.

Source code in packages/src/jeelink2mqtt/ports.py
def start_scan(self) -> None:
    """Start scanning for incoming LaCrosse sensor frames."""
    ...
stop_scan
stop_scan() -> None

Stop scanning for sensor frames.

Source code in packages/src/jeelink2mqtt/ports.py
def stop_scan(self) -> None:
    """Stop scanning for sensor frames."""
    ...
register_callback
register_callback(
    callback: Callable[[SensorReading], None],
) -> None

Register a callback invoked for each decoded sensor frame.

Parameters:

Name Type Description Default
callback Callable[[SensorReading], None]

Function called with a :class:SensorReading each time a valid frame is received.

required
Source code in packages/src/jeelink2mqtt/ports.py
def register_callback(self, callback: Callable[[SensorReading], None]) -> None:
    """Register a callback invoked for each decoded sensor frame.

    Args:
        callback: Function called with a :class:`SensorReading`
            each time a valid frame is received.
    """
    ...
set_led
set_led(enabled: bool) -> None

Control the JeeLink on-board LED.

Parameters:

Name Type Description Default
enabled bool

True to turn the LED on, False to turn it off.

required
Source code in packages/src/jeelink2mqtt/ports.py
def set_led(self, enabled: bool) -> None:
    """Control the JeeLink on-board LED.

    Args:
        enabled: ``True`` to turn the LED on, ``False`` to turn it off.
    """
    ...
SensorRegistryPort

Bases: Protocol

Read-only access to the sensor ID → name registry.

Used by telemetry-reading code to resolve ephemeral LaCrosse IDs to stable logical names and to check liveness. Write operations (adopt, assign, reset) live on the concrete registry implementation.

resolve
resolve(sensor_id: int) -> str | None

Resolve an ephemeral sensor ID to a logical name.

Returns:

Type Description
str | None

The sensor name, or None if the ID is unmapped.

Source code in packages/src/jeelink2mqtt/ports.py
def resolve(self, sensor_id: int) -> str | None:
    """Resolve an ephemeral sensor ID to a logical name.

    Returns:
        The sensor name, or ``None`` if the ID is unmapped.
    """
    ...
is_stale
is_stale(sensor_name: str) -> bool

Check whether a sensor has exceeded its staleness timeout.

Parameters:

Name Type Description Default
sensor_name str

Logical sensor name.

required

Returns:

Type Description
bool

True if the sensor hasn't reported within the timeout.

Source code in packages/src/jeelink2mqtt/ports.py
def is_stale(self, sensor_name: str) -> bool:
    """Check whether a sensor has exceeded its staleness timeout.

    Args:
        sensor_name: Logical sensor name.

    Returns:
        ``True`` if the sensor hasn't reported within the timeout.
    """
    ...
get_mapping
get_mapping(sensor_name: str) -> SensorMapping | None

Get the current mapping for a logical sensor name.

Returns:

Type Description
SensorMapping | None

The active :class:SensorMapping, or None if unmapped.

Source code in packages/src/jeelink2mqtt/ports.py
def get_mapping(self, sensor_name: str) -> SensorMapping | None:
    """Get the current mapping for a logical sensor name.

    Returns:
        The active :class:`SensorMapping`, or ``None`` if unmapped.
    """
    ...
get_all_mappings
get_all_mappings() -> dict[str, SensorMapping]

Get all active mappings.

Returns:

Type Description
dict[str, SensorMapping]

Dictionary of {sensor_name: SensorMapping}.

Source code in packages/src/jeelink2mqtt/ports.py
def get_all_mappings(self) -> dict[str, SensorMapping]:
    """Get all active mappings.

    Returns:
        Dictionary of ``{sensor_name: SensorMapping}``.
    """
    ...
get_unmapped_ids
get_unmapped_ids() -> dict[int, SensorReading]

Get recently-seen sensor IDs that are not yet mapped.

Returns:

Type Description
dict[int, SensorReading]

Dictionary of {sensor_id: last_SensorReading}.

Source code in packages/src/jeelink2mqtt/ports.py
def get_unmapped_ids(self) -> dict[int, SensorReading]:
    """Get recently-seen sensor IDs that are not yet mapped.

    Returns:
        Dictionary of ``{sensor_id: last_SensorReading}``.
    """
    ...

Calibration

jeelink2mqtt.calibration

Per-sensor calibration offset application.

Applies configurable temperature and humidity offsets to filtered sensor readings, compensating for individual sensor inaccuracies.

apply_calibration
apply_calibration(
    reading: SensorReading, config: SensorConfig
) -> SensorReading

Return a new reading with calibration offsets applied.

Temperature is offset directly; humidity uses math.floor(x + 0.5) (standard half-up rounding, not Python's default banker's rounding) and is clamped to 0–100.

Uses :func:dataclasses.replace so that any future fields added to :class:SensorReading are preserved automatically.

Source code in packages/src/jeelink2mqtt/calibration.py
def apply_calibration(reading: SensorReading, config: SensorConfig) -> SensorReading:
    """Return a new reading with calibration offsets applied.

    Temperature is offset directly; humidity uses ``math.floor(x + 0.5)``
    (standard half-up rounding, **not** Python's default banker's
    rounding) and is clamped to 0–100.

    Uses :func:`dataclasses.replace` so that any future fields added to
    :class:`SensorReading` are preserved automatically.
    """
    calibrated_temp = reading.temperature + config.temp_offset
    calibrated_humidity = int(
        max(
            0,
            min(100, reading.humidity + math.floor(config.humidity_offset + 0.5)),
        ),
    )

    return replace(
        reading,
        temperature=calibrated_temp,
        humidity=calibrated_humidity,
    )

Filters

jeelink2mqtt.filters

Per-sensor signal filtering.

Manages a bank of MedianFilter instances — one per active sensor ID — for outlier rejection on temperature and humidity readings.

FilterBank
FilterBank(window: int = 7)

Maintains per-sensor median filters for temperature and humidity.

A filter pair is lazily created the first time a sensor ID is seen. Filters can be individually reset (e.g. after a mapping change) or bulk-cleared.

Source code in packages/src/jeelink2mqtt/filters.py
def __init__(self, window: int = 7) -> None:
    self._window = window
    self._temp_filters: dict[int, MedianFilter] = {}
    self._humidity_filters: dict[int, MedianFilter] = {}
filter
filter(reading: SensorReading) -> tuple[float, float]

Apply median filtering to a sensor reading.

Returns:

Type Description
float

(filtered_temperature, filtered_humidity) after the

float

sliding-window median has been applied.

Source code in packages/src/jeelink2mqtt/filters.py
def filter(self, reading: SensorReading) -> tuple[float, float]:
    """Apply median filtering to a sensor reading.

    Returns:
        ``(filtered_temperature, filtered_humidity)`` after the
        sliding-window median has been applied.
    """
    sid = reading.sensor_id

    if sid not in self._temp_filters:
        self._temp_filters[sid] = MedianFilter(self._window)
        self._humidity_filters[sid] = MedianFilter(self._window)

    filtered_temp = self._temp_filters[sid].update(reading.temperature)
    filtered_humidity = self._humidity_filters[sid].update(
        float(reading.humidity),
    )

    return filtered_temp, filtered_humidity
reset
reset(sensor_id: int) -> None

Remove filters for a sensor ID (e.g. on mapping change).

Source code in packages/src/jeelink2mqtt/filters.py
def reset(self, sensor_id: int) -> None:
    """Remove filters for a sensor ID (e.g. on mapping change)."""
    self._temp_filters.pop(sensor_id, None)
    self._humidity_filters.pop(sensor_id, None)
reset_all
reset_all() -> None

Clear all filters.

Source code in packages/src/jeelink2mqtt/filters.py
def reset_all(self) -> None:
    """Clear all filters."""
    self._temp_filters.clear()
    self._humidity_filters.clear()