diff --git a/custom_components/bbox/__init__.py b/custom_components/bbox/__init__.py index 6939483..305f329 100644 --- a/custom_components/bbox/__init__.py +++ b/custom_components/bbox/__init__.py @@ -15,6 +15,7 @@ Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER, Platform.BUTTON, + Platform.SWITCH, ] diff --git a/custom_components/bbox/coordinator.py b/custom_components/bbox/coordinator.py index 3d22440..1ffea4f 100644 --- a/custom_components/bbox/coordinator.py +++ b/custom_components/bbox/coordinator.py @@ -43,6 +43,7 @@ async def _async_update_data(self) -> dict[str, dict[str, Any]]: bbox_info = self.check_list(await self.bbox.device.async_get_bbox_info()) devices = self.check_list(await self.bbox.lan.async_get_connected_devices()) wan_ip_stats = self.check_list(await self.bbox.wan.async_get_wan_ip_stats()) + parentalcontrol = self.check_list(await self.bbox.parentalcontrol.async_get_parental_control_service_state()) # wan = self.check_list(await self.bbox.wan.async_get_wan_ip()) # iptv_channels_infos = self.check_list(await self.bbox.iptv.async_get_iptv_info()) # lan_stats = self.check_list(await self.bbox.lan.async_get_lan_stats()) @@ -56,6 +57,7 @@ async def _async_update_data(self) -> dict[str, dict[str, Any]]: "info": bbox_info, "devices": devices, "wan_ip_stats": wan_ip_stats, + "parentalcontrol": parentalcontrol, # "wan": wan, # "iptv_channels_infos": iptv_channels_infos, # "lan_stats": lan_stats, diff --git a/custom_components/bbox/device_tracker.py b/custom_components/bbox/device_tracker.py index e710630..2e499d8 100644 --- a/custom_components/bbox/device_tracker.py +++ b/custom_components/bbox/device_tracker.py @@ -12,9 +12,8 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from . import BBoxConfigEntry -from .const import DOMAIN from .coordinator import BboxDataUpdateCoordinator -from .entity import BboxEntity +from .entity import BboxDeviceEntity _LOGGER = logging.getLogger(__name__) @@ -34,7 +33,7 @@ async def async_setup_entry( async_add_entities(entities) -class BboxDeviceTracker(BboxEntity, ScannerEntity): +class BboxDeviceTracker(BboxDeviceEntity, ScannerEntity): """Representation of a tracked device.""" _attr_has_entity_name = True @@ -47,13 +46,10 @@ def __init__( device: dict[str, Any], ) -> None: """Initialize.""" - super().__init__(coordinator, description) - self._device = device + super().__init__(coordinator, description, device) - @property - def unique_id(self): - """Return unique_id.""" - return self._device["macaddress"] + self._attr_name = self._device_name + self._attr_unique_id = f"{self._device_key}_device_tracker" @property def source_type(self): @@ -73,37 +69,4 @@ def ip_address(self): @property def is_connected(self): """Return connecting status.""" - for device in ( - self.coordinator.data.get("devices", {}).get("hosts", {}).get("list", []) - ): - if device["macaddress"] == self._device["macaddress"]: - return device.get("active", 0) == 1 - - @property - def name(self): - """Return name.""" - if self._device.get("userfriendlyname", "") != "": - name = self._device["userfriendlyname"] - elif self._device.get("hostname") != "": - name = self._device["hostname"] - else: - name = self._device["macaddress"] - return name - - @property - def device_info(self): - """Return the device info.""" - return { - "name": self.name, - "identifiers": {(DOMAIN, self.unique_id)}, - "via_device": (DOMAIN, self.box_id), - } - - @property - def extra_state_attributes(self): - """Return extra attributes.""" - return { - "link": self._device.get("link"), - "last_seen": self._device.get("lastseen"), - "ip_address": self._device.get("ipaddress"), - } + return self.coordinator_data.get("active", 0) == 1 diff --git a/custom_components/bbox/entity.py b/custom_components/bbox/entity.py index 7bb4228..08029b8 100644 --- a/custom_components/bbox/entity.py +++ b/custom_components/bbox/entity.py @@ -2,7 +2,9 @@ from __future__ import annotations import logging +from typing import Any +from homeassistant.helpers import device_registry from homeassistant.helpers.entity import Entity, EntityDescription from homeassistant.helpers.update_coordinator import CoordinatorEntity @@ -14,7 +16,7 @@ class BboxEntity(CoordinatorEntity[BboxDataUpdateCoordinator], Entity): - """Base class for all entities.""" + """Base class for all Bbox entities.""" _attr_has_entity_name = True @@ -36,3 +38,48 @@ def __init__( "sw_version": device.get("main", {}).get("version"), "configuration_url": f"https://{coordinator.config_entry.data[CONF_HOST]}", } + +class BboxDeviceEntity(BboxEntity): + """Base class for all device's entities connected to the Bbox""" + + def __init__( + self, + coordinator: BboxDataUpdateCoordinator, + description: EntityDescription, + device: dict[str, Any], + ) -> None: + """Initialize.""" + super().__init__(coordinator, description) + self._device = device + self._device_key = f"{self.box_id}_{device['macaddress'].replace(':', '_')}" + if self._device.get("userfriendlyname", "") != "": + self._device_name = device["userfriendlyname"] + elif self._device.get("hostname") != "": + self._device_name = device["hostname"] + else: + self._device_name = device["macaddress"] + self._attr_device_info = { + "name": self._device_name, + "identifiers": {(DOMAIN, self._device_key)}, + "connections": {(device_registry.CONNECTION_NETWORK_MAC, device["macaddress"])}, + "via_device": (DOMAIN, self.box_id), + } + + @property + def coordinator_data(self): + """Return connecting status.""" + for device in ( + self.coordinator.data.get("devices", {}).get("hosts", {}).get("list", []) + ): + if device["macaddress"] == self._device["macaddress"]: + return device + return {} + + @property + def extra_state_attributes(self): + """Return extra attributes.""" + return { + "link": self._device.get("link"), + "last_seen": self._device.get("lastseen"), + "ip_address": self._device.get("ipaddress"), + } \ No newline at end of file diff --git a/custom_components/bbox/switch.py b/custom_components/bbox/switch.py new file mode 100644 index 0000000..349b612 --- /dev/null +++ b/custom_components/bbox/switch.py @@ -0,0 +1,142 @@ +"""Button for Bbox router.""" + +import asyncio +import logging + +from typing import Any + +from bboxpy.exceptions import BboxException + +from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from . import BBoxConfigEntry, BboxDataUpdateCoordinator +from .entity import BboxEntity, BboxDeviceEntity +from .helpers import finditem + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, entry: BBoxConfigEntry, async_add_entities: AddEntitiesCallback +) -> None: + """Set up sensor.""" + coordinator = entry.runtime_data + description = SwitchEntityDescription( + key="parental_control", + translation_key="parental_control", + name="Parental control", + icon="mdi:security" + ) + devices = coordinator.data.get("devices", {}).get("hosts", {}).get("list", []) + entities = [ + DeviceParentalControlSwitch(coordinator, description, device) + for device in devices + if device.get("macaddress") + ] + entities += [ParentalControlServiceSwitch(coordinator, description)] + async_add_entities(entities) + + +class ParentalControlServiceSwitch(BboxEntity, SwitchEntity): + """Representation of a switch for Bbox parental control service state.""" + + waiting_delay_after_toggle = 5 + + def __init__( + self, + coordinator: BboxDataUpdateCoordinator, + description: SwitchEntityDescription, + ) -> None: + """Initialize.""" + super().__init__(coordinator, description) + + self._attr_name = "Parental control" + + @property + def is_on(self) -> bool: + """Return true if parental control service is currently enabled.""" + return bool( + finditem(self.coordinator.data, "parentalcontrol.parentalcontrol.scheduler.enable") + ) + + async def async_turn_on(self, **kwargs) -> None: + """Turn the switch on.""" + try: + await self.coordinator.bbox.parentalcontrol.async_set_parental_control_service_state( + enable=True + ) + except BboxException as error: + _LOGGER.error(error) + return + _LOGGER.debug( + "Request sent, we need to wait a bit (%ds) before updating state...", + self.waiting_delay_after_toggle + ) + await asyncio.sleep(self.waiting_delay_after_toggle) + _LOGGER.debug("Updating state") + await self.coordinator.async_request_refresh() + + async def async_turn_off(self, **kwargs) -> None: + """Turn the switch off.""" + try: + await self.coordinator.bbox.parentalcontrol.async_set_parental_control_service_state( + enable=False + ) + except BboxException as error: + _LOGGER.error(error) + return + _LOGGER.debug( + "Request sent, we need to wait a bit (%ds) before updating state...", + self.waiting_delay_after_toggle + ) + await asyncio.sleep(self.waiting_delay_after_toggle) + _LOGGER.debug("Updating state") + await self.coordinator.async_request_refresh() + + +class DeviceParentalControlSwitch(BboxDeviceEntity, SwitchEntity): + """Representation of a switch for device parental control state.""" + + def __init__( + self, + coordinator: BboxDataUpdateCoordinator, + description: SwitchEntityDescription, + device: dict[str, Any], + ) -> None: + """Initialize.""" + super().__init__(coordinator, description, device) + + self._attr_name = "Parental control" + self._attr_unique_id = f"{self._device_key}_parental_control" + + @property + def is_on(self) -> bool: + """Return true if device parental control is currently enabled.""" + return bool(finditem(self.coordinator_data, "parentalcontrol.enable")) + + async def async_turn_on(self, **kwargs) -> None: + """Turn the switch on.""" + try: + await self.coordinator.bbox.parentalcontrol.async_set_device_parental_control_state( + macaddress= self._device["macaddress"], + enable=True + ) + except BboxException as error: + _LOGGER.error(error) + return + await self.coordinator.async_request_refresh() + + async def async_turn_off(self, **kwargs) -> None: + """Turn the switch off.""" + try: + await self.coordinator.bbox.parentalcontrol.async_set_device_parental_control_state( + macaddress= self._device["macaddress"], + enable=False + ) + except BboxException as error: + _LOGGER.error(error) + return + await self.coordinator.async_request_refresh() +