home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Switch onkyo to pyeiscp, making it local_push (#120026)

Browse Source 
* Switch onkyo to pyeiscp, making it local_push

Major rewrite of the integration to use pyeiscp.
This facilitates use of the async push updates.
Streamline the code dealing with zones.
Handle sound mode.
Add myself to codeowners.

* Add types

* Add more types

* Address feedback

* Remove sound mode support for now

* Fix zone detection

* Keep legacy unique_id
This commit is contained in:
Artur Pragacz
2024-06-26 10:28:06 +02:00
committed by GitHub
parent 0a48cc29b6
commit a4ba346dfc
5 changed files with 364 additions and 413 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CODEOWNERS
View File

@ -997,6 +997,7 @@ build.json @home-assistant/supervisor
/tests/components/ondilo_ico/ @JeromeHXP
/homeassistant/components/onewire/ @garbled1 @epenet
/tests/components/onewire/ @garbled1 @epenet
/homeassistant/components/onkyo/ @arturpragacz
/homeassistant/components/onvif/ @hunterjm
/tests/components/onvif/ @hunterjm
/homeassistant/components/open_meteo/ @frenck

8
homeassistant/components/onkyo/manifest.json
View File

@ -1,9 +1,9 @@
{
"domain": "onkyo",
"name": "Onkyo",
"codeowners": [],
"codeowners": ["@arturpragacz"],
"documentation": "https://www.home-assistant.io/integrations/onkyo",
"iot_class": "local_polling",
"loggers": ["eiscp"],
"requirements": ["onkyo-eiscp==1.2.7"]
"iot_class": "local_push",
"loggers": ["pyeiscp"],
"requirements": ["pyeiscp==0.0.7"]
}

760
homeassistant/components/onkyo/media_player.py
View File

@ -2,11 +2,11 @@
from __future__ import annotations
import asyncio
import logging
from typing import Any
import eiscp
from eiscp import eISCP
import pyeiscp
import voluptuous as vol
from homeassistant.components.media_player import (
@ -17,9 +17,14 @@ from homeassistant.components.media_player import (
MediaPlayerState,
MediaType,
)
from homeassistant.const import ATTR_ENTITY_ID, CONF_HOST, CONF_NAME
from homeassistant.core import HomeAssistant, ServiceCall
import homeassistant.helpers.config_validation as cv
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_HOST,
CONF_NAME,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
@ -32,13 +37,12 @@ CONF_RECEIVER_MAX_VOLUME = "receiver_max_volume"
DEFAULT_NAME = "Onkyo Receiver"
SUPPORTED_MAX_VOLUME = 100
DEFAULT_RECEIVER_MAX_VOLUME = 80
ZONES = {"zone2": "Zone 2", "zone3": "Zone 3", "zone4": "Zone 4"}
SUPPORT_ONKYO_WO_VOLUME = (
MediaPlayerEntityFeature.TURN_ON
| MediaPlayerEntityFeature.TURN_OFF
| MediaPlayerEntityFeature.SELECT_SOURCE
| MediaPlayerEntityFeature.PLAY
| MediaPlayerEntityFeature.PLAY_MEDIA
)
SUPPORT_ONKYO = (
@ -49,6 +53,7 @@ SUPPORT_ONKYO = (
)
KNOWN_HOSTS: list[str] = []
DEFAULT_SOURCES = {
"tv": "TV",
"bd": "Bluray",
@ -63,7 +68,6 @@ DEFAULT_SOURCES = {
"video7": "Video 7",
"fm": "Radio",
}
DEFAULT_PLAYABLE_SOURCES = ("fm", "am", "tuner")
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
@ -80,15 +84,39 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
}
)
TIMEOUT_MESSAGE = "Timeout waiting for response."
ATTR_HDMI_OUTPUT = "hdmi_output"
ATTR_PRESET = "preset"
ATTR_AUDIO_INFORMATION = "audio_information"
ATTR_VIDEO_INFORMATION = "video_information"
ATTR_VIDEO_OUT = "video_out"
AUDIO_VIDEO_INFORMATION_UPDATE_WAIT_TIME = 8
AUDIO_INFORMATION_MAPPING = [
"audio_input_port",
"input_signal_format",
"input_frequency",
"input_channels",
"listening_mode",
"output_channels",
"output_frequency",
"precision_quartz_lock_system",
"auto_phase_control_delay",
"auto_phase_control_phase",
]
VIDEO_INFORMATION_MAPPING = [
"video_input_port",
"input_resolution",
"input_color_schema",
"input_color_depth",
"video_output_port",
"output_resolution",
"output_color_schema",
"output_color_depth",
"picture_mode",
]
ACCEPTED_VALUES = [
"no",
"analog",
@ -106,415 +134,187 @@ ONKYO_SELECT_OUTPUT_SCHEMA = vol.Schema(
vol.Required(ATTR_HDMI_OUTPUT): vol.In(ACCEPTED_VALUES),
}
)
SERVICE_SELECT_HDMI_OUTPUT = "onkyo_select_hdmi_output"
def _parse_onkyo_payload(payload):
"""Parse a payload returned from the eiscp library."""
if isinstance(payload, bool):
# command not supported by the device
return False
if len(payload) < 2:
# no value
return None
if isinstance(payload[1], str):
return payload[1].split(",")
return payload[1]
def _tuple_get(tup, index, default=None):
"""Return a tuple item at index or a default value if it doesn't exist."""
return (tup[index : index + 1] or [default])[0]
def determine_zones(receiver):
"""Determine what zones are available for the receiver."""
out = {"zone2": False, "zone3": False}
try:
_LOGGER.debug("Checking for zone 2 capability")
response = receiver.raw("ZPWQSTN")
if response != "ZPWN/A": # Zone 2 Available
out["zone2"] = True
else:
_LOGGER.debug("Zone 2 not available")
except ValueError as error:
if str(error) != TIMEOUT_MESSAGE:
raise
_LOGGER.debug("Zone 2 timed out, assuming no functionality")
try:
_LOGGER.debug("Checking for zone 3 capability")
response = receiver.raw("PW3QSTN")
if response != "PW3N/A":
out["zone3"] = True
else:
_LOGGER.debug("Zone 3 not available")
except ValueError as error:
if str(error) != TIMEOUT_MESSAGE:
raise
_LOGGER.debug("Zone 3 timed out, assuming no functionality")
except AssertionError:
_LOGGER.error("Zone 3 detection failed")
return out
def setup_platform(
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Onkyo platform."""
hosts: list[OnkyoDevice] = []
receivers: dict[str, pyeiscp.Connection] = {} # indexed by host
entities: dict[str, dict[str, OnkyoMediaPlayer]] = {} # indexed by host and zone
def service_handle(service: ServiceCall) -> None:
async def async_service_handle(service: ServiceCall) -> None:
"""Handle for services."""
entity_ids = service.data[ATTR_ENTITY_ID]
devices = [d for d in hosts if d.entity_id in entity_ids]
targets = [
entity
for h in entities.values()
for entity in h.values()
if entity.entity_id in entity_ids
]
for device in devices:
for target in targets:
if service.service == SERVICE_SELECT_HDMI_OUTPUT:
device.select_output(service.data[ATTR_HDMI_OUTPUT])
await target.async_select_output(service.data[ATTR_HDMI_OUTPUT])
hass.services.register(
hass.services.async_register(
DOMAIN,
SERVICE_SELECT_HDMI_OUTPUT,
service_handle,
async_service_handle,
schema=ONKYO_SELECT_OUTPUT_SCHEMA,
)
if CONF_HOST in config and (host := config[CONF_HOST]) not in KNOWN_HOSTS:
try:
receiver = eiscp.eISCP(host)
hosts.append(
OnkyoDevice(
receiver,
config.get(CONF_SOURCES),
name=config.get(CONF_NAME),
max_volume=config.get(CONF_MAX_VOLUME),
receiver_max_volume=config.get(CONF_RECEIVER_MAX_VOLUME),
)
host = config.get(CONF_HOST)
name = config[CONF_NAME]
max_volume = config[CONF_MAX_VOLUME]
receiver_max_volume = config[CONF_RECEIVER_MAX_VOLUME]
sources = config[CONF_SOURCES]
@callback
def async_onkyo_update_callback(message: tuple[str, str, Any], origin: str) -> None:
"""Process new message from receiver."""
receiver = receivers[origin]
_LOGGER.debug("Received update callback from %s: %s", receiver.name, message)
zone, _, value = message
entity = entities[origin].get(zone)
if entity is not None:
if entity.enabled:
entity.process_update(message)
elif zone in ZONES and value != "N/A":
# When we receive the status for a zone, and the value is not "N/A",
# then zone is available on the receiver, so we create the entity for it.
_LOGGER.debug("Discovered %s on %s", ZONES[zone], receiver.name)
zone_entity = OnkyoMediaPlayer(
receiver, sources, zone, max_volume, receiver_max_volume
)
KNOWN_HOSTS.append(host)
entities[origin][zone] = zone_entity
async_add_entities([zone_entity])
zones = determine_zones(receiver)
@callback
def async_onkyo_connect_callback(origin: str) -> None:
"""Receiver (re)connected."""
receiver = receivers[origin]
_LOGGER.debug("Receiver (re)connected: %s (%s)", receiver.name, receiver.host)
# Add Zone2 if available
if zones["zone2"]:
_LOGGER.debug("Setting up zone 2")
hosts.append(
OnkyoDeviceZone(
"2",
receiver,
config.get(CONF_SOURCES),
name=f"{config[CONF_NAME]} Zone 2",
max_volume=config.get(CONF_MAX_VOLUME),
receiver_max_volume=config.get(CONF_RECEIVER_MAX_VOLUME),
)
)
# Add Zone3 if available
if zones["zone3"]:
_LOGGER.debug("Setting up zone 3")
hosts.append(
OnkyoDeviceZone(
"3",
receiver,
config.get(CONF_SOURCES),
name=f"{config[CONF_NAME]} Zone 3",
max_volume=config.get(CONF_MAX_VOLUME),
receiver_max_volume=config.get(CONF_RECEIVER_MAX_VOLUME),
)
)
except OSError:
_LOGGER.error("Unable to connect to receiver at %s", host)
for entity in entities[origin].values():
entity.backfill_state()
def setup_receiver(receiver: pyeiscp.Connection) -> None:
KNOWN_HOSTS.append(receiver.host)
# Store the receiver object and create a dictionary to store its entities.
receivers[receiver.host] = receiver
entities[receiver.host] = {}
# Discover what zones are available for the receiver by querying the power.
# If we get a response for the specific zone, it means it is available.
for zone in ZONES:
receiver.query_property(zone, "power")
# Add the main zone to entities, since it is always active.
_LOGGER.debug("Adding Main Zone on %s", receiver.name)
main_entity = OnkyoMediaPlayer(
receiver, sources, "main", max_volume, receiver_max_volume
)
entities[receiver.host]["main"] = main_entity
async_add_entities([main_entity])
if host is not None and host not in KNOWN_HOSTS:
_LOGGER.debug("Manually creating receiver: %s (%s)", name, host)
receiver = await pyeiscp.Connection.create(
host=host,
update_callback=async_onkyo_update_callback,
connect_callback=async_onkyo_connect_callback,
)
# The library automatically adds a name and identifier only on discovered hosts,
# so manually add them here instead.
receiver.name = name
receiver.identifier = None
setup_receiver(receiver)
else:
for receiver in eISCP.discover():
@callback
async def async_onkyo_discovery_callback(receiver: pyeiscp.Connection):
"""Receiver discovered, connection not yet active."""
_LOGGER.debug("Receiver discovered: %s (%s)", receiver.name, receiver.host)
if receiver.host not in KNOWN_HOSTS:
hosts.append(OnkyoDevice(receiver, config.get(CONF_SOURCES)))
KNOWN_HOSTS.append(receiver.host)
add_entities(hosts, True)
await receiver.connect()
setup_receiver(receiver)
_LOGGER.debug("Discovering receivers")
await pyeiscp.Connection.discover(
update_callback=async_onkyo_update_callback,
connect_callback=async_onkyo_connect_callback,
discovery_callback=async_onkyo_discovery_callback,
)
@callback
def close_receiver(_event):
for receiver in receivers.values():
receiver.close()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, close_receiver)
class OnkyoDevice(MediaPlayerEntity):
"""Representation of an Onkyo device."""
class OnkyoMediaPlayer(MediaPlayerEntity):
"""Representation of an Onkyo Receiver Media Player (one per each zone)."""
_attr_supported_features = SUPPORT_ONKYO
_attr_should_poll = False
_supports_volume: bool = False
_supports_audio_info: bool = False
_supports_video_info: bool = False
_query_timer: asyncio.TimerHandle | None = None
def __init__(
self,
receiver,
sources,
name=None,
max_volume=SUPPORTED_MAX_VOLUME,
receiver_max_volume=DEFAULT_RECEIVER_MAX_VOLUME,
):
receiver: pyeiscp.Connection,
sources: dict[str, str],
zone: str,
max_volume: int,
receiver_max_volume: int,
) -> None:
"""Initialize the Onkyo Receiver."""
self._receiver = receiver
self._attr_is_volume_muted = False
self._attr_volume_level = 0
self._attr_state = MediaPlayerState.OFF
if name:
# not discovered
self._attr_name = name
else:
name = receiver.name
self._attr_name = f"{name}{' ' + ZONES[zone] if zone != 'main' else ''}"
identifier = receiver.identifier
if identifier is not None:
# discovered
self._attr_unique_id = (
f"{receiver.info['model_name']}_{receiver.info['identifier']}"
)
self._attr_name = self._attr_unique_id
if zone == "main":
# keep legacy unique_id
self._attr_unique_id = f"{name}_{identifier}"
else:
self._attr_unique_id = f"{identifier}_{zone}"
else:
# not discovered
self._attr_unique_id = None
self._max_volume = max_volume
self._receiver_max_volume = receiver_max_volume
self._attr_source_list = list(sources.values())
self._zone = zone
self._source_mapping = sources
self._reverse_mapping = {value: key for key, value in sources.items()}
self._max_volume = max_volume
self._receiver_max_volume = receiver_max_volume
self._attr_source_list = list(sources.values())
self._attr_extra_state_attributes = {}
self._hdmi_out_supported = True
self._audio_info_supported = True
self._video_info_supported = True
def command(self, command):
"""Run an eiscp command and catch connection errors."""
try:
result = self._receiver.command(command)
except (ValueError, OSError, AttributeError, AssertionError):
if self._receiver.command_socket:
self._receiver.command_socket = None
_LOGGER.debug("Resetting connection to %s", self.name)
else:
_LOGGER.info("%s is disconnected. Attempting to reconnect", self.name)
return False
_LOGGER.debug("Result for %s: %s", command, result)
return result
async def async_added_to_hass(self) -> None:
"""Entity has been added to hass."""
self.backfill_state()
def update(self) -> None:
"""Get the latest state from the device."""
status = self.command("system-power query")
if not status:
return
if status[1] == "on":
self._attr_state = MediaPlayerState.ON
else:
self._attr_state = MediaPlayerState.OFF
self._attr_extra_state_attributes.pop(ATTR_AUDIO_INFORMATION, None)
self._attr_extra_state_attributes.pop(ATTR_VIDEO_INFORMATION, None)
self._attr_extra_state_attributes.pop(ATTR_PRESET, None)
self._attr_extra_state_attributes.pop(ATTR_VIDEO_OUT, None)
return
volume_raw = self.command("volume query")
mute_raw = self.command("audio-muting query")
current_source_raw = self.command("input-selector query")
# If the following command is sent to a device with only one HDMI out,
# the display shows 'Not Available'.
# We avoid this by checking if HDMI out is supported
if self._hdmi_out_supported:
hdmi_out_raw = self.command("hdmi-output-selector query")
else:
hdmi_out_raw = []
preset_raw = self.command("preset query")
if self._audio_info_supported:
audio_information_raw = self.command("audio-information query")
self._parse_audio_information(audio_information_raw)
if self._video_info_supported:
video_information_raw = self.command("video-information query")
self._parse_video_information(video_information_raw)
if not (volume_raw and mute_raw and current_source_raw):
return
sources = _parse_onkyo_payload(current_source_raw)
for source in sources:
if source in self._source_mapping:
self._attr_source = self._source_mapping[source]
break
self._attr_source = "_".join(sources)
if preset_raw and self.source and self.source.lower() == "radio":
self._attr_extra_state_attributes[ATTR_PRESET] = preset_raw[1]
elif ATTR_PRESET in self._attr_extra_state_attributes:
del self._attr_extra_state_attributes[ATTR_PRESET]
self._attr_is_volume_muted = bool(mute_raw[1] == "on")
# AMP_VOL / (MAX_RECEIVER_VOL * (MAX_VOL / 100))
self._attr_volume_level = volume_raw[1] / (
self._receiver_max_volume * self._max_volume / 100
)
if not hdmi_out_raw:
return
self._attr_extra_state_attributes[ATTR_VIDEO_OUT] = ",".join(hdmi_out_raw[1])
if hdmi_out_raw[1] == "N/A":
self._hdmi_out_supported = False
def turn_off(self) -> None:
"""Turn the media player off."""
self.command("system-power standby")
def set_volume_level(self, volume: float) -> None:
"""Set volume level, input is range 0..1.
However full volume on the amp is usually far too loud so allow the user to
specify the upper range with CONF_MAX_VOLUME. We change as per max_volume
set by user. This means that if max volume is 80 then full volume in HA will
give 80% volume on the receiver. Then we convert that to the correct scale
for the receiver.
"""
# HA_VOL * (MAX VOL / 100) * MAX_RECEIVER_VOL
self.command(
"volume"
f" {int(volume * (self._max_volume / 100) * self._receiver_max_volume)}"
)
def volume_up(self) -> None:
"""Increase volume by 1 step."""
self.command("volume level-up")
def volume_down(self) -> None:
"""Decrease volume by 1 step."""
self.command("volume level-down")
def mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player."""
if mute:
self.command("audio-muting on")
else:
self.command("audio-muting off")
def turn_on(self) -> None:
"""Turn the media player on."""
self.command("system-power on")
def select_source(self, source: str) -> None:
"""Set the input source."""
if self.source_list and source in self.source_list:
source = self._reverse_mapping[source]
self.command(f"input-selector {source}")
def play_media(
self, media_type: MediaType | str, media_id: str, **kwargs: Any
) -> None:
"""Play radio station by preset number."""
source = self._reverse_mapping[self._attr_source]
if media_type.lower() == "radio" and source in DEFAULT_PLAYABLE_SOURCES:
self.command(f"preset {media_id}")
def select_output(self, output):
"""Set hdmi-out."""
self.command(f"hdmi-output-selector={output}")
def _parse_audio_information(self, audio_information_raw):
values = _parse_onkyo_payload(audio_information_raw)
if values is False:
self._audio_info_supported = False
return
if values:
info = {
"format": _tuple_get(values, 1),
"input_frequency": _tuple_get(values, 2),
"input_channels": _tuple_get(values, 3),
"listening_mode": _tuple_get(values, 4),
"output_channels": _tuple_get(values, 5),
"output_frequency": _tuple_get(values, 6),
}
self._attr_extra_state_attributes[ATTR_AUDIO_INFORMATION] = info
else:
self._attr_extra_state_attributes.pop(ATTR_AUDIO_INFORMATION, None)
def _parse_video_information(self, video_information_raw):
values = _parse_onkyo_payload(video_information_raw)
if values is False:
self._video_info_supported = False
return
if values:
info = {
"input_resolution": _tuple_get(values, 1),
"input_color_schema": _tuple_get(values, 2),
"input_color_depth": _tuple_get(values, 3),
"output_resolution": _tuple_get(values, 5),
"output_color_schema": _tuple_get(values, 6),
"output_color_depth": _tuple_get(values, 7),
"picture_mode": _tuple_get(values, 8),
"dynamic_range": _tuple_get(values, 9),
}
self._attr_extra_state_attributes[ATTR_VIDEO_INFORMATION] = info
else:
self._attr_extra_state_attributes.pop(ATTR_VIDEO_INFORMATION, None)
class OnkyoDeviceZone(OnkyoDevice):
"""Representation of an Onkyo device's extra zone."""
def __init__(
self,
zone,
receiver,
sources,
name=None,
max_volume=SUPPORTED_MAX_VOLUME,
receiver_max_volume=DEFAULT_RECEIVER_MAX_VOLUME,
):
"""Initialize the Zone with the zone identifier."""
self._zone = zone
self._supports_volume = True
super().__init__(receiver, sources, name, max_volume, receiver_max_volume)
def update(self) -> None:
"""Get the latest state from the device."""
status = self.command(f"zone{self._zone}.power=query")
if not status:
return
if status[1] == "on":
self._attr_state = MediaPlayerState.ON
else:
self._attr_state = MediaPlayerState.OFF
return
volume_raw = self.command(f"zone{self._zone}.volume=query")
mute_raw = self.command(f"zone{self._zone}.muting=query")
current_source_raw = self.command(f"zone{self._zone}.selector=query")
preset_raw = self.command(f"zone{self._zone}.preset=query")
# If we received a source value, but not a volume value
# it's likely this zone permanently does not support volume.
if current_source_raw and not volume_raw:
self._supports_volume = False
if not (volume_raw and mute_raw and current_source_raw):
return
# It's possible for some players to have zones set to HDMI with
# no sound control. In this case, the string `N/A` is returned.
self._supports_volume = isinstance(volume_raw[1], (float, int))
# eiscp can return string or tuple. Make everything tuples.
if isinstance(current_source_raw[1], str):
current_source_tuples = (current_source_raw[0], (current_source_raw[1],))
else:
current_source_tuples = current_source_raw
for source in current_source_tuples[1]:
if source in self._source_mapping:
self._attr_source = self._source_mapping[source]
break
self._attr_source = "_".join(current_source_tuples[1])
self._attr_is_volume_muted = bool(mute_raw[1] == "on")
if preset_raw and self.source and self.source.lower() == "radio":
self._attr_extra_state_attributes[ATTR_PRESET] = preset_raw[1]
elif ATTR_PRESET in self._attr_extra_state_attributes:
del self._attr_extra_state_attributes[ATTR_PRESET]
if self._supports_volume:
# AMP_VOL / (MAX_RECEIVER_VOL * (MAX_VOL / 100))
self._attr_volume_level = volume_raw[1] / (
self._receiver_max_volume * self._max_volume / 100
)
async def async_will_remove_from_hass(self) -> None:
"""Cancel the query timer when the entity is removed."""
if self._query_timer:
self._query_timer.cancel()
self._query_timer = None
@property
def supported_features(self) -> MediaPlayerEntityFeature:
@ -523,12 +323,26 @@ class OnkyoDeviceZone(OnkyoDevice):
return SUPPORT_ONKYO
return SUPPORT_ONKYO_WO_VOLUME
def turn_off(self) -> None:
"""Turn the media player off."""
self.command(f"zone{self._zone}.power=standby")
@callback
def _update_receiver(self, propname: str, value: Any) -> None:
"""Update a property in the receiver."""
self._receiver.update_property(self._zone, propname, value)
def set_volume_level(self, volume: float) -> None:
"""Set volume level, input is range 0..1.
@callback
def _query_receiver(self, propname: str) -> None:
"""Cause the receiver to send an update about a property."""
self._receiver.query_property(self._zone, propname)
async def async_turn_on(self) -> None:
"""Turn the media player on."""
self._update_receiver("power", "on")
async def async_turn_off(self) -> None:
"""Turn the media player off."""
self._update_receiver("power", "standby")
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1.
However full volume on the amp is usually far too loud so allow the user to
specify the upper range with CONF_MAX_VOLUME. We change as per max_volume
@ -537,31 +351,167 @@ class OnkyoDeviceZone(OnkyoDevice):
scale for the receiver.
"""
# HA_VOL * (MAX VOL / 100) * MAX_RECEIVER_VOL
self.command(
f"zone{self._zone}.volume={int(volume * (self._max_volume / 100) * self._receiver_max_volume)}"
self._update_receiver(
"volume", int(volume * (self._max_volume / 100) * self._receiver_max_volume)
)
def volume_up(self) -> None:
async def async_volume_up(self) -> None:
"""Increase volume by 1 step."""
self.command(f"zone{self._zone}.volume=level-up")
self._update_receiver("volume", "level-up")
def volume_down(self) -> None:
async def async_volume_down(self) -> None:
"""Decrease volume by 1 step."""
self.command(f"zone{self._zone}.volume=level-down")
self._update_receiver("volume", "level-down")
def mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player."""
if mute:
self.command(f"zone{self._zone}.muting=on")
else:
self.command(f"zone{self._zone}.muting=off")
async def async_mute_volume(self, mute: bool) -> None:
"""Mute the volume."""
self._update_receiver(
"audio-muting" if self._zone == "main" else "muting",
"on" if mute else "off",
)
def turn_on(self) -> None:
"""Turn the media player on."""
self.command(f"zone{self._zone}.power=on")
def select_source(self, source: str) -> None:
"""Set the input source."""
async def async_select_source(self, source: str) -> None:
"""Select input source."""
if self.source_list and source in self.source_list:
source = self._reverse_mapping[source]
self.command(f"zone{self._zone}.selector={source}")
self._update_receiver(
"input-selector" if self._zone == "main" else "selector", source
)
async def async_select_output(self, hdmi_output: str) -> None:
"""Set hdmi-out."""
self._update_receiver("hdmi-output-selector", hdmi_output)
async def async_play_media(
self, media_type: MediaType | str, media_id: str, **kwargs: Any
) -> None:
"""Play radio station by preset number."""
if self.source is not None:
source = self._reverse_mapping[self.source]
if media_type.lower() == "radio" and source in DEFAULT_PLAYABLE_SOURCES:
self._update_receiver("preset", media_id)
@callback
def backfill_state(self) -> None:
"""Get the receiver to send all the info we care about.
Usually run only on connect, as we can otherwise rely on the
receiver to keep us informed of changes.
"""
self._query_receiver("power")
self._query_receiver("volume")
self._query_receiver("preset")
if self._zone == "main":
self._query_receiver("hdmi-output-selector")
self._query_receiver("audio-muting")
self._query_receiver("input-selector")
self._query_receiver("listening-mode")
self._query_receiver("audio-information")
self._query_receiver("video-information")
else:
self._query_receiver("muting")
self._query_receiver("selector")
@callback
def process_update(self, update: tuple[str, str, Any]) -> None:
"""Store relevant updates so they can be queried later."""
zone, command, value = update
if zone != self._zone:
return
if command in ["system-power", "power"]:
if value == "on":
self._attr_state = MediaPlayerState.ON
else:
self._attr_state = MediaPlayerState.OFF
self._attr_extra_state_attributes.pop(ATTR_AUDIO_INFORMATION, None)
self._attr_extra_state_attributes.pop(ATTR_VIDEO_INFORMATION, None)
self._attr_extra_state_attributes.pop(ATTR_PRESET, None)
self._attr_extra_state_attributes.pop(ATTR_VIDEO_OUT, None)
elif command in ["volume", "master-volume"] and value != "N/A":
self._supports_volume = True
# AMP_VOL / (MAX_RECEIVER_VOL * (MAX_VOL / 100))
self._attr_volume_level = value / (
self._receiver_max_volume * self._max_volume / 100
)
elif command in ["muting", "audio-muting"]:
self._attr_is_volume_muted = bool(value == "on")
elif command in ["selector", "input-selector"]:
self._parse_source(value)
self._query_av_info_delayed()
elif command == "hdmi-output-selector":
self._attr_extra_state_attributes[ATTR_VIDEO_OUT] = ",".join(value)
elif command == "preset":
if self.source is not None and self.source.lower() == "radio":
self._attr_extra_state_attributes[ATTR_PRESET] = value
elif ATTR_PRESET in self._attr_extra_state_attributes:
del self._attr_extra_state_attributes[ATTR_PRESET]
elif command == "audio-information":
self._supports_audio_info = True
self._parse_audio_information(value)
elif command == "video-information":
self._supports_video_info = True
self._parse_video_information(value)
elif command == "fl-display-information":
self._query_av_info_delayed()
self.async_write_ha_state()
@callback
def _parse_source(self, source):
# source is either a tuple of values or a single value,
# so we convert to a tuple, when it is a single value.
if not isinstance(source, tuple):
source = (source,)
for value in source:
if value in self._source_mapping:
self._attr_source = self._source_mapping[value]
break
self._attr_source = "_".join(source)
@callback
def _parse_audio_information(self, audio_information):
# If audio information is not available, N/A is returned,
# so only update the audio information, when it is not N/A.
if audio_information == "N/A":
self._attr_extra_state_attributes.pop(ATTR_AUDIO_INFORMATION, None)
return
self._attr_extra_state_attributes[ATTR_AUDIO_INFORMATION] = {
name: value
for name, value in zip(
AUDIO_INFORMATION_MAPPING, audio_information, strict=False
)
if len(value) > 0
}
@callback
def _parse_video_information(self, video_information):
# If video information is not available, N/A is returned,
# so only update the video information, when it is not N/A.
if video_information == "N/A":
self._attr_extra_state_attributes.pop(ATTR_VIDEO_INFORMATION, None)
return
self._attr_extra_state_attributes[ATTR_VIDEO_INFORMATION] = {
name: value
for name, value in zip(
VIDEO_INFORMATION_MAPPING, video_information, strict=False
)
if len(value) > 0
}
def _query_av_info_delayed(self):
if self._zone == "main" and not self._query_timer:
@callback
def _query_av_info():
if self._supports_audio_info:
self._query_receiver("audio-information")
if self._supports_video_info:
self._query_receiver("video-information")
self._query_timer = None
self._query_timer = self.hass.loop.call_later(
AUDIO_VIDEO_INFORMATION_UPDATE_WAIT_TIME, _query_av_info
)

2
homeassistant/generated/integrations.json
View File

@ -4243,7 +4243,7 @@
"name": "Onkyo",
"integration_type": "hub",
"config_flow": false,
"iot_class": "local_polling"
"iot_class": "local_push"
},
"onvif": {
"name": "ONVIF",

6
requirements_all.txt
View File

@ -1470,9 +1470,6 @@ omnilogic==0.4.5
# homeassistant.components.ondilo_ico
ondilo==0.5.0
# homeassistant.components.onkyo
onkyo-eiscp==1.2.7
# homeassistant.components.onvif
onvif-zeep-async==3.1.12
@ -1826,6 +1823,9 @@ pyefergy==22.5.0
# homeassistant.components.energenie_power_sockets
pyegps==0.2.5
# homeassistant.components.onkyo
pyeiscp==0.0.7
# homeassistant.components.emoncms
pyemoncms==0.0.7