github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-08-02 23:40:32 +02:00
Code Releases Activity

Code cleanups for lookin (#64106)

Browse Source
This commit is contained in:
J. Nick Koston 2022-01-14 14:22:06 -10:00 committed by GitHub
parent b949199866
commit 06329a2f43
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 149 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
homeassistant/components/lookin/__init__.py
View File

@ -2,31 +2,59 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine
from datetime import timedelta
import logging
from typing import Any
import aiohttp
from aiolookin import (
Climate,
LookInHttpProtocol,
LookinUDPSubscriptions,
MeteoSensor,
Remote,
start_lookin_udp,
)
from aiolookin.models import UDPCommandType, UDPEvent
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.const import CONF_HOST, Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, PLATFORMS
from .const import DOMAIN, PLATFORMS, TYPE_TO_PLATFORM
from .models import LookinData
LOGGER = logging.getLogger(__name__)
def _async_climate_updater(
lookin_protocol: LookInHttpProtocol,
uuid: str,
) -> Callable[[], Coroutine[None, Any, Remote]]:
"""Create a function to capture the cell variable."""
async def _async_update() -> Climate:
return await lookin_protocol.get_conditioner(uuid)
return _async_update
def _async_remote_updater(
lookin_protocol: LookInHttpProtocol,
uuid: str,
) -> Callable[[], Coroutine[None, Any, Remote]]:
"""Create a function to capture the cell variable."""
async def _async_update() -> Remote:
return await lookin_protocol.get_remote(uuid)
return _async_update
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up lookin from a config entry."""
@ -52,6 +80,27 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
await meteo_coordinator.async_config_entry_first_refresh()
device_coordinators: dict[str, DataUpdateCoordinator] = {}
for remote in devices:
if (platform := TYPE_TO_PLATFORM.get(remote["Type"])) is None:
continue
uuid = remote["UUID"]
if platform == Platform.CLIMATE:
updater = _async_climate_updater(lookin_protocol, uuid)
else:
updater = _async_remote_updater(lookin_protocol, uuid)
coordinator = DataUpdateCoordinator(
hass,
LOGGER,
name=f"{entry.title} {uuid}",
update_method=updater,
update_interval=timedelta(
seconds=60
), # Updates are pushed (fallback is polling)
)
await coordinator.async_config_entry_first_refresh()
device_coordinators[uuid] = coordinator
@callback
def _async_meteo_push_update(event: UDPEvent) -> None:
"""Process an update pushed via UDP."""
@ -66,6 +115,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
lookin_device.id, UDPCommandType.meteo, None, _async_meteo_push_update
)
)
entry.async_on_unload(await start_lookin_udp(lookin_udp_subs, lookin_device.id))
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = LookinData(
@ -74,6 +124,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
meteo_coordinator=meteo_coordinator,
devices=devices,
lookin_protocol=lookin_protocol,
device_coordinators=device_coordinators,
)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)

35
homeassistant/components/lookin/climate.py
View File

@ -1,8 +1,6 @@
"""The lookin integration climate platform."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from datetime import timedelta
import logging
from typing import Any, Final, cast
@ -29,12 +27,17 @@ from homeassistant.components.climate.const import (
SWING_OFF,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_TEMPERATURE, PRECISION_WHOLE, TEMP_CELSIUS
from homeassistant.const import (
ATTR_TEMPERATURE,
PRECISION_WHOLE,
TEMP_CELSIUS,
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
from .const import DOMAIN, TYPE_TO_PLATFORM
from .entity import LookinCoordinatorEntity
from .models import LookinData
@ -77,30 +80,10 @@ async def async_setup_entry(
entities = []
for remote in lookin_data.devices:
if remote["Type"] != "EF":
if TYPE_TO_PLATFORM.get(remote["Type"]) != Platform.CLIMATE:
continue
uuid = remote["UUID"]
def _wrap_async_update(
uuid: str,
) -> Callable[[], Coroutine[None, Any, Climate]]:
"""Create a function to capture the uuid cell variable."""
async def _async_update() -> Climate:
return await lookin_data.lookin_protocol.get_conditioner(uuid)
return _async_update
coordinator = DataUpdateCoordinator(
hass,
LOGGER,
name=f"{config_entry.title} {uuid}",
update_method=_wrap_async_update(uuid),
update_interval=timedelta(
seconds=60
), # Updates are pushed (fallback is polling)
)
await coordinator.async_refresh()
coordinator = lookin_data.device_coordinators[uuid]
device: Climate = coordinator.data
entities.append(
ConditionerEntity(

8
homeassistant/components/lookin/const.py
View File

@ -14,3 +14,11 @@ PLATFORMS: Final = [
Platform.MEDIA_PLAYER,
Platform.SENSOR,
]
TYPE_TO_PLATFORM = {
"01": Platform.MEDIA_PLAYER,
"02": Platform.MEDIA_PLAYER,
"03": Platform.LIGHT,
"EF": Platform.CLIMATE,
}

63
homeassistant/components/lookin/entity.py
View File

@ -1,8 +1,12 @@
"""The lookin integration entity."""
from __future__ import annotations
from abc import abstractmethod
import logging
from typing import cast
from aiolookin import POWER_CMD, POWER_OFF_CMD, POWER_ON_CMD, Climate, Remote
from aiolookin.models import Device
from aiolookin.models import Device, UDPCommandType, UDPEvent
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import (
@ -13,6 +17,8 @@ from homeassistant.helpers.update_coordinator import (
from .const import DOMAIN, MODEL_NAMES
from .models import LookinData
LOGGER = logging.getLogger(__name__)
def _lookin_device_to_device_info(lookin_device: Device) -> DeviceInfo:
"""Convert a lookin device into DeviceInfo."""
@ -124,3 +130,58 @@ class LookinPowerEntity(LookinCoordinatorEntity):
self._power_on_command = POWER_ON_CMD
if POWER_OFF_CMD in self._function_names:
self._power_off_command = POWER_OFF_CMD
class LookinPowerPushRemoteEntity(LookinPowerEntity):
"""A Lookin entity that has a power on and power off command with push updates."""
def __init__(
self,
coordinator: DataUpdateCoordinator,
uuid: str,
device: Remote,
lookin_data: LookinData,
) -> None:
"""Init the entity."""
super().__init__(coordinator, uuid, device, lookin_data)
self._update_from_status(self._remote.status)
self._attr_name = self._remote.name
@property
def _remote(self) -> Remote:
return cast(Remote, self.coordinator.data)
@abstractmethod
def _update_from_status(self, status: str) -> None:
"""Update properties from status."""
def _async_push_update(self, event: UDPEvent) -> None:
"""Process an update pushed via UDP."""
LOGGER.debug("Processing push message for %s: %s", self.entity_id, event)
self._update_from_status(event.value)
self.coordinator.async_set_updated_data(self._remote)
async def _async_push_update_device(self, event: UDPEvent) -> None:
"""Process an update pushed via UDP."""
LOGGER.debug("Processing push message for %s: %s", self.entity_id, event)
await self.coordinator.async_refresh()
self._attr_name = self._remote.name
async def async_added_to_hass(self) -> None:
"""Call when the entity is added to hass."""
self.async_on_remove(
self._lookin_udp_subs.subscribe_event(
self._lookin_device.id,
UDPCommandType.ir,
self._uuid,
self._async_push_update,
)
)
self.async_on_remove(
self._lookin_udp_subs.subscribe_event(
self._lookin_device.id,
UDPCommandType.data,
self._uuid,
self._async_push_update_device,
)
)

87
homeassistant/components/lookin/light.py
View File

@ -1,22 +1,19 @@
"""The lookin integration light platform."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from datetime import timedelta
import logging
from typing import Any, cast
from typing import Any
from aiolookin import Remote
from aiolookin.models import UDPCommandType, UDPEvent
from homeassistant.components.light import COLOR_MODE_ONOFF, LightEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
from .entity import LookinPowerEntity
from .const import DOMAIN, TYPE_TO_PLATFORM
from .entity import LookinPowerPushRemoteEntity
from .models import LookinData
LOGGER = logging.getLogger(__name__)
@ -32,65 +29,29 @@ async def async_setup_entry(
entities = []
for remote in lookin_data.devices:
if remote["Type"] != "03":
if TYPE_TO_PLATFORM.get(remote["Type"]) != Platform.LIGHT:
continue
uuid = remote["UUID"]
def _wrap_async_update(
uuid: str,
) -> Callable[[], Coroutine[None, Any, Remote]]:
"""Create a function to capture the uuid cell variable."""
async def _async_update() -> Remote:
return await lookin_data.lookin_protocol.get_remote(uuid)
return _async_update
coordinator = DataUpdateCoordinator(
hass,
LOGGER,
name=f"{config_entry.title} {uuid}",
update_method=_wrap_async_update(uuid),
update_interval=timedelta(
seconds=60
), # Updates are pushed (fallback is polling)
)
await coordinator.async_refresh()
coordinator = lookin_data.device_coordinators[uuid]
device: Remote = coordinator.data
entities.append(
LookinLightEntity(
coordinator=coordinator,
uuid=uuid,
device=device,
lookin_data=lookin_data,
coordinator=coordinator,
)
)
async_add_entities(entities)
class LookinLightEntity(LookinPowerEntity, LightEntity):
class LookinLightEntity(LookinPowerPushRemoteEntity, LightEntity):
"""A lookin IR controlled light."""
_attr_supported_color_modes = {COLOR_MODE_ONOFF}
_attr_color_mode = COLOR_MODE_ONOFF
def __init__(
self,
uuid: str,
device: Remote,
lookin_data: LookinData,
coordinator: DataUpdateCoordinator,
) -> None:
"""Init the light."""
super().__init__(coordinator, uuid, device, lookin_data)
self._attr_is_on = False
@property
def _remote(self) -> Remote:
return cast(Remote, self.coordinator.data)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the light."""
await self._async_send_command(self._power_on_command)
@ -114,35 +75,3 @@ class LookinLightEntity(LookinPowerEntity, LightEntity):
state = status[0]
self._attr_is_on = state == "1"
def _async_push_update(self, event: UDPEvent) -> None:
"""Process an update pushed via UDP."""
LOGGER.debug("Processing push message for %s: %s", self.entity_id, event)
self._update_from_status(event.value)
self.coordinator.async_set_updated_data(self._remote)
self.async_write_ha_state()
async def _async_push_update_device(self, event: UDPEvent) -> None:
"""Process an update pushed via UDP."""
LOGGER.debug("Processing push message for %s: %s", self.entity_id, event)
await self.coordinator.async_refresh()
self._attr_name = self._remote.name
async def async_added_to_hass(self) -> None:
"""Call when the entity is added to hass."""
self.async_on_remove(
self._lookin_udp_subs.subscribe_event(
self._lookin_device.id,
UDPCommandType.ir,
self._uuid,
self._async_push_update,
)
)
self.async_on_remove(
self._lookin_udp_subs.subscribe_event(
self._lookin_device.id,
UDPCommandType.data,
self._uuid,
self._async_push_update_device,
)
)

85
homeassistant/components/lookin/media_player.py
View File

@ -1,13 +1,9 @@
"""The lookin integration light platform."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from datetime import timedelta
import logging
from typing import Any, cast
from aiolookin import Remote
from aiolookin.models import UDPCommandType, UDPEvent
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
@ -22,13 +18,13 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_STEP,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_ON, STATE_STANDBY
from homeassistant.const import STATE_ON, STATE_STANDBY, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
from .entity import LookinPowerEntity
from .const import DOMAIN, TYPE_TO_PLATFORM
from .entity import LookinPowerPushRemoteEntity
from .models import LookinData
LOGGER = logging.getLogger(__name__)
@ -59,73 +55,44 @@ async def async_setup_entry(
entities = []
for remote in lookin_data.devices:
if remote["Type"] not in _TYPE_TO_DEVICE_CLASS:
if TYPE_TO_PLATFORM.get(remote["Type"]) != Platform.MEDIA_PLAYER:
continue
uuid = remote["UUID"]
def _wrap_async_update(
uuid: str,
) -> Callable[[], Coroutine[None, Any, Remote]]:
"""Create a function to capture the uuid cell variable."""
async def _async_update() -> Remote:
return await lookin_data.lookin_protocol.get_remote(uuid)
return _async_update
coordinator = DataUpdateCoordinator(
hass,
LOGGER,
name=f"{config_entry.title} {uuid}",
update_method=_wrap_async_update(uuid),
update_interval=timedelta(
seconds=60
), # Updates are pushed (fallback is polling)
)
await coordinator.async_refresh()
coordinator = lookin_data.device_coordinators[uuid]
device: Remote = coordinator.data
entities.append(
LookinMedia(
coordinator=coordinator,
uuid=uuid,
device=device,
lookin_data=lookin_data,
device_class=_TYPE_TO_DEVICE_CLASS[remote["Type"]],
coordinator=coordinator,
)
)
async_add_entities(entities)
class LookinMedia(LookinPowerEntity, MediaPlayerEntity):
class LookinMedia(LookinPowerPushRemoteEntity, MediaPlayerEntity):
"""A lookin media player."""
_attr_should_poll = False
def __init__(
self,
coordinator: DataUpdateCoordinator,
uuid: str,
device: Remote,
lookin_data: LookinData,
device_class: str,
coordinator: DataUpdateCoordinator,
) -> None:
"""Init the lookin media player."""
self._attr_device_class = device_class
self._attr_supported_features: int = 0
self._attr_state = None
self._attr_is_volume_muted: bool = False
super().__init__(coordinator, uuid, device, lookin_data)
for function_name, feature in _FUNCTION_NAME_TO_FEATURE.items():
if function_name in self._function_names:
self._attr_supported_features |= feature
self._attr_name = self._remote.name
self._async_update_from_data()
@property
def _remote(self) -> Remote:
return cast(Remote, self.coordinator.data)
async def async_volume_up(self) -> None:
"""Turn volume up for media player."""
@ -177,39 +144,3 @@ class LookinMedia(LookinPowerEntity, MediaPlayerEntity):
self._attr_state = STATE_ON if state == "1" else STATE_STANDBY
self._attr_is_volume_muted = mute == "0"
def _async_push_update(self, event: UDPEvent) -> None:
"""Process an update pushed via UDP."""
LOGGER.debug("Processing push message for %s: %s", self.entity_id, event)
self._update_from_status(event.value)
self.coordinator.async_set_updated_data(self._remote)
self.async_write_ha_state()
async def _async_push_update_device(self, event: UDPEvent) -> None:
"""Process an update pushed via UDP."""
LOGGER.debug("Processing push message for %s: %s", self.entity_id, event)
await self.coordinator.async_refresh()
self._attr_name = self._remote.name
async def async_added_to_hass(self) -> None:
"""Call when the entity is added to hass."""
self.async_on_remove(
self._lookin_udp_subs.subscribe_event(
self._lookin_device.id,
UDPCommandType.ir,
self._uuid,
self._async_push_update,
)
)
self.async_on_remove(
self._lookin_udp_subs.subscribe_event(
self._lookin_device.id,
UDPCommandType.data,
self._uuid,
self._async_push_update_device,
)
)
def _async_update_from_data(self) -> None:
"""Update attrs from data."""
self._update_from_status(self._remote.status)

1
homeassistant/components/lookin/models.py
View File

@ -18,3 +18,4 @@ class LookinData:
meteo_coordinator: DataUpdateCoordinator
devices: list[dict[str, Any]]
lookin_protocol: LookInHttpProtocol
device_coordinators: dict[str, DataUpdateCoordinator]