Split models and helpers from coordinator module in AVM Fritz!Box tools (#147412)

* split models from coordinator

* split helpers from coordinator

homeassistant/components/fritz/binary_sensor.py

@@ -15,8 +15,9 @@ from homeassistant.const import EntityCategory
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from .coordinator import ConnectionInfo, FritzConfigEntry
+from .coordinator import FritzConfigEntry
 from .entity import FritzBoxBaseCoordinatorEntity, FritzEntityDescription
+from .models import ConnectionInfo
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/fritz/button.py

@@ -19,15 +19,10 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
 from .const import BUTTON_TYPE_WOL, CONNECTION_TYPE_LAN, MeshRoles
-from .coordinator import (
-    FRITZ_DATA_KEY,
-    AvmWrapper,
-    FritzConfigEntry,
-    FritzData,
-    FritzDevice,
-    _is_tracked,
-)
+from .coordinator import FRITZ_DATA_KEY, AvmWrapper, FritzConfigEntry, FritzData
 from .entity import FritzDeviceBase
+from .helpers import _is_tracked
+from .models import FritzDevice
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/fritz/coordinator.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping, ValuesView
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
 from functools import partial
@@ -34,7 +34,6 @@ from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
 from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.typing import StateType
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
-from homeassistant.util import dt as dt_util
 from homeassistant.util.hass_dict import HassKey
 
 from .const import (
@@ -48,6 +47,15 @@ from .const import (
     FRITZ_EXCEPTIONS,
     MeshRoles,
 )
+from .helpers import _ha_is_stopping
+from .models import (
+    ConnectionInfo,
+    Device,
+    FritzDevice,
+    HostAttributes,
+    HostInfo,
+    Interface,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -56,33 +64,13 @@ FRITZ_DATA_KEY: HassKey[FritzData] = HassKey(DOMAIN)
 type FritzConfigEntry = ConfigEntry[AvmWrapper]
 
 
-def _is_tracked(mac: str, current_devices: ValuesView[set[str]]) -> bool:
-    """Check if device is already tracked."""
-    return any(mac in tracked for tracked in current_devices)
+@dataclass
+class FritzData:
+    """Storage class for platform global data."""
 
-
-def device_filter_out_from_trackers(
-    mac: str,
-    device: FritzDevice,
-    current_devices: ValuesView[set[str]],
-) -> bool:
-    """Check if device should be filtered out from trackers."""
-    reason: str | None = None
-    if device.ip_address == "":
-        reason = "Missing IP"
-    elif _is_tracked(mac, current_devices):
-        reason = "Already tracked"
-
-    if reason:
-        _LOGGER.debug(
-            "Skip adding device %s [%s], reason: %s", device.hostname, mac, reason
-        )
-    return bool(reason)
-
-
-def _ha_is_stopping(activity: str) -> None:
-    """Inform that HA is stopping."""
-    _LOGGER.warning("Cannot execute %s: HomeAssistant is shutting down", activity)
+    tracked: dict[str, set[str]] = field(default_factory=dict)
+    profile_switches: dict[str, set[str]] = field(default_factory=dict)
+    wol_buttons: dict[str, set[str]] = field(default_factory=dict)
 
 
 class ClassSetupMissing(Exception):
@@ -93,68 +81,6 @@ class ClassSetupMissing(Exception):
         super().__init__("Function called before Class setup")
 
 
-@dataclass
-class Device:
-    """FRITZ!Box device class."""
-
-    connected: bool
-    connected_to: str
-    connection_type: str
-    ip_address: str
-    name: str
-    ssid: str | None
-    wan_access: bool | None = None
-
-
-class Interface(TypedDict):
-    """Interface details."""
-
-    device: str
-    mac: str
-    op_mode: str
-    ssid: str | None
-    type: str
-
-
-HostAttributes = TypedDict(
-    "HostAttributes",
-    {
-        "Index": int,
-        "IPAddress": str,
-        "MACAddress": str,
-        "Active": bool,
-        "HostName": str,
-        "InterfaceType": str,
-        "X_AVM-DE_Port": int,
-        "X_AVM-DE_Speed": int,
-        "X_AVM-DE_UpdateAvailable": bool,
-        "X_AVM-DE_UpdateSuccessful": str,
-        "X_AVM-DE_InfoURL": str | None,
-        "X_AVM-DE_MACAddressList": str | None,
-        "X_AVM-DE_Model": str | None,
-        "X_AVM-DE_URL": str | None,
-        "X_AVM-DE_Guest": bool,
-        "X_AVM-DE_RequestClient": str,
-        "X_AVM-DE_VPN": bool,
-        "X_AVM-DE_WANAccess": str,
-        "X_AVM-DE_Disallow": bool,
-        "X_AVM-DE_IsMeshable": str,
-        "X_AVM-DE_Priority": str,
-        "X_AVM-DE_FriendlyName": str,
-        "X_AVM-DE_FriendlyNameIsWriteable": str,
-    },
-)
-
-
-class HostInfo(TypedDict):
-    """FRITZ!Box host info class."""
-
-    mac: str
-    name: str
-    ip: str
-    status: bool
-
-
 class UpdateCoordinatorDataType(TypedDict):
     """Update coordinator data type."""
 
@@ -898,120 +824,3 @@ class AvmWrapper(FritzBoxTools):
             "X_AVM-DE_WakeOnLANByMACAddress",
             NewMACAddress=mac_address,
         )
-
-
-@dataclass
-class FritzData:
-    """Storage class for platform global data."""
-
-    tracked: dict[str, set[str]] = field(default_factory=dict)
-    profile_switches: dict[str, set[str]] = field(default_factory=dict)
-    wol_buttons: dict[str, set[str]] = field(default_factory=dict)
-
-
-class FritzDevice:
-    """Representation of a device connected to the FRITZ!Box."""
-
-    def __init__(self, mac: str, name: str) -> None:
-        """Initialize device info."""
-        self._connected = False
-        self._connected_to: str | None = None
-        self._connection_type: str | None = None
-        self._ip_address: str | None = None
-        self._last_activity: datetime | None = None
-        self._mac = mac
-        self._name = name
-        self._ssid: str | None = None
-        self._wan_access: bool | None = False
-
-    def update(self, dev_info: Device, consider_home: float) -> None:
-        """Update device info."""
-        utc_point_in_time = dt_util.utcnow()
-
-        if self._last_activity:
-            consider_home_evaluated = (
-                utc_point_in_time - self._last_activity
-            ).total_seconds() < consider_home
-        else:
-            consider_home_evaluated = dev_info.connected
-
-        if not self._name:
-            self._name = dev_info.name or self._mac.replace(":", "_")
-
-        self._connected = dev_info.connected or consider_home_evaluated
-
-        if dev_info.connected:
-            self._last_activity = utc_point_in_time
-
-        self._connected_to = dev_info.connected_to
-        self._connection_type = dev_info.connection_type
-        self._ip_address = dev_info.ip_address
-        self._ssid = dev_info.ssid
-        self._wan_access = dev_info.wan_access
-
-    @property
-    def connected_to(self) -> str | None:
-        """Return connected status."""
-        return self._connected_to
-
-    @property
-    def connection_type(self) -> str | None:
-        """Return connected status."""
-        return self._connection_type
-
-    @property
-    def is_connected(self) -> bool:
-        """Return connected status."""
-        return self._connected
-
-    @property
-    def mac_address(self) -> str:
-        """Get MAC address."""
-        return self._mac
-
-    @property
-    def hostname(self) -> str:
-        """Get Name."""
-        return self._name
-
-    @property
-    def ip_address(self) -> str | None:
-        """Get IP address."""
-        return self._ip_address
-
-    @property
-    def last_activity(self) -> datetime | None:
-        """Return device last activity."""
-        return self._last_activity
-
-    @property
-    def ssid(self) -> str | None:
-        """Return device connected SSID."""
-        return self._ssid
-
-    @property
-    def wan_access(self) -> bool | None:
-        """Return device wan access."""
-        return self._wan_access
-
-
-class SwitchInfo(TypedDict):
-    """FRITZ!Box switch info class."""
-
-    description: str
-    friendly_name: str
-    icon: str
-    type: str
-    callback_update: Callable
-    callback_switch: Callable
-    init_state: bool
-
-
-@dataclass
-class ConnectionInfo:
-    """Fritz sensor connection information class."""
-
-    connection: str
-    mesh_role: MeshRoles
-    wan_enabled: bool
-    ipv6_active: bool

homeassistant/components/fritz/device_tracker.py

@@ -10,15 +10,10 @@ from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
-from .coordinator import (
-    FRITZ_DATA_KEY,
-    AvmWrapper,
-    FritzConfigEntry,
-    FritzData,
-    FritzDevice,
-    device_filter_out_from_trackers,
-)
+from .coordinator import FRITZ_DATA_KEY, AvmWrapper, FritzConfigEntry, FritzData
 from .entity import FritzDeviceBase
+from .helpers import device_filter_out_from_trackers
+from .models import FritzDevice
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/fritz/entity.py

@@ -14,7 +14,8 @@ from homeassistant.helpers.entity import EntityDescription
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DEFAULT_DEVICE_NAME, DOMAIN
-from .coordinator import AvmWrapper, FritzDevice
+from .coordinator import AvmWrapper
+from .models import FritzDevice
 
 
 class FritzDeviceBase(CoordinatorEntity[AvmWrapper]):

homeassistant/components/fritz/helpers.py

@@ -0,0 +1,39 @@
+"""Helpers for AVM FRITZ!Box."""
+
+from __future__ import annotations
+
+from collections.abc import ValuesView
+import logging
+
+from .models import FritzDevice
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _is_tracked(mac: str, current_devices: ValuesView[set[str]]) -> bool:
+    """Check if device is already tracked."""
+    return any(mac in tracked for tracked in current_devices)
+
+
+def device_filter_out_from_trackers(
+    mac: str,
+    device: FritzDevice,
+    current_devices: ValuesView[set[str]],
+) -> bool:
+    """Check if device should be filtered out from trackers."""
+    reason: str | None = None
+    if device.ip_address == "":
+        reason = "Missing IP"
+    elif _is_tracked(mac, current_devices):
+        reason = "Already tracked"
+
+    if reason:
+        _LOGGER.debug(
+            "Skip adding device %s [%s], reason: %s", device.hostname, mac, reason
+        )
+    return bool(reason)
+
+
+def _ha_is_stopping(activity: str) -> None:
+    """Inform that HA is stopping."""
+    _LOGGER.warning("Cannot execute %s: HomeAssistant is shutting down", activity)

homeassistant/components/fritz/models.py

@@ -0,0 +1,182 @@
+"""Models for AVM FRITZ!Box."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import datetime
+from typing import TypedDict
+
+from homeassistant.util import dt as dt_util
+
+from .const import MeshRoles
+
+
+@dataclass
+class Device:
+    """FRITZ!Box device class."""
+
+    connected: bool
+    connected_to: str
+    connection_type: str
+    ip_address: str
+    name: str
+    ssid: str | None
+    wan_access: bool | None = None
+
+
+class Interface(TypedDict):
+    """Interface details."""
+
+    device: str
+    mac: str
+    op_mode: str
+    ssid: str | None
+    type: str
+
+
+HostAttributes = TypedDict(
+    "HostAttributes",
+    {
+        "Index": int,
+        "IPAddress": str,
+        "MACAddress": str,
+        "Active": bool,
+        "HostName": str,
+        "InterfaceType": str,
+        "X_AVM-DE_Port": int,
+        "X_AVM-DE_Speed": int,
+        "X_AVM-DE_UpdateAvailable": bool,
+        "X_AVM-DE_UpdateSuccessful": str,
+        "X_AVM-DE_InfoURL": str | None,
+        "X_AVM-DE_MACAddressList": str | None,
+        "X_AVM-DE_Model": str | None,
+        "X_AVM-DE_URL": str | None,
+        "X_AVM-DE_Guest": bool,
+        "X_AVM-DE_RequestClient": str,
+        "X_AVM-DE_VPN": bool,
+        "X_AVM-DE_WANAccess": str,
+        "X_AVM-DE_Disallow": bool,
+        "X_AVM-DE_IsMeshable": str,
+        "X_AVM-DE_Priority": str,
+        "X_AVM-DE_FriendlyName": str,
+        "X_AVM-DE_FriendlyNameIsWriteable": str,
+    },
+)
+
+
+class HostInfo(TypedDict):
+    """FRITZ!Box host info class."""
+
+    mac: str
+    name: str
+    ip: str
+    status: bool
+
+
+class FritzDevice:
+    """Representation of a device connected to the FRITZ!Box."""
+
+    def __init__(self, mac: str, name: str) -> None:
+        """Initialize device info."""
+        self._connected = False
+        self._connected_to: str | None = None
+        self._connection_type: str | None = None
+        self._ip_address: str | None = None
+        self._last_activity: datetime | None = None
+        self._mac = mac
+        self._name = name
+        self._ssid: str | None = None
+        self._wan_access: bool | None = False
+
+    def update(self, dev_info: Device, consider_home: float) -> None:
+        """Update device info."""
+        utc_point_in_time = dt_util.utcnow()
+
+        if self._last_activity:
+            consider_home_evaluated = (
+                utc_point_in_time - self._last_activity
+            ).total_seconds() < consider_home
+        else:
+            consider_home_evaluated = dev_info.connected
+
+        if not self._name:
+            self._name = dev_info.name or self._mac.replace(":", "_")
+
+        self._connected = dev_info.connected or consider_home_evaluated
+
+        if dev_info.connected:
+            self._last_activity = utc_point_in_time
+
+        self._connected_to = dev_info.connected_to
+        self._connection_type = dev_info.connection_type
+        self._ip_address = dev_info.ip_address
+        self._ssid = dev_info.ssid
+        self._wan_access = dev_info.wan_access
+
+    @property
+    def connected_to(self) -> str | None:
+        """Return connected status."""
+        return self._connected_to
+
+    @property
+    def connection_type(self) -> str | None:
+        """Return connected status."""
+        return self._connection_type
+
+    @property
+    def is_connected(self) -> bool:
+        """Return connected status."""
+        return self._connected
+
+    @property
+    def mac_address(self) -> str:
+        """Get MAC address."""
+        return self._mac
+
+    @property
+    def hostname(self) -> str:
+        """Get Name."""
+        return self._name
+
+    @property
+    def ip_address(self) -> str | None:
+        """Get IP address."""
+        return self._ip_address
+
+    @property
+    def last_activity(self) -> datetime | None:
+        """Return device last activity."""
+        return self._last_activity
+
+    @property
+    def ssid(self) -> str | None:
+        """Return device connected SSID."""
+        return self._ssid
+
+    @property
+    def wan_access(self) -> bool | None:
+        """Return device wan access."""
+        return self._wan_access
+
+
+class SwitchInfo(TypedDict):
+    """FRITZ!Box switch info class."""
+
+    description: str
+    friendly_name: str
+    icon: str
+    type: str
+    callback_update: Callable
+    callback_switch: Callable
+    init_state: bool
+
+
+@dataclass
+class ConnectionInfo:
+    """Fritz sensor connection information class."""
+
+    connection: str
+    mesh_role: MeshRoles
+    wan_enabled: bool
+    ipv6_active: bool

homeassistant/components/fritz/sensor.py

@@ -27,8 +27,9 @@ from homeassistant.helpers.typing import StateType
 from homeassistant.util.dt import utcnow
 
 from .const import DSL_CONNECTION, UPTIME_DEVIATION
-from .coordinator import ConnectionInfo, FritzConfigEntry
+from .coordinator import FritzConfigEntry
 from .entity import FritzBoxBaseCoordinatorEntity, FritzEntityDescription
+from .models import ConnectionInfo
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/fritz/switch.py

@@ -25,16 +25,10 @@ from .const import (
     WIFI_STANDARD,
     MeshRoles,
 )
-from .coordinator import (
-    FRITZ_DATA_KEY,
-    AvmWrapper,
-    FritzConfigEntry,
-    FritzData,
-    FritzDevice,
-    SwitchInfo,
-    device_filter_out_from_trackers,
-)
+from .coordinator import FRITZ_DATA_KEY, AvmWrapper, FritzConfigEntry, FritzData
 from .entity import FritzBoxBaseEntity, FritzDeviceBase
+from .helpers import device_filter_out_from_trackers
+from .models import FritzDevice, SwitchInfo
 
 _LOGGER = logging.getLogger(__name__)