Create coordinator

Author: Andre0512

README.md

@@ -9,8 +9,9 @@ Telekom Speedport Integration for Home Assistant based on [speedport-api](https:
 | ⚠️ **Work in progress...** ⚠️ |
 |------------------------------|
 
-Currently, this integration can be used to track devices in network.
-
+- Track presence of connected devices
+- Turn on/off wifi (guest/office/normal)
+- Reconnect, reboot, wps on
 
 ## Installation
 #### Installing via HACS

custom_components/speedport/__init__.py

@@ -20,6 +20,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         host=entry.data["host"], password=entry.data["password"], session=session
     ).create()
     hass.data[DOMAIN][entry.entry_id] = speedport
+    hass.data[DOMAIN]["coordinators"] = {}
 
     for platform in PLATFORMS:
         hass.async_create_task(

custom_components/speedport/button.py

@@ -5,9 +5,11 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity import EntityCategory
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import HomeAssistantType
 from speedport import Speedport
 
 from .const import DOMAIN
+from .device import SpeedportEntity
 
 
 async def async_setup_entry(
@@ -18,53 +20,56 @@ async def async_setup_entry(
     coordinator = hass.data[DOMAIN][entry.entry_id]
     async_add_entities(
         [
-            SpeedportReconnectButton(coordinator),
-            SpeedportRebootButton(coordinator),
-            SpeedportWPSButton(coordinator),
+            SpeedportReconnectButton(hass, coordinator),
+            SpeedportRebootButton(hass, coordinator),
+            SpeedportWPSButton(hass, coordinator),
         ]
     )
 
 
-class SpeedportReconnectButton(ButtonEntity):
+class SpeedportReconnectButton(ButtonEntity, SpeedportEntity):
     _attr_device_class = ButtonDeviceClass.RESTART
     _attr_entity_category = EntityCategory.CONFIG
-    _attr_name = "Reconnect"
 
-    def __init__(self, coordinator) -> None:
+    def __init__(self, hass: HomeAssistantType, speedport: Speedport) -> None:
         """Initialize the button entity."""
-        self.coordinator = coordinator
+        super().__init__(hass, speedport)
+        self._attr_name = "Reconnect"
+        self._speedport = speedport
         self._attr_unique_id = "speedport_reconnect"
 
     async def async_press(self) -> None:
         """Send out a restart command."""
-        await self.coordinator.reconnect()
+        await self._speedport.reconnect()
 
 
-class SpeedportRebootButton(ButtonEntity):
+class SpeedportRebootButton(ButtonEntity, SpeedportEntity):
     _attr_device_class = ButtonDeviceClass.RESTART
     _attr_entity_category = EntityCategory.CONFIG
-    _attr_name = "Reboot"
 
-    def __init__(self, coordinator) -> None:
+    def __init__(self, hass: HomeAssistantType, speedport: Speedport) -> None:
         """Initialize the button entity."""
-        self.coordinator = coordinator
+        super().__init__(hass, speedport)
+        self._attr_name = "Reboot"
+        self._speedport = speedport
         self._attr_unique_id = "speedport_reboot"
 
     async def async_press(self) -> None:
         """Send out a restart command."""
-        await self.coordinator.reboot()
+        await self._speedport.reboot()
 
 
-class SpeedportWPSButton(ButtonEntity):
+class SpeedportWPSButton(ButtonEntity, SpeedportEntity):
     _attr_device_class = ButtonDeviceClass.IDENTIFY
     _attr_entity_category = EntityCategory.CONFIG
-    _attr_name = "WPS on"
 
-    def __init__(self, coordinator) -> None:
+    def __init__(self, hass: HomeAssistantType, speedport: Speedport) -> None:
         """Initialize the button entity."""
-        self.coordinator: Speedport = coordinator
+        super().__init__(hass, speedport)
+        self._attr_name = "WPS on"
+        self._speedport: Speedport = speedport
         self._attr_unique_id = "speedport_wps"
 
     async def async_press(self) -> None:
         """Send out a restart command."""
-        await self.coordinator.wps_on()
+        await self._speedport.wps_on()

custom_components/speedport/const.py

@@ -1,3 +1,4 @@
 """Constants for the Speedport integration."""
 
 DOMAIN = "speedport"
+UPDATE_INTERVAL = 5

custom_components/speedport/device.py

@@ -0,0 +1,71 @@
+import logging
+from datetime import timedelta
+
+from homeassistant.core import callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.typing import HomeAssistantType
+from homeassistant.helpers.update_coordinator import (
+    DataUpdateCoordinator,
+    CoordinatorEntity,
+)
+from speedport import Speedport
+
+from .const import DOMAIN, UPDATE_INTERVAL
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class SpeedportCoordinator(DataUpdateCoordinator[None]):
+    def __init__(self, hass: HomeAssistantType, device: Speedport):
+        """Initialize my coordinator."""
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=device.device_name,
+            update_interval=timedelta(seconds=UPDATE_INTERVAL),
+        )
+        self._device = device
+
+    async def _async_update_data(self) -> None:
+        return await self._device.update_status()
+
+
+class SpeedportEntity(CoordinatorEntity[SpeedportCoordinator]):
+    _attr_has_entity_name = True
+
+    def __init__(self, hass: HomeAssistantType, speedport: Speedport) -> None:
+        coordinator = get_coordinator(hass, speedport)
+        super().__init__(coordinator)
+
+        self._coordinator = coordinator
+        self._speedport: Speedport = speedport
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        return DeviceInfo(
+            identifiers={(DOMAIN, self._speedport.serial_number)},
+            manufacturer="Telekom",
+            name="Speedport",
+            model=self._speedport.device_name,
+            sw_version=self._speedport.firmware_version,
+            configuration_url=self._speedport.api.url,
+        )
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        self.async_write_ha_state()
+
+
+def get_coordinator(
+    hass: HomeAssistantType, speedport: Speedport
+) -> SpeedportCoordinator:
+    coordinators = hass.data[DOMAIN]["coordinators"]
+    if speedport.serial_number in coordinators:
+        coordinator: SpeedportCoordinator = hass.data[DOMAIN]["coordinators"][
+            speedport.serial_number
+        ]
+    else:
+        coordinator = SpeedportCoordinator(hass, speedport)
+        hass.data[DOMAIN]["coordinators"][speedport.serial_number] = coordinator
+    return coordinator

custom_components/speedport/manifest.json

@@ -12,7 +12,7 @@
   "iot_class": "local_polling",
   "issue_tracker": "https://github.com/Andre0512/speedport/issues",
   "requirements": [
-    "speedport-api==0.5.0"
+    "speedport-api==0.5.1"
   ],
-  "version": "0.2.0"
+  "version": "0.2.1"
 }

custom_components/speedport/switch.py

@@ -3,13 +3,15 @@
 import logging
 from typing import Any
 
-from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
+from homeassistant.components.switch import SwitchEntity
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import HomeAssistantType
 from speedport import Speedport
 
 from .const import DOMAIN
+from .device import SpeedportEntity
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -23,23 +25,25 @@ async def async_setup_entry(
     await speedport.update_status()
     async_add_entities(
         [
-            SpeedportWifiSwitch(speedport),
-            SpeedportGuestWifiSwitch(speedport),
-            SpeedportOfficeWifiSwitch(speedport),
+            SpeedportWifiSwitch(hass, speedport),
+            SpeedportGuestWifiSwitch(hass, speedport),
+            SpeedportOfficeWifiSwitch(hass, speedport),
         ]
     )
 
 
-class SpeedportWifiSwitch(SwitchEntity):
+class SpeedportWifiSwitch(SwitchEntity, SpeedportEntity):
     _attr_is_on: bool | None = False
 
-    def __init__(self, speedport: Speedport) -> None:
+    def __init__(self, hass: HomeAssistantType, speedport: Speedport) -> None:
+        super().__init__(hass, speedport)
         self._speedport: Speedport = speedport
         self._attr_icon = "mdi:wifi"
-        self._attr_name = speedport.wlan_ssid
+        self._attr_name = f"WLAN {speedport.wlan_ssid}"
         self._attr_unique_id = "wifi"
 
-    async def is_on(self) -> bool | None:
+    @property
+    def is_on(self) -> bool | None:
         return self._speedport.wlan_active
 
     async def async_turn_on(self, **kwargs: Any) -> None:
@@ -51,16 +55,18 @@ async def async_turn_off(self, **kwargs: Any) -> None:
         await self._speedport.wifi_off()
 
 
-class SpeedportGuestWifiSwitch(SwitchEntity):
+class SpeedportGuestWifiSwitch(SwitchEntity, SpeedportEntity):
     _attr_is_on: bool | None = False
 
-    def __init__(self, speedport: Speedport) -> None:
+    def __init__(self, hass: HomeAssistantType, speedport: Speedport) -> None:
+        super().__init__(hass, speedport)
         self._speedport: Speedport = speedport
         self._attr_icon = "mdi:wifi"
-        self._attr_name = speedport.wlan_guest_ssid
+        self._attr_name = f"WLAN {speedport.wlan_guest_ssid}"
         self._attr_unique_id = "wifi_guest"
 
-    async def is_on(self) -> bool | None:
+    @property
+    def is_on(self) -> bool | None:
         return self._speedport.wlan_guest_active
 
     async def async_turn_on(self, **kwargs: Any) -> None:
@@ -72,16 +78,18 @@ async def async_turn_off(self, **kwargs: Any) -> None:
         await self._speedport.wifi_guest_off()
 
 
-class SpeedportOfficeWifiSwitch(SwitchEntity):
+class SpeedportOfficeWifiSwitch(SwitchEntity, SpeedportEntity):
     _attr_is_on: bool | None = False
 
-    def __init__(self, speedport: Speedport) -> None:
+    def __init__(self, hass: HomeAssistantType, speedport: Speedport) -> None:
+        super().__init__(hass, speedport)
         self._speedport: Speedport = speedport
         self._attr_icon = "mdi:wifi"
-        self._attr_name = speedport.wlan_office_ssid
+        self._attr_name = f"WLAN {speedport.wlan_office_ssid}"
         self._attr_unique_id = "wifi_office"
 
-    async def is_on(self) -> bool | None:
+    @property
+    def is_on(self) -> bool | None:
         return self._speedport.wlan_office_ssid
 
     async def async_turn_on(self, **kwargs: Any) -> None:

requirements.txt

@@ -1,2 +1,2 @@
-speedport-api==0.5.0
+speedport-api==0.5.1
 homeassistant~=2023.10

requirements_dev.txt

@@ -1,5 +1,4 @@
-homeassistant~=2023.10.0
-black~=23.7.0
-flake8~=6.0.0
-mypy~=1.4.1
-pylint~=2.17.4
+black~=23.9
+flake8~=6.1
+mypy~=1.6
+pylint~=3.0