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

Fibaro integration refactorings (#139624) 

* Fibaro integration refactorings

* Fix execute_action

* Add test

* more tests

* Add tests

* Fix test

* More tests
pull/137215/merge
rappenze 2025-03-02 14:38:56 +01:00 committed by GitHub
parent b7bedd4b8f
commit 5ac3fe6ee1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 287 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
homeassistant/components/fibaro/__init__.py
View File

@ -12,6 +12,7 @@ from pyfibaro.fibaro_client import (
FibaroClient,
FibaroConnectFailed,
)
from pyfibaro.fibaro_data_helper import read_rooms
from pyfibaro.fibaro_device import DeviceModel
from pyfibaro.fibaro_info import InfoModel
from pyfibaro.fibaro_scene import SceneModel
@ -83,7 +84,7 @@ class FibaroController:
# Whether to import devices from plugins
self._import_plugins = import_plugins
# Mapping roomId to room object
self._room_map = {room.fibaro_id: room for room in self._client.read_rooms()}
self._room_map = read_rooms(fibaro_client)
self._device_map: dict[int, DeviceModel] # Mapping deviceId to device object
self.fibaro_devices: dict[Platform, list[DeviceModel]] = defaultdict(
list
@ -269,9 +270,7 @@ class FibaroController:
def get_room_name(self, room_id: int) -> str | None:
"""Get the room name by room id."""
assert self._room_map
room = self._room_map.get(room_id)
return room.name if room else None
return self._room_map.get(room_id)
def read_scenes(self) -> list[SceneModel]:
"""Return list of scenes."""
@ -294,20 +293,17 @@ class FibaroController:
for device in devices:
try:
device.fibaro_controller = self
if device.room_id == 0:
room_name = self.get_room_name(device.room_id)
if not room_name:
room_name = "Unknown"
else:
room_name = self._room_map[device.room_id].name
device.room_name = room_name
device.friendly_name = f"{room_name} {device.name}"
device.ha_id = (
f"{slugify(room_name)}_{slugify(device.name)}_{device.fibaro_id}"
)
if device.enabled and (not device.is_plugin or self._import_plugins):
device.mapped_platform = self._map_device_to_platform(device)
else:
device.mapped_platform = None
if (platform := device.mapped_platform) is None:
platform = self._map_device_to_platform(device)
if platform is None:
continue
device.unique_id_str = f"{slugify(self.hub_serial)}.{device.fibaro_id}"
self._create_device_info(device, devices)

96
homeassistant/components/fibaro/climate.py
View File

@ -129,13 +129,13 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
def __init__(self, fibaro_device: DeviceModel) -> None:
"""Initialize the Fibaro device."""
super().__init__(fibaro_device)
self._temp_sensor_device: FibaroEntity | None = None
self._target_temp_device: FibaroEntity | None = None
self._op_mode_device: FibaroEntity | None = None
self._fan_mode_device: FibaroEntity | None = None
self._temp_sensor_device: DeviceModel | None = None
self._target_temp_device: DeviceModel | None = None
self._op_mode_device: DeviceModel | None = None
self._fan_mode_device: DeviceModel | None = None
self.entity_id = ENTITY_ID_FORMAT.format(self.ha_id)
siblings = fibaro_device.fibaro_controller.get_siblings(fibaro_device)
siblings = self.controller.get_siblings(fibaro_device)
_LOGGER.debug("%s siblings: %s", fibaro_device.ha_id, siblings)
tempunit = "C"
for device in siblings:
@ -147,23 +147,23 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
and (device.value.has_value or device.has_heating_thermostat_setpoint)
and device.unit in ("C", "F")
):
self._temp_sensor_device = FibaroEntity(device)
self._temp_sensor_device = device
tempunit = device.unit
if any(
action for action in TARGET_TEMP_ACTIONS if action in device.actions
):
self._target_temp_device = FibaroEntity(device)
self._target_temp_device = device
self._attr_supported_features |= ClimateEntityFeature.TARGET_TEMPERATURE
if device.has_unit:
tempunit = device.unit
if any(action for action in OP_MODE_ACTIONS if action in device.actions):
self._op_mode_device = FibaroEntity(device)
self._op_mode_device = device
self._attr_supported_features |= ClimateEntityFeature.PRESET_MODE
if "setFanMode" in device.actions:
self._fan_mode_device = FibaroEntity(device)
self._fan_mode_device = device
self._attr_supported_features |= ClimateEntityFeature.FAN_MODE
if tempunit == "F":
@ -172,7 +172,7 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
self._attr_temperature_unit = UnitOfTemperature.CELSIUS
if self._fan_mode_device:
fan_modes = self._fan_mode_device.fibaro_device.supported_modes
fan_modes = self._fan_mode_device.supported_modes
self._attr_fan_modes = []
for mode in fan_modes:
if mode not in FANMODES:
@ -184,7 +184,7 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
if self._op_mode_device:
self._attr_preset_modes = []
self._attr_hvac_modes: list[HVACMode] = []
device = self._op_mode_device.fibaro_device
device = self._op_mode_device
if device.has_supported_thermostat_modes:
for mode in device.supported_thermostat_modes:
try:
@ -222,15 +222,15 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
"- _fan_mode_device %s"
),
self.ha_id,
self._temp_sensor_device.ha_id if self._temp_sensor_device else "None",
self._target_temp_device.ha_id if self._target_temp_device else "None",
self._op_mode_device.ha_id if self._op_mode_device else "None",
self._fan_mode_device.ha_id if self._fan_mode_device else "None",
self._temp_sensor_device.fibaro_id if self._temp_sensor_device else "None",
self._target_temp_device.fibaro_id if self._target_temp_device else "None",
self._op_mode_device.fibaro_id if self._op_mode_device else "None",
self._fan_mode_device.fibaro_id if self._fan_mode_device else "None",
)
await super().async_added_to_hass()
# Register update callback for child devices
siblings = self.fibaro_device.fibaro_controller.get_siblings(self.fibaro_device)
siblings = self.controller.get_siblings(self.fibaro_device)
for device in siblings:
if device != self.fibaro_device:
self.controller.register(device.fibaro_id, self._update_callback)
@ -240,14 +240,14 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
"""Return the fan setting."""
if not self._fan_mode_device:
return None
mode = self._fan_mode_device.fibaro_device.mode
mode = self._fan_mode_device.mode
return FANMODES[mode]
def set_fan_mode(self, fan_mode: str) -> None:
"""Set new target fan mode."""
if not self._fan_mode_device:
return
self._fan_mode_device.action("setFanMode", HA_FANMODES[fan_mode])
self._fan_mode_device.execute_action("setFanMode", [HA_FANMODES[fan_mode]])
@property
def fibaro_op_mode(self) -> str | int:
@ -255,7 +255,7 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
if not self._op_mode_device:
return HA_OPMODES_HVAC[HVACMode.AUTO]
device = self._op_mode_device.fibaro_device
device = self._op_mode_device
if device.has_operating_mode:
return device.operating_mode
@ -281,17 +281,17 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
if not self._op_mode_device:
return
if "setOperatingMode" in self._op_mode_device.fibaro_device.actions:
self._op_mode_device.action("setOperatingMode", HA_OPMODES_HVAC[hvac_mode])
elif "setThermostatMode" in self._op_mode_device.fibaro_device.actions:
device = self._op_mode_device.fibaro_device
device = self._op_mode_device
if "setOperatingMode" in device.actions:
device.execute_action("setOperatingMode", [HA_OPMODES_HVAC[hvac_mode]])
elif "setThermostatMode" in device.actions:
if device.has_supported_thermostat_modes:
for mode in device.supported_thermostat_modes:
if mode.lower() == hvac_mode:
self._op_mode_device.action("setThermostatMode", mode)
device.execute_action("setThermostatMode", [mode])
break
elif "setMode" in self._op_mode_device.fibaro_device.actions:
self._op_mode_device.action("setMode", HA_OPMODES_HVAC[hvac_mode])
elif "setMode" in device.actions:
device.execute_action("setMode", [HA_OPMODES_HVAC[hvac_mode]])
@property
def hvac_action(self) -> HVACAction | None:
@ -299,7 +299,7 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
if not self._op_mode_device:
return None
device = self._op_mode_device.fibaro_device
device = self._op_mode_device
if device.has_thermostat_operating_state:
with suppress(ValueError):
return HVACAction(device.thermostat_operating_state.lower())
@ -315,15 +315,15 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
if not self._op_mode_device:
return None
if self._op_mode_device.fibaro_device.has_thermostat_mode:
mode = self._op_mode_device.fibaro_device.thermostat_mode
if self._op_mode_device.has_thermostat_mode:
mode = self._op_mode_device.thermostat_mode
if self.preset_modes is not None and mode in self.preset_modes:
return mode
return None
if self._op_mode_device.fibaro_device.has_operating_mode:
mode = self._op_mode_device.fibaro_device.operating_mode
if self._op_mode_device.has_operating_mode:
mode = self._op_mode_device.operating_mode
else:
mode = self._op_mode_device.fibaro_device.mode
mode = self._op_mode_device.mode
if mode not in OPMODES_PRESET:
return None
@ -334,20 +334,22 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
if self._op_mode_device is None:
return
if "setThermostatMode" in self._op_mode_device.fibaro_device.actions:
self._op_mode_device.action("setThermostatMode", preset_mode)
elif "setOperatingMode" in self._op_mode_device.fibaro_device.actions:
self._op_mode_device.action(
"setOperatingMode", HA_OPMODES_PRESET[preset_mode]
if "setThermostatMode" in self._op_mode_device.actions:
self._op_mode_device.execute_action("setThermostatMode", [preset_mode])
elif "setOperatingMode" in self._op_mode_device.actions:
self._op_mode_device.execute_action(
"setOperatingMode", [HA_OPMODES_PRESET[preset_mode]]
)
elif "setMode" in self._op_mode_device.actions:
self._op_mode_device.execute_action(
"setMode", [HA_OPMODES_PRESET[preset_mode]]
)
elif "setMode" in self._op_mode_device.fibaro_device.actions:
self._op_mode_device.action("setMode", HA_OPMODES_PRESET[preset_mode])
@property
def current_temperature(self) -> float | None:
"""Return the current temperature."""
if self._temp_sensor_device:
device = self._temp_sensor_device.fibaro_device
device = self._temp_sensor_device
if device.has_heating_thermostat_setpoint:
return device.heating_thermostat_setpoint
return device.value.float_value()
@ -357,7 +359,7 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
if self._target_temp_device:
device = self._target_temp_device.fibaro_device
device = self._target_temp_device
if device.has_heating_thermostat_setpoint_future:
return device.heating_thermostat_setpoint_future
return device.target_level
@ -368,9 +370,11 @@ class FibaroThermostat(FibaroEntity, ClimateEntity):
temperature = kwargs.get(ATTR_TEMPERATURE)
target = self._target_temp_device
if target is not None and temperature is not None:
if "setThermostatSetpoint" in target.fibaro_device.actions:
target.action("setThermostatSetpoint", self.fibaro_op_mode, temperature)
elif "setHeatingThermostatSetpoint" in target.fibaro_device.actions:
target.action("setHeatingThermostatSetpoint", temperature)
if "setThermostatSetpoint" in target.actions:
target.execute_action(
"setThermostatSetpoint", [self.fibaro_op_mode, temperature]
)
elif "setHeatingThermostatSetpoint" in target.actions:
target.execute_action("setHeatingThermostatSetpoint", [temperature])
else:
target.action("setTargetLevel", temperature)
target.execute_action("setTargetLevel", [temperature])

19
homeassistant/components/fibaro/entity.py
View File

@ -11,6 +11,8 @@ from pyfibaro.fibaro_device import DeviceModel
from homeassistant.const import ATTR_ARMED, ATTR_BATTERY_LEVEL
from homeassistant.helpers.entity import Entity
from . import FibaroController
_LOGGER = logging.getLogger(__name__)
@ -22,7 +24,7 @@ class FibaroEntity(Entity):
def __init__(self, fibaro_device: DeviceModel) -> None:
"""Initialize the device."""
self.fibaro_device = fibaro_device
self.controller = fibaro_device.fibaro_controller
self.controller: FibaroController = fibaro_device.fibaro_controller
self.ha_id = fibaro_device.ha_id
self._attr_name = fibaro_device.friendly_name
self._attr_unique_id = fibaro_device.unique_id_str
@ -54,15 +56,6 @@ class FibaroEntity(Entity):
return self.fibaro_device.value_2.int_value()
return None
def dont_know_message(self, cmd: str) -> None:
"""Make a warning in case we don't know how to perform an action."""
_LOGGER.warning(
"Not sure how to %s: %s (available actions: %s)",
cmd,
str(self.ha_id),
str(self.fibaro_device.actions),
)
def set_level(self, level: int) -> None:
"""Set the level of Fibaro device."""
self.action("setValue", level)
@ -97,11 +90,7 @@ class FibaroEntity(Entity):
def action(self, cmd: str, *args: Any) -> None:
"""Perform an action on the Fibaro HC."""
if cmd in self.fibaro_device.actions:
self.fibaro_device.execute_action(cmd, args)
_LOGGER.debug("-> %s.%s%s called", str(self.ha_id), str(cmd), str(args))
else:
self.dont_know_message(cmd)
self.fibaro_device.execute_action(cmd, args)
@property
def current_binary_state(self) -> bool:

54
tests/components/fibaro/conftest.py
View File

@ -157,12 +157,31 @@ def mock_thermostat() -> Mock:
return climate
@pytest.fixture
def mock_thermostat_parent() -> Mock:
"""Fixture for a thermostat."""
climate = Mock()
climate.fibaro_id = 5
climate.parent_fibaro_id = 0
climate.name = "Test climate"
climate.room_id = 1
climate.dead = False
climate.visible = True
climate.enabled = True
climate.type = "com.fibaro.device"
climate.base_type = "com.fibaro.device"
climate.properties = {"manufacturer": ""}
climate.actions = []
return climate
@pytest.fixture
def mock_thermostat_with_operating_mode() -> Mock:
"""Fixture for a thermostat."""
climate = Mock()
climate.fibaro_id = 4
climate.parent_fibaro_id = 0
climate.fibaro_id = 6
climate.endpoint_id = 1
climate.parent_fibaro_id = 5
climate.name = "Test climate"
climate.room_id = 1
climate.dead = False
@ -171,20 +190,47 @@ def mock_thermostat_with_operating_mode() -> Mock:
climate.type = "com.fibaro.thermostatDanfoss"
climate.base_type = "com.fibaro.device"
climate.properties = {"manufacturer": ""}
climate.actions = {"setOperationMode": 1}
climate.actions = {"setOperatingMode": 1, "setTargetLevel": 1}
climate.supported_features = {}
climate.has_supported_operating_modes = True
climate.supported_operating_modes = [0, 1, 15]
climate.has_operating_mode = True
climate.operating_mode = 15
climate.has_supported_thermostat_modes = False
climate.has_thermostat_mode = False
climate.has_unit = True
climate.unit = "C"
climate.has_heating_thermostat_setpoint = False
climate.has_heating_thermostat_setpoint_future = False
climate.target_level = 23
value_mock = Mock()
value_mock.has_value = True
value_mock.int_value.return_value = 20
value_mock.float_value.return_value = 20
climate.value = value_mock
return climate
@pytest.fixture
def mock_fan_device() -> Mock:
"""Fixture for a fan endpoint of a thermostat device."""
climate = Mock()
climate.fibaro_id = 7
climate.endpoint_id = 1
climate.parent_fibaro_id = 5
climate.name = "Test fan"
climate.room_id = 1
climate.dead = False
climate.visible = True
climate.enabled = True
climate.type = "com.fibaro.fan"
climate.base_type = "com.fibaro.device"
climate.properties = {"manufacturer": ""}
climate.actions = {"setFanMode": 1}
climate.supported_modes = [0, 1, 2]
climate.mode = 1
return climate
@pytest.fixture
def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Return the default mocked config entry."""

150
tests/components/fibaro/test_climate.py
View File

@ -130,5 +130,153 @@ async def test_hvac_mode_with_operation_mode_support(
# Act
await init_integration(hass, mock_config_entry)
# Assert
state = hass.states.get("climate.room_1_test_climate_4")
state = hass.states.get("climate.room_1_test_climate_6")
assert state.state == HVACMode.AUTO
async def test_set_hvac_mode_with_operation_mode_support(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_thermostat_with_operating_mode: Mock,
mock_room: Mock,
) -> None:
"""Test that set_hvac_mode() works."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [mock_thermostat_with_operating_mode]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.CLIMATE]):
# Act
await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"climate",
"set_hvac_mode",
{"entity_id": "climate.room_1_test_climate_6", "hvac_mode": HVACMode.HEAT},
blocking=True,
)
# Assert
mock_thermostat_with_operating_mode.execute_action.assert_called_once()
async def test_fan_mode(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_thermostat_parent: Mock,
mock_thermostat_with_operating_mode: Mock,
mock_fan_device: Mock,
mock_room: Mock,
) -> None:
"""Test that operating mode works."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [
mock_thermostat_parent,
mock_thermostat_with_operating_mode,
mock_fan_device,
]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.CLIMATE]):
# Act
await init_integration(hass, mock_config_entry)
# Assert
state = hass.states.get("climate.room_1_test_climate_6")
assert state.attributes["fan_mode"] == "low"
assert state.attributes["fan_modes"] == ["off", "low", "auto_high"]
async def test_set_fan_mode(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_thermostat_parent: Mock,
mock_thermostat_with_operating_mode: Mock,
mock_fan_device: Mock,
mock_room: Mock,
) -> None:
"""Test that set_fan_mode() works."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [
mock_thermostat_parent,
mock_thermostat_with_operating_mode,
mock_fan_device,
]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.CLIMATE]):
# Act
await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"climate",
"set_fan_mode",
{"entity_id": "climate.room_1_test_climate_6", "fan_mode": "off"},
blocking=True,
)
# Assert
mock_fan_device.execute_action.assert_called_once()
async def test_target_temperature(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_thermostat_parent: Mock,
mock_thermostat_with_operating_mode: Mock,
mock_fan_device: Mock,
mock_room: Mock,
) -> None:
"""Test that operating mode works."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [
mock_thermostat_parent,
mock_thermostat_with_operating_mode,
mock_fan_device,
]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.CLIMATE]):
# Act
await init_integration(hass, mock_config_entry)
# Assert
state = hass.states.get("climate.room_1_test_climate_6")
assert state.attributes["temperature"] == 23
async def test_set_target_temperature(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_thermostat_parent: Mock,
mock_thermostat_with_operating_mode: Mock,
mock_fan_device: Mock,
mock_room: Mock,
) -> None:
"""Test that set_fan_mode() works."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [
mock_thermostat_parent,
mock_thermostat_with_operating_mode,
mock_fan_device,
]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.CLIMATE]):
# Act
await init_integration(hass, mock_config_entry)
await hass.services.async_call(
"climate",
"set_temperature",
{"entity_id": "climate.room_1_test_climate_6", "temperature": 25.5},
blocking=True,
)
# Assert
mock_thermostat_with_operating_mode.execute_action.assert_called_once()

28
tests/components/fibaro/test_light.py
View File

@ -2,7 +2,8 @@
from unittest.mock import Mock, patch
from homeassistant.const import Platform
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, SERVICE_TURN_OFF, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@ -55,3 +56,28 @@ async def test_light_brightness(
state = hass.states.get("light.room_1_test_light_3")
assert state.attributes["brightness"] == 51
assert state.state == "on"
async def test_light_turn_off(
hass: HomeAssistant,
mock_fibaro_client: Mock,
mock_config_entry: MockConfigEntry,
mock_light: Mock,
mock_room: Mock,
) -> None:
"""Test activate scene is called."""
# Arrange
mock_fibaro_client.read_rooms.return_value = [mock_room]
mock_fibaro_client.read_devices.return_value = [mock_light]
with patch("homeassistant.components.fibaro.PLATFORMS", [Platform.LIGHT]):
await init_integration(hass, mock_config_entry)
# Act
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.room_1_test_light_3"},
blocking=True,
)
# Assert
assert mock_light.execute_action.call_count == 1