Reference¶

@field_validator("name")
@classmethod
def name_must_be_valid_topic_segment(cls, value: str) -> str:
    """Reject names that would form invalid MQTT topic segments."""
    if not re.fullmatch(r"[A-Za-z0-9_-]+", value):
        raise ValueError(
            "Sensor name must be a non-empty MQTT topic segment "
            "matching [A-Za-z0-9_-]+ (no '/', '+', '#', whitespace, "
            "or control chars)"
        )
    return value

async def open(self) -> None:
    """Open the serial connection to the JeeLink receiver."""
    ...

async def close(self) -> None:
    """Close the serial connection."""
    ...

async def start_scan(self) -> None:
    """Start scanning for incoming LaCrosse sensor frames."""
    ...

async def stop_scan(self) -> None:
    """Stop scanning for sensor frames."""
    ...

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.
    """
    ...

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.
    """
    ...

def __aiter__(self) -> AsyncIterator[SensorReading]:
    """Return an async iterator over sensor readings."""
    ...

async def __anext__(self) -> SensorReading:
    """Get the next sensor reading."""
    ...

async def __aenter__(self) -> JeeLinkPort:
    """Enter async context manager."""
    ...

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit async context manager."""
    ...

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.
    """
    ...

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.
    """
    ...

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.
    """
    ...

def get_all_mappings(self) -> dict[str, SensorMapping]:
    """Get all active mappings.

    Returns:
        Dictionary of ``{sensor_name: SensorMapping}``.
    """
    ...

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}``.
    """
    ...

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

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)

def reset_all(self) -> None:
    """Clear all filters."""
    self._temp_filters.clear()
    self._humidity_filters.clear()