Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Freebox Home binary sensors (#92196) 

Co-authored-by: Quentame <polletquentin74@me.com>
pull/101013/head
nachonam 2023-09-27 17:11:31 +02:00 committed by GitHub
parent 8b5bfd8cee
commit 9fdc8494b6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 249 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
homeassistant/components/freebox/binary_sensor.py
View File

@ -15,11 +15,13 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .const import DOMAIN, FreeboxHomeCategory
from .home_base import FreeboxHomeEntity
from .router import FreeboxRouter
_LOGGER = logging.getLogger(__name__)
RAID_SENSORS: tuple[BinarySensorEntityDescription, ...] = (
BinarySensorEntityDescription(
key="raid_degraded",
@ -33,21 +35,105 @@ RAID_SENSORS: tuple[BinarySensorEntityDescription, ...] = (
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the binary sensors."""
"""Set up binary sensors."""
router: FreeboxRouter = hass.data[DOMAIN][entry.unique_id]
_LOGGER.debug("%s - %s - %s raid(s)", router.name, router.mac, len(router.raids))
binary_entities = [
binary_entities: list[BinarySensorEntity] = [
FreeboxRaidDegradedSensor(router, raid, description)
for raid in router.raids.values()
for description in RAID_SENSORS
]
for node in router.home_devices.values():
if node["category"] == FreeboxHomeCategory.PIR:
binary_entities.append(FreeboxPirSensor(hass, router, node))
elif node["category"] == FreeboxHomeCategory.DWS:
binary_entities.append(FreeboxDwsSensor(hass, router, node))
for endpoint in node["show_endpoints"]:
if (
endpoint["name"] == "cover"
and endpoint["ep_type"] == "signal"
and endpoint.get("value") is not None
):
binary_entities.append(FreeboxCoverSensor(hass, router, node))
if binary_entities:
async_add_entities(binary_entities, True)
class FreeboxHomeBinarySensor(FreeboxHomeEntity, BinarySensorEntity):
"""Representation of a Freebox binary sensor."""
_sensor_name = "trigger"
def __init__(
self,
hass: HomeAssistant,
router: FreeboxRouter,
node: dict[str, Any],
sub_node: dict[str, Any] | None = None,
) -> None:
"""Initialize a Freebox binary sensor."""
super().__init__(hass, router, node, sub_node)
self._command_id = self.get_command_id(
node["type"]["endpoints"], "signal", self._sensor_name
)
self._attr_is_on = self._edit_state(self.get_value("signal", self._sensor_name))
async def async_update_signal(self):
"""Update name & state."""
self._attr_is_on = self._edit_state(
await self.get_home_endpoint_value(self._command_id)
)
await FreeboxHomeEntity.async_update_signal(self)
def _edit_state(self, state: bool | None) -> bool | None:
"""Edit state depending on sensor name."""
if state is None:
return None
if self._sensor_name == "trigger":
return not state
return state
class FreeboxPirSensor(FreeboxHomeBinarySensor):
"""Representation of a Freebox motion binary sensor."""
_attr_device_class = BinarySensorDeviceClass.MOTION
class FreeboxDwsSensor(FreeboxHomeBinarySensor):
"""Representation of a Freebox door opener binary sensor."""
_attr_device_class = BinarySensorDeviceClass.DOOR
class FreeboxCoverSensor(FreeboxHomeBinarySensor):
"""Representation of a cover Freebox plastic removal cover binary sensor (for some sensors: motion detector, door opener detector...)."""
_attr_device_class = BinarySensorDeviceClass.SAFETY
_attr_entity_category = EntityCategory.DIAGNOSTIC
_attr_entity_registry_enabled_default = False
_sensor_name = "cover"
def __init__(
self, hass: HomeAssistant, router: FreeboxRouter, node: dict[str, Any]
) -> None:
"""Initialize a cover for another device."""
cover_node = next(
filter(
lambda x: (x["name"] == self._sensor_name and x["ep_type"] == "signal"),
node["type"]["endpoints"],
),
None,
)
super().__init__(hass, router, node, cover_node)
class FreeboxRaidDegradedSensor(BinarySensorEntity):
"""Representation of a Freebox raid sensor."""

12
homeassistant/components/freebox/camera.py
View File

@ -80,27 +80,27 @@ class FreeboxCamera(FreeboxHomeEntity, FFmpegCamera):
)
self._command_motion_detection = self.get_command_id(
node["type"]["endpoints"], ATTR_DETECTION
node["type"]["endpoints"], "slot", ATTR_DETECTION
)
self._attr_extra_state_attributes = {}
self.update_node(node)
async def async_enable_motion_detection(self) -> None:
"""Enable motion detection in the camera."""
await self.set_home_endpoint_value(self._command_motion_detection, True)
self._attr_motion_detection_enabled = True
if await self.set_home_endpoint_value(self._command_motion_detection, True):
self._attr_motion_detection_enabled = True
async def async_disable_motion_detection(self) -> None:
"""Disable motion detection in camera."""
await self.set_home_endpoint_value(self._command_motion_detection, False)
self._attr_motion_detection_enabled = False
if await self.set_home_endpoint_value(self._command_motion_detection, False):
self._attr_motion_detection_enabled = False
async def async_update_signal(self) -> None:
"""Update the camera node."""
self.update_node(self._router.home_devices[self._id])
self.async_write_ha_state()
def update_node(self, node):
def update_node(self, node: dict[str, Any]) -> None:
"""Update params."""
self._name = node["label"].strip()

2
homeassistant/components/freebox/const.py
View File

@ -86,6 +86,8 @@ CATEGORY_TO_MODEL = {
HOME_COMPATIBLE_CATEGORIES = [
FreeboxHomeCategory.CAMERA,
FreeboxHomeCategory.DWS,
FreeboxHomeCategory.IOHOME,
FreeboxHomeCategory.KFB,
FreeboxHomeCategory.PIR,
FreeboxHomeCategory.RTS,
]

27
homeassistant/components/freebox/home_base.py
View File

@ -77,23 +77,36 @@ class FreeboxHomeEntity(Entity):
)
self.async_write_ha_state()
async def set_home_endpoint_value(self, command_id: Any, value=None) -> None:
async def set_home_endpoint_value(self, command_id: Any, value=None) -> bool:
"""Set Home endpoint value."""
if command_id is None:
_LOGGER.error("Unable to SET a value through the API. Command is None")
return
return False
await self._router.home.set_home_endpoint_value(
self._id, command_id, {"value": value}
)
return True
def get_command_id(self, nodes, name) -> int | None:
async def get_home_endpoint_value(self, command_id: Any) -> Any | None:
"""Get Home endpoint value."""
if command_id is None:
_LOGGER.error("Unable to GET a value through the API. Command is None")
return None
node = await self._router.home.get_home_endpoint_value(self._id, command_id)
return node.get("value")
def get_command_id(self, nodes, ep_type, name) -> int | None:
"""Get the command id."""
node = next(
filter(lambda x: (x["name"] == name), nodes),
filter(lambda x: (x["name"] == name and x["ep_type"] == ep_type), nodes),
None,
)
if not node:
_LOGGER.warning("The Freebox Home device has no value for: %s", name)
_LOGGER.warning(
"The Freebox Home device has no command value for: %s/%s", name, ep_type
)
return None
return node["id"]
@ -115,7 +128,7 @@ class FreeboxHomeEntity(Entity):
"""Register state update callback."""
self._remove_signal_update = dispacher
def get_value(self, ep_type, name):
def get_value(self, ep_type: str, name: str):
"""Get the value."""
node = next(
filter(
@ -126,7 +139,7 @@ class FreeboxHomeEntity(Entity):
)
if not node:
_LOGGER.warning(
"The Freebox Home device has no node for: %s/%s", ep_type, name
"The Freebox Home device has no node value for: %s/%s", ep_type, name
)
return None
return node.get("value")

3
homeassistant/components/freebox/router.py
View File

@ -118,6 +118,7 @@ class FreeboxRouter:
async def update_sensors(self) -> None:
"""Update Freebox sensors."""
# System sensors
syst_datas: dict[str, Any] = await self._api.system.get_config()
@ -145,7 +146,6 @@ class FreeboxRouter:
self.call_list = await self._api.call.get_calls_log()
await self._update_disks_sensors()
await self._update_raids_sensors()
async_dispatcher_send(self.hass, self.signal_sensor_update)
@ -165,6 +165,7 @@ class FreeboxRouter:
async def _update_raids_sensors(self) -> None:
"""Update Freebox raids."""
# None at first request
if not self.supports_raid:
return

24
tests/components/freebox/conftest.py
View File

@ -1,5 +1,5 @@
"""Test helpers for Freebox."""
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, PropertyMock, patch
import pytest
@ -10,6 +10,7 @@ from .const import (
DATA_CALL_GET_CALLS_LOG,
DATA_CONNECTION_GET_STATUS,
DATA_HOME_GET_NODES,
DATA_HOME_GET_VALUES,
DATA_LAN_GET_HOSTS_LIST,
DATA_STORAGE_GET_DISKS,
DATA_STORAGE_GET_RAIDS,
@ -27,6 +28,16 @@ def mock_path():
yield
@pytest.fixture(autouse=True)
def enable_all_entities():
"""Make sure all entities are enabled."""
with patch(
"homeassistant.helpers.entity.Entity.entity_registry_enabled_default",
PropertyMock(return_value=True),
):
yield
@pytest.fixture
def mock_device_registry_devices(hass: HomeAssistant, device_registry):
"""Create device registry devices so the device tracker entities are enabled."""
@ -56,18 +67,21 @@ def mock_router(mock_device_registry_devices):
instance = service_mock.return_value
instance.open = AsyncMock()
instance.system.get_config = AsyncMock(return_value=DATA_SYSTEM_GET_CONFIG)
# device_tracker
instance.lan.get_hosts_list = AsyncMock(return_value=DATA_LAN_GET_HOSTS_LIST)
# sensor
instance.call.get_calls_log = AsyncMock(return_value=DATA_CALL_GET_CALLS_LOG)
instance.storage.get_disks = AsyncMock(return_value=DATA_STORAGE_GET_DISKS)
instance.storage.get_raids = AsyncMock(return_value=DATA_STORAGE_GET_RAIDS)
# home devices
instance.home.get_home_nodes = AsyncMock(return_value=DATA_HOME_GET_NODES)
instance.connection.get_status = AsyncMock(
return_value=DATA_CONNECTION_GET_STATUS
)
# switch
instance.wifi.get_global_config = AsyncMock(return_value=WIFI_GET_GLOBAL_CONFIG)
# device_tracker
instance.lan.get_hosts_list = AsyncMock(return_value=DATA_LAN_GET_HOSTS_LIST)
# home devices
instance.home.get_home_nodes = AsyncMock(return_value=DATA_HOME_GET_NODES)
instance.home.get_home_endpoint_value = AsyncMock(
return_value=DATA_HOME_GET_VALUES
)
instance.close = AsyncMock()
yield service_mock

60
tests/components/freebox/const.py
View File

@ -513,7 +513,22 @@ DATA_LAN_GET_HOSTS_LIST = [
},
]
# Home
# PIR node id 26, endpoint id 6
DATA_HOME_GET_VALUES = {
"category": "",
"ep_type": "signal",
"id": 6,
"label": "Détection",
"name": "trigger",
"ui": {"access": "w", "display": "toggle"},
"value": False,
"value_type": "bool",
"visibility": "normal",
}
# Home
# ALL
DATA_HOME_GET_NODES = [
{
"adapter": 2,
@ -2110,6 +2125,22 @@ DATA_HOME_GET_NODES = [
"value_type": "bool",
"visibility": "normal",
},
{
"category": "",
"ep_type": "signal",
"id": 7,
"label": "Couvercle",
"name": "cover",
"refresh": 2000,
"ui": {
"access": "r",
"display": "warning",
"icon_url": "/resources/images/home/pictos/warning.png",
},
"value": False,
"value_type": "bool",
"visibility": "normal",
},
{
"category": "",
"ep_type": "signal",
@ -2211,7 +2242,7 @@ DATA_HOME_GET_NODES = [
"ep_type": "signal",
"id": 7,
"label": "Couvercle",
"name": "1cover",
"name": "cover",
"param_type": "void",
"value_type": "bool",
"visibility": "normal",
@ -2302,6 +2333,33 @@ DATA_HOME_GET_NODES = [
"value_type": "bool",
"visibility": "normal",
},
{
"category": "",
"ep_type": "signal",
"id": 6,
"label": "Détection",
"name": "trigger",
"ui": {"access": "w", "display": "toggle"},
"value": False,
"value_type": "bool",
"visibility": "normal",
},
{
"category": "",
"ep_type": "signal",
"id": 7,
"label": "Couvercle",
"name": "cover",
"refresh": 2000,
"ui": {
"access": "r",
"display": "warning",
"icon_url": "/resources/images/home/pictos/warning.png",
},
"value": False,
"value_type": "bool",
"visibility": "normal",
},
{
"category": "",
"ep_type": "signal",

52
tests/components/freebox/test_binary_sensor.py
View File

@ -4,12 +4,16 @@ from unittest.mock import Mock
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.binary_sensor import (
DOMAIN as BINARY_SENSOR_DOMAIN,
BinarySensorDeviceClass,
)
from homeassistant.components.freebox import SCAN_INTERVAL
from homeassistant.const import ATTR_DEVICE_CLASS
from homeassistant.core import HomeAssistant
from .common import setup_platform
from .const import DATA_STORAGE_GET_RAIDS
from .const import DATA_HOME_GET_VALUES, DATA_STORAGE_GET_RAIDS
from tests.common import async_fire_time_changed
@ -38,3 +42,47 @@ async def test_raid_array_degraded(
hass.states.get("binary_sensor.freebox_server_r2_raid_array_0_degraded").state
== "on"
)
async def test_home(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, router: Mock
) -> None:
"""Test home binary sensors."""
await setup_platform(hass, BINARY_SENSOR_DOMAIN)
# Device class
assert (
hass.states.get("binary_sensor.detecteur").attributes[ATTR_DEVICE_CLASS]
== BinarySensorDeviceClass.MOTION
)
assert (
hass.states.get("binary_sensor.ouverture_porte").attributes[ATTR_DEVICE_CLASS]
== BinarySensorDeviceClass.DOOR
)
assert (
hass.states.get("binary_sensor.ouverture_porte_couvercle").attributes[
ATTR_DEVICE_CLASS
]
== BinarySensorDeviceClass.SAFETY
)
# Initial state
assert hass.states.get("binary_sensor.detecteur").state == "on"
assert hass.states.get("binary_sensor.detecteur_couvercle").state == "off"
assert hass.states.get("binary_sensor.ouverture_porte").state == "unknown"
assert hass.states.get("binary_sensor.ouverture_porte_couvercle").state == "off"
# Now simulate a changed status
data_home_get_values_changed = deepcopy(DATA_HOME_GET_VALUES)
data_home_get_values_changed["value"] = True
router().home.get_home_endpoint_value.return_value = data_home_get_values_changed
# Simulate an update
freezer.tick(SCAN_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("binary_sensor.detecteur").state == "off"
assert hass.states.get("binary_sensor.detecteur_couvercle").state == "on"
assert hass.states.get("binary_sensor.ouverture_porte").state == "off"
assert hass.states.get("binary_sensor.ouverture_porte_couvercle").state == "on"

4
tests/components/freebox/test_sensor.py
View File

@ -104,8 +104,8 @@ async def test_battery(
# Simulate a changed battery
data_home_get_nodes_changed = deepcopy(DATA_HOME_GET_NODES)
data_home_get_nodes_changed[2]["show_endpoints"][3]["value"] = 25
data_home_get_nodes_changed[3]["show_endpoints"][3]["value"] = 50
data_home_get_nodes_changed[4]["show_endpoints"][3]["value"] = 75
data_home_get_nodes_changed[3]["show_endpoints"][4]["value"] = 50
data_home_get_nodes_changed[4]["show_endpoints"][5]["value"] = 75
router().home.get_home_nodes.return_value = data_home_get_nodes_changed
# Simulate an update
freezer.tick(SCAN_INTERVAL)