23143225975

Committed 16 Mar 2026 12:15PM UTC coverage: 97.406% (+0.2%) from 97.211%

Build # 23143225975

Build Type

push

github

Committed by

derliebemarcus

Commit Message

change: Detection and behaviour of binary_sensors and switches had to be updated.

Coverage Stats

19 of 20 new or added lines in 2 files covered. (95.0%)

1690 of 1735 relevant lines covered (97.41%)

0.97 hits per line

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

97.5

/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, device_info in bundle.inventory.data.items():
            unique = f"{device_id}_online"
            if unique not in known:
                known.add(unique)
                new_entities.append(RmsOnlineBinarySensor(bundle, device_id))

            port_configs = {
                str(p.get("id")): p
                for p in bundle.port_config.data.get(device_id, [])
                if p.get("id")
            }

            model = device_info.get("model", "UNKNOWN")
            is_switch_device = model.startswith("TSW") or model.startswith("SWM")

            if is_switch_device and not port_configs:
                for i in range(1, 9):
                    port_configs[f"port{i}"] = {"id": f"port{i}"}
                for i in range(1, 3):
                    port_configs[f"sfp{i}"] = {"id": f"sfp{i}"}

            for port in bundle.port_scan.data.get(device_id, []):
                port_id = str(port.get("name") or "").strip()
                if port_id and port_id not in port_configs:
                    port_configs[port_id] = {"id": port_id}

            for port_id in port_configs:
                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

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

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, device_info in bundle.inventory.data.items():	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	port_configs = {	1✔
36	str(p.get("id")): p
37	for p in bundle.port_config.data.get(device_id, [])
38	if p.get("id")
39	}
40
41	model = device_info.get("model", "UNKNOWN")	1✔
42	is_switch_device = model.startswith("TSW") or model.startswith("SWM")	1✔
43
44	if is_switch_device and not port_configs:	1✔
45	for i in range(1, 9):	1✔
46	port_configs[f"port{i}"] = {"id": f"port{i}"}	1✔
47	for i in range(1, 3):	1✔
48	port_configs[f"sfp{i}"] = {"id": f"sfp{i}"}	1✔
49
50	for port in bundle.port_scan.data.get(device_id, []):	1✔
51	port_id = str(port.get("name") or "").strip()	1✔
52	if port_id and port_id not in port_configs:	1✔
NEW 53	port_configs[port_id] = {"id": port_id}	×
54
55	for port_id in port_configs:	1✔
56	unique_port = f"{device_id}_{port_id}_link"	1✔
57	if unique_port not in known:	1✔
58	known.add(unique_port)	1✔
59	new_entities.append(RmsPortLinkBinarySensor(bundle, device_id, port_id))	1✔
60	if new_entities:	1✔
61	async_add_entities(new_entities)	1✔
62
63	_add_new_entities()	1✔
64	entry.async_on_unload(bundle.inventory.async_add_listener(_add_new_entities))	1✔
65	entry.async_on_unload(bundle.port_scan.async_add_listener(_add_new_entities))	1✔
66
67
68	class RmsOnlineBinarySensor(TeltonikaRmsEntity, BinarySensorEntity):	1✔
69	"""Connectivity state for an RMS device."""
70
71	_attr_translation_key = "online"	1✔
72	_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY	1✔
73	_attr_icon = "mdi:router-network"	1✔
74
75	def __init__(self, bundle: CoordinatorBundle, device_id: str) -> None:	1✔
76	super().__init__(bundle, device_id)	1✔
77	self._attr_unique_id = f"{device_id}_online"	1✔
78
79	@property	1✔
80	def is_on(self) -> bool \| None:	1✔
81	normalized = self._normalized	1✔
82	return normalized.online if normalized else None	1✔
83
84
85	class RmsPortLinkBinarySensor(TeltonikaRmsEntity, BinarySensorEntity):	1✔
86	"""Link state for an ethernet port."""
87
88	_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY	1✔
89	_attr_icon = "mdi:ethernet-cable"	1✔
90
91	def __init__(self, bundle: CoordinatorBundle, device_id: str, port_id: str) -> None:	1✔
92	super().__init__(bundle, device_id, coordinator=bundle.port_scan)	1✔
93	self._port_id = port_id	1✔
94	self._attr_unique_id = f"{device_id}_{port_id}_link"	1✔
95	self._attr_name = f"{port_id.upper()} Link"	1✔
96
97	@property	1✔
98	def _port(self) -> dict[str, Any] \| None:	1✔
99	for port in self._bundle.port_scan.data.get(self.device_id, []):	1✔
100	if str(port.get("name") or "").strip() == self._port_id:	1✔
101	return port	1✔
102	return None	1✔
103
104	@property	1✔
105	def available(self) -> bool:	1✔
106	return super().available	1✔
107
108	@property	1✔
109	def is_on(self) -> bool \| None:	1✔
110	port = self._port	1✔
111	if port is None:	1✔
112	return False	1✔
113	state = port.get("state")	1✔
114	if state is not None:	1✔
115	return str(state).lower() == "up"	×
116	return True	1✔

derliebemarcus / teltonika_rms / 23143225975

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