Skip to content

Commit

Permalink
Add switch entities to control parental control service & devices state
Browse files Browse the repository at this point in the history
  • Loading branch information
brenard committed Oct 22, 2024
1 parent 304d53c commit 9e62922
Show file tree
Hide file tree
Showing 5 changed files with 199 additions and 44 deletions.
1 change: 1 addition & 0 deletions custom_components/bbox/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
Platform.BINARY_SENSOR,
Platform.DEVICE_TRACKER,
Platform.BUTTON,
Platform.SWITCH,
]


Expand Down
2 changes: 2 additions & 0 deletions custom_components/bbox/coordinator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand All @@ -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,
Expand Down
49 changes: 6 additions & 43 deletions custom_components/bbox/device_tracker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand All @@ -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
Expand All @@ -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):
Expand All @@ -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
49 changes: 48 additions & 1 deletion custom_components/bbox/entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

Expand All @@ -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"),
}
142 changes: 142 additions & 0 deletions custom_components/bbox/switch.py
Original file line number Diff line number Diff line change
@@ -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()

0 comments on commit 9e62922

Please sign in to comment.