23186357956

Committed 17 Mar 2026 08:59AM UTC coverage: 97.433% (+0.03%) from 97.406%

Build # 23186357956

Build Type

push

github

Committed by

derliebemarcus

Commit Message

fix: ensure NIL ports are filtered before generating fallback switch ports

Coverage Stats

1746 of 1792 relevant lines covered (97.43%)

0.97 hits per line

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

97.62

/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") and str(p.get("id")) != "NIL"
            }

            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"switch_port{i}"] = {"id": f"switch_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 == "NIL":
                    continue
                if port_id and port_id not in port_configs:
                    port_configs[port_id] = {"id": port_id}

            for port_id in list(port_configs.keys()):
                if port_id == "NIL":
                    port_configs.pop(port_id)
                else:
                    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") and str(p.get("id")) != "NIL"
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"switch_port{i}"] = {"id": f"switch_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 == "NIL":	1✔
53	continue	1✔
54	if port_id and port_id not in port_configs:	1✔
55	port_configs[port_id] = {"id": port_id}	1✔
56
57	for port_id in list(port_configs.keys()):	1✔
58	if port_id == "NIL":	1✔
59	port_configs.pop(port_id)	×
60	else:
61	unique_port = f"{device_id}_{port_id}_link"	1✔
62	if unique_port not in known:	1✔
63	known.add(unique_port)	1✔
64	new_entities.append(RmsPortLinkBinarySensor(bundle, device_id, port_id))	1✔
65	if new_entities:	1✔
66	async_add_entities(new_entities)	1✔
67
68	_add_new_entities()	1✔
69	entry.async_on_unload(bundle.inventory.async_add_listener(_add_new_entities))	1✔
70	entry.async_on_unload(bundle.port_scan.async_add_listener(_add_new_entities))	1✔
71
72
73	class RmsOnlineBinarySensor(TeltonikaRmsEntity, BinarySensorEntity):	1✔
74	"""Connectivity state for an RMS device."""
75
76	_attr_translation_key = "online"	1✔
77	_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY	1✔
78	_attr_icon = "mdi:router-network"	1✔
79
80	def __init__(self, bundle: CoordinatorBundle, device_id: str) -> None:	1✔
81	super().__init__(bundle, device_id)	1✔
82	self._attr_unique_id = f"{device_id}_online"	1✔
83
84	@property	1✔
85	def is_on(self) -> bool \| None:	1✔
86	normalized = self._normalized	1✔
87	return normalized.online if normalized else None	1✔
88
89
90	class RmsPortLinkBinarySensor(TeltonikaRmsEntity, BinarySensorEntity):	1✔
91	"""Link state for an ethernet port."""
92
93	_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY	1✔
94	_attr_icon = "mdi:ethernet-cable"	1✔
95
96	def __init__(self, bundle: CoordinatorBundle, device_id: str, port_id: str) -> None:	1✔
97	super().__init__(bundle, device_id, coordinator=bundle.port_scan)	1✔
98	self._port_id = port_id	1✔
99	self._attr_unique_id = f"{device_id}_{port_id}_link"	1✔
100	self._attr_name = f"{port_id.upper()} Link"	1✔
101
102	@property	1✔
103	def _port(self) -> dict[str, Any] \| None:	1✔
104	for port in self._bundle.port_scan.data.get(self.device_id, []):	1✔
105	if str(port.get("name") or "").strip() == self._port_id:	1✔
106	return port	1✔
107	return None	1✔
108
109	@property	1✔
110	def available(self) -> bool:	1✔
111	return super().available	1✔
112
113	@property	1✔
114	def is_on(self) -> bool \| None:	1✔
115	port = self._port	1✔
116	if port is None:	1✔
117	return False	1✔
118	state = port.get("state")	1✔
119	if state is not None:	1✔
120	return str(state).lower() == "up"	×
121	return True	1✔

derliebemarcus / teltonika_rms / 23186357956

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