Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00
Code Activity

Add a DALI line into the device hierarchy with a broadcast entity (#156570)

Browse Source 
Co-authored-by: Tom <CoMPaTech@users.noreply.github.com>
This commit is contained in:
MoonDevLT
2025-12-19 14:57:51 +01:00
committed by GitHub
parent ddb1ae371d
commit 6f9dc2e5a2
8 changed files with 383 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
homeassistant/components/lunatone/__init__.py
View File

@@ -2,7 +2,7 @@
from typing import Final
from lunatone_rest_api_client import Auth, Devices, Info
from lunatone_rest_api_client import Auth, DALIBroadcast, Devices, Info
from homeassistant.const import CONF_URL, Platform
from homeassistant.core import HomeAssistant
@@ -42,19 +42,27 @@ async def async_setup_entry(hass: HomeAssistant, entry: LunatoneConfigEntry) ->
name=info_api.name,
manufacturer="Lunatone",
sw_version=info_api.version,
hw_version=info_api.data.device.pcb,
hw_version=coordinator_info.data.device.pcb,
configuration_url=entry.data[CONF_URL],
serial_number=str(info_api.serial_number),
model=info_api.product_name,
model_id=(
f"{info_api.data.device.article_number}{info_api.data.device.article_info}"
f"{coordinator_info.data.device.article_number}{coordinator_info.data.device.article_info}"
),
)
coordinator_devices = LunatoneDevicesDataUpdateCoordinator(hass, entry, devices_api)
await coordinator_devices.async_config_entry_first_refresh()
entry.runtime_data = LunatoneData(coordinator_info, coordinator_devices)
dali_line_broadcasts = [
DALIBroadcast(auth_api, int(line)) for line in coordinator_info.data.lines
]
entry.runtime_data = LunatoneData(
coordinator_info,
coordinator_devices,
dali_line_broadcasts,
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True

5
homeassistant/components/lunatone/coordinator.py
View File

@@ -7,7 +7,7 @@ from datetime import timedelta
import logging
import aiohttp
from lunatone_rest_api_client import Device, Devices, Info
from lunatone_rest_api_client import DALIBroadcast, Device, Devices, Info
from lunatone_rest_api_client.models import InfoData
from homeassistant.config_entries import ConfigEntry
@@ -18,6 +18,7 @@ from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
DEFAULT_INFO_SCAN_INTERVAL = timedelta(seconds=60)
DEFAULT_DEVICES_SCAN_INTERVAL = timedelta(seconds=10)
@@ -27,6 +28,7 @@ class LunatoneData:
coordinator_info: LunatoneInfoDataUpdateCoordinator
coordinator_devices: LunatoneDevicesDataUpdateCoordinator
dali_line_broadcasts: list[DALIBroadcast]
type LunatoneConfigEntry = ConfigEntry[LunatoneData]
@@ -47,6 +49,7 @@ class LunatoneInfoDataUpdateCoordinator(DataUpdateCoordinator[InfoData]):
config_entry=config_entry,
name=f"{DOMAIN}-info",
always_update=False,
update_interval=DEFAULT_INFO_SCAN_INTERVAL,
)
self.info_api = info_api

114
homeassistant/components/lunatone/light.py
View File

@@ -5,6 +5,9 @@ from __future__ import annotations
import asyncio
from typing import Any
from lunatone_rest_api_client import DALIBroadcast
from lunatone_rest_api_client.models import LineStatus
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ColorMode,
@@ -18,7 +21,11 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.util.color import brightness_to_value, value_to_brightness
from .const import DOMAIN
from .coordinator import LunatoneConfigEntry, LunatoneDevicesDataUpdateCoordinator
from .coordinator import (
LunatoneConfigEntry,
LunatoneDevicesDataUpdateCoordinator,
LunatoneInfoDataUpdateCoordinator,
)
PARALLEL_UPDATES = 0
STATUS_UPDATE_DELAY = 0.04
@@ -32,8 +39,15 @@ async def async_setup_entry(
"""Set up the Lunatone Light platform."""
coordinator_info = config_entry.runtime_data.coordinator_info
coordinator_devices = config_entry.runtime_data.coordinator_devices
dali_line_broadcasts = config_entry.runtime_data.dali_line_broadcasts
async_add_entities(
entities: list[LightEntity] = [
LunatoneLineBroadcastLight(
coordinator_info, coordinator_devices, dali_line_broadcast
)
for dali_line_broadcast in dali_line_broadcasts
]
entities.extend(
[
LunatoneLight(
coordinator_devices, device_id, coordinator_info.data.device.serial
@@ -42,6 +56,8 @@ async def async_setup_entry(
]
)
async_add_entities(entities)
class LunatoneLight(
CoordinatorEntity[LunatoneDevicesDataUpdateCoordinator], LightEntity
@@ -62,22 +78,24 @@ class LunatoneLight(
device_id: int,
interface_serial_number: int,
) -> None:
"""Initialize a LunatoneLight."""
super().__init__(coordinator=coordinator)
"""Initialize a Lunatone light."""
super().__init__(coordinator)
self._device_id = device_id
self._interface_serial_number = interface_serial_number
self._device = self.coordinator.data.get(self._device_id)
self._device = self.coordinator.data[self._device_id]
self._attr_unique_id = f"{interface_serial_number}-device{device_id}"
@property
def device_info(self) -> DeviceInfo:
"""Return the device info."""
assert self.unique_id
name = self._device.name if self._device is not None else None
return DeviceInfo(
identifiers={(DOMAIN, self.unique_id)},
name=name,
via_device=(DOMAIN, str(self._interface_serial_number)),
name=self._device.name,
via_device=(
DOMAIN,
f"{self._interface_serial_number}-line{self._device.data.line}",
),
)
@property
@@ -93,8 +111,6 @@ class LunatoneLight(
@property
def brightness(self) -> int:
"""Return the brightness of this light between 0..255."""
if self._device is None:
return 0
return value_to_brightness(self.BRIGHTNESS_SCALE, self._device.brightness)
@property
@@ -112,17 +128,17 @@ class LunatoneLight(
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._device = self.coordinator.data.get(self._device_id)
self._device = self.coordinator.data[self._device_id]
self.async_write_ha_state()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Instruct the light to turn on."""
assert self._device
if brightness_supported(self.supported_color_modes):
brightness = kwargs.get(ATTR_BRIGHTNESS, self._last_brightness)
await self._device.fade_to_brightness(
brightness_to_value(self.BRIGHTNESS_SCALE, brightness)
brightness_to_value(
self.BRIGHTNESS_SCALE,
kwargs.get(ATTR_BRIGHTNESS, self._last_brightness),
)
)
else:
await self._device.switch_on()
@@ -132,8 +148,6 @@ class LunatoneLight(
async def async_turn_off(self, **kwargs: Any) -> None:
"""Instruct the light to turn off."""
assert self._device
if brightness_supported(self.supported_color_modes):
self._last_brightness = self.brightness
await self._device.fade_to_brightness(0)
@@ -142,3 +156,69 @@ class LunatoneLight(
await asyncio.sleep(STATUS_UPDATE_DELAY)
await self.coordinator.async_refresh()
class LunatoneLineBroadcastLight(
CoordinatorEntity[LunatoneInfoDataUpdateCoordinator], LightEntity
):
"""Representation of a Lunatone line broadcast light."""
BRIGHTNESS_SCALE = (1, 100)
_attr_assumed_state = True
_attr_color_mode = ColorMode.BRIGHTNESS
_attr_supported_color_modes = {ColorMode.BRIGHTNESS}
def __init__(
self,
coordinator_info: LunatoneInfoDataUpdateCoordinator,
coordinator_devices: LunatoneDevicesDataUpdateCoordinator,
broadcast: DALIBroadcast,
) -> None:
"""Initialize a Lunatone line broadcast light."""
super().__init__(coordinator_info)
self._coordinator_devices = coordinator_devices
self._broadcast = broadcast
line = broadcast.line
self._attr_unique_id = f"{coordinator_info.data.device.serial}-line{line}"
line_device = self.coordinator.data.lines[str(line)].device
extra_info: dict = {}
if line_device.serial != coordinator_info.data.device.serial:
extra_info.update(
serial_number=str(line_device.serial),
hw_version=line_device.pcb,
model_id=f"{line_device.article_number}{line_device.article_info}",
)
assert self.unique_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self.unique_id)},
name=f"DALI Line {line}",
via_device=(DOMAIN, str(coordinator_info.data.device.serial)),
**extra_info,
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
line_status = self.coordinator.data.lines[str(self._broadcast.line)].line_status
return super().available and line_status == LineStatus.OK
async def async_turn_on(self, **kwargs: Any) -> None:
"""Instruct the line to turn on."""
await self._broadcast.fade_to_brightness(
brightness_to_value(self.BRIGHTNESS_SCALE, kwargs.get(ATTR_BRIGHTNESS, 255))
)
await asyncio.sleep(STATUS_UPDATE_DELAY)
await self._coordinator_devices.async_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Instruct the line to turn off."""
await self._broadcast.fade_to_brightness(0)
await asyncio.sleep(STATUS_UPDATE_DELAY)
await self._coordinator_devices.async_refresh()

44
tests/components/lunatone/__init__.py
View File

@@ -3,11 +3,13 @@
from typing import Final
from lunatone_rest_api_client.models import (
DALIBusData,
DeviceData,
DeviceInfoData,
DevicesData,
FeaturesStatus,
InfoData,
LineStatus,
)
from lunatone_rest_api_client.models.common import ColorRGBData, ColorWAFData, Status
from lunatone_rest_api_client.models.devices import DeviceStatus
@@ -54,17 +56,43 @@ DEVICE_DATA_LIST: Final[list[DeviceData]] = [
),
]
DEVICES_DATA: Final[DevicesData] = DevicesData(devices=DEVICE_DATA_LIST)
DEVICE_INFO_DATA: Final[DeviceInfoData] = DeviceInfoData(
serial=SERIAL_NUMBER,
gtin=192837465,
pcb="2a",
articleNumber=87654321,
productionYear=20,
productionWeek=1,
)
INFO_DATA: Final[InfoData] = InfoData(
name="Test",
version=VERSION,
device=DeviceInfoData(
serial=SERIAL_NUMBER,
gtin=192837465,
pcb="2a",
articleNumber=87654321,
productionYear=20,
productionWeek=1,
),
device=DEVICE_INFO_DATA,
lines={
"0": DALIBusData(
sendBlockedInitialize=False,
sendBlockedQuiescent=False,
sendBlockedMacroRunning=False,
sendBufferFull=False,
lineStatus=LineStatus.OK,
device=DEVICE_INFO_DATA,
),
"1": DALIBusData(
sendBlockedInitialize=False,
sendBlockedQuiescent=False,
sendBlockedMacroRunning=False,
sendBufferFull=False,
lineStatus=LineStatus.OK,
device=DeviceInfoData(
serial=54321,
gtin=101010101,
pcb="1a",
articleNumber=12345678,
productionYear=22,
productionWeek=10,
),
),
},
)

14
tests/components/lunatone/conftest.py
View File

@@ -31,6 +31,8 @@ def mock_lunatone_devices() -> Generator[AsyncMock]:
def build_devices_mock(devices: Devices):
device_list = []
if devices.data is None:
return device_list
for device_data in devices.data.devices:
device = AsyncMock(spec=Device)
device.data = device_data
@@ -78,6 +80,18 @@ def mock_lunatone_info() -> Generator[AsyncMock]:
yield info
@pytest.fixture
def mock_lunatone_dali_broadcast() -> Generator[AsyncMock]:
"""Mock a Lunatone DALI broadcast object."""
with patch(
"homeassistant.components.lunatone.DALIBroadcast",
autospec=True,
) as mock_dali_broadcast:
dali_broadcast = mock_dali_broadcast.return_value
dali_broadcast.line = 0
yield dali_broadcast
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""

116
tests/components/lunatone/snapshots/test_light.ambr
View File

@@ -113,3 +113,119 @@
'state': 'off',
})
# ---
# name: test_setup[light.lunatone_12345_line0-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'supported_color_modes': list([
<ColorMode.BRIGHTNESS: 'brightness'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'light',
'entity_category': None,
'entity_id': 'light.lunatone_12345_line0',
'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': None,
'platform': 'lunatone',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '12345-line0',
'unit_of_measurement': None,
})
# ---
# name: test_setup[light.lunatone_12345_line0-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'assumed_state': True,
'brightness': None,
'color_mode': None,
'supported_color_modes': list([
<ColorMode.BRIGHTNESS: 'brightness'>,
]),
'supported_features': <LightEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'light.lunatone_12345_line0',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_setup[light.lunatone_12345_line1-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'supported_color_modes': list([
<ColorMode.BRIGHTNESS: 'brightness'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'light',
'entity_category': None,
'entity_id': 'light.lunatone_12345_line1',
'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': None,
'platform': 'lunatone',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '12345-line1',
'unit_of_measurement': None,
})
# ---
# name: test_setup[light.lunatone_12345_line1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'assumed_state': True,
'brightness': None,
'color_mode': None,
'supported_color_modes': list([
<ColorMode.BRIGHTNESS: 'brightness'>,
]),
'supported_features': <LightEntityFeature: 0>,
}),
'context': <ANY>,
'entity_id': 'light.lunatone_12345_line1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

1
tests/components/lunatone/test_init.py
View File

@@ -25,6 +25,7 @@ async def test_load_unload_config_entry(
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_config_entry.unique_id
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, mock_config_entry.unique_id)}

103
tests/components/lunatone/test_light.py
View File

@@ -1,7 +1,9 @@
"""Tests for the Lunatone integration."""
import copy
from unittest.mock import AsyncMock
from lunatone_rest_api_client.models import LineStatus
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.light import ATTR_BRIGHTNESS, DOMAIN as LIGHT_DOMAIN
@@ -38,6 +40,7 @@ async def test_setup(
entities = hass.states.async_all(Platform.LIGHT)
for entity_state in entities:
entity_entry = entity_registry.async_get(entity_state.entity_id)
assert entity_entry
assert entity_entry == snapshot(name=f"{entity_entry.entity_id}-entry")
assert entity_state == snapshot(name=f"{entity_entry.entity_id}-state")
@@ -65,6 +68,7 @@ async def test_turn_on_off(
)
state = hass.states.get(TEST_ENTITY_ID)
assert state
assert state.state == STATE_ON
await hass.services.async_call(
@@ -75,6 +79,7 @@ async def test_turn_on_off(
)
state = hass.states.get(TEST_ENTITY_ID)
assert state
assert state.state == STATE_OFF
@@ -108,6 +113,7 @@ async def test_turn_on_off_with_brightness(
)
state = hass.states.get(TEST_ENTITY_ID)
assert state
assert state.state == STATE_ON
assert state.attributes["brightness"] == expected_brightness
@@ -119,6 +125,7 @@ async def test_turn_on_off_with_brightness(
)
state = hass.states.get(TEST_ENTITY_ID)
assert state
assert state.state == STATE_OFF
assert not state.attributes["brightness"]
@@ -130,5 +137,101 @@ async def test_turn_on_off_with_brightness(
)
state = hass.states.get(TEST_ENTITY_ID)
assert state
assert state.state == STATE_ON
assert state.attributes["brightness"] == expected_brightness
async def test_turn_on_off_broadcast(
hass: HomeAssistant,
mock_lunatone_info: AsyncMock,
mock_lunatone_devices: AsyncMock,
mock_lunatone_dali_broadcast: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the broadcast light can be turned on and off."""
entity_id = (
f"light.{mock_config_entry.domain}_{mock_config_entry.unique_id}"
f"_line{mock_lunatone_dali_broadcast.line}"
)
await setup_integration(hass, mock_config_entry)
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert mock_lunatone_dali_broadcast.fade_to_brightness.await_count == 1
mock_lunatone_dali_broadcast.fade_to_brightness.assert_awaited()
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id, ATTR_BRIGHTNESS: 128},
blocking=True,
)
assert mock_lunatone_dali_broadcast.fade_to_brightness.await_count == 2
mock_lunatone_dali_broadcast.fade_to_brightness.assert_awaited()
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert mock_lunatone_dali_broadcast.fade_to_brightness.await_count == 3
mock_lunatone_dali_broadcast.fade_to_brightness.assert_awaited()
async def test_line_broadcast_available_status(
hass: HomeAssistant,
mock_lunatone_info: AsyncMock,
mock_lunatone_devices: AsyncMock,
mock_lunatone_dali_broadcast: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test if the broadcast light is available."""
entity_id = (
f"light.{mock_config_entry.domain}_{mock_config_entry.unique_id}"
f"_line{mock_lunatone_dali_broadcast.line}"
)
await setup_integration(hass, mock_config_entry)
async def fake_update():
info_data = copy.deepcopy(mock_lunatone_info.data)
info_data.lines["0"].line_status = LineStatus.NOT_REACHABLE
mock_lunatone_info.data = info_data
mock_lunatone_info.async_update.side_effect = fake_update
state = hass.states.get(entity_id)
assert state
assert state.state != "unavailable"
await mock_config_entry.runtime_data.coordinator_info.async_refresh()
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "unavailable"
async def test_line_broadcast_line_present(
hass: HomeAssistant,
mock_lunatone_info: AsyncMock,
mock_lunatone_devices: AsyncMock,
mock_lunatone_dali_broadcast: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test if the broadcast light line is present."""
mock_lunatone_dali_broadcast.line = None
await setup_integration(hass, mock_config_entry)
assert not hass.states.async_entity_ids("light")