starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fix sensors for Mikrotik (#178604)

Browse Source
This commit is contained in:
Simone Chemelli
2026-08-14 14:24:15 +02:00
committed by Bram Kragten
parent ef53f48ac0
commit 7eddf01f5b
6 changed files with 186 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/mikrotik/coordinator.py
+40 -7
View File
@@ -47,7 +47,7 @@ from .const import (
)
from .device import Device
from .errors import CannotConnect, LoginError
from .utils import mikrotik_config_entry_errors
from .utils import calculate_uptime, mikrotik_config_entry_errors, percentage
type MikrotikConfigEntry = ConfigEntry[MikrotikDataUpdateCoordinator]
@@ -90,6 +90,43 @@ class MikrotikData:
or [{}]
)[0]
def _get_health_details(self) -> None:
"""Retrieve health details from Mikrotik API."""
health_data = (
self.command(MIKROTIK_SERVICES[HEALTH], suppress_errors=True) or []
)
self.sensors[HEALTH] = {
entry["name"]: entry["value"]
for entry in health_data
if "name" in entry and "value" in entry
}
def _get_resource_details(self) -> None:
"""Retrieve resource details from Mikrotik API."""
resource_data = (
self.command(MIKROTIK_SERVICES[RESOURCE], suppress_errors=True) or [{}]
)[0]
self.sensors[RESOURCE] = (
{
"cpu-load": resource_data.get("cpu-load"),
"memory-usage": percentage(
resource_data.get("total-memory", 0),
resource_data.get("free-memory", 0),
),
"disk-usage": percentage(
resource_data.get("total-hdd-space", 0),
resource_data.get("free-hdd-space", 0),
),
"uptime": (
calculate_uptime(resource_data["uptime"])
if resource_data.get("uptime")
else None
),
}
if resource_data
else {}
)
@staticmethod
def load_mac(devices: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""Load dictionary using MAC address as key."""
@@ -189,12 +226,8 @@ class MikrotikData:
self.command(MIKROTIK_SERVICES[UPDATE], suppress_errors=True) or [{}]
)[0]
self.sensors[HEALTH] = (
self.command(MIKROTIK_SERVICES[HEALTH], suppress_errors=True) or []
)
self.sensors[RESOURCE] = (
self.command(MIKROTIK_SERVICES[RESOURCE], suppress_errors=True) or []
)
self._get_health_details()
self._get_resource_details()
if not device_list:
return
homeassistant/components/mikrotik/sensor.py
+7 -67
View File
@@ -1,9 +1,8 @@
"""Support for Mikrotik routers sensors."""
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Any, Final, override
from datetime import datetime
from typing import Final, cast, override
from homeassistant.components.sensor import (
SensorDeviceClass,
@@ -20,9 +19,8 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util.dt import utcnow
from .const import HEALTH, LOGGER, RESOURCE
from .const import HEALTH, RESOURCE
from .coordinator import MikrotikConfigEntry
from .entity import MikrotikEntity
@@ -33,44 +31,7 @@ PARALLEL_UPDATES = 0
class MikrotikSensorEntityDescription(SensorEntityDescription):
"""Shared Mikrotik Sensors entity description."""
value: Callable[[dict[str, Any]], StateType | datetime]
type: str
index: int
def _calculate_uptime(data: dict[str, Any]) -> datetime | None:
"""Calculate uptime."""
# e.g. 1d3h39m30s
uptime_string = data["uptime"]
total = 0
num = 0
for ch in uptime_string.strip():
if ch.isdigit():
num = num * 10 + int(ch)
else:
if ch == "w":
total += num * (60 * 60 * 24 * 7)
elif ch == "d":
total += num * (60 * 60 * 24)
elif ch == "h":
total += num * (60 * 60)
elif ch == "m":
total += num * 60
elif ch == "s":
total += num
else:
LOGGER.warning("Unknown uptime format: %s", uptime_string)
return None
num = 0
if num != 0:
LOGGER.warning("Unknown uptime format: %s", uptime_string)
return None
return utcnow() - timedelta(seconds=total)
SENSORS: Final = (
@@ -79,18 +40,14 @@ SENSORS: Final = (
device_class=SensorDeviceClass.TEMPERATURE,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
value=lambda _data: _data["value"],
type=HEALTH,
index=1,
),
MikrotikSensorEntityDescription(
key="voltage",
device_class=SensorDeviceClass.VOLTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
value=lambda _data: _data["value"],
type=HEALTH,
index=0,
),
MikrotikSensorEntityDescription(
key="cpu-load",
@@ -99,9 +56,7 @@ SENSORS: Final = (
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfRatio.PERCENTAGE,
suggested_display_precision=2,
value=lambda _data: _data["cpu-load"],
type=RESOURCE,
index=0,
),
MikrotikSensorEntityDescription(
key="memory-usage",
@@ -110,13 +65,7 @@ SENSORS: Final = (
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfRatio.PERCENTAGE,
suggested_display_precision=2,
value=lambda _data: (
None
if (total := _data.get("total-memory", 0)) == 0
else (total - _data.get("free-memory", 0)) / total * 100
),
type=RESOURCE,
index=0,
),
MikrotikSensorEntityDescription(
key="disk-usage",
@@ -125,20 +74,12 @@ SENSORS: Final = (
entity_category=EntityCategory.DIAGNOSTIC,
native_unit_of_measurement=UnitOfRatio.PERCENTAGE,
suggested_display_precision=2,
value=lambda _data: (
None
if (total := _data.get("total-hdd-space", 0)) == 0
else (total - _data.get("free-hdd-space", 0)) / total * 100
),
type=RESOURCE,
index=0,
),
MikrotikSensorEntityDescription(
key="uptime",
device_class=SensorDeviceClass.UPTIME,
value=_calculate_uptime,
type=RESOURCE,
index=0,
),
)
@@ -155,8 +96,8 @@ async def async_setup_entry(
sensors_list = [
MikrotikSensorEntity(coordinator, sensor_desc)
for sensor_desc in SENSORS
if len(coordinator.api.sensors.get(sensor_desc.type, []))
>= (sensor_desc.index + 1)
if coordinator.api.sensors.get(sensor_desc.type, {}).get(sensor_desc.key)
is not None
]
async_add_entities(sensors_list)
@@ -173,7 +114,6 @@ class MikrotikSensorEntity(
@override
def native_value(self) -> StateType | datetime:
"""Return the state of the sensor."""
data_list = self.coordinator.api.sensors[self.entity_description.type]
data_entry = data_list[self.entity_description.index]
data = self.coordinator.api.sensors[self.entity_description.type]
return self.entity_description.value(data_entry)
return cast(StateType | datetime, data.get(self.entity_description.key))
homeassistant/components/mikrotik/utils.py
+42 -1
View File
@@ -2,6 +2,7 @@
from collections.abc import Generator
from contextlib import contextmanager
from datetime import datetime, timedelta
from librouteros.exceptions import ConnectionClosed, LibRouterosError
@@ -11,11 +12,51 @@ from homeassistant.exceptions import (
HomeAssistantError,
)
from homeassistant.helpers.update_coordinator import UpdateFailed
from homeassistant.util.dt import utcnow
from .const import DOMAIN
from .const import DOMAIN, LOGGER
from .errors import CannotConnect, LoginError
def percentage(total: float, free: float) -> float | None:
"""Return the used percentage for a total/free pair, or None if total is zero."""
if total == 0:
return None
return (total - free) / total * 100
def calculate_uptime(uptime_string: str) -> datetime | None:
"""Calculate uptime from a RouterOS duration string, e.g. "1d3h39m30s"."""
total = 0
num = 0
for ch in uptime_string.strip():
if ch.isdigit():
num = num * 10 + int(ch)
else:
if ch == "w":
total += num * (60 * 60 * 24 * 7)
elif ch == "d":
total += num * (60 * 60 * 24)
elif ch == "h":
total += num * (60 * 60)
elif ch == "m":
total += num * 60
elif ch == "s":
total += num
else:
LOGGER.warning("Unknown uptime format: %s", uptime_string)
return None
num = 0
if num != 0:
LOGGER.warning("Unknown uptime format: %s", uptime_string)
return None
return utcnow() - timedelta(seconds=total)
@contextmanager
def mikrotik_config_entry_errors(
suppress_errors: bool = False, during_setup: bool = False
tests/components/mikrotik/snapshots/test_diagnostics.ambr
+10 -20
View File
@@ -7,26 +7,16 @@
'last_update success': True,
'model': 'RB5009',
'sensors': dict({
'health': list([
dict({
'name': 'voltage',
'value': 24.2,
}),
dict({
'name': 'temperature',
'value': 50.0,
}),
]),
'resource': list([
dict({
'cpu-load': 15,
'free-hdd-space': 25,
'free-memory': 200,
'total-hdd-space': 100,
'total-memory': 1000,
'uptime': '1w2d3h4m5s',
}),
]),
'health': dict({
'temperature': 50.0,
'voltage': 24.2,
}),
'resource': dict({
'cpu-load': 15,
'disk-usage': 75.0,
'memory-usage': 80.0,
'uptime': '2025-12-23T08:55:55+00:00',
}),
}),
'support_capman': False,
'support_wifi': False,
tests/components/mikrotik/test_diagnostics.py
+2
View File
@@ -1,5 +1,6 @@
"""Tests for Mikrotik diagnostics platform."""
import pytest
from syrupy.assertion import SnapshotAssertion
from syrupy.filters import props
@@ -11,6 +12,7 @@ from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
@pytest.mark.freeze_time("2026-01-01T12:00:00+00:00")
async def test_entry_diagnostics(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
tests/components/mikrotik/test_sensor.py
+85 -45
View File
@@ -1,12 +1,13 @@
"""Tests for the Mikrotik sensor platform."""
from typing import Any
from unittest.mock import patch
from freezegun import freeze_time
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_UNKNOWN, Platform
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@@ -28,40 +29,70 @@ async def test_sensor_entities_created(
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
async def test_sensor_wrong_data(hass: HomeAssistant) -> None:
"""Test Mikrotik sensor entities handle missing data gracefully."""
await setup_mikrotik_entry(
hass,
health_data=[
{"name": "voltage", "value": 24.2},
],
system_data=[
@pytest.mark.parametrize(
("health_data", "system_data", "existing_states", "missing_entities"),
[
pytest.param(
[{"name": "voltage", "value": 24.2}],
[
{
"cpu-load": 15,
"total-memory": 0,
"free-memory": 200,
"total-hdd-space": 0,
"free-hdd-space": 25,
"uptime": None,
}
],
{
"cpu-load": 15,
"total-memory": 0,
"free-memory": 200,
"total-hdd-space": 0,
"free-hdd-space": 25,
"uptime": None,
}
],
)
"sensor.mikrotik_voltage": "24.2",
"sensor.mikrotik_cpu_usage": "15",
},
[
"sensor.mikrotik_temperature",
"sensor.mikrotik_memory_usage",
"sensor.mikrotik_disk_usage",
"sensor.mikrotik_uptime",
],
id="degenerate_data",
),
pytest.param(
[],
[],
{},
[
"sensor.mikrotik_voltage",
"sensor.mikrotik_temperature",
"sensor.mikrotik_cpu_usage",
"sensor.mikrotik_memory_usage",
"sensor.mikrotik_disk_usage",
"sensor.mikrotik_uptime",
],
id="no_data",
),
],
)
async def test_sensor_missing_or_wrong_data(
hass: HomeAssistant,
health_data: list[dict[str, Any]],
system_data: list[dict[str, Any]],
existing_states: dict[str, str],
missing_entities: list[str],
) -> None:
"""Test Mikrotik sensor entities handle missing/wrong data gracefully.
assert (state := hass.states.get("sensor.mikrotik_voltage"))
assert state.state == "24.2"
memory-usage/disk-usage can't be computed when the reported totals are
zero, and uptime can't be computed without a raw uptime string, so those
sensors are not created at all.
"""
await setup_mikrotik_entry(hass, health_data=health_data, system_data=system_data)
assert (state := hass.states.get("sensor.mikrotik_temperature")) is None
for entity_id, expected_state in existing_states.items():
assert (state := hass.states.get(entity_id))
assert state.state == expected_state
assert (state := hass.states.get("sensor.mikrotik_cpu_usage"))
assert state.state == "15"
assert (state := hass.states.get("sensor.mikrotik_memory_usage"))
assert state.state == STATE_UNKNOWN
assert (state := hass.states.get("sensor.mikrotik_disk_usage"))
assert state.state == STATE_UNKNOWN
assert (state := hass.states.get("sensor.mikrotik_uptime")) is None
for entity_id in missing_entities:
assert hass.states.get(entity_id) is None
@pytest.mark.parametrize(
@@ -71,13 +102,16 @@ async def test_sensor_wrong_data(hass: HomeAssistant) -> None:
pytest.param("2h30", id="missing_unit"),
],
)
@freeze_time("2026-01-01T12:00:00+00:00")
async def test_sensor_bad_uptime_data(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
uptime_api: str,
) -> None:
"""Test Mikrotik sensor entities handle missing data gracefully."""
"""Test Mikrotik sensor entities handle missing data gracefully.
An uptime string that can't be parsed computes to None, so the sensor is
not created at all, but the parsing failure is still logged.
"""
await setup_mikrotik_entry(
hass,
@@ -95,21 +129,27 @@ async def test_sensor_bad_uptime_data(
assert f"Unknown uptime format: {uptime_api}" in caplog.text
assert (state := hass.states.get("sensor.mikrotik_uptime"))
assert state.state == STATE_UNKNOWN
assert hass.states.get("sensor.mikrotik_uptime") is None
async def test_sensor_no_data(hass: HomeAssistant) -> None:
"""Test Mikrotik sensor entities handle missing data gracefully."""
async def test_sensor_health_data_reordered(hass: HomeAssistant) -> None:
"""Test voltage/temperature are matched by name, not list position.
Some devices (e.g. netPower 16P) report additional health items such as
PoE power consumption ahead of voltage, so the sensors must not assume a
fixed ordering. https://github.com/home-assistant/core/issues/178392
"""
await setup_mikrotik_entry(
hass,
health_data=[],
system_data=[],
health_data=[
{"name": "power-consumption", "value": 8.0},
{"name": "temperature", "value": 50.0},
{"name": "voltage", "value": 52.0},
],
)
assert hass.states.get("sensor.mikrotik_voltage") is None
assert hass.states.get("sensor.mikrotik_temperature") is None
assert hass.states.get("sensor.mikrotik_cpu_usage") is None
assert hass.states.get("sensor.mikrotik_memory_usage") is None
assert hass.states.get("sensor.mikrotik_disk_usage") is None
assert hass.states.get("sensor.mikrotik_uptime") is None
assert (state := hass.states.get("sensor.mikrotik_voltage"))
assert state.state == "52.0"
assert (state := hass.states.get("sensor.mikrotik_temperature"))
assert state.state == "50.0"