mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 16:57:53 +00:00
Add config flow to Onkyo (#117319)
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com> Co-authored-by: Artur Pragacz <49985303+arturpragacz@users.noreply.github.com> Co-authored-by: Artur Pragacz <artur@pragacz.com> Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Nebula83
GitHub
parent
66a7b508b2
commit
937dbdc71f
@ -1047,6 +1047,7 @@ build.json @home-assistant/supervisor
|
||||
/homeassistant/components/onewire/ @garbled1 @epenet
|
||||
/tests/components/onewire/ @garbled1 @epenet
|
||||
/homeassistant/components/onkyo/ @arturpragacz
|
||||
/tests/components/onkyo/ @arturpragacz
|
||||
/homeassistant/components/onvif/ @hunterjm
|
||||
/tests/components/onvif/ @hunterjm
|
||||
/homeassistant/components/open_meteo/ @frenck
|
||||
|
||||
@ -1 +1,76 @@
|
||||
"""The onkyo component."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .const import DOMAIN, OPTION_INPUT_SOURCES, InputSource
|
||||
from .receiver import Receiver, async_interview
|
||||
from .services import DATA_MP_ENTITIES, async_register_services
|
||||
|
||||
PLATFORMS = [Platform.MEDIA_PLAYER]
|
||||
|
||||
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
|
||||
|
||||
|
||||
@dataclass
|
||||
class OnkyoData:
|
||||
"""Config Entry data."""
|
||||
|
||||
receiver: Receiver
|
||||
sources: dict[InputSource, str]
|
||||
|
||||
|
||||
type OnkyoConfigEntry = ConfigEntry[OnkyoData]
|
||||
|
||||
|
||||
async def async_setup(hass: HomeAssistant, _: ConfigType) -> bool:
|
||||
"""Set up Onkyo component."""
|
||||
await async_register_services(hass)
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: OnkyoConfigEntry) -> bool:
|
||||
"""Set up the Onkyo config entry."""
|
||||
entry.async_on_unload(entry.add_update_listener(update_listener))
|
||||
|
||||
host = entry.data[CONF_HOST]
|
||||
|
||||
info = await async_interview(host)
|
||||
if info is None:
|
||||
raise ConfigEntryNotReady(f"Unable to connect to: {host}")
|
||||
|
||||
receiver = await Receiver.async_create(info)
|
||||
|
||||
sources_store: dict[str, str] = entry.options[OPTION_INPUT_SOURCES]
|
||||
sources = {InputSource(k): v for k, v in sources_store.items()}
|
||||
|
||||
entry.runtime_data = OnkyoData(receiver, sources)
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
await receiver.conn.connect()
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: OnkyoConfigEntry) -> bool:
|
||||
"""Unload Onkyo config entry."""
|
||||
del hass.data[DATA_MP_ENTITIES][entry.entry_id]
|
||||
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
receiver = entry.runtime_data.receiver
|
||||
receiver.conn.close()
|
||||
|
||||
return unload_ok
|
||||
|
||||
|
||||
async def update_listener(hass: HomeAssistant, entry: OnkyoConfigEntry) -> None:
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(entry.entry_id)
|
||||
|
||||
311
homeassistant/components/onkyo/config_flow.py
Normal file
311
homeassistant/components/onkyo/config_flow.py
Normal file
@ -0,0 +1,311 @@
|
||||
"""Config flow for Onkyo."""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import (
|
||||
ConfigEntry,
|
||||
ConfigFlow,
|
||||
ConfigFlowResult,
|
||||
OptionsFlow,
|
||||
OptionsFlowWithConfigEntry,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.selector import (
|
||||
NumberSelector,
|
||||
NumberSelectorConfig,
|
||||
NumberSelectorMode,
|
||||
Selector,
|
||||
SelectSelector,
|
||||
SelectSelectorConfig,
|
||||
SelectSelectorMode,
|
||||
TextSelector,
|
||||
)
|
||||
|
||||
from .const import (
|
||||
CONF_RECEIVER_MAX_VOLUME,
|
||||
CONF_SOURCES,
|
||||
DOMAIN,
|
||||
OPTION_INPUT_SOURCES,
|
||||
OPTION_MAX_VOLUME,
|
||||
OPTION_MAX_VOLUME_DEFAULT,
|
||||
OPTION_VOLUME_RESOLUTION,
|
||||
OPTION_VOLUME_RESOLUTION_DEFAULT,
|
||||
VOLUME_RESOLUTION_ALLOWED,
|
||||
InputSource,
|
||||
)
|
||||
from .receiver import ReceiverInfo, async_discover, async_interview
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONF_DEVICE = "device"
|
||||
|
||||
INPUT_SOURCES_ALL_MEANINGS = [
|
||||
input_source.value_meaning for input_source in InputSource
|
||||
]
|
||||
STEP_CONFIGURE_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(
|
||||
OPTION_VOLUME_RESOLUTION,
|
||||
default=OPTION_VOLUME_RESOLUTION_DEFAULT,
|
||||
): vol.In(VOLUME_RESOLUTION_ALLOWED),
|
||||
vol.Required(OPTION_INPUT_SOURCES, default=[]): SelectSelector(
|
||||
SelectSelectorConfig(
|
||||
options=INPUT_SOURCES_ALL_MEANINGS,
|
||||
multiple=True,
|
||||
mode=SelectSelectorMode.DROPDOWN,
|
||||
)
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class OnkyoConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Onkyo config flow."""
|
||||
|
||||
_receiver_info: ReceiverInfo
|
||||
_discovered_infos: dict[str, ReceiverInfo]
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle a flow initialized by the user."""
|
||||
return self.async_show_menu(
|
||||
step_id="user", menu_options=["manual", "eiscp_discovery"]
|
||||
)
|
||||
|
||||
async def async_step_manual(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle manual device entry."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
host = user_input[CONF_HOST]
|
||||
_LOGGER.debug("Config flow start manual: %s", host)
|
||||
try:
|
||||
info = await async_interview(host)
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
if info is None:
|
||||
errors["base"] = "cannot_connect"
|
||||
else:
|
||||
self._receiver_info = info
|
||||
await self.async_set_unique_id(
|
||||
info.identifier, raise_on_progress=False
|
||||
)
|
||||
self._abort_if_unique_id_configured(updates=user_input)
|
||||
return await self.async_step_configure_receiver()
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="manual",
|
||||
data_schema=vol.Schema({vol.Required(CONF_HOST): str}),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_eiscp_discovery(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Start eiscp discovery and handle user device selection."""
|
||||
if user_input is not None:
|
||||
self._receiver_info = self._discovered_infos[user_input[CONF_DEVICE]]
|
||||
await self.async_set_unique_id(
|
||||
self._receiver_info.identifier, raise_on_progress=False
|
||||
)
|
||||
self._abort_if_unique_id_configured(
|
||||
updates={CONF_HOST: self._receiver_info.host}
|
||||
)
|
||||
return await self.async_step_configure_receiver()
|
||||
|
||||
_LOGGER.debug("Config flow start eiscp discovery")
|
||||
|
||||
try:
|
||||
infos = await async_discover()
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
return self.async_abort(reason="unknown")
|
||||
|
||||
_LOGGER.debug("Discovered devices: %s", infos)
|
||||
|
||||
self._discovered_infos = {}
|
||||
discovered_names = {}
|
||||
current_unique_ids = self._async_current_ids()
|
||||
for info in infos:
|
||||
if info.identifier in current_unique_ids:
|
||||
continue
|
||||
self._discovered_infos[info.identifier] = info
|
||||
device_name = f"{info.model_name} ({info.host})"
|
||||
discovered_names[info.identifier] = device_name
|
||||
|
||||
_LOGGER.debug("Discovered new devices: %s", self._discovered_infos)
|
||||
|
||||
if not discovered_names:
|
||||
return self.async_abort(reason="no_devices_found")
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="eiscp_discovery",
|
||||
data_schema=vol.Schema(
|
||||
{vol.Required(CONF_DEVICE): vol.In(discovered_names)}
|
||||
),
|
||||
)
|
||||
|
||||
async def async_step_configure_receiver(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the configuration of a single receiver."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
source_meanings: list[str] = user_input[OPTION_INPUT_SOURCES]
|
||||
if not source_meanings:
|
||||
errors[OPTION_INPUT_SOURCES] = "empty_input_source_list"
|
||||
else:
|
||||
sources_store: dict[str, str] = {}
|
||||
for source_meaning in source_meanings:
|
||||
source = InputSource.from_meaning(source_meaning)
|
||||
sources_store[source.value] = source_meaning
|
||||
|
||||
result = self.async_create_entry(
|
||||
title=self._receiver_info.model_name,
|
||||
data={
|
||||
CONF_HOST: self._receiver_info.host,
|
||||
},
|
||||
options={
|
||||
OPTION_VOLUME_RESOLUTION: user_input[OPTION_VOLUME_RESOLUTION],
|
||||
OPTION_MAX_VOLUME: OPTION_MAX_VOLUME_DEFAULT,
|
||||
OPTION_INPUT_SOURCES: sources_store,
|
||||
},
|
||||
)
|
||||
_LOGGER.debug("Configured receiver, result: %s", result)
|
||||
return result
|
||||
|
||||
_LOGGER.debug("Configuring receiver, info: %s", self._receiver_info)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="configure_receiver",
|
||||
data_schema=STEP_CONFIGURE_SCHEMA,
|
||||
errors=errors,
|
||||
description_placeholders={
|
||||
"name": f"{self._receiver_info.model_name} ({self._receiver_info.host})"
|
||||
},
|
||||
)
|
||||
|
||||
async def async_step_import(self, user_input: dict[str, Any]) -> ConfigFlowResult:
|
||||
"""Import the yaml config."""
|
||||
_LOGGER.debug("Import flow user input: %s", user_input)
|
||||
|
||||
host: str = user_input[CONF_HOST]
|
||||
name: str | None = user_input.get(CONF_NAME)
|
||||
user_max_volume: int = user_input[OPTION_MAX_VOLUME]
|
||||
user_volume_resolution: int = user_input[CONF_RECEIVER_MAX_VOLUME]
|
||||
user_sources: dict[InputSource, str] = user_input[CONF_SOURCES]
|
||||
|
||||
info: ReceiverInfo | None = user_input.get("info")
|
||||
if info is None:
|
||||
try:
|
||||
info = await async_interview(host)
|
||||
except Exception:
|
||||
_LOGGER.exception("Import flow interview error for host %s", host)
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
|
||||
if info is None:
|
||||
_LOGGER.error("Import flow interview error for host %s", host)
|
||||
return self.async_abort(reason="cannot_connect")
|
||||
|
||||
unique_id = info.identifier
|
||||
await self.async_set_unique_id(unique_id)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
name = name or info.model_name
|
||||
|
||||
volume_resolution = VOLUME_RESOLUTION_ALLOWED[-1]
|
||||
for volume_resolution_allowed in VOLUME_RESOLUTION_ALLOWED:
|
||||
if user_volume_resolution <= volume_resolution_allowed:
|
||||
volume_resolution = volume_resolution_allowed
|
||||
break
|
||||
|
||||
max_volume = min(
|
||||
100, user_max_volume * user_volume_resolution / volume_resolution
|
||||
)
|
||||
|
||||
sources_store: dict[str, str] = {}
|
||||
for source, source_name in user_sources.items():
|
||||
sources_store[source.value] = source_name
|
||||
|
||||
return self.async_create_entry(
|
||||
title=name,
|
||||
data={
|
||||
CONF_HOST: host,
|
||||
},
|
||||
options={
|
||||
OPTION_VOLUME_RESOLUTION: volume_resolution,
|
||||
OPTION_MAX_VOLUME: max_volume,
|
||||
OPTION_INPUT_SOURCES: sources_store,
|
||||
},
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(
|
||||
config_entry: ConfigEntry,
|
||||
) -> OptionsFlow:
|
||||
"""Return the options flow."""
|
||||
return OnkyoOptionsFlowHandler(config_entry)
|
||||
|
||||
|
||||
class OnkyoOptionsFlowHandler(OptionsFlowWithConfigEntry):
|
||||
"""Handle an options flow for Onkyo."""
|
||||
|
||||
def __init__(self, config_entry: ConfigEntry) -> None:
|
||||
"""Initialize options flow."""
|
||||
super().__init__(config_entry)
|
||||
|
||||
sources_store: dict[str, str] = self.options[OPTION_INPUT_SOURCES]
|
||||
sources = {InputSource(k): v for k, v in sources_store.items()}
|
||||
self.options[OPTION_INPUT_SOURCES] = sources
|
||||
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Manage the options."""
|
||||
if user_input is not None:
|
||||
sources_store: dict[str, str] = {}
|
||||
for source_meaning, source_name in user_input.items():
|
||||
if source_meaning in INPUT_SOURCES_ALL_MEANINGS:
|
||||
source = InputSource.from_meaning(source_meaning)
|
||||
sources_store[source.value] = source_name
|
||||
|
||||
return self.async_create_entry(
|
||||
data={
|
||||
OPTION_VOLUME_RESOLUTION: self.options[OPTION_VOLUME_RESOLUTION],
|
||||
OPTION_MAX_VOLUME: user_input[OPTION_MAX_VOLUME],
|
||||
OPTION_INPUT_SOURCES: sources_store,
|
||||
}
|
||||
)
|
||||
|
||||
schema_dict: dict[Any, Selector] = {}
|
||||
|
||||
max_volume: float = self.options[OPTION_MAX_VOLUME]
|
||||
schema_dict[vol.Required(OPTION_MAX_VOLUME, default=max_volume)] = (
|
||||
NumberSelector(
|
||||
NumberSelectorConfig(min=1, max=100, mode=NumberSelectorMode.BOX)
|
||||
)
|
||||
)
|
||||
|
||||
sources: dict[InputSource, str] = self.options[OPTION_INPUT_SOURCES]
|
||||
for source in sources:
|
||||
schema_dict[vol.Required(source.value_meaning, default=sources[source])] = (
|
||||
TextSelector()
|
||||
)
|
||||
|
||||
schema = vol.Schema(schema_dict)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="init",
|
||||
data_schema=schema,
|
||||
)
|
||||
141
homeassistant/components/onkyo/const.py
Normal file
141
homeassistant/components/onkyo/const.py
Normal file
@ -0,0 +1,141 @@
|
||||
"""Constants for the Onkyo integration."""
|
||||
|
||||
from enum import Enum
|
||||
import typing
|
||||
from typing import ClassVar, Literal, Self
|
||||
|
||||
import pyeiscp
|
||||
|
||||
DOMAIN = "onkyo"
|
||||
|
||||
DEVICE_INTERVIEW_TIMEOUT = 5
|
||||
DEVICE_DISCOVERY_TIMEOUT = 5
|
||||
|
||||
CONF_SOURCES = "sources"
|
||||
CONF_RECEIVER_MAX_VOLUME = "receiver_max_volume"
|
||||
|
||||
type VolumeResolution = Literal[50, 80, 100, 200]
|
||||
OPTION_VOLUME_RESOLUTION = "volume_resolution"
|
||||
OPTION_VOLUME_RESOLUTION_DEFAULT: VolumeResolution = 50
|
||||
VOLUME_RESOLUTION_ALLOWED: tuple[VolumeResolution, ...] = typing.get_args(
|
||||
VolumeResolution.__value__
|
||||
)
|
||||
|
||||
OPTION_MAX_VOLUME = "max_volume"
|
||||
OPTION_MAX_VOLUME_DEFAULT = 100.0
|
||||
|
||||
OPTION_INPUT_SOURCES = "input_sources"
|
||||
|
||||
_INPUT_SOURCE_MEANINGS = {
|
||||
"00": "VIDEO1 ··· VCR/DVR ··· STB/DVR",
|
||||
"01": "VIDEO2 ··· CBL/SAT",
|
||||
"02": "VIDEO3 ··· GAME/TV ··· GAME",
|
||||
"03": "VIDEO4 ··· AUX",
|
||||
"04": "VIDEO5 ··· AUX2 ··· GAME2",
|
||||
"05": "VIDEO6 ··· PC",
|
||||
"06": "VIDEO7",
|
||||
"07": "HIDDEN1 ··· EXTRA1",
|
||||
"08": "HIDDEN2 ··· EXTRA2",
|
||||
"09": "HIDDEN3 ··· EXTRA3",
|
||||
"10": "DVD ··· BD/DVD",
|
||||
"11": "STRM BOX",
|
||||
"12": "TV",
|
||||
"20": "TAPE ··· TV/TAPE",
|
||||
"21": "TAPE2",
|
||||
"22": "PHONO",
|
||||
"23": "CD ··· TV/CD",
|
||||
"24": "FM",
|
||||
"25": "AM",
|
||||
"26": "TUNER",
|
||||
"27": "MUSIC SERVER ··· P4S ··· DLNA",
|
||||
"28": "INTERNET RADIO ··· IRADIO FAVORITE",
|
||||
"29": "USB ··· USB(FRONT)",
|
||||
"2A": "USB(REAR)",
|
||||
"2B": "NETWORK ··· NET",
|
||||
"2D": "AIRPLAY",
|
||||
"2E": "BLUETOOTH",
|
||||
"2F": "USB DAC IN",
|
||||
"30": "MULTI CH",
|
||||
"31": "XM",
|
||||
"32": "SIRIUS",
|
||||
"33": "DAB",
|
||||
"40": "UNIVERSAL PORT",
|
||||
"41": "LINE",
|
||||
"42": "LINE2",
|
||||
"44": "OPTICAL",
|
||||
"45": "COAXIAL",
|
||||
"55": "HDMI 5",
|
||||
"56": "HDMI 6",
|
||||
"57": "HDMI 7",
|
||||
"80": "MAIN SOURCE",
|
||||
}
|
||||
|
||||
|
||||
class InputSource(Enum):
|
||||
"""Receiver input source."""
|
||||
|
||||
DVR = "00"
|
||||
CBL = "01"
|
||||
GAME = "02"
|
||||
AUX = "03"
|
||||
GAME2 = "04"
|
||||
PC = "05"
|
||||
VIDEO7 = "06"
|
||||
EXTRA1 = "07"
|
||||
EXTRA2 = "08"
|
||||
EXTRA3 = "09"
|
||||
DVD = "10"
|
||||
STRM_BOX = "11"
|
||||
TV = "12"
|
||||
TAPE = "20"
|
||||
TAPE2 = "21"
|
||||
PHONO = "22"
|
||||
CD = "23"
|
||||
FM = "24"
|
||||
AM = "25"
|
||||
TUNER = "26"
|
||||
MUSIC_SERVER = "27"
|
||||
INTERNET_RADIO = "28"
|
||||
USB = "29"
|
||||
USB_REAR = "2A"
|
||||
NETWORK = "2B"
|
||||
AIRPLAY = "2D"
|
||||
BLUETOOTH = "2E"
|
||||
USB_DAC_IN = "2F"
|
||||
MULTI_CH = "30"
|
||||
XM = "31"
|
||||
SIRIUS = "32"
|
||||
DAB = "33"
|
||||
UNIVERSAL_PORT = "40"
|
||||
LINE = "41"
|
||||
LINE2 = "42"
|
||||
OPTICAL = "44"
|
||||
COAXIAL = "45"
|
||||
HDMI_5 = "55"
|
||||
HDMI_6 = "56"
|
||||
HDMI_7 = "57"
|
||||
MAIN_SOURCE = "80"
|
||||
|
||||
__meaning_mapping: ClassVar[dict[str, Self]] = {} # type: ignore[misc]
|
||||
|
||||
value_meaning: str
|
||||
|
||||
def __new__(cls, value: str) -> Self:
|
||||
"""Create InputSource enum."""
|
||||
obj = object.__new__(cls)
|
||||
obj._value_ = value
|
||||
obj.value_meaning = _INPUT_SOURCE_MEANINGS[value]
|
||||
|
||||
cls.__meaning_mapping[obj.value_meaning] = obj
|
||||
|
||||
return obj
|
||||
|
||||
@classmethod
|
||||
def from_meaning(cls, meaning: str) -> Self:
|
||||
"""Get InputSource enum from its meaning."""
|
||||
return cls.__meaning_mapping[meaning]
|
||||
|
||||
|
||||
ZONES = {"main": "Main", "zone2": "Zone 2", "zone3": "Zone 3", "zone4": "Zone 4"}
|
||||
|
||||
PYEISCP_COMMANDS = pyeiscp.commands.COMMANDS
|
||||
@ -2,7 +2,9 @@
|
||||
"domain": "onkyo",
|
||||
"name": "Onkyo",
|
||||
"codeowners": ["@arturpragacz"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/onkyo",
|
||||
"integration_type": "device",
|
||||
"iot_class": "local_push",
|
||||
"loggers": ["pyeiscp"],
|
||||
"requirements": ["pyeiscp==0.0.7"]
|
||||
|
||||
@ -6,45 +6,73 @@ import asyncio
|
||||
import logging
|
||||
from typing import Any, Literal
|
||||
|
||||
import pyeiscp
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.media_player import (
|
||||
DOMAIN as MEDIA_PLAYER_DOMAIN,
|
||||
PLATFORM_SCHEMA as MEDIA_PLAYER_PLATFORM_SCHEMA,
|
||||
MediaPlayerEntity,
|
||||
MediaPlayerEntityFeature,
|
||||
MediaPlayerState,
|
||||
MediaType,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_ID,
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
EVENT_HOMEASSISTANT_STOP,
|
||||
)
|
||||
from homeassistant.core import Event, HomeAssistant, ServiceCall, callback
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.config_entries import SOURCE_IMPORT
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME
|
||||
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, callback
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
from homeassistant.helpers import config_validation as cv, entity_registry as er
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
from homeassistant.util.hass_dict import HassKey
|
||||
|
||||
from .receiver import Receiver, ReceiverInfo
|
||||
from . import OnkyoConfigEntry
|
||||
from .const import (
|
||||
CONF_RECEIVER_MAX_VOLUME,
|
||||
CONF_SOURCES,
|
||||
DOMAIN,
|
||||
OPTION_MAX_VOLUME,
|
||||
OPTION_VOLUME_RESOLUTION,
|
||||
PYEISCP_COMMANDS,
|
||||
ZONES,
|
||||
InputSource,
|
||||
VolumeResolution,
|
||||
)
|
||||
from .receiver import Receiver, async_discover
|
||||
from .services import DATA_MP_ENTITIES
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DOMAIN = "onkyo"
|
||||
CONF_MAX_VOLUME_DEFAULT = 100
|
||||
CONF_RECEIVER_MAX_VOLUME_DEFAULT = 80
|
||||
CONF_SOURCES_DEFAULT = {
|
||||
"tv": "TV",
|
||||
"bd": "Bluray",
|
||||
"game": "Game",
|
||||
"aux1": "Aux1",
|
||||
"video1": "Video 1",
|
||||
"video2": "Video 2",
|
||||
"video3": "Video 3",
|
||||
"video4": "Video 4",
|
||||
"video5": "Video 5",
|
||||
"video6": "Video 6",
|
||||
"video7": "Video 7",
|
||||
"fm": "Radio",
|
||||
}
|
||||
|
||||
DATA_MP_ENTITIES: HassKey[list[dict[str, OnkyoMediaPlayer]]] = HassKey(DOMAIN)
|
||||
|
||||
CONF_SOURCES = "sources"
|
||||
CONF_MAX_VOLUME = "max_volume"
|
||||
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"}
|
||||
PLATFORM_SCHEMA = MEDIA_PLAYER_PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Optional(CONF_HOST): cv.string,
|
||||
vol.Optional(CONF_NAME): cv.string,
|
||||
vol.Optional(OPTION_MAX_VOLUME, default=CONF_MAX_VOLUME_DEFAULT): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=1, max=100)
|
||||
),
|
||||
vol.Optional(
|
||||
CONF_RECEIVER_MAX_VOLUME, default=CONF_RECEIVER_MAX_VOLUME_DEFAULT
|
||||
): cv.positive_int,
|
||||
vol.Optional(CONF_SOURCES, default=CONF_SOURCES_DEFAULT): {
|
||||
cv.string: cv.string
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
SUPPORT_ONKYO_WO_VOLUME = (
|
||||
MediaPlayerEntityFeature.TURN_ON
|
||||
@ -59,39 +87,12 @@ SUPPORT_ONKYO = (
|
||||
| MediaPlayerEntityFeature.VOLUME_STEP
|
||||
)
|
||||
|
||||
KNOWN_HOSTS: list[str] = []
|
||||
|
||||
DEFAULT_SOURCES = {
|
||||
"tv": "TV",
|
||||
"bd": "Bluray",
|
||||
"game": "Game",
|
||||
"aux1": "Aux1",
|
||||
"video1": "Video 1",
|
||||
"video2": "Video 2",
|
||||
"video3": "Video 3",
|
||||
"video4": "Video 4",
|
||||
"video5": "Video 5",
|
||||
"video6": "Video 6",
|
||||
"video7": "Video 7",
|
||||
"fm": "Radio",
|
||||
}
|
||||
DEFAULT_PLAYABLE_SOURCES = ("fm", "am", "tuner")
|
||||
|
||||
PLATFORM_SCHEMA = MEDIA_PLAYER_PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Optional(CONF_HOST): cv.string,
|
||||
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
|
||||
vol.Optional(CONF_MAX_VOLUME, default=SUPPORTED_MAX_VOLUME): vol.All(
|
||||
vol.Coerce(int), vol.Range(min=1, max=100)
|
||||
),
|
||||
vol.Optional(
|
||||
CONF_RECEIVER_MAX_VOLUME, default=DEFAULT_RECEIVER_MAX_VOLUME
|
||||
): cv.positive_int,
|
||||
vol.Optional(CONF_SOURCES, default=DEFAULT_SOURCES): {cv.string: cv.string},
|
||||
}
|
||||
DEFAULT_PLAYABLE_SOURCES = (
|
||||
InputSource.from_meaning("FM"),
|
||||
InputSource.from_meaning("AM"),
|
||||
InputSource.from_meaning("TUNER"),
|
||||
)
|
||||
|
||||
ATTR_HDMI_OUTPUT = "hdmi_output"
|
||||
ATTR_PRESET = "preset"
|
||||
ATTR_AUDIO_INFORMATION = "audio_information"
|
||||
ATTR_VIDEO_INFORMATION = "video_information"
|
||||
@ -123,52 +124,17 @@ VIDEO_INFORMATION_MAPPING = [
|
||||
"output_color_depth",
|
||||
"picture_mode",
|
||||
]
|
||||
ISSUE_URL_PLACEHOLDER = "/config/integrations/dashboard/add?domain=onkyo"
|
||||
|
||||
ACCEPTED_VALUES = [
|
||||
"no",
|
||||
"analog",
|
||||
"yes",
|
||||
"out",
|
||||
"out-sub",
|
||||
"sub",
|
||||
"hdbaset",
|
||||
"both",
|
||||
"up",
|
||||
]
|
||||
ONKYO_SELECT_OUTPUT_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(ATTR_ENTITY_ID): cv.entity_ids,
|
||||
vol.Required(ATTR_HDMI_OUTPUT): vol.In(ACCEPTED_VALUES),
|
||||
}
|
||||
)
|
||||
SERVICE_SELECT_HDMI_OUTPUT = "onkyo_select_hdmi_output"
|
||||
type InputLibValue = str | tuple[str, ...]
|
||||
|
||||
|
||||
async def async_register_services(hass: HomeAssistant) -> None:
|
||||
"""Register Onkyo services."""
|
||||
|
||||
async def async_service_handle(service: ServiceCall) -> None:
|
||||
"""Handle for services."""
|
||||
entity_ids = service.data[ATTR_ENTITY_ID]
|
||||
|
||||
targets: list[OnkyoMediaPlayer] = []
|
||||
for receiver_entities in hass.data[DATA_MP_ENTITIES]:
|
||||
targets.extend(
|
||||
entity
|
||||
for entity in receiver_entities.values()
|
||||
if entity.entity_id in entity_ids
|
||||
)
|
||||
|
||||
for target in targets:
|
||||
if service.service == SERVICE_SELECT_HDMI_OUTPUT:
|
||||
await target.async_select_output(service.data[ATTR_HDMI_OUTPUT])
|
||||
|
||||
hass.services.async_register(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_SELECT_HDMI_OUTPUT,
|
||||
async_service_handle,
|
||||
schema=ONKYO_SELECT_OUTPUT_SCHEMA,
|
||||
)
|
||||
_cmds: dict[str, InputLibValue] = {
|
||||
k: v["name"]
|
||||
for k, v in {
|
||||
**PYEISCP_COMMANDS["main"]["SLI"]["values"],
|
||||
**PYEISCP_COMMANDS["zone2"]["SLZ"]["values"],
|
||||
}.items()
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_platform(
|
||||
@ -177,130 +143,170 @@ async def async_setup_platform(
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the Onkyo platform."""
|
||||
await async_register_services(hass)
|
||||
|
||||
receivers: dict[str, Receiver] = {} # indexed by host
|
||||
all_entities = hass.data.setdefault(DATA_MP_ENTITIES, [])
|
||||
|
||||
"""Import config from yaml."""
|
||||
host = config.get(CONF_HOST)
|
||||
name = config.get(CONF_NAME)
|
||||
max_volume = config[CONF_MAX_VOLUME]
|
||||
receiver_max_volume = config[CONF_RECEIVER_MAX_VOLUME]
|
||||
sources = config[CONF_SOURCES]
|
||||
|
||||
async def async_setup_receiver(
|
||||
info: ReceiverInfo, discovered: bool, name: str | None
|
||||
) -> None:
|
||||
entities: dict[str, OnkyoMediaPlayer] = {}
|
||||
all_entities.append(entities)
|
||||
source_mapping: dict[str, InputSource] = {}
|
||||
for value, source_lib in _cmds.items():
|
||||
try:
|
||||
source = InputSource(value)
|
||||
except ValueError:
|
||||
continue
|
||||
if isinstance(source_lib, str):
|
||||
source_mapping.setdefault(source_lib, source)
|
||||
else:
|
||||
for source_lib_single in source_lib:
|
||||
source_mapping.setdefault(source_lib_single, source)
|
||||
|
||||
@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
|
||||
)
|
||||
sources: dict[InputSource, str] = {}
|
||||
for source_lib_single, source_name in config[CONF_SOURCES].items():
|
||||
user_source = source_mapping.get(source_lib_single.lower())
|
||||
if user_source is not None:
|
||||
sources[user_source] = source_name
|
||||
|
||||
zone, _, value = message
|
||||
entity = entities.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
|
||||
)
|
||||
entities[zone] = zone_entity
|
||||
async_add_entities([zone_entity])
|
||||
|
||||
@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.conn.host
|
||||
)
|
||||
|
||||
for entity in entities.values():
|
||||
entity.backfill_state()
|
||||
|
||||
_LOGGER.debug("Creating receiver: %s (%s)", info.model_name, info.host)
|
||||
connection = await pyeiscp.Connection.create(
|
||||
host=info.host,
|
||||
port=info.port,
|
||||
update_callback=async_onkyo_update_callback,
|
||||
connect_callback=async_onkyo_connect_callback,
|
||||
)
|
||||
|
||||
receiver = Receiver(
|
||||
conn=connection,
|
||||
model_name=info.model_name,
|
||||
identifier=info.identifier,
|
||||
name=name or info.model_name,
|
||||
discovered=discovered,
|
||||
)
|
||||
|
||||
receivers[connection.host] = receiver
|
||||
|
||||
# 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.conn.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["main"] = main_entity
|
||||
async_add_entities([main_entity])
|
||||
config[CONF_SOURCES] = sources
|
||||
|
||||
results = []
|
||||
if host is not None:
|
||||
if host in KNOWN_HOSTS:
|
||||
return
|
||||
|
||||
_LOGGER.debug("Manually creating receiver: %s (%s)", name, host)
|
||||
|
||||
async def async_onkyo_interview_callback(conn: pyeiscp.Connection) -> None:
|
||||
"""Receiver interviewed, connection not yet active."""
|
||||
info = ReceiverInfo(conn.host, conn.port, conn.name, conn.identifier)
|
||||
_LOGGER.debug("Receiver interviewed: %s (%s)", info.model_name, info.host)
|
||||
if info.host not in KNOWN_HOSTS:
|
||||
KNOWN_HOSTS.append(info.host)
|
||||
await async_setup_receiver(info, False, name)
|
||||
|
||||
await pyeiscp.Connection.discover(
|
||||
host=host,
|
||||
discovery_callback=async_onkyo_interview_callback,
|
||||
_LOGGER.debug("Importing yaml single: %s", host)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data=config
|
||||
)
|
||||
results.append((host, result))
|
||||
else:
|
||||
_LOGGER.debug("Discovering receivers")
|
||||
for info in await async_discover():
|
||||
host = info.host
|
||||
|
||||
async def async_onkyo_discovery_callback(conn: pyeiscp.Connection) -> None:
|
||||
"""Receiver discovered, connection not yet active."""
|
||||
info = ReceiverInfo(conn.host, conn.port, conn.name, conn.identifier)
|
||||
_LOGGER.debug("Receiver discovered: %s (%s)", info.model_name, info.host)
|
||||
if info.host not in KNOWN_HOSTS:
|
||||
KNOWN_HOSTS.append(info.host)
|
||||
await async_setup_receiver(info, True, None)
|
||||
# Migrate legacy entities.
|
||||
registry = er.async_get(hass)
|
||||
old_unique_id = f"{info.model_name}_{info.identifier}"
|
||||
new_unique_id = f"{info.identifier}_main"
|
||||
entity_id = registry.async_get_entity_id(
|
||||
"media_player", DOMAIN, old_unique_id
|
||||
)
|
||||
if entity_id is not None:
|
||||
_LOGGER.debug(
|
||||
"Migrating unique_id from [%s] to [%s] for entity %s",
|
||||
old_unique_id,
|
||||
new_unique_id,
|
||||
entity_id,
|
||||
)
|
||||
registry.async_update_entity(entity_id, new_unique_id=new_unique_id)
|
||||
|
||||
await pyeiscp.Connection.discover(
|
||||
discovery_callback=async_onkyo_discovery_callback,
|
||||
_LOGGER.debug("Importing yaml discover: %s", info.host)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=config | {CONF_HOST: info.host} | {"info": info},
|
||||
)
|
||||
results.append((host, result))
|
||||
|
||||
_LOGGER.debug("Importing yaml results: %s", results)
|
||||
if not results:
|
||||
async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
"deprecated_yaml_import_issue_no_discover",
|
||||
breaks_in_ha_version="2025.5.0",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml_import_issue_no_discover",
|
||||
translation_placeholders={"url": ISSUE_URL_PLACEHOLDER},
|
||||
)
|
||||
|
||||
@callback
|
||||
def close_receiver(_event: Event) -> None:
|
||||
for receiver in receivers.values():
|
||||
receiver.conn.close()
|
||||
all_successful = True
|
||||
for host, result in results:
|
||||
if (
|
||||
result.get("type") == FlowResultType.CREATE_ENTRY
|
||||
or result.get("reason") == "already_configured"
|
||||
):
|
||||
continue
|
||||
if error := result.get("reason"):
|
||||
all_successful = False
|
||||
async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
f"deprecated_yaml_import_issue_{host}_{error}",
|
||||
breaks_in_ha_version="2025.5.0",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key=f"deprecated_yaml_import_issue_{error}",
|
||||
translation_placeholders={
|
||||
"host": host,
|
||||
"url": ISSUE_URL_PLACEHOLDER,
|
||||
},
|
||||
)
|
||||
|
||||
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, close_receiver)
|
||||
if all_successful:
|
||||
async_create_issue(
|
||||
hass,
|
||||
HOMEASSISTANT_DOMAIN,
|
||||
f"deprecated_yaml_{DOMAIN}",
|
||||
is_fixable=False,
|
||||
issue_domain=DOMAIN,
|
||||
breaks_in_ha_version="2025.5.0",
|
||||
severity=IssueSeverity.WARNING,
|
||||
translation_key="deprecated_yaml",
|
||||
translation_placeholders={
|
||||
"domain": DOMAIN,
|
||||
"integration_title": "onkyo",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: OnkyoConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up MediaPlayer for config entry."""
|
||||
data = entry.runtime_data
|
||||
|
||||
receiver = data.receiver
|
||||
all_entities = hass.data[DATA_MP_ENTITIES]
|
||||
|
||||
entities: dict[str, OnkyoMediaPlayer] = {}
|
||||
all_entities[entry.entry_id] = entities
|
||||
|
||||
volume_resolution: VolumeResolution = entry.options[OPTION_VOLUME_RESOLUTION]
|
||||
max_volume: float = entry.options[OPTION_MAX_VOLUME]
|
||||
sources = data.sources
|
||||
|
||||
def connect_callback(receiver: Receiver) -> None:
|
||||
if not receiver.first_connect:
|
||||
for entity in entities.values():
|
||||
if entity.enabled:
|
||||
entity.backfill_state()
|
||||
|
||||
def update_callback(receiver: Receiver, message: tuple[str, str, Any]) -> None:
|
||||
zone, _, value = message
|
||||
entity = entities.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 (%s)",
|
||||
ZONES[zone],
|
||||
receiver.model_name,
|
||||
receiver.host,
|
||||
)
|
||||
zone_entity = OnkyoMediaPlayer(
|
||||
receiver,
|
||||
zone,
|
||||
volume_resolution=volume_resolution,
|
||||
max_volume=max_volume,
|
||||
sources=sources,
|
||||
)
|
||||
entities[zone] = zone_entity
|
||||
async_add_entities([zone_entity])
|
||||
|
||||
receiver.callbacks.connect.append(connect_callback)
|
||||
receiver.callbacks.update.append(update_callback)
|
||||
|
||||
|
||||
class OnkyoMediaPlayer(MediaPlayerEntity):
|
||||
@ -316,27 +322,27 @@ class OnkyoMediaPlayer(MediaPlayerEntity):
|
||||
def __init__(
|
||||
self,
|
||||
receiver: Receiver,
|
||||
sources: dict[str, str],
|
||||
zone: str,
|
||||
max_volume: int,
|
||||
volume_resolution: int,
|
||||
*,
|
||||
volume_resolution: VolumeResolution,
|
||||
max_volume: float,
|
||||
sources: dict[InputSource, str],
|
||||
) -> None:
|
||||
"""Initialize the Onkyo Receiver."""
|
||||
self._receiver = receiver
|
||||
name = receiver.name
|
||||
name = receiver.model_name
|
||||
identifier = receiver.identifier
|
||||
self._attr_name = f"{name}{' ' + ZONES[zone] if zone != 'main' else ''}"
|
||||
if receiver.discovered and zone == "main":
|
||||
# keep legacy unique_id
|
||||
self._attr_unique_id = f"{name}_{identifier}"
|
||||
else:
|
||||
self._attr_unique_id = f"{identifier}_{zone}"
|
||||
self._attr_unique_id = f"{identifier}_{zone}"
|
||||
|
||||
self._zone = zone
|
||||
|
||||
self._volume_resolution = volume_resolution
|
||||
self._max_volume = max_volume
|
||||
|
||||
self._source_mapping = sources
|
||||
self._reverse_mapping = {value: key for key, value in sources.items()}
|
||||
self._max_volume = max_volume
|
||||
self._volume_resolution = volume_resolution
|
||||
self._lib_mapping = {_cmds[source.value]: source for source in InputSource}
|
||||
|
||||
self._attr_source_list = list(sources.values())
|
||||
self._attr_extra_state_attributes = {}
|
||||
@ -408,9 +414,13 @@ class OnkyoMediaPlayer(MediaPlayerEntity):
|
||||
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]
|
||||
source_lib = _cmds[self._reverse_mapping[source].value]
|
||||
if isinstance(source_lib, str):
|
||||
source_lib_single = source_lib
|
||||
else:
|
||||
source_lib_single = source_lib[0]
|
||||
self._update_receiver(
|
||||
"input-selector" if self._zone == "main" else "selector", source
|
||||
"input-selector" if self._zone == "main" else "selector", source_lib_single
|
||||
)
|
||||
|
||||
async def async_select_output(self, hdmi_output: str) -> None:
|
||||
@ -466,9 +476,10 @@ class OnkyoMediaPlayer(MediaPlayerEntity):
|
||||
elif command in ["volume", "master-volume"] and value != "N/A":
|
||||
self._supports_volume = True
|
||||
# AMP_VOL / (VOL_RESOLUTION * (MAX_VOL / 100))
|
||||
self._attr_volume_level = value / (
|
||||
volume_level: float = value / (
|
||||
self._volume_resolution * self._max_volume / 100
|
||||
)
|
||||
self._attr_volume_level = min(1, volume_level)
|
||||
elif command in ["muting", "audio-muting"]:
|
||||
self._attr_is_volume_muted = bool(value == "on")
|
||||
elif command in ["selector", "input-selector"]:
|
||||
@ -493,18 +504,17 @@ class OnkyoMediaPlayer(MediaPlayerEntity):
|
||||
self.async_write_ha_state()
|
||||
|
||||
@callback
|
||||
def _parse_source(self, source_raw: str | int | tuple[str]) -> None:
|
||||
# source is either a tuple of values or a single value,
|
||||
# so we convert to a tuple, when it is a single value.
|
||||
if isinstance(source_raw, str | int):
|
||||
source = (str(source_raw),)
|
||||
else:
|
||||
source = source_raw
|
||||
for value in source:
|
||||
if value in self._source_mapping:
|
||||
self._attr_source = self._source_mapping[value]
|
||||
return
|
||||
self._attr_source = "_".join(source)
|
||||
def _parse_source(self, source_lib: InputLibValue) -> None:
|
||||
source = self._lib_mapping[source_lib]
|
||||
if source in self._source_mapping:
|
||||
self._attr_source = self._source_mapping[source]
|
||||
return
|
||||
|
||||
source_meaning = source.value_meaning
|
||||
_LOGGER.error(
|
||||
'Input source "%s" not in source list: %s', source_meaning, self.entity_id
|
||||
)
|
||||
self._attr_source = source_meaning
|
||||
|
||||
@callback
|
||||
def _parse_audio_information(
|
||||
|
||||
@ -2,10 +2,29 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import asyncio
|
||||
from collections.abc import Callable, Iterable
|
||||
import contextlib
|
||||
from dataclasses import dataclass, field
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import pyeiscp
|
||||
|
||||
from .const import DEVICE_DISCOVERY_TIMEOUT, DEVICE_INTERVIEW_TIMEOUT, ZONES
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Callbacks:
|
||||
"""Onkyo Receiver Callbacks."""
|
||||
|
||||
connect: list[Callable[[Receiver], None]] = field(default_factory=list)
|
||||
update: list[Callable[[Receiver, tuple[str, str, Any]], None]] = field(
|
||||
default_factory=list
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Receiver:
|
||||
@ -14,8 +33,62 @@ class Receiver:
|
||||
conn: pyeiscp.Connection
|
||||
model_name: str
|
||||
identifier: str
|
||||
name: str
|
||||
discovered: bool
|
||||
host: str
|
||||
first_connect: bool = True
|
||||
callbacks: Callbacks = field(default_factory=Callbacks)
|
||||
|
||||
@classmethod
|
||||
async def async_create(cls, info: ReceiverInfo) -> Receiver:
|
||||
"""Set up Onkyo Receiver."""
|
||||
|
||||
receiver: Receiver | None = None
|
||||
|
||||
def on_connect(_origin: str) -> None:
|
||||
assert receiver is not None
|
||||
receiver.on_connect()
|
||||
|
||||
def on_update(message: tuple[str, str, Any], _origin: str) -> None:
|
||||
assert receiver is not None
|
||||
receiver.on_update(message)
|
||||
|
||||
_LOGGER.debug("Creating receiver: %s (%s)", info.model_name, info.host)
|
||||
|
||||
connection = await pyeiscp.Connection.create(
|
||||
host=info.host,
|
||||
port=info.port,
|
||||
connect_callback=on_connect,
|
||||
update_callback=on_update,
|
||||
auto_connect=False,
|
||||
)
|
||||
|
||||
return (
|
||||
receiver := cls(
|
||||
conn=connection,
|
||||
model_name=info.model_name,
|
||||
identifier=info.identifier,
|
||||
host=info.host,
|
||||
)
|
||||
)
|
||||
|
||||
def on_connect(self) -> None:
|
||||
"""Receiver (re)connected."""
|
||||
_LOGGER.debug("Receiver (re)connected: %s (%s)", self.model_name, self.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:
|
||||
self.conn.query_property(zone, "power")
|
||||
|
||||
for callback in self.callbacks.connect:
|
||||
callback(self)
|
||||
|
||||
self.first_connect = False
|
||||
|
||||
def on_update(self, message: tuple[str, str, Any]) -> None:
|
||||
"""Process new message from the receiver."""
|
||||
_LOGGER.debug("Received update callback from %s: %s", self.model_name, message)
|
||||
for callback in self.callbacks.update:
|
||||
callback(self, message)
|
||||
|
||||
|
||||
@dataclass
|
||||
@ -26,3 +99,53 @@ class ReceiverInfo:
|
||||
port: int
|
||||
model_name: str
|
||||
identifier: str
|
||||
|
||||
|
||||
async def async_interview(host: str) -> ReceiverInfo | None:
|
||||
"""Interview Onkyo Receiver."""
|
||||
_LOGGER.debug("Interviewing receiver: %s", host)
|
||||
|
||||
receiver_info: ReceiverInfo | None = None
|
||||
|
||||
event = asyncio.Event()
|
||||
|
||||
async def _callback(conn: pyeiscp.Connection) -> None:
|
||||
"""Receiver interviewed, connection not yet active."""
|
||||
nonlocal receiver_info
|
||||
if receiver_info is None:
|
||||
info = ReceiverInfo(host, conn.port, conn.name, conn.identifier)
|
||||
_LOGGER.debug("Receiver interviewed: %s (%s)", info.model_name, info.host)
|
||||
receiver_info = info
|
||||
event.set()
|
||||
|
||||
timeout = DEVICE_INTERVIEW_TIMEOUT
|
||||
|
||||
await pyeiscp.Connection.discover(
|
||||
host=host, discovery_callback=_callback, timeout=timeout
|
||||
)
|
||||
|
||||
with contextlib.suppress(asyncio.TimeoutError):
|
||||
await asyncio.wait_for(event.wait(), timeout)
|
||||
|
||||
return receiver_info
|
||||
|
||||
|
||||
async def async_discover() -> Iterable[ReceiverInfo]:
|
||||
"""Discover Onkyo Receivers."""
|
||||
_LOGGER.debug("Discovering receivers")
|
||||
|
||||
receiver_infos: list[ReceiverInfo] = []
|
||||
|
||||
async def _callback(conn: pyeiscp.Connection) -> None:
|
||||
"""Receiver discovered, connection not yet active."""
|
||||
info = ReceiverInfo(conn.host, conn.port, conn.name, conn.identifier)
|
||||
_LOGGER.debug("Receiver discovered: %s (%s)", info.model_name, info.host)
|
||||
receiver_infos.append(info)
|
||||
|
||||
timeout = DEVICE_DISCOVERY_TIMEOUT
|
||||
|
||||
await pyeiscp.Connection.discover(discovery_callback=_callback, timeout=timeout)
|
||||
|
||||
await asyncio.sleep(timeout)
|
||||
|
||||
return receiver_infos
|
||||
|
||||
69
homeassistant/components/onkyo/services.py
Normal file
69
homeassistant/components/onkyo/services.py
Normal file
@ -0,0 +1,69 @@
|
||||
"""Onkyo services."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.media_player import DOMAIN as MEDIA_PLAYER_DOMAIN
|
||||
from homeassistant.const import ATTR_ENTITY_ID
|
||||
from homeassistant.core import HomeAssistant, ServiceCall
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.util.hass_dict import HassKey
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .media_player import OnkyoMediaPlayer
|
||||
|
||||
DATA_MP_ENTITIES: HassKey[dict[str, dict[str, OnkyoMediaPlayer]]] = HassKey(DOMAIN)
|
||||
|
||||
ATTR_HDMI_OUTPUT = "hdmi_output"
|
||||
ACCEPTED_VALUES = [
|
||||
"no",
|
||||
"analog",
|
||||
"yes",
|
||||
"out",
|
||||
"out-sub",
|
||||
"sub",
|
||||
"hdbaset",
|
||||
"both",
|
||||
"up",
|
||||
]
|
||||
ONKYO_SELECT_OUTPUT_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Required(ATTR_ENTITY_ID): cv.entity_ids,
|
||||
vol.Required(ATTR_HDMI_OUTPUT): vol.In(ACCEPTED_VALUES),
|
||||
}
|
||||
)
|
||||
SERVICE_SELECT_HDMI_OUTPUT = "onkyo_select_hdmi_output"
|
||||
|
||||
|
||||
async def async_register_services(hass: HomeAssistant) -> None:
|
||||
"""Register Onkyo services."""
|
||||
|
||||
hass.data.setdefault(DATA_MP_ENTITIES, {})
|
||||
|
||||
async def async_service_handle(service: ServiceCall) -> None:
|
||||
"""Handle for services."""
|
||||
entity_ids = service.data[ATTR_ENTITY_ID]
|
||||
|
||||
targets: list[OnkyoMediaPlayer] = []
|
||||
for receiver_entities in hass.data[DATA_MP_ENTITIES].values():
|
||||
targets.extend(
|
||||
entity
|
||||
for entity in receiver_entities.values()
|
||||
if entity.entity_id in entity_ids
|
||||
)
|
||||
|
||||
for target in targets:
|
||||
if service.service == SERVICE_SELECT_HDMI_OUTPUT:
|
||||
await target.async_select_output(service.data[ATTR_HDMI_OUTPUT])
|
||||
|
||||
hass.services.async_register(
|
||||
MEDIA_PLAYER_DOMAIN,
|
||||
SERVICE_SELECT_HDMI_OUTPUT,
|
||||
async_service_handle,
|
||||
schema=ONKYO_SELECT_OUTPUT_SCHEMA,
|
||||
)
|
||||
58
homeassistant/components/onkyo/strings.json
Normal file
58
homeassistant/components/onkyo/strings.json
Normal file
@ -0,0 +1,58 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"menu_options": {
|
||||
"manual": "Manual entry",
|
||||
"eiscp_discovery": "Onkyo discovery"
|
||||
}
|
||||
},
|
||||
"manual": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]"
|
||||
}
|
||||
},
|
||||
"eiscp_discovery": {
|
||||
"data": {
|
||||
"device": "[%key:common::config_flow::data::device%]"
|
||||
}
|
||||
},
|
||||
"configure_receiver": {
|
||||
"description": "Configure {name}",
|
||||
"data": {
|
||||
"volume_resolution": "Number of steps it takes for the receiver to go from the lowest to the highest possible volume",
|
||||
"input_sources": "List of input sources supported by the receiver"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"empty_input_source_list": "Input source list cannot be empty",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"data": {
|
||||
"max_volume": "Maximum volume limit (%)"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"deprecated_yaml_import_issue_no_discover": {
|
||||
"title": "The Onkyo YAML configuration import failed",
|
||||
"description": "Configuring Onkyo using YAML is being removed but no receivers were discovered when importing your YAML configuration.\n\nEnsure the connection to the receiver works and restart Home Assistant to try again or remove the Onkyo YAML configuration from your configuration.yaml file and continue to [set up the integration]({url}) manually."
|
||||
},
|
||||
"deprecated_yaml_import_issue_cannot_connect": {
|
||||
"title": "The Onkyo YAML configuration import failed",
|
||||
"description": "Configuring Onkyo using YAML is being removed but there was a connection error when importing your YAML configuration for host {host}.\n\nEnsure the connection to the receiver works and restart Home Assistant to try again or remove the Onkyo YAML configuration from your configuration.yaml file and continue to [set up the integration]({url}) manually."
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -418,6 +418,7 @@ FLOWS = {
|
||||
"oncue",
|
||||
"ondilo_ico",
|
||||
"onewire",
|
||||
"onkyo",
|
||||
"onvif",
|
||||
"open_meteo",
|
||||
"openai_conversation",
|
||||
|
||||
@ -4319,8 +4319,8 @@
|
||||
},
|
||||
"onkyo": {
|
||||
"name": "Onkyo",
|
||||
"integration_type": "hub",
|
||||
"config_flow": false,
|
||||
"integration_type": "device",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_push"
|
||||
},
|
||||
"onvif": {
|
||||
|
||||
@ -1511,6 +1511,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
|
||||
|
||||
|
||||
60
tests/components/onkyo/__init__.py
Normal file
60
tests/components/onkyo/__init__.py
Normal file
@ -0,0 +1,60 @@
|
||||
"""Tests for the Onkyo integration."""
|
||||
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
from homeassistant.components.onkyo.receiver import Receiver, ReceiverInfo
|
||||
from homeassistant.const import CONF_HOST
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
def create_receiver_info(id: int) -> ReceiverInfo:
|
||||
"""Create an empty receiver info object for testing."""
|
||||
return ReceiverInfo(
|
||||
host=f"host {id}",
|
||||
port=id,
|
||||
model_name=f"type {id}",
|
||||
identifier=f"id{id}",
|
||||
)
|
||||
|
||||
|
||||
def create_empty_config_entry() -> MockConfigEntry:
|
||||
"""Create an empty config entry for use in unit tests."""
|
||||
config = {CONF_HOST: ""}
|
||||
options = {
|
||||
"volume_resolution": 80,
|
||||
"input_sources": {"12": "tv"},
|
||||
"max_volume": 100,
|
||||
}
|
||||
|
||||
return MockConfigEntry(
|
||||
data=config,
|
||||
options=options,
|
||||
title="Unit test Onkyo",
|
||||
domain="onkyo",
|
||||
unique_id="onkyo_unique_id",
|
||||
)
|
||||
|
||||
|
||||
async def setup_integration(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry, receiver_info: ReceiverInfo
|
||||
) -> None:
|
||||
"""Fixture for setting up the component."""
|
||||
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
mock_receiver = AsyncMock()
|
||||
mock_receiver.conn.close = Mock()
|
||||
mock_receiver.callbacks.connect = Mock()
|
||||
mock_receiver.callbacks.update = Mock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.onkyo.async_interview",
|
||||
return_value=receiver_info,
|
||||
),
|
||||
patch.object(Receiver, "async_create", return_value=mock_receiver),
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
30
tests/components/onkyo/conftest.py
Normal file
30
tests/components/onkyo/conftest.py
Normal file
@ -0,0 +1,30 @@
|
||||
"""Configure tests for the Onkyo integration."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.onkyo.const import DOMAIN
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Override async_setup_entry."""
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture(name="config_entry")
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
"""Create Onkyo entry in Home Assistant."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="Onkyo",
|
||||
data={},
|
||||
)
|
||||
459
tests/components/onkyo/test_config_flow.py
Normal file
459
tests/components/onkyo/test_config_flow.py
Normal file
@ -0,0 +1,459 @@
|
||||
"""Test Onkyo config flow."""
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.onkyo import InputSource
|
||||
from homeassistant.components.onkyo.config_flow import OnkyoConfigFlow
|
||||
from homeassistant.components.onkyo.const import DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_USER
|
||||
from homeassistant.const import CONF_HOST
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType, InvalidData
|
||||
|
||||
from . import create_empty_config_entry, create_receiver_info, setup_integration
|
||||
|
||||
from tests.common import Mock, MockConfigEntry
|
||||
|
||||
|
||||
async def test_user_initial_menu(hass: HomeAssistant) -> None:
|
||||
"""Test initial menu."""
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
assert init_result["type"] is FlowResultType.MENU
|
||||
# Check if the values are there, but ignore order
|
||||
assert not set(init_result["menu_options"]) ^ {"manual", "eiscp_discovery"}
|
||||
|
||||
|
||||
async def test_manual_valid_host(hass: HomeAssistant) -> None:
|
||||
"""Test valid host entered."""
|
||||
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "manual"},
|
||||
)
|
||||
|
||||
mock_info = Mock()
|
||||
mock_info.identifier = "mock_id"
|
||||
mock_info.host = "mock_host"
|
||||
mock_info.model_name = "mock_model"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_interview",
|
||||
return_value=mock_info,
|
||||
):
|
||||
select_result = await hass.config_entries.flow.async_configure(
|
||||
form_result["flow_id"],
|
||||
user_input={CONF_HOST: "sample-host-name"},
|
||||
)
|
||||
|
||||
assert select_result["step_id"] == "configure_receiver"
|
||||
assert (
|
||||
select_result["description_placeholders"]["name"]
|
||||
== "mock_model (mock_host)"
|
||||
)
|
||||
|
||||
|
||||
async def test_manual_invalid_host(hass: HomeAssistant) -> None:
|
||||
"""Test invalid host entered."""
|
||||
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "manual"},
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_interview", return_value=None
|
||||
):
|
||||
host_result = await hass.config_entries.flow.async_configure(
|
||||
form_result["flow_id"],
|
||||
user_input={CONF_HOST: "sample-host-name"},
|
||||
)
|
||||
|
||||
assert host_result["step_id"] == "manual"
|
||||
assert host_result["errors"]["base"] == "cannot_connect"
|
||||
|
||||
|
||||
async def test_manual_valid_host_unexpected_error(hass: HomeAssistant) -> None:
|
||||
"""Test valid host entered."""
|
||||
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "manual"},
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_interview",
|
||||
side_effect=Exception(),
|
||||
):
|
||||
host_result = await hass.config_entries.flow.async_configure(
|
||||
form_result["flow_id"],
|
||||
user_input={CONF_HOST: "sample-host-name"},
|
||||
)
|
||||
|
||||
assert host_result["step_id"] == "manual"
|
||||
assert host_result["errors"]["base"] == "unknown"
|
||||
|
||||
|
||||
async def test_discovery_and_no_devices_discovered(hass: HomeAssistant) -> None:
|
||||
"""Test initial menu."""
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_discover", return_value=[]
|
||||
):
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "eiscp_discovery"},
|
||||
)
|
||||
|
||||
assert form_result["type"] is FlowResultType.ABORT
|
||||
assert form_result["reason"] == "no_devices_found"
|
||||
|
||||
|
||||
async def test_discovery_with_exception(hass: HomeAssistant) -> None:
|
||||
"""Test discovery which throws an unexpected exception."""
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_discover",
|
||||
side_effect=Exception(),
|
||||
):
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "eiscp_discovery"},
|
||||
)
|
||||
|
||||
assert form_result["type"] is FlowResultType.ABORT
|
||||
assert form_result["reason"] == "unknown"
|
||||
|
||||
|
||||
async def test_discovery_with_new_and_existing_found(hass: HomeAssistant) -> None:
|
||||
"""Test discovery with a new and an existing entry."""
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
infos = [create_receiver_info(1), create_receiver_info(2)]
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_discover",
|
||||
return_value=infos,
|
||||
),
|
||||
# Fake it like the first entry was already added
|
||||
patch.object(OnkyoConfigFlow, "_async_current_ids", return_value=["id1"]),
|
||||
):
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "eiscp_discovery"},
|
||||
)
|
||||
|
||||
assert form_result["type"] is FlowResultType.FORM
|
||||
|
||||
assert form_result["data_schema"] is not None
|
||||
schema = form_result["data_schema"].schema
|
||||
container = schema["device"].container
|
||||
assert container == {"id2": "type 2 (host 2)"}
|
||||
|
||||
|
||||
async def test_discovery_with_one_selected(hass: HomeAssistant) -> None:
|
||||
"""Test discovery after a selection."""
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
infos = [create_receiver_info(42), create_receiver_info(0)]
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_discover",
|
||||
return_value=infos,
|
||||
),
|
||||
):
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "eiscp_discovery"},
|
||||
)
|
||||
|
||||
select_result = await hass.config_entries.flow.async_configure(
|
||||
form_result["flow_id"],
|
||||
user_input={"device": "id42"},
|
||||
)
|
||||
|
||||
assert select_result["step_id"] == "configure_receiver"
|
||||
assert select_result["description_placeholders"]["name"] == "type 42 (host 42)"
|
||||
|
||||
|
||||
async def test_configure_empty_source_list(hass: HomeAssistant) -> None:
|
||||
"""Test receiver configuration with no sources set."""
|
||||
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "manual"},
|
||||
)
|
||||
|
||||
mock_info = Mock()
|
||||
mock_info.identifier = "mock_id"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_interview",
|
||||
return_value=mock_info,
|
||||
):
|
||||
select_result = await hass.config_entries.flow.async_configure(
|
||||
form_result["flow_id"],
|
||||
user_input={CONF_HOST: "sample-host-name"},
|
||||
)
|
||||
|
||||
configure_result = await hass.config_entries.flow.async_configure(
|
||||
select_result["flow_id"],
|
||||
user_input={"input_sources": []},
|
||||
)
|
||||
|
||||
assert configure_result["errors"] == {
|
||||
"input_sources": "empty_input_source_list"
|
||||
}
|
||||
|
||||
|
||||
async def test_configure_no_resolution(hass: HomeAssistant) -> None:
|
||||
"""Test receiver configure with no resolution set."""
|
||||
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "manual"},
|
||||
)
|
||||
|
||||
mock_info = Mock()
|
||||
mock_info.identifier = "mock_id"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_interview",
|
||||
return_value=mock_info,
|
||||
):
|
||||
select_result = await hass.config_entries.flow.async_configure(
|
||||
form_result["flow_id"],
|
||||
user_input={CONF_HOST: "sample-host-name"},
|
||||
)
|
||||
|
||||
configure_result = await hass.config_entries.flow.async_configure(
|
||||
select_result["flow_id"],
|
||||
user_input={"input_sources": ["TV"]},
|
||||
)
|
||||
|
||||
assert configure_result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert configure_result["options"]["volume_resolution"] == 50
|
||||
|
||||
|
||||
async def test_configure_resolution_set(hass: HomeAssistant) -> None:
|
||||
"""Test receiver configure with specified resolution."""
|
||||
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "manual"},
|
||||
)
|
||||
|
||||
mock_info = Mock()
|
||||
mock_info.identifier = "mock_id"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_interview",
|
||||
return_value=mock_info,
|
||||
):
|
||||
select_result = await hass.config_entries.flow.async_configure(
|
||||
form_result["flow_id"],
|
||||
user_input={CONF_HOST: "sample-host-name"},
|
||||
)
|
||||
|
||||
configure_result = await hass.config_entries.flow.async_configure(
|
||||
select_result["flow_id"],
|
||||
user_input={"volume_resolution": 200, "input_sources": ["TV"]},
|
||||
)
|
||||
|
||||
assert configure_result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert configure_result["options"]["volume_resolution"] == 200
|
||||
|
||||
|
||||
async def test_configure_invalid_resolution_set(hass: HomeAssistant) -> None:
|
||||
"""Test receiver configure with invalid resolution."""
|
||||
|
||||
init_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_USER},
|
||||
)
|
||||
|
||||
form_result = await hass.config_entries.flow.async_configure(
|
||||
init_result["flow_id"],
|
||||
{"next_step_id": "manual"},
|
||||
)
|
||||
|
||||
mock_info = Mock()
|
||||
mock_info.identifier = "mock_id"
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_interview",
|
||||
return_value=mock_info,
|
||||
):
|
||||
select_result = await hass.config_entries.flow.async_configure(
|
||||
form_result["flow_id"],
|
||||
user_input={CONF_HOST: "sample-host-name"},
|
||||
)
|
||||
|
||||
with pytest.raises(InvalidData):
|
||||
await hass.config_entries.flow.async_configure(
|
||||
select_result["flow_id"],
|
||||
user_input={"volume_resolution": 42, "input_sources": ["TV"]},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("user_input", "exception", "error"),
|
||||
[
|
||||
(
|
||||
# No host, and thus no host reachable
|
||||
{
|
||||
CONF_HOST: None,
|
||||
"receiver_max_volume": 100,
|
||||
"max_volume": 100,
|
||||
"sources": {},
|
||||
},
|
||||
None,
|
||||
"cannot_connect",
|
||||
),
|
||||
(
|
||||
# No host, and connection exception
|
||||
{
|
||||
CONF_HOST: None,
|
||||
"receiver_max_volume": 100,
|
||||
"max_volume": 100,
|
||||
"sources": {},
|
||||
},
|
||||
Exception(),
|
||||
"cannot_connect",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_import_fail(
|
||||
hass: HomeAssistant,
|
||||
user_input: dict[str, Any],
|
||||
exception: Exception,
|
||||
error: str,
|
||||
) -> None:
|
||||
"""Test import flow failed."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.onkyo.config_flow.async_interview",
|
||||
return_value=None,
|
||||
side_effect=exception,
|
||||
),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=user_input
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == error
|
||||
|
||||
|
||||
async def test_import_success(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test import flow succeeded."""
|
||||
info = create_receiver_info(1)
|
||||
|
||||
user_input = {
|
||||
CONF_HOST: info.host,
|
||||
"receiver_max_volume": 80,
|
||||
"max_volume": 110,
|
||||
"sources": {
|
||||
InputSource("00"): "Auxiliary",
|
||||
InputSource("01"): "Video",
|
||||
},
|
||||
"info": info,
|
||||
}
|
||||
|
||||
import_result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_IMPORT}, data=user_input
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert import_result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert import_result["data"]["host"] == "host 1"
|
||||
assert import_result["options"]["volume_resolution"] == 80
|
||||
assert import_result["options"]["max_volume"] == 100
|
||||
assert import_result["options"]["input_sources"] == {
|
||||
"00": "Auxiliary",
|
||||
"01": "Video",
|
||||
}
|
||||
|
||||
|
||||
async def test_options_flow(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
|
||||
"""Test options flow."""
|
||||
|
||||
receiver_info = create_receiver_info(1)
|
||||
config_entry = create_empty_config_entry()
|
||||
await setup_integration(hass, config_entry, receiver_info)
|
||||
|
||||
result = await hass.config_entries.options.async_init(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
"max_volume": 42,
|
||||
"TV": "television",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["data"] == {
|
||||
"volume_resolution": 80,
|
||||
"max_volume": 42.0,
|
||||
"input_sources": {
|
||||
"12": "television",
|
||||
},
|
||||
}
|
||||
72
tests/components/onkyo/test_init.py
Normal file
72
tests/components/onkyo/test_init.py
Normal file
@ -0,0 +1,72 @@
|
||||
"""Test Onkyo component setup process."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.onkyo import async_setup_entry
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
|
||||
from . import create_empty_config_entry, create_receiver_info, setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_load_unload_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test load and unload entry."""
|
||||
|
||||
config_entry = create_empty_config_entry()
|
||||
receiver_info = create_receiver_info(1)
|
||||
await setup_integration(hass, config_entry, receiver_info)
|
||||
|
||||
assert config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_update_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test update options."""
|
||||
|
||||
with patch.object(hass.config_entries, "async_reload", return_value=True):
|
||||
config_entry = create_empty_config_entry()
|
||||
receiver_info = create_receiver_info(1)
|
||||
await setup_integration(hass, config_entry, receiver_info)
|
||||
|
||||
# Force option change
|
||||
assert hass.config_entries.async_update_entry(
|
||||
config_entry, options={"option": "new_value"}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
hass.config_entries.async_reload.assert_called_with(config_entry.entry_id)
|
||||
|
||||
|
||||
async def test_no_connection(
|
||||
hass: HomeAssistant,
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test update options."""
|
||||
|
||||
config_entry = create_empty_config_entry()
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.onkyo.async_interview",
|
||||
return_value=None,
|
||||
),
|
||||
pytest.raises(ConfigEntryNotReady),
|
||||
):
|
||||
await async_setup_entry(hass, config_entry)
|
||||
Loading…
x
Reference in New Issue
Block a user