mirror of
https://github.com/home-assistant/core.git
synced 2025-07-29 08:07:45 +00:00
Make connected stations coordinator a dict in devolo Home Network (#147042)
This commit is contained in:
parent
2ea09ff37a
commit
8d82e34ba5
@ -207,7 +207,7 @@ class DevoloUptimeGetCoordinator(DevoloDataUpdateCoordinator[int]):
|
||||
|
||||
|
||||
class DevoloWifiConnectedStationsGetCoordinator(
|
||||
DevoloDataUpdateCoordinator[list[ConnectedStationInfo]]
|
||||
DevoloDataUpdateCoordinator[dict[str, ConnectedStationInfo]]
|
||||
):
|
||||
"""Class to manage fetching data from the WifiGuestAccessGet endpoint."""
|
||||
|
||||
@ -230,10 +230,11 @@ class DevoloWifiConnectedStationsGetCoordinator(
|
||||
)
|
||||
self.update_method = self.async_get_wifi_connected_station
|
||||
|
||||
async def async_get_wifi_connected_station(self) -> list[ConnectedStationInfo]:
|
||||
async def async_get_wifi_connected_station(self) -> dict[str, ConnectedStationInfo]:
|
||||
"""Fetch data from API endpoint."""
|
||||
assert self.device.device
|
||||
return await self.device.device.async_get_wifi_connected_station()
|
||||
clients = await self.device.device.async_get_wifi_connected_station()
|
||||
return {client.mac_address: client for client in clients}
|
||||
|
||||
|
||||
class DevoloWifiGuestAccessGetCoordinator(
|
||||
|
@ -28,9 +28,9 @@ async def async_setup_entry(
|
||||
) -> None:
|
||||
"""Get all devices and sensors and setup them via config entry."""
|
||||
device = entry.runtime_data.device
|
||||
coordinators: dict[str, DevoloDataUpdateCoordinator[list[ConnectedStationInfo]]] = (
|
||||
entry.runtime_data.coordinators
|
||||
)
|
||||
coordinators: dict[
|
||||
str, DevoloDataUpdateCoordinator[dict[str, ConnectedStationInfo]]
|
||||
] = entry.runtime_data.coordinators
|
||||
registry = er.async_get(hass)
|
||||
tracked = set()
|
||||
|
||||
@ -38,16 +38,16 @@ async def async_setup_entry(
|
||||
def new_device_callback() -> None:
|
||||
"""Add new devices if needed."""
|
||||
new_entities = []
|
||||
for station in coordinators[CONNECTED_WIFI_CLIENTS].data:
|
||||
if station.mac_address in tracked:
|
||||
for mac_address in coordinators[CONNECTED_WIFI_CLIENTS].data:
|
||||
if mac_address in tracked:
|
||||
continue
|
||||
|
||||
new_entities.append(
|
||||
DevoloScannerEntity(
|
||||
coordinators[CONNECTED_WIFI_CLIENTS], device, station.mac_address
|
||||
coordinators[CONNECTED_WIFI_CLIENTS], device, mac_address
|
||||
)
|
||||
)
|
||||
tracked.add(station.mac_address)
|
||||
tracked.add(mac_address)
|
||||
async_add_entities(new_entities)
|
||||
|
||||
@callback
|
||||
@ -82,7 +82,7 @@ async def async_setup_entry(
|
||||
|
||||
# The pylint disable is needed because of https://github.com/pylint-dev/pylint/issues/9138
|
||||
class DevoloScannerEntity( # pylint: disable=hass-enforce-class-module
|
||||
CoordinatorEntity[DevoloDataUpdateCoordinator[list[ConnectedStationInfo]]],
|
||||
CoordinatorEntity[DevoloDataUpdateCoordinator[dict[str, ConnectedStationInfo]]],
|
||||
ScannerEntity,
|
||||
):
|
||||
"""Representation of a devolo device tracker."""
|
||||
@ -92,7 +92,7 @@ class DevoloScannerEntity( # pylint: disable=hass-enforce-class-module
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: DevoloDataUpdateCoordinator[list[ConnectedStationInfo]],
|
||||
coordinator: DevoloDataUpdateCoordinator[dict[str, ConnectedStationInfo]],
|
||||
device: Device,
|
||||
mac: str,
|
||||
) -> None:
|
||||
@ -109,14 +109,8 @@ class DevoloScannerEntity( # pylint: disable=hass-enforce-class-module
|
||||
if not self.coordinator.data:
|
||||
return {}
|
||||
|
||||
station = next(
|
||||
(
|
||||
station
|
||||
for station in self.coordinator.data
|
||||
if station.mac_address == self.mac_address
|
||||
),
|
||||
None,
|
||||
)
|
||||
assert self.mac_address
|
||||
station = self.coordinator.data.get(self.mac_address)
|
||||
if station:
|
||||
attrs["wifi"] = WIFI_APTYPE.get(station.vap_type, STATE_UNKNOWN)
|
||||
attrs["band"] = (
|
||||
@ -129,11 +123,8 @@ class DevoloScannerEntity( # pylint: disable=hass-enforce-class-module
|
||||
@property
|
||||
def is_connected(self) -> bool:
|
||||
"""Return true if the device is connected to the network."""
|
||||
return any(
|
||||
station
|
||||
for station in self.coordinator.data
|
||||
if station.mac_address == self.mac_address
|
||||
)
|
||||
assert self.mac_address
|
||||
return self.coordinator.data.get(self.mac_address) is not None
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
|
@ -21,7 +21,7 @@ from .coordinator import DevoloDataUpdateCoordinator, DevoloHomeNetworkConfigEnt
|
||||
type _DataType = (
|
||||
LogicalNetwork
|
||||
| DataRate
|
||||
| list[ConnectedStationInfo]
|
||||
| dict[str, ConnectedStationInfo]
|
||||
| list[NeighborAPInfo]
|
||||
| WifiGuestAccessGet
|
||||
| bool
|
||||
|
@ -47,7 +47,11 @@ def _last_restart(runtime: int) -> datetime:
|
||||
|
||||
|
||||
type _CoordinatorDataType = (
|
||||
LogicalNetwork | DataRate | list[ConnectedStationInfo] | list[NeighborAPInfo] | int
|
||||
LogicalNetwork
|
||||
| DataRate
|
||||
| dict[str, ConnectedStationInfo]
|
||||
| list[NeighborAPInfo]
|
||||
| int
|
||||
)
|
||||
type _SensorDataType = int | float | datetime
|
||||
|
||||
@ -79,7 +83,7 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription[Any, Any]] = {
|
||||
),
|
||||
),
|
||||
CONNECTED_WIFI_CLIENTS: DevoloSensorEntityDescription[
|
||||
list[ConnectedStationInfo], int
|
||||
dict[str, ConnectedStationInfo], int
|
||||
](
|
||||
key=CONNECTED_WIFI_CLIENTS,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
|
Loading…
x
Reference in New Issue
Block a user