Bump sfrbox-api to 0.0.10 (#125405)

* bump sfr_box requirement to 0.0.10

* upate manifest file

* Handle None values

---------

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>
pull/125421/head
Alexandre TRUPIN 2024-09-06 16:18:47 +02:00 committed by GitHub
parent e58cf00a96
commit c4cc158a77
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 68 additions and 40 deletions

View File

@ -3,6 +3,7 @@
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING
from sfrbox_api.bridge import SFRBox
from sfrbox_api.exceptions import SFRBoxAuthenticationError, SFRBoxError
@ -46,6 +47,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Preload system information
await data.system.async_config_entry_first_refresh()
system_info = data.system.data
if TYPE_CHECKING:
assert system_info is not None
# Preload other coordinators (based on net infrastructure)
tasks = [data.wan.async_config_entry_first_refresh()]

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING
from sfrbox_api.models import DslInfo, FtthInfo, SystemInfo, WanInfo
@ -65,19 +66,22 @@ async def async_setup_entry(
) -> None:
"""Set up the sensors."""
data: DomainData = hass.data[DOMAIN][entry.entry_id]
system_info = data.system.data
if TYPE_CHECKING:
assert system_info is not None
entities: list[SFRBoxBinarySensor] = [
SFRBoxBinarySensor(data.wan, description, data.system.data)
SFRBoxBinarySensor(data.wan, description, system_info)
for description in WAN_SENSOR_TYPES
]
if (net_infra := data.system.data.net_infra) == "adsl":
if (net_infra := system_info.net_infra) == "adsl":
entities.extend(
SFRBoxBinarySensor(data.dsl, description, data.system.data)
SFRBoxBinarySensor(data.dsl, description, system_info)
for description in DSL_SENSOR_TYPES
)
elif net_infra == "ftth":
entities.extend(
SFRBoxBinarySensor(data.ftth, description, data.system.data)
SFRBoxBinarySensor(data.ftth, description, system_info)
for description in FTTH_SENSOR_TYPES
)
@ -111,4 +115,6 @@ class SFRBoxBinarySensor[_T](
@property
def is_on(self) -> bool | None:
"""Return the native value of the device."""
if self.coordinator.data is None:
return None
return self.entity_description.value_fn(self.coordinator.data)

View File

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from functools import wraps
from typing import Any, Concatenate
from typing import TYPE_CHECKING, Any, Concatenate
from sfrbox_api.bridge import SFRBox
from sfrbox_api.exceptions import SFRBoxError
@ -69,10 +69,12 @@ async def async_setup_entry(
) -> None:
"""Set up the buttons."""
data: DomainData = hass.data[DOMAIN][entry.entry_id]
system_info = data.system.data
if TYPE_CHECKING:
assert system_info is not None
entities = [
SFRBoxButton(data.box, description, data.system.data)
for description in BUTTON_TYPES
SFRBoxButton(data.box, description, system_info) for description in BUTTON_TYPES
]
async_add_entities(entities)

View File

@ -3,7 +3,7 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from typing import TYPE_CHECKING, Any
from sfrbox_api.bridge import SFRBox
from sfrbox_api.exceptions import SFRBoxAuthenticationError, SFRBoxError
@ -51,6 +51,8 @@ class SFRBoxFlowHandler(ConfigFlow, domain=DOMAIN):
except SFRBoxError:
errors["base"] = "cannot_connect"
else:
if TYPE_CHECKING:
assert system_info is not None
await self.async_set_unique_id(system_info.mac_addr)
self._abort_if_unique_id_configured()
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})

View File

@ -15,7 +15,7 @@ _LOGGER = logging.getLogger(__name__)
_SCAN_INTERVAL = timedelta(minutes=1)
class SFRDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
class SFRDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT | None]):
"""Coordinator to manage data updates."""
def __init__(
@ -23,14 +23,14 @@ class SFRDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
hass: HomeAssistant,
box: SFRBox,
name: str,
method: Callable[[SFRBox], Coroutine[Any, Any, _DataT]],
method: Callable[[SFRBox], Coroutine[Any, Any, _DataT | None]],
) -> None:
"""Initialize coordinator."""
self.box = box
self._method = method
super().__init__(hass, _LOGGER, name=name, update_interval=_SCAN_INTERVAL)
async def _async_update_data(self) -> _DataT:
async def _async_update_data(self) -> _DataT | None:
"""Update data."""
try:
return await self._method(self.box)

View File

@ -3,7 +3,7 @@
from __future__ import annotations
import dataclasses
from typing import Any
from typing import TYPE_CHECKING, Any
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.config_entries import ConfigEntry
@ -12,9 +12,18 @@ from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .models import DomainData
if TYPE_CHECKING:
from _typeshed import DataclassInstance
TO_REDACT = {"mac_addr", "serial_number", "ip_addr", "ipv6_addr"}
def _async_redact_data(obj: DataclassInstance | None) -> dict[str, Any] | None:
if obj is None:
return None
return async_redact_data(dataclasses.asdict(obj), TO_REDACT)
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ConfigEntry
) -> dict[str, Any]:
@ -27,21 +36,9 @@ async def async_get_config_entry_diagnostics(
"data": dict(entry.data),
},
"data": {
"dsl": async_redact_data(
dataclasses.asdict(await data.system.box.dsl_get_info()),
TO_REDACT,
),
"ftth": async_redact_data(
dataclasses.asdict(await data.system.box.ftth_get_info()),
TO_REDACT,
),
"system": async_redact_data(
dataclasses.asdict(await data.system.box.system_get_info()),
TO_REDACT,
),
"wan": async_redact_data(
dataclasses.asdict(await data.system.box.wan_get_info()),
TO_REDACT,
),
"dsl": _async_redact_data(await data.system.box.dsl_get_info()),
"ftth": _async_redact_data(await data.system.box.ftth_get_info()),
"system": _async_redact_data(await data.system.box.system_get_info()),
"wan": _async_redact_data(await data.system.box.wan_get_info()),
},
}

View File

@ -6,5 +6,5 @@
"documentation": "https://www.home-assistant.io/integrations/sfr_box",
"integration_type": "device",
"iot_class": "local_polling",
"requirements": ["sfrbox-api==0.0.8"]
"requirements": ["sfrbox-api==0.0.10"]
}

View File

@ -2,6 +2,7 @@
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING
from sfrbox_api.models import DslInfo, SystemInfo, WanInfo
@ -129,7 +130,7 @@ DSL_SENSOR_TYPES: tuple[SFRBoxSensorEntityDescription[DslInfo], ...] = (
"unknown",
],
translation_key="dsl_line_status",
value_fn=lambda x: x.line_status.lower().replace(" ", "_"),
value_fn=lambda x: _value_to_option(x.line_status),
),
SFRBoxSensorEntityDescription[DslInfo](
key="training",
@ -149,7 +150,7 @@ DSL_SENSOR_TYPES: tuple[SFRBoxSensorEntityDescription[DslInfo], ...] = (
"unknown",
],
translation_key="dsl_training",
value_fn=lambda x: x.training.lower().replace(" ", "_").replace(".", "_"),
value_fn=lambda x: _value_to_option(x.training),
),
)
SYSTEM_SENSOR_TYPES: tuple[SFRBoxSensorEntityDescription[SystemInfo], ...] = (
@ -181,7 +182,7 @@ SYSTEM_SENSOR_TYPES: tuple[SFRBoxSensorEntityDescription[SystemInfo], ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
value_fn=lambda x: None if x.temperature is None else x.temperature / 1000,
value_fn=lambda x: _get_temperature(x.temperature),
),
)
WAN_SENSOR_TYPES: tuple[SFRBoxSensorEntityDescription[WanInfo], ...] = (
@ -203,23 +204,38 @@ WAN_SENSOR_TYPES: tuple[SFRBoxSensorEntityDescription[WanInfo], ...] = (
)
def _value_to_option(value: str | None) -> str | None:
if value is None:
return value
return value.lower().replace(" ", "_").replace(".", "_")
def _get_temperature(value: float | None) -> float | None:
if value is None or value < 1000:
return value
return value / 1000
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the sensors."""
data: DomainData = hass.data[DOMAIN][entry.entry_id]
system_info = data.system.data
if TYPE_CHECKING:
assert system_info is not None
entities: list[SFRBoxSensor] = [
SFRBoxSensor(data.system, description, data.system.data)
SFRBoxSensor(data.system, description, system_info)
for description in SYSTEM_SENSOR_TYPES
]
entities.extend(
SFRBoxSensor(data.wan, description, data.system.data)
SFRBoxSensor(data.wan, description, system_info)
for description in WAN_SENSOR_TYPES
)
if data.system.data.net_infra == "adsl":
if system_info.net_infra == "adsl":
entities.extend(
SFRBoxSensor(data.dsl, description, data.system.data)
SFRBoxSensor(data.dsl, description, system_info)
for description in DSL_SENSOR_TYPES
)
@ -251,4 +267,6 @@ class SFRBoxSensor[_T](CoordinatorEntity[SFRDataUpdateCoordinator[_T]], SensorEn
@property
def native_value(self) -> StateType:
"""Return the native value of the device."""
if self.coordinator.data is None:
return None
return self.entity_description.value_fn(self.coordinator.data)

View File

@ -2616,7 +2616,7 @@ sensoterra==2.0.1
sentry-sdk==1.40.3
# homeassistant.components.sfr_box
sfrbox-api==0.0.8
sfrbox-api==0.0.10
# homeassistant.components.sharkiq
sharkiq==1.0.2

View File

@ -2074,7 +2074,7 @@ sensoterra==2.0.1
sentry-sdk==1.40.3
# homeassistant.components.sfr_box
sfrbox-api==0.0.8
sfrbox-api==0.0.10
# homeassistant.components.sharkiq
sharkiq==1.0.2

View File

@ -31,7 +31,7 @@
'product_id': 'NB6VAC-FXC-r0',
'refclient': '',
'serial_number': '**REDACTED**',
'temperature': 27560,
'temperature': 27560.0,
'uptime': 2353575,
'version_bootloader': 'NB6VAC-BOOTLOADER-R4.0.8',
'version_dsldriver': 'NB6VAC-XDSL-A2pv6F039p',
@ -90,7 +90,7 @@
'product_id': 'NB6VAC-FXC-r0',
'refclient': '',
'serial_number': '**REDACTED**',
'temperature': 27560,
'temperature': 27560.0,
'uptime': 2353575,
'version_bootloader': 'NB6VAC-BOOTLOADER-R4.0.8',
'version_dsldriver': 'NB6VAC-XDSL-A2pv6F039p',