23139347238

Committed 16 Mar 2026 10:32AM UTC coverage: 97.211% (-0.2%) from 97.377%

Build # 23139347238

Build Type

push

github

Committed by

derliebemarcus

Commit Message

fix: decouple port link state from switch administration state

add: introduce per-port link state binary sensors
change: default state polling interval to 300s and inventory polling to 3600s
change: rename serial sensor friendly name to Serial Number
change: make device tags, status, and OpenAPI spec path options optional
remove: drop deprecated aggregate used ethernet port sensors

Coverage Stats

39 of 44 new or added lines in 3 files covered. (88.64%)

1 existing line in 1 file now uncovered.

1673 of 1721 relevant lines covered (97.21%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.96

/binary_sensor.py

"""Binary sensors for Teltonika RMS."""

from __future__ import annotations

from typing import Any

from homeassistant.components.binary_sensor import BinarySensorDeviceClass, BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from . import TeltonikaRmsRuntime
from .coordinator import CoordinatorBundle
from .entity import TeltonikaRmsEntity


async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    runtime: TeltonikaRmsRuntime = entry.runtime_data
    bundle: CoordinatorBundle = runtime.bundle
    known: set[str] = set()

    @callback
    def _add_new_entities() -> None:
        new_entities: list[BinarySensorEntity] = []
        for device_id in bundle.inventory.data:
            unique = f"{device_id}_online"
            if unique not in known:
                known.add(unique)
                new_entities.append(RmsOnlineBinarySensor(bundle, device_id))

            for port in bundle.port_scan.data.get(device_id, []):
                port_id = str(port.get("name") or "").strip()
                if not port_id:
                    continue
                unique_port = f"{device_id}_{port_id}_link"
                if unique_port not in known:
                    known.add(unique_port)
                    new_entities.append(RmsPortLinkBinarySensor(bundle, device_id, port_id))

        if new_entities:
            async_add_entities(new_entities)

    _add_new_entities()
    entry.async_on_unload(bundle.inventory.async_add_listener(_add_new_entities))
    entry.async_on_unload(bundle.port_scan.async_add_listener(_add_new_entities))


class RmsOnlineBinarySensor(TeltonikaRmsEntity, BinarySensorEntity):
    """Connectivity state for an RMS device."""

    _attr_translation_key = "online"
    _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
    _attr_icon = "mdi:router-network"

    def __init__(self, bundle: CoordinatorBundle, device_id: str) -> None:
        super().__init__(bundle, device_id)
        self._attr_unique_id = f"{device_id}_online"

    @property
    def is_on(self) -> bool | None:
        normalized = self._normalized
        return normalized.online if normalized else None


class RmsPortLinkBinarySensor(TeltonikaRmsEntity, BinarySensorEntity):
    """Link state for an ethernet port."""

    _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
    _attr_icon = "mdi:ethernet-cable"

    def __init__(self, bundle: CoordinatorBundle, device_id: str, port_id: str) -> None:
        super().__init__(bundle, device_id, coordinator=bundle.port_scan)
        self._port_id = port_id
        self._attr_unique_id = f"{device_id}_{port_id}_link"
        self._attr_name = f"{port_id.upper()} Link"

    @property
    def _port(self) -> dict[str, Any] | None:
        for port in self._bundle.port_scan.data.get(self.device_id, []):
            if str(port.get("name") or "").strip() == self._port_id:
                return port
        return None

    @property
    def available(self) -> bool:
        return super().available and self._port is not None

    @property
    def is_on(self) -> bool | None:
        port = self._port
        if port is None:
            return None
        state = port.get("state")
        if state is not None:
            return str(state).lower() == "up"
        return False

1	"""Binary sensors for Teltonika RMS."""
2
3	from __future__ import annotations	1✔
4
5	from typing import Any	1✔
6
7	from homeassistant.components.binary_sensor import BinarySensorDeviceClass, BinarySensorEntity	1✔
8	from homeassistant.config_entries import ConfigEntry	1✔
9	from homeassistant.core import HomeAssistant, callback	1✔
10	from homeassistant.helpers.entity_platform import AddEntitiesCallback	1✔
11
12	from . import TeltonikaRmsRuntime	1✔
13	from .coordinator import CoordinatorBundle	1✔
14	from .entity import TeltonikaRmsEntity	1✔
15
16
17	async def async_setup_entry(	1✔
18	hass: HomeAssistant,
19	entry: ConfigEntry,
20	async_add_entities: AddEntitiesCallback,
21	) -> None:
22	runtime: TeltonikaRmsRuntime = entry.runtime_data	1✔
23	bundle: CoordinatorBundle = runtime.bundle	1✔
24	known: set[str] = set()	1✔
25
26	@callback	1✔
27	def _add_new_entities() -> None:	1✔
28	new_entities: list[BinarySensorEntity] = []	1✔
29	for device_id in bundle.inventory.data:	1✔
30	unique = f"{device_id}_online"	1✔
31	if unique not in known:	1✔
32	known.add(unique)	1✔
33	new_entities.append(RmsOnlineBinarySensor(bundle, device_id))	1✔
34
35	for port in bundle.port_scan.data.get(device_id, []):	1✔
36	port_id = str(port.get("name") or "").strip()	1✔
37	if not port_id:	1✔
NEW 38	continue	×
39	unique_port = f"{device_id}_{port_id}_link"	1✔
40	if unique_port not in known:	1✔
41	known.add(unique_port)	1✔
42	new_entities.append(RmsPortLinkBinarySensor(bundle, device_id, port_id))	1✔
43
44	if new_entities:	1✔
45	async_add_entities(new_entities)	1✔
46
47	_add_new_entities()	1✔
48	entry.async_on_unload(bundle.inventory.async_add_listener(_add_new_entities))	1✔
49	entry.async_on_unload(bundle.port_scan.async_add_listener(_add_new_entities))	1✔
50
51
52	class RmsOnlineBinarySensor(TeltonikaRmsEntity, BinarySensorEntity):	1✔
53	"""Connectivity state for an RMS device."""
54
55	_attr_translation_key = "online"	1✔
56	_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY	1✔
57	_attr_icon = "mdi:router-network"	1✔
58
59	def __init__(self, bundle: CoordinatorBundle, device_id: str) -> None:	1✔
60	super().__init__(bundle, device_id)	1✔
61	self._attr_unique_id = f"{device_id}_online"	1✔
62
63	@property	1✔
64	def is_on(self) -> bool \| None:	1✔
65	normalized = self._normalized	1✔
66	return normalized.online if normalized else None	1✔
67
68
69	class RmsPortLinkBinarySensor(TeltonikaRmsEntity, BinarySensorEntity):	1✔
70	"""Link state for an ethernet port."""
71
72	_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY	1✔
73	_attr_icon = "mdi:ethernet-cable"	1✔
74
75	def __init__(self, bundle: CoordinatorBundle, device_id: str, port_id: str) -> None:	1✔
76	super().__init__(bundle, device_id, coordinator=bundle.port_scan)	1✔
77	self._port_id = port_id	1✔
78	self._attr_unique_id = f"{device_id}_{port_id}_link"	1✔
79	self._attr_name = f"{port_id.upper()} Link"	1✔
80
81	@property	1✔
82	def _port(self) -> dict[str, Any] \| None:	1✔
83	for port in self._bundle.port_scan.data.get(self.device_id, []):	1✔
84	if str(port.get("name") or "").strip() == self._port_id:	1✔
85	return port	1✔
NEW 86	return None	×
87
88	@property	1✔
89	def available(self) -> bool:	1✔
NEW 90	return super().available and self._port is not None	×
91
92	@property	1✔
93	def is_on(self) -> bool \| None:	1✔
94	port = self._port	1✔
95	if port is None:	1✔
NEW 96	return None	×
97	state = port.get("state")	1✔
98	if state is not None:	1✔
NEW 99	return str(state).lower() == "up"	×
100	return False	1✔

derliebemarcus / teltonika_rms / 23139347238

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous