mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 03:07:37 +00:00
Huawei LTE sensor improvements (#84019)
* Use `None` for unknown states consistently * Use huawei_lte_api NetworkModeEnum instead of magic strings * Recognize some new sensor items * Exclude current day duration sensor * Fix current month upload/download types * Add current day transfer * Extract lambdas used in multiple spots to named functions * Formatter naming consistency improvements
This commit is contained in:
Ville Skyttä
GitHub
parent
7953c4a6d5
commit
a5a079fb06
@ -2,11 +2,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from bisect import bisect
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Callable, Sequence
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
import re
|
||||
|
||||
from huawei_lte_api.enums.net import NetworkModeEnum
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
DOMAIN as SENSOR_DOMAIN,
|
||||
SensorDeviceClass,
|
||||
@ -17,7 +20,6 @@ from homeassistant.components.sensor import (
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
PERCENTAGE,
|
||||
STATE_UNKNOWN,
|
||||
UnitOfDataRate,
|
||||
UnitOfFrequency,
|
||||
UnitOfInformation,
|
||||
@ -62,6 +64,45 @@ def format_default(value: StateType) -> tuple[StateType, str | None]:
|
||||
return value, unit
|
||||
|
||||
|
||||
def format_freq_mhz(value: StateType) -> tuple[StateType, UnitOfFrequency]:
|
||||
"""Format a frequency value for which source is in tens of MHz."""
|
||||
return (
|
||||
round(int(value) / 10) if value is not None else None,
|
||||
UnitOfFrequency.MEGAHERTZ,
|
||||
)
|
||||
|
||||
|
||||
def format_last_reset_elapsed_seconds(value: str | None) -> datetime | None:
|
||||
"""Convert elapsed seconds to last reset datetime."""
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
last_reset = datetime.now() - timedelta(seconds=int(value))
|
||||
last_reset.replace(microsecond=0)
|
||||
return last_reset
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def signal_icon(limits: Sequence[int], value: StateType) -> str:
|
||||
"""Get signal icon."""
|
||||
return (
|
||||
"mdi:signal-cellular-outline",
|
||||
"mdi:signal-cellular-1",
|
||||
"mdi:signal-cellular-2",
|
||||
"mdi:signal-cellular-3",
|
||||
)[bisect(limits, value if value is not None else -1000)]
|
||||
|
||||
|
||||
def bandwidth_icon(limits: Sequence[int], value: StateType) -> str:
|
||||
"""Get bandwidth icon."""
|
||||
return (
|
||||
"mdi:speedometer-slow",
|
||||
"mdi:speedometer-medium",
|
||||
"mdi:speedometer",
|
||||
)[bisect(limits, value if value is not None else -1000)]
|
||||
|
||||
|
||||
@dataclass
|
||||
class HuaweiSensorGroup:
|
||||
"""Class describing Huawei LTE sensor groups."""
|
||||
@ -75,8 +116,10 @@ class HuaweiSensorGroup:
|
||||
class HuaweiSensorEntityDescription(SensorEntityDescription):
|
||||
"""Class describing Huawei LTE sensor entities."""
|
||||
|
||||
formatter: Callable[[str], tuple[StateType, str | None]] = format_default
|
||||
format_fn: Callable[[str], tuple[StateType, str | None]] = format_default
|
||||
icon_fn: Callable[[StateType], str] | None = None
|
||||
last_reset_item: str | None = None
|
||||
last_reset_format_fn: Callable[[str | None], datetime | None] | None = None
|
||||
|
||||
|
||||
SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
@ -114,11 +157,21 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
#
|
||||
KEY_DEVICE_SIGNAL: HuaweiSensorGroup(
|
||||
descriptions={
|
||||
"arfcn": HuaweiSensorEntityDescription(
|
||||
key="arfcn",
|
||||
name="ARFCN",
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
"band": HuaweiSensorEntityDescription(
|
||||
key="band",
|
||||
name="Band",
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
"bsic": HuaweiSensorEntityDescription(
|
||||
key="bsic",
|
||||
name="Base station identity code",
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
"cell_id": HuaweiSensorEntityDescription(
|
||||
key="cell_id",
|
||||
name="Cell ID",
|
||||
@ -144,11 +197,13 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
"dlbandwidth": HuaweiSensorEntityDescription(
|
||||
key="dlbandwidth",
|
||||
name="Downlink bandwidth",
|
||||
icon_fn=lambda x: (
|
||||
"mdi:speedometer-slow",
|
||||
"mdi:speedometer-medium",
|
||||
"mdi:speedometer",
|
||||
)[bisect((8, 15), x if x is not None else -1000)],
|
||||
icon_fn=lambda x: bandwidth_icon((8, 15), x),
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
"dlfrequency": HuaweiSensorEntityDescription(
|
||||
key="dlfrequency",
|
||||
name="Downlink frequency",
|
||||
device_class=SensorDeviceClass.FREQUENCY,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
"earfcn": HuaweiSensorEntityDescription(
|
||||
@ -161,12 +216,7 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
name="EC/IO",
|
||||
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
|
||||
# https://wiki.teltonika.lt/view/EC/IO
|
||||
icon_fn=lambda x: (
|
||||
"mdi:signal-cellular-outline",
|
||||
"mdi:signal-cellular-1",
|
||||
"mdi:signal-cellular-2",
|
||||
"mdi:signal-cellular-3",
|
||||
)[bisect((-20, -10, -6), x if x is not None else -1000)],
|
||||
icon_fn=lambda x: signal_icon((-20, -10, -6), x),
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
@ -183,29 +233,23 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
),
|
||||
"ltedlfreq": HuaweiSensorEntityDescription(
|
||||
key="ltedlfreq",
|
||||
name="Downlink frequency",
|
||||
formatter=lambda x: (
|
||||
round(int(x) / 10) if x is not None else None,
|
||||
UnitOfFrequency.MEGAHERTZ,
|
||||
),
|
||||
name="LTE downlink frequency",
|
||||
format_fn=format_freq_mhz,
|
||||
device_class=SensorDeviceClass.FREQUENCY,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
"lteulfreq": HuaweiSensorEntityDescription(
|
||||
key="lteulfreq",
|
||||
name="Uplink frequency",
|
||||
formatter=lambda x: (
|
||||
round(int(x) / 10) if x is not None else None,
|
||||
UnitOfFrequency.MEGAHERTZ,
|
||||
),
|
||||
name="LTE uplink frequency",
|
||||
format_fn=format_freq_mhz,
|
||||
device_class=SensorDeviceClass.FREQUENCY,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
"mode": HuaweiSensorEntityDescription(
|
||||
key="mode",
|
||||
name="Mode",
|
||||
formatter=lambda x: (
|
||||
{"0": "2G", "2": "3G", "7": "4G"}.get(x, "Unknown"),
|
||||
format_fn=lambda x: (
|
||||
{"0": "2G", "2": "3G", "7": "4G"}.get(x),
|
||||
None,
|
||||
),
|
||||
icon_fn=lambda x: (
|
||||
@ -244,12 +288,7 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
name="RSCP",
|
||||
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
|
||||
# https://wiki.teltonika.lt/view/RSCP
|
||||
icon_fn=lambda x: (
|
||||
"mdi:signal-cellular-outline",
|
||||
"mdi:signal-cellular-1",
|
||||
"mdi:signal-cellular-2",
|
||||
"mdi:signal-cellular-3",
|
||||
)[bisect((-95, -85, -75), x if x is not None else -1000)],
|
||||
icon_fn=lambda x: signal_icon((-95, -85, -75), x),
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
@ -258,12 +297,7 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
name="RSRP",
|
||||
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
|
||||
# http://www.lte-anbieter.info/technik/rsrp.php
|
||||
icon_fn=lambda x: (
|
||||
"mdi:signal-cellular-outline",
|
||||
"mdi:signal-cellular-1",
|
||||
"mdi:signal-cellular-2",
|
||||
"mdi:signal-cellular-3",
|
||||
)[bisect((-110, -95, -80), x if x is not None else -1000)],
|
||||
icon_fn=lambda x: signal_icon((-110, -95, -80), x),
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=True,
|
||||
@ -273,12 +307,7 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
name="RSRQ",
|
||||
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
|
||||
# http://www.lte-anbieter.info/technik/rsrq.php
|
||||
icon_fn=lambda x: (
|
||||
"mdi:signal-cellular-outline",
|
||||
"mdi:signal-cellular-1",
|
||||
"mdi:signal-cellular-2",
|
||||
"mdi:signal-cellular-3",
|
||||
)[bisect((-11, -8, -5), x if x is not None else -1000)],
|
||||
icon_fn=lambda x: signal_icon((-11, -8, -5), x),
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=True,
|
||||
@ -288,12 +317,7 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
name="RSSI",
|
||||
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
|
||||
# https://eyesaas.com/wi-fi-signal-strength/
|
||||
icon_fn=lambda x: (
|
||||
"mdi:signal-cellular-outline",
|
||||
"mdi:signal-cellular-1",
|
||||
"mdi:signal-cellular-2",
|
||||
"mdi:signal-cellular-3",
|
||||
)[bisect((-80, -70, -60), x if x is not None else -1000)],
|
||||
icon_fn=lambda x: signal_icon((-80, -70, -60), x),
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=True,
|
||||
@ -303,12 +327,7 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
name="SINR",
|
||||
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
|
||||
# http://www.lte-anbieter.info/technik/sinr.php
|
||||
icon_fn=lambda x: (
|
||||
"mdi:signal-cellular-outline",
|
||||
"mdi:signal-cellular-1",
|
||||
"mdi:signal-cellular-2",
|
||||
"mdi:signal-cellular-3",
|
||||
)[bisect((0, 5, 10), x if x is not None else -1000)],
|
||||
icon_fn=lambda x: signal_icon((0, 5, 10), x),
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=True,
|
||||
@ -343,11 +362,13 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
"ulbandwidth": HuaweiSensorEntityDescription(
|
||||
key="ulbandwidth",
|
||||
name="Uplink bandwidth",
|
||||
icon_fn=lambda x: (
|
||||
"mdi:speedometer-slow",
|
||||
"mdi:speedometer-medium",
|
||||
"mdi:speedometer",
|
||||
)[bisect((8, 15), x if x is not None else -1000)],
|
||||
icon_fn=lambda x: bandwidth_icon((8, 15), x),
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
"ulfrequency": HuaweiSensorEntityDescription(
|
||||
key="ulfrequency",
|
||||
name="Uplink frequency",
|
||||
device_class=SensorDeviceClass.FREQUENCY,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
),
|
||||
}
|
||||
@ -367,15 +388,29 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
},
|
||||
),
|
||||
KEY_MONITORING_MONTH_STATISTICS: HuaweiSensorGroup(
|
||||
exclude=re.compile(r"^month(duration|lastcleartime)$", re.IGNORECASE),
|
||||
exclude=re.compile(
|
||||
r"^(currentday|month)(duration|lastcleartime)$", re.IGNORECASE
|
||||
),
|
||||
descriptions={
|
||||
"CurrentDayUsed": HuaweiSensorEntityDescription(
|
||||
key="CurrentDayUsed",
|
||||
name="Current day transfer",
|
||||
native_unit_of_measurement=UnitOfInformation.BYTES,
|
||||
device_class=SensorDeviceClass.DATA_SIZE,
|
||||
icon="mdi:arrow-up-down-bold",
|
||||
state_class=SensorStateClass.TOTAL,
|
||||
last_reset_item="CurrentDayDuration",
|
||||
last_reset_format_fn=format_last_reset_elapsed_seconds,
|
||||
),
|
||||
"CurrentMonthDownload": HuaweiSensorEntityDescription(
|
||||
key="CurrentMonthDownload",
|
||||
name="Current month download",
|
||||
native_unit_of_measurement=UnitOfInformation.BYTES,
|
||||
device_class=SensorDeviceClass.DATA_SIZE,
|
||||
icon="mdi:download",
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
state_class=SensorStateClass.TOTAL,
|
||||
last_reset_item="MonthDuration",
|
||||
last_reset_format_fn=format_last_reset_elapsed_seconds,
|
||||
),
|
||||
"CurrentMonthUpload": HuaweiSensorEntityDescription(
|
||||
key="CurrentMonthUpload",
|
||||
@ -383,7 +418,9 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
native_unit_of_measurement=UnitOfInformation.BYTES,
|
||||
device_class=SensorDeviceClass.DATA_SIZE,
|
||||
icon="mdi:upload",
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
state_class=SensorStateClass.TOTAL,
|
||||
last_reset_item="MonthDuration",
|
||||
last_reset_format_fn=format_last_reset_elapsed_seconds,
|
||||
),
|
||||
},
|
||||
),
|
||||
@ -521,8 +558,8 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
"State": HuaweiSensorEntityDescription(
|
||||
key="State",
|
||||
name="Operator search mode",
|
||||
formatter=lambda x: (
|
||||
{"0": "Auto", "1": "Manual"}.get(x, "Unknown"),
|
||||
format_fn=lambda x: (
|
||||
{"0": "Auto", "1": "Manual"}.get(x),
|
||||
None,
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
@ -535,16 +572,16 @@ SENSOR_META: dict[str, HuaweiSensorGroup] = {
|
||||
"NetworkMode": HuaweiSensorEntityDescription(
|
||||
key="NetworkMode",
|
||||
name="Preferred mode",
|
||||
formatter=lambda x: (
|
||||
format_fn=lambda x: (
|
||||
{
|
||||
"00": "4G/3G/2G",
|
||||
"01": "2G",
|
||||
"02": "3G",
|
||||
"03": "4G",
|
||||
"0301": "4G/2G",
|
||||
"0302": "4G/3G",
|
||||
"0201": "3G/2G",
|
||||
}.get(x, "Unknown"),
|
||||
NetworkModeEnum.MODE_AUTO.value: "4G/3G/2G",
|
||||
NetworkModeEnum.MODE_4G_3G_AUTO.value: "4G/3G",
|
||||
NetworkModeEnum.MODE_4G_2G_AUTO.value: "4G/2G",
|
||||
NetworkModeEnum.MODE_4G_ONLY.value: "4G",
|
||||
NetworkModeEnum.MODE_3G_2G_AUTO.value: "3G/2G",
|
||||
NetworkModeEnum.MODE_3G_ONLY.value: "3G",
|
||||
NetworkModeEnum.MODE_2G_ONLY.value: "2G",
|
||||
}.get(x),
|
||||
None,
|
||||
),
|
||||
entity_category=EntityCategory.CONFIG,
|
||||
@ -660,8 +697,9 @@ class HuaweiLteSensor(HuaweiLteBaseEntityWithDevice, SensorEntity):
|
||||
item: str
|
||||
entity_description: HuaweiSensorEntityDescription
|
||||
|
||||
_state: StateType = field(default=STATE_UNKNOWN, init=False)
|
||||
_state: StateType = field(default=None, init=False)
|
||||
_unit: str | None = field(default=None, init=False)
|
||||
_last_reset: datetime | None = field(default=None, init=False)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
"""Initialize remaining attributes."""
|
||||
@ -671,11 +709,19 @@ class HuaweiLteSensor(HuaweiLteBaseEntityWithDevice, SensorEntity):
|
||||
"""Subscribe to needed data on add."""
|
||||
await super().async_added_to_hass()
|
||||
self.router.subscriptions[self.key].add(f"{SENSOR_DOMAIN}/{self.item}")
|
||||
if self.entity_description.last_reset_item:
|
||||
self.router.subscriptions[self.key].add(
|
||||
f"{SENSOR_DOMAIN}/{self.entity_description.last_reset_item}"
|
||||
)
|
||||
|
||||
async def async_will_remove_from_hass(self) -> None:
|
||||
"""Unsubscribe from needed data on remove."""
|
||||
await super().async_will_remove_from_hass()
|
||||
self.router.subscriptions[self.key].remove(f"{SENSOR_DOMAIN}/{self.item}")
|
||||
if self.entity_description.last_reset_item:
|
||||
self.router.subscriptions[self.key].remove(
|
||||
f"{SENSOR_DOMAIN}/{self.entity_description.last_reset_item}"
|
||||
)
|
||||
|
||||
@property
|
||||
def _device_unique_id(self) -> str:
|
||||
@ -698,6 +744,11 @@ class HuaweiLteSensor(HuaweiLteBaseEntityWithDevice, SensorEntity):
|
||||
return self.entity_description.icon_fn(self.state)
|
||||
return self.entity_description.icon
|
||||
|
||||
@property
|
||||
def last_reset(self) -> datetime | None:
|
||||
"""Return the time when the sensor was last reset, if any."""
|
||||
return self._last_reset
|
||||
|
||||
async def async_update(self) -> None:
|
||||
"""Update state."""
|
||||
try:
|
||||
@ -706,7 +757,26 @@ class HuaweiLteSensor(HuaweiLteBaseEntityWithDevice, SensorEntity):
|
||||
_LOGGER.debug("%s[%s] not in data", self.key, self.item)
|
||||
value = None
|
||||
|
||||
formatter = self.entity_description.formatter
|
||||
last_reset = None
|
||||
if (
|
||||
self.entity_description.last_reset_item
|
||||
and self.entity_description.last_reset_format_fn
|
||||
):
|
||||
try:
|
||||
last_reset_value = self.router.data[self.key][
|
||||
self.entity_description.last_reset_item
|
||||
]
|
||||
except KeyError:
|
||||
_LOGGER.debug(
|
||||
"%s[%s] not in data",
|
||||
self.key,
|
||||
self.entity_description.last_reset_item,
|
||||
)
|
||||
else:
|
||||
last_reset = self.entity_description.last_reset_format_fn(
|
||||
last_reset_value
|
||||
)
|
||||
|
||||
self._state, self._unit = formatter(value)
|
||||
self._state, self._unit = self.entity_description.format_fn(value)
|
||||
self._last_reset = last_reset
|
||||
self._available = value is not None
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user