Add device info to fibaro integration (#73352)

This commit is contained in:
rappenze 2022-06-30 23:57:35 +02:00 committed by GitHub
parent 73a0197cac
commit 7eae3691c2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 93 additions and 11 deletions

View File

@ -29,8 +29,9 @@ from homeassistant.const import (
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import device_registry as dr
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity from homeassistant.helpers.entity import DeviceInfo, Entity
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType
from homeassistant.util import slugify from homeassistant.util import slugify
@ -156,16 +157,21 @@ class FibaroController:
) # List of devices by entity platform ) # List of devices by entity platform
self._callbacks: dict[Any, Any] = {} # Update value callbacks by deviceId self._callbacks: dict[Any, Any] = {} # Update value callbacks by deviceId
self._state_handler = None # Fiblary's StateHandler object self._state_handler = None # Fiblary's StateHandler object
self.hub_serial = None # Unique serial number of the hub self.hub_serial: str # Unique serial number of the hub
self.name = None # The friendly name of the hub self.hub_name: str # The friendly name of the hub
self.hub_software_version: str
self.hub_api_url: str = config[CONF_URL]
# Device infos by fibaro device id
self._device_infos: dict[int, DeviceInfo] = {}
def connect(self): def connect(self):
"""Start the communication with the Fibaro controller.""" """Start the communication with the Fibaro controller."""
try: try:
login = self._client.login.get() login = self._client.login.get()
info = self._client.info.get() info = self._client.info.get()
self.hub_serial = slugify(info.serialNumber) self.hub_serial = info.serialNumber
self.name = slugify(info.hcName) self.hub_name = info.hcName
self.hub_software_version = info.softVersion
except AssertionError: except AssertionError:
_LOGGER.error("Can't connect to Fibaro HC. Please check URL") _LOGGER.error("Can't connect to Fibaro HC. Please check URL")
return False return False
@ -305,6 +311,44 @@ class FibaroController:
platform = Platform.LIGHT platform = Platform.LIGHT
return platform return platform
def _create_device_info(self, device: Any, devices: list) -> None:
"""Create the device info. Unrooted entities are directly shown below the home center."""
# The home center is always id 1 (z-wave primary controller)
if "parentId" not in device or device.parentId <= 1:
return
master_entity: Any | None = None
if device.parentId == 1:
master_entity = device
else:
for parent in devices:
if "id" in parent and parent.id == device.parentId:
master_entity = parent
if master_entity is None:
_LOGGER.error("Parent with id %s not found", device.parentId)
return
if "zwaveCompany" in master_entity.properties:
manufacturer = master_entity.properties.zwaveCompany
else:
manufacturer = "Unknown"
self._device_infos[master_entity.id] = DeviceInfo(
identifiers={(DOMAIN, master_entity.id)},
manufacturer=manufacturer,
name=master_entity.name,
via_device=(DOMAIN, self.hub_serial),
)
def get_device_info(self, device: Any) -> DeviceInfo:
"""Get the device info by fibaro device id."""
if device.id in self._device_infos:
return self._device_infos[device.id]
if "parentId" in device and device.parentId in self._device_infos:
return self._device_infos[device.parentId]
return DeviceInfo(identifiers={(DOMAIN, self.hub_serial)})
def _read_scenes(self): def _read_scenes(self):
scenes = self._client.scenes.list() scenes = self._client.scenes.list()
self._scene_map = {} self._scene_map = {}
@ -321,14 +365,14 @@ class FibaroController:
device.ha_id = ( device.ha_id = (
f"scene_{slugify(room_name)}_{slugify(device.name)}_{device.id}" f"scene_{slugify(room_name)}_{slugify(device.name)}_{device.id}"
) )
device.unique_id_str = f"{self.hub_serial}.scene.{device.id}" device.unique_id_str = f"{slugify(self.hub_serial)}.scene.{device.id}"
self._scene_map[device.id] = device self._scene_map[device.id] = device
self.fibaro_devices[Platform.SCENE].append(device) self.fibaro_devices[Platform.SCENE].append(device)
_LOGGER.debug("%s scene -> %s", device.ha_id, device) _LOGGER.debug("%s scene -> %s", device.ha_id, device)
def _read_devices(self): def _read_devices(self):
"""Read and process the device list.""" """Read and process the device list."""
devices = self._client.devices.list() devices = list(self._client.devices.list())
self._device_map = {} self._device_map = {}
last_climate_parent = None last_climate_parent = None
last_endpoint = None last_endpoint = None
@ -355,7 +399,8 @@ class FibaroController:
device.mapped_platform = None device.mapped_platform = None
if (platform := device.mapped_platform) is None: if (platform := device.mapped_platform) is None:
continue continue
device.unique_id_str = f"{self.hub_serial}.{device.id}" device.unique_id_str = f"{slugify(self.hub_serial)}.{device.id}"
self._create_device_info(device, devices)
self._device_map[device.id] = device self._device_map[device.id] = device
_LOGGER.debug( _LOGGER.debug(
"%s (%s, %s) -> %s %s", "%s (%s, %s) -> %s %s",
@ -462,6 +507,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
for platform in PLATFORMS: for platform in PLATFORMS:
devices[platform] = [*controller.fibaro_devices[platform]] devices[platform] = [*controller.fibaro_devices[platform]]
# register the hub device info separately as the hub has sometimes no entities
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, controller.hub_serial)},
manufacturer="Fibaro",
name=controller.hub_name,
model=controller.hub_serial,
sw_version=controller.hub_software_version,
configuration_url=controller.hub_api_url.removesuffix("/api/"),
)
hass.config_entries.async_setup_platforms(entry, PLATFORMS) hass.config_entries.async_setup_platforms(entry, PLATFORMS)
controller.enable_state_handler() controller.enable_state_handler()
@ -490,6 +547,7 @@ class FibaroDevice(Entity):
self.ha_id = fibaro_device.ha_id self.ha_id = fibaro_device.ha_id
self._attr_name = fibaro_device.friendly_name self._attr_name = fibaro_device.friendly_name
self._attr_unique_id = fibaro_device.unique_id_str self._attr_unique_id = fibaro_device.unique_id_str
self._attr_device_info = self.controller.get_device_info(fibaro_device)
# propagate hidden attribute set in fibaro home center to HA # propagate hidden attribute set in fibaro home center to HA
if "visible" in fibaro_device and fibaro_device.visible is False: if "visible" in fibaro_device and fibaro_device.visible is False:
self._attr_entity_registry_visible_default = False self._attr_entity_registry_visible_default = False

View File

@ -4,6 +4,7 @@ from __future__ import annotations
import logging import logging
from typing import Any from typing import Any
from slugify import slugify
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
@ -44,9 +45,12 @@ async def _validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str
_LOGGER.debug( _LOGGER.debug(
"Successfully connected to fibaro home center %s with name %s", "Successfully connected to fibaro home center %s with name %s",
controller.hub_serial, controller.hub_serial,
controller.name, controller.hub_name,
) )
return {"serial_number": controller.hub_serial, "name": controller.name} return {
"serial_number": slugify(controller.hub_serial),
"name": controller.hub_name,
}
class FibaroConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): class FibaroConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):

View File

@ -7,6 +7,7 @@ from homeassistant.components.scene import Scene
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform from homeassistant.const import Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import FIBARO_DEVICES, FibaroDevice from . import FIBARO_DEVICES, FibaroDevice
@ -33,6 +34,15 @@ async def async_setup_entry(
class FibaroScene(FibaroDevice, Scene): class FibaroScene(FibaroDevice, Scene):
"""Representation of a Fibaro scene entity.""" """Representation of a Fibaro scene entity."""
def __init__(self, fibaro_device: Any) -> None:
"""Initialize the Fibaro scene."""
super().__init__(fibaro_device)
# All scenes are shown on hub device
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self.controller.hub_serial)}
)
def activate(self, **kwargs: Any) -> None: def activate(self, **kwargs: Any) -> None:
"""Activate the scene.""" """Activate the scene."""
self.fibaro_device.start() self.fibaro_device.start()

View File

@ -14,17 +14,23 @@ TEST_NAME = "my_fibaro_home_center"
TEST_URL = "http://192.168.1.1/api/" TEST_URL = "http://192.168.1.1/api/"
TEST_USERNAME = "user" TEST_USERNAME = "user"
TEST_PASSWORD = "password" TEST_PASSWORD = "password"
TEST_VERSION = "4.360"
@pytest.fixture(name="fibaro_client", autouse=True) @pytest.fixture(name="fibaro_client", autouse=True)
def fibaro_client_fixture(): def fibaro_client_fixture():
"""Mock common methods and attributes of fibaro client.""" """Mock common methods and attributes of fibaro client."""
info_mock = Mock() info_mock = Mock()
info_mock.get.return_value = Mock(serialNumber=TEST_SERIALNUMBER, hcName=TEST_NAME) info_mock.get.return_value = Mock(
serialNumber=TEST_SERIALNUMBER, hcName=TEST_NAME, softVersion=TEST_VERSION
)
array_mock = Mock() array_mock = Mock()
array_mock.list.return_value = [] array_mock.list.return_value = []
client_mock = Mock()
client_mock.base_url.return_value = TEST_URL
with patch("fiblary3.client.v4.client.Client.__init__", return_value=None,), patch( with patch("fiblary3.client.v4.client.Client.__init__", return_value=None,), patch(
"fiblary3.client.v4.client.Client.info", "fiblary3.client.v4.client.Client.info",
info_mock, info_mock,
@ -37,6 +43,10 @@ def fibaro_client_fixture():
"fiblary3.client.v4.client.Client.scenes", "fiblary3.client.v4.client.Client.scenes",
array_mock, array_mock,
create=True, create=True,
), patch(
"fiblary3.client.v4.client.Client.client",
client_mock,
create=True,
): ):
yield yield