Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-20 02:48:57 +00:00
Code Activity

Add support for FRITZ! Smarthome routines (#158947)

Browse Source
This commit is contained in:
Michael
2025-12-18 13:09:06 +01:00
committed by GitHub
parent 5349045932
commit 3d71b6de44
5 changed files with 235 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
homeassistant/components/fritzbox/coordinator.py
View File

@@ -6,7 +6,7 @@ from dataclasses import dataclass
from datetime import timedelta
from pyfritzhome import Fritzhome, FritzhomeDevice, LoginError
from pyfritzhome.devicetypes import FritzhomeTemplate
from pyfritzhome.devicetypes import FritzhomeTemplate, FritzhomeTrigger
from requests.exceptions import ConnectionError as RequestConnectionError, HTTPError
from homeassistant.config_entries import ConfigEntry
@@ -27,6 +27,7 @@ class FritzboxCoordinatorData:
devices: dict[str, FritzhomeDevice]
templates: dict[str, FritzhomeTemplate]
triggers: dict[str, FritzhomeTrigger]
supported_color_properties: dict[str, tuple[dict, list]]
@@ -37,6 +38,7 @@ class FritzboxDataUpdateCoordinator(DataUpdateCoordinator[FritzboxCoordinatorDat
configuration_url: str
fritz: Fritzhome
has_templates: bool
has_triggers: bool
def __init__(self, hass: HomeAssistant, config_entry: FritzboxConfigEntry) -> None:
"""Initialize the Fritzbox Smarthome device coordinator."""
@@ -50,8 +52,9 @@ class FritzboxDataUpdateCoordinator(DataUpdateCoordinator[FritzboxCoordinatorDat
self.new_devices: set[str] = set()
self.new_templates: set[str] = set()
self.new_triggers: set[str] = set()
self.data = FritzboxCoordinatorData({}, {}, {})
self.data = FritzboxCoordinatorData({}, {}, {}, {})
async def async_setup(self) -> None:
"""Set up the coordinator."""
@@ -74,6 +77,11 @@ class FritzboxDataUpdateCoordinator(DataUpdateCoordinator[FritzboxCoordinatorDat
)
LOGGER.debug("enable smarthome templates: %s", self.has_templates)
self.has_triggers = await self.hass.async_add_executor_job(
self.fritz.has_triggers
)
LOGGER.debug("enable smarthome triggers: %s", self.has_triggers)
self.configuration_url = self.fritz.get_prefixed_host()
await self.async_config_entry_first_refresh()
@@ -92,7 +100,7 @@ class FritzboxDataUpdateCoordinator(DataUpdateCoordinator[FritzboxCoordinatorDat
available_main_ains = [
ain
for ain, dev in data.devices.items() | data.templates.items()
for ain, dev in (data.devices | data.templates | data.triggers).items()
if dev.device_and_unit_id[1] is None
]
device_reg = dr.async_get(self.hass)
@@ -112,6 +120,9 @@ class FritzboxDataUpdateCoordinator(DataUpdateCoordinator[FritzboxCoordinatorDat
self.fritz.update_devices(ignore_removed=False)
if self.has_templates:
self.fritz.update_templates(ignore_removed=False)
if self.has_triggers:
self.fritz.update_triggers(ignore_removed=False)
except RequestConnectionError as ex:
raise UpdateFailed from ex
except HTTPError:
@@ -123,6 +134,8 @@ class FritzboxDataUpdateCoordinator(DataUpdateCoordinator[FritzboxCoordinatorDat
self.fritz.update_devices(ignore_removed=False)
if self.has_templates:
self.fritz.update_templates(ignore_removed=False)
if self.has_triggers:
self.fritz.update_triggers(ignore_removed=False)
devices = self.fritz.get_devices()
device_data = {}
@@ -156,12 +169,20 @@ class FritzboxDataUpdateCoordinator(DataUpdateCoordinator[FritzboxCoordinatorDat
for template in templates:
template_data[template.ain] = template
trigger_data = {}
if self.has_triggers:
triggers = self.fritz.get_triggers()
for trigger in triggers:
trigger_data[trigger.ain] = trigger
self.new_devices = device_data.keys() - self.data.devices.keys()
self.new_templates = template_data.keys() - self.data.templates.keys()
self.new_triggers = trigger_data.keys() - self.data.triggers.keys()
return FritzboxCoordinatorData(
devices=device_data,
templates=template_data,
triggers=trigger_data,
supported_color_properties=supported_color_properties,
)
@@ -193,6 +214,7 @@ class FritzboxDataUpdateCoordinator(DataUpdateCoordinator[FritzboxCoordinatorDat
if (
self.data.devices.keys() - new_data.devices.keys()
or self.data.templates.keys() - new_data.templates.keys()
or self.data.triggers.keys() - new_data.triggers.keys()
):
self.cleanup_removed_devices(new_data)

62
homeassistant/components/fritzbox/switch.py
View File

@@ -4,14 +4,17 @@ from __future__ import annotations
from typing import Any
from pyfritzhome.devicetypes import FritzhomeTrigger
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import FritzboxConfigEntry
from .entity import FritzBoxDeviceEntity
from .entity import FritzBoxDeviceEntity, FritzBoxEntity
# Coordinator handles data updates, so we can allow unlimited parallel updates
PARALLEL_UPDATES = 0
@@ -26,21 +29,27 @@ async def async_setup_entry(
coordinator = entry.runtime_data
@callback
def _add_entities(devices: set[str] | None = None) -> None:
"""Add devices."""
def _add_entities(
devices: set[str] | None = None, triggers: set[str] | None = None
) -> None:
"""Add devices and triggers."""
if devices is None:
devices = coordinator.new_devices
if not devices:
if triggers is None:
triggers = coordinator.new_triggers
if not devices and not triggers:
return
async_add_entities(
entities = [
FritzboxSwitch(coordinator, ain)
for ain in devices
if coordinator.data.devices[ain].has_switch
)
] + [FritzboxTrigger(coordinator, ain) for ain in triggers]
async_add_entities(entities)
entry.async_on_unload(coordinator.async_add_listener(_add_entities))
_add_entities(set(coordinator.data.devices))
_add_entities(set(coordinator.data.devices), set(coordinator.data.triggers))
class FritzboxSwitch(FritzBoxDeviceEntity, SwitchEntity):
@@ -70,3 +79,42 @@ class FritzboxSwitch(FritzBoxDeviceEntity, SwitchEntity):
translation_domain=DOMAIN,
translation_key="manual_switching_disabled",
)
class FritzboxTrigger(FritzBoxEntity, SwitchEntity):
"""The switch class for FRITZ!SmartHome triggers."""
@property
def data(self) -> FritzhomeTrigger:
"""Return the trigger data entity."""
return self.coordinator.data.triggers[self.ain]
@property
def device_info(self) -> DeviceInfo:
"""Return device specific attributes."""
return DeviceInfo(
name=self.data.name,
identifiers={(DOMAIN, self.ain)},
configuration_url=self.coordinator.configuration_url,
manufacturer="FRITZ!",
model="SmartHome Routine",
)
@property
def is_on(self) -> bool:
"""Return true if the trigger is active."""
return self.data.active # type: ignore [no-any-return]
async def async_turn_on(self, **kwargs: Any) -> None:
"""Activate the trigger."""
await self.hass.async_add_executor_job(
self.coordinator.fritz.set_trigger_active, self.ain
)
await self.coordinator.async_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Deactivate the trigger."""
await self.hass.async_add_executor_job(
self.coordinator.fritz.set_trigger_inactive, self.ain
)
await self.coordinator.async_refresh()

20
tests/components/fritzbox/__init__.py
View File

@@ -25,6 +25,7 @@ async def setup_config_entry(
device: Mock | None = None,
fritz: Mock | None = None,
template: Mock | None = None,
trigger: Mock | None = None,
) -> MockConfigEntry:
"""Do setup of a MockConfigEntry."""
entry = MockConfigEntry(
@@ -39,6 +40,9 @@ async def setup_config_entry(
if template is not None and fritz is not None:
fritz().get_templates.return_value = [template]
if trigger is not None and fritz is not None:
fritz().get_triggers.return_value = [trigger]
await hass.config_entries.async_setup(entry.entry_id)
if device is not None:
await hass.async_block_till_done()
@@ -46,7 +50,10 @@ async def setup_config_entry(
def set_devices(
fritz: Mock, devices: list[Mock] | None = None, templates: list[Mock] | None = None
fritz: Mock,
devices: list[Mock] | None = None,
templates: list[Mock] | None = None,
triggers: list[Mock] | None = None,
) -> None:
"""Set list of devices or templates."""
if devices is not None:
@@ -55,6 +62,9 @@ def set_devices(
if templates is not None:
fritz().get_templates.return_value = templates
if triggers is not None:
fritz().get_triggers.return_value = triggers
class FritzEntityBaseMock(Mock):
"""base mock of a AVM Fritz!Box binary sensor device."""
@@ -199,3 +209,11 @@ class FritzDeviceCoverUnknownPositionMock(FritzDeviceCoverMock):
"""Mock of a AVM Fritz!Box cover device with unknown position."""
levelpercentage = None
class FritzTriggerMock(FritzEntityBaseMock):
"""Mock of a AVM Fritz!Box smarthome trigger."""
active = True
ain = "trg1234 56789"
name = "fake_trigger"

48
tests/components/fritzbox/snapshots/test_switch.ambr
View File

@@ -47,3 +47,51 @@
'state': 'on',
})
# ---
# name: test_setup[switch.fake_trigger-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.fake_trigger',
'has_entity_name': False,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'fake_trigger',
'platform': 'fritzbox',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'trg1234 56789',
'unit_of_measurement': None,
})
# ---
# name: test_setup[switch.fake_trigger-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'fake_trigger',
}),
'context': <ANY>,
'entity_id': 'switch.fake_trigger',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---

113
tests/components/fritzbox/test_switch.py
View File

@@ -23,12 +23,13 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from . import FritzDeviceSwitchMock, set_devices, setup_config_entry
from . import FritzDeviceSwitchMock, FritzTriggerMock, set_devices, setup_config_entry
from .const import CONF_FAKE_NAME, MOCK_CONFIG
from tests.common import async_fire_time_changed, snapshot_platform
ENTITY_ID = f"{SWITCH_DOMAIN}.{CONF_FAKE_NAME}"
SWITCH_ENTITY_ID = f"{SWITCH_DOMAIN}.{CONF_FAKE_NAME}"
TRIGGER_ENTITY_ID = f"{SWITCH_DOMAIN}.fake_trigger"
async def test_setup(
@@ -39,50 +40,56 @@ async def test_setup(
) -> None:
"""Test setup of platform."""
device = FritzDeviceSwitchMock()
trigger = FritzTriggerMock()
with patch("homeassistant.components.fritzbox.PLATFORMS", [Platform.SWITCH]):
entry = await setup_config_entry(
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], ENTITY_ID, device, fritz
hass,
MOCK_CONFIG[DOMAIN][CONF_DEVICES][0],
device=device,
fritz=fritz,
trigger=trigger,
)
assert entry.state is ConfigEntryState.LOADED
await snapshot_platform(hass, entity_registry, snapshot, entry.entry_id)
async def test_turn_on(hass: HomeAssistant, fritz: Mock) -> None:
"""Test turn device on."""
async def test_switch_turn_on(hass: HomeAssistant, fritz: Mock) -> None:
"""Test turn switch device on."""
device = FritzDeviceSwitchMock()
await setup_config_entry(
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], ENTITY_ID, device, fritz
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], device=device, fritz=fritz
)
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: SWITCH_ENTITY_ID}, True
)
assert device.set_switch_state_on.call_count == 1
async def test_turn_off(hass: HomeAssistant, fritz: Mock) -> None:
"""Test turn device off."""
async def test_switch_turn_off(hass: HomeAssistant, fritz: Mock) -> None:
"""Test turn switch device off."""
device = FritzDeviceSwitchMock()
await setup_config_entry(
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], ENTITY_ID, device, fritz
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], device=device, fritz=fritz
)
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: SWITCH_ENTITY_ID}, True
)
assert device.set_switch_state_off.call_count == 1
async def test_toggle_while_locked(hass: HomeAssistant, fritz: Mock) -> None:
"""Test toggling while device is locked."""
async def test_switch_toggle_while_locked(hass: HomeAssistant, fritz: Mock) -> None:
"""Test toggling while switch device is locked."""
device = FritzDeviceSwitchMock()
device.lock = True
await setup_config_entry(
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], ENTITY_ID, device, fritz
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], device=device, fritz=fritz
)
with pytest.raises(
@@ -90,7 +97,7 @@ async def test_toggle_while_locked(hass: HomeAssistant, fritz: Mock) -> None:
match="Can't toggle switch while manual switching is disabled for the device",
):
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, True
SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: SWITCH_ENTITY_ID}, True
)
with pytest.raises(
@@ -98,17 +105,23 @@ async def test_toggle_while_locked(hass: HomeAssistant, fritz: Mock) -> None:
match="Can't toggle switch while manual switching is disabled for the device",
):
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, True
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: SWITCH_ENTITY_ID}, True
)
async def test_update(hass: HomeAssistant, fritz: Mock) -> None:
"""Test update without error."""
device = FritzDeviceSwitchMock()
trigger = FritzTriggerMock()
await setup_config_entry(
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], ENTITY_ID, device, fritz
hass,
MOCK_CONFIG[DOMAIN][CONF_DEVICES][0],
device=device,
fritz=fritz,
trigger=trigger,
)
assert fritz().update_devices.call_count == 1
assert fritz().update_triggers.call_count == 1
assert fritz().login.call_count == 1
next_update = dt_util.utcnow() + timedelta(seconds=200)
@@ -116,6 +129,7 @@ async def test_update(hass: HomeAssistant, fritz: Mock) -> None:
await hass.async_block_till_done(wait_background_tasks=True)
assert fritz().update_devices.call_count == 2
assert fritz().update_triggers.call_count == 2
assert fritz().login.call_count == 1
@@ -124,7 +138,7 @@ async def test_update_error(hass: HomeAssistant, fritz: Mock) -> None:
device = FritzDeviceSwitchMock()
fritz().update_devices.side_effect = HTTPError("Boom")
entry = await setup_config_entry(
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], ENTITY_ID, device, fritz
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], device=device, fritz=fritz
)
assert entry.state is ConfigEntryState.SETUP_RETRY
assert fritz().update_devices.call_count == 2
@@ -145,10 +159,10 @@ async def test_assume_device_unavailable(hass: HomeAssistant, fritz: Mock) -> No
device.energy = 0
device.power = 0
await setup_config_entry(
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], ENTITY_ID, device, fritz
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], device=device, fritz=fritz
)
state = hass.states.get(ENTITY_ID)
state = hass.states.get(SWITCH_ENTITY_ID)
assert state
assert state.state == STATE_UNAVAILABLE
@@ -156,13 +170,19 @@ async def test_assume_device_unavailable(hass: HomeAssistant, fritz: Mock) -> No
async def test_discover_new_device(hass: HomeAssistant, fritz: Mock) -> None:
"""Test adding new discovered devices during runtime."""
device = FritzDeviceSwitchMock()
trigger = FritzTriggerMock()
await setup_config_entry(
hass, MOCK_CONFIG[DOMAIN][CONF_DEVICES][0], ENTITY_ID, device, fritz
hass,
MOCK_CONFIG[DOMAIN][CONF_DEVICES][0],
device=device,
fritz=fritz,
trigger=trigger,
)
state = hass.states.get(ENTITY_ID)
assert state
assert hass.states.get(SWITCH_ENTITY_ID)
assert hass.states.get(TRIGGER_ENTITY_ID)
# add new switch device
new_device = FritzDeviceSwitchMock()
new_device.ain = "7890 1234"
new_device.name = "new_switch"
@@ -172,5 +192,48 @@ async def test_discover_new_device(hass: HomeAssistant, fritz: Mock) -> None:
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done(wait_background_tasks=True)
state = hass.states.get(f"{SWITCH_DOMAIN}.new_switch")
assert state
assert hass.states.get(f"{SWITCH_DOMAIN}.new_switch")
# add new trigger
new_trigger = FritzTriggerMock()
new_trigger.ain = "trg7890 1234"
new_trigger.name = "new_trigger"
set_devices(fritz, triggers=[trigger, new_trigger])
next_update = dt_util.utcnow() + timedelta(seconds=200)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done(wait_background_tasks=True)
assert hass.states.get(f"{SWITCH_DOMAIN}.new_trigger")
async def test_activate_trigger(hass: HomeAssistant, fritz: Mock) -> None:
"""Test activating a FRITZ! trigger."""
trigger = FritzTriggerMock()
await setup_config_entry(
hass,
MOCK_CONFIG[DOMAIN][CONF_DEVICES][0],
fritz=fritz,
trigger=trigger,
)
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: TRIGGER_ENTITY_ID}, True
)
assert fritz().set_trigger_active.call_count == 1
async def test_deactivate_trigger(hass: HomeAssistant, fritz: Mock) -> None:
"""Test deactivating a FRITZ! trigger."""
trigger = FritzTriggerMock()
await setup_config_entry(
hass,
MOCK_CONFIG[DOMAIN][CONF_DEVICES][0],
fritz=fritz,
trigger=trigger,
)
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: TRIGGER_ENTITY_ID}, True
)
assert fritz().set_trigger_inactive.call_count == 1