evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add data update coordinator to enigma2 (#122046)

Browse source 
* Add data update coordinator to enigma2

* Apply suggestions from code review

Co-authored-by: Robert Resch <robert@resch.dev>

---------

Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
Sid 2024-07-19 17:51:34 +02:00 • committed by GitHub
parent 0cde518a89
commit e029dad0eb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 176 additions and 99 deletions
Download patch file Download diff file Expand all files Collapse all files

34
homeassistant/components/enigma2/__init__.py
View file

@ -1,45 +1,23 @@
"""Support for Enigma2 devices."""
from openwebif.api import OpenWebIfDevice
from yarl import URL
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
CONF_VERIFY_SSL,
Platform,
)
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import CONF_SOURCE_BOUQUET
from .coordinator import Enigma2UpdateCoordinator
type Enigma2ConfigEntry = ConfigEntry[OpenWebIfDevice]
type Enigma2ConfigEntry = ConfigEntry[Enigma2UpdateCoordinator]
PLATFORMS = [Platform.MEDIA_PLAYER]
async def async_setup_entry(hass: HomeAssistant, entry: Enigma2ConfigEntry) -> bool:
"""Set up Enigma2 from a config entry."""
base_url = URL.build(
scheme="http" if not entry.data[CONF_SSL] else "https",
host=entry.data[CONF_HOST],
port=entry.data[CONF_PORT],
user=entry.data.get(CONF_USERNAME),
password=entry.data.get(CONF_PASSWORD),
)
session = async_create_clientsession(
hass, verify_ssl=entry.data[CONF_VERIFY_SSL], base_url=base_url
)
coordinator = Enigma2UpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
entry.runtime_data = OpenWebIfDevice(
session, source_bouquet=entry.options.get(CONF_SOURCE_BOUQUET)
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True

5
homeassistant/components/enigma2/config_flow.py
View file

@ -68,8 +68,9 @@ CONFIG_SCHEMA = vol.Schema(
async def get_options_schema(handler: SchemaCommonFlowHandler) -> vol.Schema:
"""Get the options schema."""
entry = cast(SchemaOptionsFlowHandler, handler.parent_handler).config_entry
device: OpenWebIfDevice = entry.runtime_data
bouquets = [b[1] for b in (await device.get_all_bouquets())["bouquets"]]
bouquets = [
b[1] for b in (await entry.runtime_data.device.get_all_bouquets())["bouquets"]
]
return vol.Schema(
{

84
homeassistant/components/enigma2/coordinator.py Normal file
View file

@ -0,0 +1,84 @@
"""Data update coordinator for the Enigma2 integration."""
import logging
from openwebif.api import OpenWebIfDevice, OpenWebIfStatus
from yarl import URL
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_CONNECTIONS,
ATTR_IDENTIFIERS,
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_SSL,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant.helpers.device_registry import (
CONNECTION_NETWORK_MAC,
DeviceInfo,
format_mac,
)
from homeassistant.helpers.entity_component import DEFAULT_SCAN_INTERVAL
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import CONF_SOURCE_BOUQUET, DOMAIN
LOGGER = logging.getLogger(__package__)
class Enigma2UpdateCoordinator(DataUpdateCoordinator[OpenWebIfStatus]):
"""The Enigma2 data update coordinator."""
device: OpenWebIfDevice
def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Initialize the Enigma2 data update coordinator."""
super().__init__(
hass, logger=LOGGER, name=DOMAIN, update_interval=DEFAULT_SCAN_INTERVAL
)
base_url = URL.build(
scheme="http" if not config_entry.data[CONF_SSL] else "https",
host=config_entry.data[CONF_HOST],
port=config_entry.data[CONF_PORT],
user=config_entry.data.get(CONF_USERNAME),
password=config_entry.data.get(CONF_PASSWORD),
)
session = async_create_clientsession(
hass, verify_ssl=config_entry.data[CONF_VERIFY_SSL], base_url=base_url
)
self.device = OpenWebIfDevice(
session, source_bouquet=config_entry.options.get(CONF_SOURCE_BOUQUET)
)
self.device_info = DeviceInfo(
configuration_url=base_url,
name=config_entry.data[CONF_HOST],
)
async def _async_setup(self) -> None:
"""Provide needed data to the device info."""
about = await self.device.get_about()
self.device.mac_address = about["info"]["ifaces"][0]["mac"]
self.device_info["model"] = about["info"]["model"]
self.device_info["manufacturer"] = about["info"]["brand"]
self.device_info[ATTR_IDENTIFIERS] = {
(DOMAIN, format_mac(iface["mac"])) for iface in about["info"]["ifaces"]
}
self.device_info[ATTR_CONNECTIONS] = {
(CONNECTION_NETWORK_MAC, format_mac(iface["mac"]))
for iface in about["info"]["ifaces"]
}
async def _async_update_data(self) -> OpenWebIfStatus:
await self.device.update()
return self.device.status

137
homeassistant/components/enigma2/media_player.py
View file

@ -4,9 +4,9 @@ from __future__ import annotations
import contextlib
from logging import getLogger
from typing import cast
from aiohttp.client_exceptions import ClientConnectorError, ServerDisconnectedError
from openwebif.api import OpenWebIfDevice
from aiohttp.client_exceptions import ServerDisconnectedError
from openwebif.enums import PowerState, RemoteControlCodes, SetVolumeOption
import voluptuous as vol
@ -26,11 +26,11 @@ from homeassistant.const import (
CONF_SSL,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import Enigma2ConfigEntry
from .const import (
@ -49,6 +49,7 @@ from .const import (
DEFAULT_USERNAME,
DOMAIN,
)
from .coordinator import Enigma2UpdateCoordinator
ATTR_MEDIA_CURRENTLY_RECORDING = "media_currently_recording"
ATTR_MEDIA_DESCRIPTION = "media_description"
@ -107,15 +108,10 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Enigma2 media player platform."""
device = entry.runtime_data
about = await device.get_about()
device.mac_address = about["info"]["ifaces"][0]["mac"]
entity = Enigma2Device(entry, device, about)
async_add_entities([entity])
async_add_entities([Enigma2Device(entry.runtime_data)])
class Enigma2Device(MediaPlayerEntity):
class Enigma2Device(CoordinatorEntity[Enigma2UpdateCoordinator], MediaPlayerEntity):
"""Representation of an Enigma2 box."""
_attr_has_entity_name = True
@ -135,118 +131,125 @@ class Enigma2Device(MediaPlayerEntity):
| MediaPlayerEntityFeature.SELECT_SOURCE
)
def __init__(
self, entry: ConfigEntry, device: OpenWebIfDevice, about: dict
) -> None:
def __init__(self, coordinator: Enigma2UpdateCoordinator) -> None:
"""Initialize the Enigma2 device."""
self._device: OpenWebIfDevice = device
self._entry = entry
self._attr_unique_id = device.mac_address or entry.entry_id
super().__init__(coordinator)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._attr_unique_id)},
manufacturer=about["info"]["brand"],
model=about["info"]["model"],
configuration_url=device.base,
name=entry.data[CONF_HOST],
self._attr_unique_id = (
coordinator.device.mac_address
or cast(ConfigEntry, coordinator.config_entry).entry_id
)
self._attr_device_info = coordinator.device_info
async def async_turn_off(self) -> None:
"""Turn off media player."""
if self._device.turn_off_to_deep:
if self.coordinator.device.turn_off_to_deep:
with contextlib.suppress(ServerDisconnectedError):
await self._device.set_powerstate(PowerState.DEEP_STANDBY)
await self.coordinator.device.set_powerstate(PowerState.DEEP_STANDBY)
self._attr_available = False
else:
await self._device.set_powerstate(PowerState.STANDBY)
await self.coordinator.device.set_powerstate(PowerState.STANDBY)
await self.coordinator.async_refresh()
async def async_turn_on(self) -> None:
"""Turn the media player on."""
await self._device.turn_on()
await self.coordinator.device.turn_on()
await self.coordinator.async_refresh()
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
await self._device.set_volume(int(volume * 100))
await self.coordinator.device.set_volume(int(volume * 100))
await self.coordinator.async_refresh()
async def async_volume_up(self) -> None:
"""Volume up the media player."""
await self._device.set_volume(SetVolumeOption.UP)
await self.coordinator.device.set_volume(SetVolumeOption.UP)
await self.coordinator.async_refresh()
async def async_volume_down(self) -> None:
"""Volume down media player."""
await self._device.set_volume(SetVolumeOption.DOWN)
await self.coordinator.device.set_volume(SetVolumeOption.DOWN)
await self.coordinator.async_refresh()
async def async_media_stop(self) -> None:
"""Send stop command."""
await self._device.send_remote_control_action(RemoteControlCodes.STOP)
await self.coordinator.device.send_remote_control_action(
RemoteControlCodes.STOP
)
await self.coordinator.async_refresh()
async def async_media_play(self) -> None:
"""Play media."""
await self._device.send_remote_control_action(RemoteControlCodes.PLAY)
await self.coordinator.device.send_remote_control_action(
RemoteControlCodes.PLAY
)
await self.coordinator.async_refresh()
async def async_media_pause(self) -> None:
"""Pause the media player."""
await self._device.send_remote_control_action(RemoteControlCodes.PAUSE)
await self.coordinator.device.send_remote_control_action(
RemoteControlCodes.PAUSE
)
await self.coordinator.async_refresh()
async def async_media_next_track(self) -> None:
"""Send next track command."""
await self._device.send_remote_control_action(RemoteControlCodes.CHANNEL_UP)
await self.coordinator.device.send_remote_control_action(
RemoteControlCodes.CHANNEL_UP
)
await self.coordinator.async_refresh()
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
await self._device.send_remote_control_action(RemoteControlCodes.CHANNEL_DOWN)
await self.coordinator.device.send_remote_control_action(
RemoteControlCodes.CHANNEL_DOWN
)
await self.coordinator.async_refresh()
async def async_mute_volume(self, mute: bool) -> None:
"""Mute or unmute."""
if mute != self._device.status.muted:
await self._device.toggle_mute()
if mute != self.coordinator.data.muted:
await self.coordinator.device.toggle_mute()
await self.coordinator.async_refresh()
async def async_select_source(self, source: str) -> None:
"""Select input source."""
await self._device.zap(self._device.sources[source])
await self.coordinator.device.zap(self.coordinator.device.sources[source])
await self.coordinator.async_refresh()
async def async_update(self) -> None:
@callback
def _handle_coordinator_update(self) -> None:
"""Update state of the media_player."""
try:
await self._device.update()
except ClientConnectorError as err:
if self._attr_available:
_LOGGER.warning(
"%s is unavailable. Error: %s", self._device.base.host, err
)
self._attr_available = False
return
if not self._attr_available:
_LOGGER.debug("%s is available", self._device.base.host)
self._attr_available = True
if not self._device.status.in_standby:
if not self.coordinator.data.in_standby:
self._attr_extra_state_attributes = {
ATTR_MEDIA_CURRENTLY_RECORDING: self._device.status.is_recording,
ATTR_MEDIA_DESCRIPTION: self._device.status.currservice.fulldescription,
ATTR_MEDIA_START_TIME: self._device.status.currservice.begin,
ATTR_MEDIA_END_TIME: self._device.status.currservice.end,
ATTR_MEDIA_CURRENTLY_RECORDING: self.coordinator.data.is_recording,
ATTR_MEDIA_DESCRIPTION: self.coordinator.data.currservice.fulldescription,
ATTR_MEDIA_START_TIME: self.coordinator.data.currservice.begin,
ATTR_MEDIA_END_TIME: self.coordinator.data.currservice.end,
}
else:
self._attr_extra_state_attributes = {}
self._attr_media_title = self._device.status.currservice.station
self._attr_media_series_title = self._device.status.currservice.name
self._attr_media_channel = self._device.status.currservice.station
self._attr_is_volume_muted = self._device.status.muted
self._attr_media_content_id = self._device.status.currservice.serviceref
self._attr_media_image_url = self._device.picon_url
self._attr_source = self._device.status.currservice.station
self._attr_source_list = self._device.source_list
self._attr_media_title = self.coordinator.data.currservice.station
self._attr_media_series_title = self.coordinator.data.currservice.name
self._attr_media_channel = self.coordinator.data.currservice.station
self._attr_is_volume_muted = self.coordinator.data.muted
self._attr_media_content_id = self.coordinator.data.currservice.serviceref
self._attr_media_image_url = self.coordinator.device.picon_url
self._attr_source = self.coordinator.data.currservice.station
self._attr_source_list = self.coordinator.device.source_list
if self._device.status.in_standby:
if self.coordinator.data.in_standby:
self._attr_state = MediaPlayerState.OFF
else:
self._attr_state = MediaPlayerState.ON
if (volume_level := self._device.status.volume) is not None:
if (volume_level := self.coordinator.data.volume) is not None:
self._attr_volume_level = volume_level / 100
else:
self._attr_volume_level = None
self.async_write_ha_state()

13
tests/components/enigma2/conftest.py
View file

@ -1,5 +1,7 @@
"""Test the Enigma2 config flow."""
from openwebif.api import OpenWebIfServiceEvent, OpenWebIfStatus
from homeassistant.components.enigma2.const import (
CONF_DEEP_STANDBY,
CONF_MAC_ADDRESS,
@ -66,6 +68,10 @@ class MockDevice:
mac_address: str | None = "12:34:56:78:90:ab"
_base = "http://1.1.1.1"
def __init__(self) -> None:
"""Initialize the mock Enigma2 device."""
self.status = OpenWebIfStatus(currservice=OpenWebIfServiceEvent())
async def _call_api(self, url: str) -> dict:
if url.endswith("/api/about"):
return {
@ -74,7 +80,9 @@ class MockDevice:
{
"mac": self.mac_address,
}
]
],
"model": "Mock Enigma2",
"brand": "Enigma2",
}
}
@ -97,5 +105,8 @@ class MockDevice:
]
}
async def update(self) -> None:
"""Mock update."""
async def close(self):
"""Mock close."""

2
tests/components/enigma2/test_init.py
View file

@ -15,7 +15,7 @@ async def test_unload_entry(hass: HomeAssistant) -> None:
"""Test successful unload of entry."""
with (
patch(
"homeassistant.components.enigma2.OpenWebIfDevice.__new__",
"homeassistant.components.enigma2.coordinator.OpenWebIfDevice.__new__",
return_value=MockDevice(),
),
patch(