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

Add Room Audio Control to Control4 Integration (#87821)

Browse source 
* Add control4 room based media player (#13)

* update attribute names (#14)

* change to data class and set off to idle (#15)

Co-authored-by: nalin29 <nalin29@github.com>

---------

Co-authored-by: nalin29 <nalin29@github.com>
This commit is contained in:
Nalin Mahajan 2024-03-19 05:44:52 -05:00 • committed by GitHub
parent 4381780936
commit 18ef76a018
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 408 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.coveragerc
View file

@ -202,6 +202,7 @@ omit =
homeassistant/components/control4/__init__.py
homeassistant/components/control4/director_utils.py
homeassistant/components/control4/light.py
homeassistant/components/control4/media_player.py
homeassistant/components/coolmaster/coordinator.py
homeassistant/components/cppm_tracker/device_tracker.py
homeassistant/components/crownstone/__init__.py

12
homeassistant/components/control4/__init__.py
View file

@ -4,6 +4,7 @@ from __future__ import annotations
import json
import logging
from typing import Any
from aiohttp import client_exceptions
from pyControl4.account import C4Account
@ -36,13 +37,14 @@ from .const import (
CONF_DIRECTOR_ALL_ITEMS,
CONF_DIRECTOR_MODEL,
CONF_DIRECTOR_SW_VERSION,
CONF_UI_CONFIGURATION,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
)
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.LIGHT]
PLATFORMS = [Platform.LIGHT, Platform.MEDIA_PLAYER]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@ -105,6 +107,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
director_all_items = json.loads(director_all_items)
entry_data[CONF_DIRECTOR_ALL_ITEMS] = director_all_items
entry_data[CONF_UI_CONFIGURATION] = json.loads(await director.getUiConfiguration())
# Load options from config entry
entry_data[CONF_SCAN_INTERVAL] = entry.options.get(
CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL
@ -145,14 +149,14 @@ async def get_items_of_category(hass: HomeAssistant, entry: ConfigEntry, categor
]
class Control4Entity(CoordinatorEntity):
class Control4Entity(CoordinatorEntity[Any]):
"""Base entity for Control4."""
def __init__(
self,
entry_data: dict,
coordinator: DataUpdateCoordinator,
name: str,
coordinator: DataUpdateCoordinator[Any],
name: str | None,
idx: int,
device_name: str | None,
device_manufacturer: str | None,

1
homeassistant/components/control4/const.py
View file

@ -10,6 +10,7 @@ CONF_DIRECTOR = "director"
CONF_DIRECTOR_SW_VERSION = "director_sw_version"
CONF_DIRECTOR_MODEL = "director_model"
CONF_DIRECTOR_ALL_ITEMS = "director_all_items"
CONF_UI_CONFIGURATION = "ui_configuration"
CONF_CONTROLLER_UNIQUE_ID = "controller_unique_id"
CONF_CONFIG_LISTENER = "config_listener"

12
homeassistant/components/control4/light.py
View file

@ -45,7 +45,7 @@ async def async_setup_entry(
scan_interval,
)
async def async_update_data_non_dimmer():
async def async_update_data_non_dimmer() -> dict[int, dict[str, Any]]:
"""Fetch data from Control4 director for non-dimmer lights."""
try:
return await update_variables_for_config_entry(
@ -54,7 +54,7 @@ async def async_setup_entry(
except C4Exception as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
async def async_update_data_dimmer():
async def async_update_data_dimmer() -> dict[int, dict[str, Any]]:
"""Fetch data from Control4 director for dimmer lights."""
try:
return await update_variables_for_config_entry(
@ -63,14 +63,14 @@ async def async_setup_entry(
except C4Exception as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
non_dimmer_coordinator = DataUpdateCoordinator(
non_dimmer_coordinator = DataUpdateCoordinator[dict[int, dict[str, Any]]](
hass,
_LOGGER,
name="light",
update_method=async_update_data_non_dimmer,
update_interval=timedelta(seconds=scan_interval),
)
dimmer_coordinator = DataUpdateCoordinator(
dimmer_coordinator = DataUpdateCoordinator[dict[int, dict[str, Any]]](
hass,
_LOGGER,
name="light",
@ -149,10 +149,12 @@ async def async_setup_entry(
class Control4Light(Control4Entity, LightEntity):
"""Control4 light entity."""
_attr_has_entity_name = True
def __init__(
self,
entry_data: dict,
coordinator: DataUpdateCoordinator,
coordinator: DataUpdateCoordinator[dict[int, dict[str, Any]]],
name: str,
idx: int,
device_name: str | None,

391
homeassistant/components/control4/media_player.py Normal file
View file

@ -0,0 +1,391 @@
"""Platform for Control4 Rooms Media Players."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
import enum
import logging
from typing import Any
from pyControl4.error_handling import C4Exception
from pyControl4.room import C4Room
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_SCAN_INTERVAL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from . import Control4Entity
from .const import CONF_DIRECTOR, CONF_DIRECTOR_ALL_ITEMS, CONF_UI_CONFIGURATION, DOMAIN
from .director_utils import update_variables_for_config_entry
_LOGGER = logging.getLogger(__name__)
CONTROL4_POWER_STATE = "POWER_STATE"
CONTROL4_VOLUME_STATE = "CURRENT_VOLUME"
CONTROL4_MUTED_STATE = "IS_MUTED"
CONTROL4_CURRENT_VIDEO_DEVICE = "CURRENT_VIDEO_DEVICE"
CONTROL4_PLAYING = "PLAYING"
CONTROL4_PAUSED = "PAUSED"
CONTROL4_STOPPED = "STOPPED"
CONTROL4_MEDIA_INFO = "CURRENT MEDIA INFO"
CONTROL4_PARENT_ID = "parentId"
VARIABLES_OF_INTEREST = {
CONTROL4_POWER_STATE,
CONTROL4_VOLUME_STATE,
CONTROL4_MUTED_STATE,
CONTROL4_CURRENT_VIDEO_DEVICE,
CONTROL4_MEDIA_INFO,
CONTROL4_PLAYING,
CONTROL4_PAUSED,
CONTROL4_STOPPED,
}
class _SourceType(enum.Enum):
AUDIO = 1
VIDEO = 2
@dataclass
class _RoomSource:
"""Class for Room Source."""
source_type: set[_SourceType]
idx: int
name: str
async def get_rooms(hass: HomeAssistant, entry: ConfigEntry):
"""Return a list of all Control4 rooms."""
director_all_items = hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR_ALL_ITEMS]
return [
item
for item in director_all_items
if "typeName" in item and item["typeName"] == "room"
]
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Control4 rooms from a config entry."""
all_rooms = await get_rooms(hass, entry)
if not all_rooms:
return
entry_data = hass.data[DOMAIN][entry.entry_id]
scan_interval = entry_data[CONF_SCAN_INTERVAL]
_LOGGER.debug("Scan interval = %s", scan_interval)
async def async_update_data() -> dict[int, dict[str, Any]]:
"""Fetch data from Control4 director."""
try:
return await update_variables_for_config_entry(
hass, entry, VARIABLES_OF_INTEREST
)
except C4Exception as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
coordinator = DataUpdateCoordinator[dict[int, dict[str, Any]]](
hass,
_LOGGER,
name="room",
update_method=async_update_data,
update_interval=timedelta(seconds=scan_interval),
)
# Fetch initial data so we have data when entities subscribe
await coordinator.async_refresh()
items_by_id = {
item["id"]: item
for item in hass.data[DOMAIN][entry.entry_id][CONF_DIRECTOR_ALL_ITEMS]
}
item_to_parent_map = {
k: item["parentId"]
for k, item in items_by_id.items()
if "parentId" in item and k > 1
}
ui_config = entry_data[CONF_UI_CONFIGURATION]
entity_list = []
for room in all_rooms:
room_id = room["id"]
sources: dict[int, _RoomSource] = {}
for exp in ui_config["experiences"]:
if room_id == exp["room_id"]:
exp_type = exp["type"]
if exp_type not in ("listen", "watch"):
continue
dev_type = (
_SourceType.AUDIO if exp_type == "listen" else _SourceType.VIDEO
)
for source in exp["sources"]["source"]:
dev_id = source["id"]
name = items_by_id.get(dev_id, {}).get(
"name", f"Unknown Device - {dev_id}"
)
if dev_id in sources:
sources[dev_id].source_type.add(dev_type)
else:
sources[dev_id] = _RoomSource(
source_type={dev_type}, idx=dev_id, name=name
)
try:
hidden = room["roomHidden"]
entity_list.append(
Control4Room(
entry_data,
coordinator,
room["name"],
room_id,
item_to_parent_map,
sources,
hidden,
)
)
except KeyError:
_LOGGER.exception(
"Unknown device properties received from Control4: %s",
room,
)
continue
async_add_entities(entity_list, True)
class Control4Room(Control4Entity, MediaPlayerEntity):
"""Control4 Room entity."""
_attr_has_entity_name = True
def __init__(
self,
entry_data: dict,
coordinator: DataUpdateCoordinator[dict[int, dict[str, Any]]],
name: str,
room_id: int,
id_to_parent: dict[int, int],
sources: dict[int, _RoomSource],
room_hidden: bool,
) -> None:
"""Initialize Control4 room entity."""
super().__init__(
entry_data,
coordinator,
None,
room_id,
device_name=name,
device_manufacturer=None,
device_model=None,
device_id=room_id,
)
self._attr_entity_registry_enabled_default = not room_hidden
self._id_to_parent = id_to_parent
self._sources = sources
self._attr_supported_features = (
MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.STOP
| MediaPlayerEntityFeature.VOLUME_MUTE
| MediaPlayerEntityFeature.VOLUME_SET
| MediaPlayerEntityFeature.VOLUME_STEP
| MediaPlayerEntityFeature.TURN_OFF
| MediaPlayerEntityFeature.SELECT_SOURCE
)
def _create_api_object(self):
"""Create a pyControl4 device object.
This exists so the director token used is always the latest one, without needing to re-init the entire entity.
"""
return C4Room(self.entry_data[CONF_DIRECTOR], self._idx)
def _get_device_from_variable(self, var: str) -> int | None:
current_device = self.coordinator.data[self._idx][var]
if current_device == 0:
return None
return current_device
def _get_current_video_device_id(self) -> int | None:
return self._get_device_from_variable(CONTROL4_CURRENT_VIDEO_DEVICE)
def _get_current_playing_device_id(self) -> int | None:
media_info = self._get_media_info()
if media_info:
if "medSrcDev" in media_info:
return media_info["medSrcDev"]
if "deviceid" in media_info:
return media_info["deviceid"]
return 0
def _get_media_info(self) -> dict | None:
"""Get the Media Info Dictionary if populated."""
media_info = self.coordinator.data[self._idx][CONTROL4_MEDIA_INFO]
if "mediainfo" in media_info:
return media_info["mediainfo"]
return None
def _get_current_source_state(self) -> str | None:
current_source = self._get_current_playing_device_id()
while current_source:
current_data = self.coordinator.data.get(current_source, None)
if current_data:
if current_data.get(CONTROL4_PLAYING, None):
return MediaPlayerState.PLAYING
if current_data.get(CONTROL4_PAUSED, None):
return MediaPlayerState.PAUSED
if current_data.get(CONTROL4_STOPPED, None):
return MediaPlayerState.ON
current_source = self._id_to_parent.get(current_source, None)
return None
@property
def device_class(self) -> MediaPlayerDeviceClass | None:
"""Return the class of this entity."""
for avail_source in self._sources.values():
if _SourceType.VIDEO in avail_source.source_type:
return MediaPlayerDeviceClass.TV
return MediaPlayerDeviceClass.SPEAKER
@property
def state(self):
"""Return whether this room is on or idle."""
if source_state := self._get_current_source_state():
return source_state
if self.coordinator.data[self._idx][CONTROL4_POWER_STATE]:
return MediaPlayerState.ON
return MediaPlayerState.IDLE
@property
def source(self):
"""Get the current source."""
current_source = self._get_current_playing_device_id()
if not current_source or current_source not in self._sources:
return None
return self._sources[current_source].name
@property
def media_title(self) -> str | None:
"""Get the Media Title."""
media_info = self._get_media_info()
if not media_info:
return None
if "title" in media_info:
return media_info["title"]
current_source = self._get_current_playing_device_id()
if not current_source or current_source not in self._sources:
return None
return self._sources[current_source].name
@property
def media_content_type(self):
"""Get current content type if available."""
current_source = self._get_current_playing_device_id()
if not current_source:
return None
if current_source == self._get_current_video_device_id():
return MediaType.VIDEO
return MediaType.MUSIC
async def async_media_play_pause(self):
"""If possible, toggle the current play/pause state.
Not every source supports play/pause.
Unfortunately MediaPlayer capabilities are not dynamic,
so we must determine if play/pause is supported here
"""
if self._get_current_source_state():
await super().async_media_play_pause()
@property
def source_list(self) -> list[str]:
"""Get the available source."""
return [x.name for x in self._sources.values()]
@property
def volume_level(self):
"""Get the volume level."""
return self.coordinator.data[self._idx][CONTROL4_VOLUME_STATE] / 100
@property
def is_volume_muted(self):
"""Check if the volume is muted."""
return bool(self.coordinator.data[self._idx][CONTROL4_MUTED_STATE])
async def async_select_source(self, source):
"""Select a new source."""
for avail_source in self._sources.values():
if avail_source.name == source:
audio_only = _SourceType.VIDEO not in avail_source.source_type
if audio_only:
await self._create_api_object().setAudioSource(avail_source.idx)
else:
await self._create_api_object().setVideoAndAudioSource(
avail_source.idx
)
break
await self.coordinator.async_request_refresh()
async def async_turn_off(self):
"""Turn off the room."""
await self._create_api_object().setRoomOff()
await self.coordinator.async_request_refresh()
async def async_mute_volume(self, mute):
"""Mute the room."""
if mute:
await self._create_api_object().setMuteOn()
else:
await self._create_api_object().setMuteOff()
await self.coordinator.async_request_refresh()
async def async_set_volume_level(self, volume):
"""Set room volume, 0-1 scale."""
await self._create_api_object().setVolume(int(volume * 100))
await self.coordinator.async_request_refresh()
async def async_volume_up(self):
"""Increase the volume by 1."""
await self._create_api_object().setIncrementVolume()
await self.coordinator.async_request_refresh()
async def async_volume_down(self):
"""Decrease the volume by 1."""
await self._create_api_object().setDecrementVolume()
await self.coordinator.async_request_refresh()
async def async_media_pause(self):
"""Issue a pause command."""
await self._create_api_object().setPause()
await self.coordinator.async_request_refresh()
async def async_media_play(self):
"""Issue a play command."""
await self._create_api_object().setPlay()
await self.coordinator.async_request_refresh()
async def async_media_stop(self):
"""Issue a stop command."""
await self._create_api_object().setStop()
await self.coordinator.async_request_refresh()