jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reorganize input sources in Onkyo options (#133511)

Browse Source
This commit is contained in:
Artur Pragacz 2025-01-24 15:45:53 +01:00 committed by GitHub
parent 51bc56929b
commit fc9ad40ac8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 215 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

209
homeassistant/components/onkyo/config_flow.py
View File

@ -15,6 +15,7 @@ from homeassistant.config_entries import (
)
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import callback
from homeassistant.data_entry_flow import section
from homeassistant.helpers.selector import (
NumberSelector,
NumberSelectorConfig,
@ -49,9 +50,13 @@ INPUT_SOURCES_ALL_MEANINGS = [
input_source.value_meaning for input_source in InputSource
]
STEP_MANUAL_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str})
STEP_CONFIGURE_SCHEMA = vol.Schema(
STEP_RECONFIGURE_SCHEMA = vol.Schema(
{
vol.Required(OPTION_VOLUME_RESOLUTION): vol.In(VOLUME_RESOLUTION_ALLOWED),
}
)
STEP_CONFIGURE_SCHEMA = STEP_RECONFIGURE_SCHEMA.extend(
{
vol.Required(OPTION_INPUT_SOURCES): SelectSelector(
SelectSelectorConfig(
options=INPUT_SOURCES_ALL_MEANINGS,
@ -216,55 +221,52 @@ class OnkyoConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle the configuration of a single receiver."""
errors = {}
entry = None
entry_options = None
reconfigure_entry = None
schema = STEP_CONFIGURE_SCHEMA
if self.source == SOURCE_RECONFIGURE:
entry = self._get_reconfigure_entry()
entry_options = entry.options
schema = STEP_RECONFIGURE_SCHEMA
reconfigure_entry = self._get_reconfigure_entry()
if user_input is not None:
source_meanings: list[str] = user_input[OPTION_INPUT_SOURCES]
if not source_meanings:
volume_resolution = user_input[OPTION_VOLUME_RESOLUTION]
if reconfigure_entry is not None:
entry_options = reconfigure_entry.options
result = self.async_update_reload_and_abort(
reconfigure_entry,
data={
CONF_HOST: self._receiver_info.host,
},
options={
OPTION_VOLUME_RESOLUTION: volume_resolution,
OPTION_MAX_VOLUME: entry_options[OPTION_MAX_VOLUME],
OPTION_INPUT_SOURCES: entry_options[OPTION_INPUT_SOURCES],
},
)
_LOGGER.debug("Reconfigured receiver, result: %s", result)
return result
input_source_meanings: list[str] = user_input[OPTION_INPUT_SOURCES]
if not input_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)
input_sources_store: dict[str, str] = {}
for input_source_meaning in input_source_meanings:
input_source = InputSource.from_meaning(input_source_meaning)
input_sources_store[input_source.value] = input_source_meaning
source_name = source_meaning
if entry_options is not None:
source_name = entry_options[OPTION_INPUT_SOURCES].get(
source.value, source_name
)
sources_store[source.value] = source_name
volume_resolution = user_input[OPTION_VOLUME_RESOLUTION]
if entry_options is None:
result = self.async_create_entry(
title=self._receiver_info.model_name,
data={
CONF_HOST: self._receiver_info.host,
},
options={
OPTION_VOLUME_RESOLUTION: volume_resolution,
OPTION_MAX_VOLUME: OPTION_MAX_VOLUME_DEFAULT,
OPTION_INPUT_SOURCES: sources_store,
},
)
else:
assert entry is not None
result = self.async_update_reload_and_abort(
entry,
data={
CONF_HOST: self._receiver_info.host,
},
options={
OPTION_VOLUME_RESOLUTION: volume_resolution,
OPTION_MAX_VOLUME: entry_options[OPTION_MAX_VOLUME],
OPTION_INPUT_SOURCES: sources_store,
},
)
result = self.async_create_entry(
title=self._receiver_info.model_name,
data={
CONF_HOST: self._receiver_info.host,
},
options={
OPTION_VOLUME_RESOLUTION: volume_resolution,
OPTION_MAX_VOLUME: OPTION_MAX_VOLUME_DEFAULT,
OPTION_INPUT_SOURCES: input_sources_store,
},
)
_LOGGER.debug("Configured receiver, result: %s", result)
return result
@ -273,12 +275,13 @@ class OnkyoConfigFlow(ConfigFlow, domain=DOMAIN):
suggested_values = user_input
if suggested_values is None:
if entry_options is None:
if reconfigure_entry is None:
suggested_values = {
OPTION_VOLUME_RESOLUTION: OPTION_VOLUME_RESOLUTION_DEFAULT,
OPTION_INPUT_SOURCES: [],
}
else:
entry_options = reconfigure_entry.options
suggested_values = {
OPTION_VOLUME_RESOLUTION: entry_options[OPTION_VOLUME_RESOLUTION],
OPTION_INPUT_SOURCES: [
@ -289,9 +292,7 @@ class OnkyoConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="configure_receiver",
data_schema=self.add_suggested_values_to_schema(
STEP_CONFIGURE_SCHEMA, suggested_values
),
data_schema=self.add_suggested_values_to_schema(schema, suggested_values),
errors=errors,
description_placeholders={
"name": f"{self._receiver_info.model_name} ({self._receiver_info.host})"
@ -360,57 +361,107 @@ class OnkyoConfigFlow(ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> OptionsFlow:
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
"""Return the options flow."""
return OnkyoOptionsFlowHandler(config_entry)
return OnkyoOptionsFlowHandler()
OPTIONS_STEP_INIT_SCHEMA = vol.Schema(
{
vol.Required(OPTION_MAX_VOLUME): NumberSelector(
NumberSelectorConfig(min=1, max=100, mode=NumberSelectorMode.BOX)
),
vol.Required(OPTION_INPUT_SOURCES): SelectSelector(
SelectSelectorConfig(
options=INPUT_SOURCES_ALL_MEANINGS,
multiple=True,
mode=SelectSelectorMode.DROPDOWN,
)
),
}
)
class OnkyoOptionsFlowHandler(OptionsFlow):
"""Handle an options flow for Onkyo."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize options flow."""
sources_store: dict[str, str] = config_entry.options[OPTION_INPUT_SOURCES]
self._input_sources = {InputSource(k): v for k, v in sources_store.items()}
_data: dict[str, Any]
_input_sources: dict[InputSource, str]
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options."""
errors = {}
entry_options = self.config_entry.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
self._input_sources = {}
for input_source_meaning in user_input[OPTION_INPUT_SOURCES]:
input_source = InputSource.from_meaning(input_source_meaning)
input_source_name = entry_options[OPTION_INPUT_SOURCES].get(
input_source.value, input_source_meaning
)
self._input_sources[input_source] = input_source_name
if not self._input_sources:
errors[OPTION_INPUT_SOURCES] = "empty_input_source_list"
else:
self._data = {
OPTION_VOLUME_RESOLUTION: entry_options[OPTION_VOLUME_RESOLUTION],
OPTION_MAX_VOLUME: user_input[OPTION_MAX_VOLUME],
}
return await self.async_step_names()
suggested_values = user_input
if suggested_values is None:
suggested_values = {
OPTION_MAX_VOLUME: entry_options[OPTION_MAX_VOLUME],
OPTION_INPUT_SOURCES: [
InputSource(input_source).value_meaning
for input_source in entry_options[OPTION_INPUT_SOURCES]
],
}
return self.async_show_form(
step_id="init",
data_schema=self.add_suggested_values_to_schema(
OPTIONS_STEP_INIT_SCHEMA, suggested_values
),
errors=errors,
)
async def async_step_names(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Configure names."""
if user_input is not None:
input_sources_store: dict[str, str] = {}
for input_source_meaning, input_source_name in user_input[
"input_sources"
].items():
input_source = InputSource.from_meaning(input_source_meaning)
input_sources_store[input_source.value] = input_source_name
return self.async_create_entry(
data={
OPTION_VOLUME_RESOLUTION: self.config_entry.options[
OPTION_VOLUME_RESOLUTION
],
OPTION_MAX_VOLUME: user_input[OPTION_MAX_VOLUME],
OPTION_INPUT_SOURCES: sources_store,
**self._data,
OPTION_INPUT_SOURCES: input_sources_store,
}
)
schema_dict: dict[Any, Selector] = {}
max_volume: float = self.config_entry.options[OPTION_MAX_VOLUME]
schema_dict[vol.Required(OPTION_MAX_VOLUME, default=max_volume)] = (
NumberSelector(
NumberSelectorConfig(min=1, max=100, mode=NumberSelectorMode.BOX)
)
)
for source, source_name in self._input_sources.items():
schema_dict[vol.Required(source.value_meaning, default=source_name)] = (
TextSelector()
)
for input_source, input_source_name in self._input_sources.items():
schema_dict[
vol.Required(input_source.value_meaning, default=input_source_name)
] = TextSelector()
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(schema_dict),
step_id="names",
data_schema=vol.Schema(
{vol.Required("input_sources"): section(vol.Schema(schema_dict))}
),
)

23
homeassistant/components/onkyo/strings.json
View File

@ -27,17 +27,17 @@
"description": "Configure {name}",
"data": {
"volume_resolution": "Volume resolution",
"input_sources": "Input sources"
"input_sources": "[%key:component::onkyo::options::step::init::data::input_sources%]"
},
"data_description": {
"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."
"input_sources": "[%key:component::onkyo::options::step::init::data_description::input_sources%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"empty_input_source_list": "Input source list cannot be empty",
"empty_input_source_list": "[%key:component::onkyo::options::error::empty_input_source_list%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
@ -52,12 +52,25 @@
"step": {
"init": {
"data": {
"max_volume": "Maximum volume limit (%)"
"max_volume": "Maximum volume limit (%)",
"input_sources": "Input sources"
},
"data_description": {
"max_volume": "Maximum volume limit as a percentage. This will associate Home Assistant's maximum volume to this value on the receiver, i.e., if you set this to 50%, then setting the volume to 100% in Home Assistant will cause the volume on the receiver to be set to 50% of its maximum value."
"max_volume": "Maximum volume limit as a percentage. This will associate Home Assistant's maximum volume to this value on the receiver, i.e., if you set this to 50%, then setting the volume to 100% in Home Assistant will cause the volume on the receiver to be set to 50% of its maximum value.",
"input_sources": "List of input sources supported by the receiver."
}
},
"names": {
"sections": {
"input_sources": {
"name": "Input source names",
"description": "Mappings of receiver's input sources to their names."
}
}
}
},
"error": {
"empty_input_source_list": "Input source list cannot be empty"
}
},
"issues": {

110
tests/components/onkyo/test_config_flow.py
View File

@ -10,6 +10,7 @@ from homeassistant.components.onkyo import InputSource
from homeassistant.components.onkyo.config_flow import OnkyoConfigFlow
from homeassistant.components.onkyo.const import (
DOMAIN,
OPTION_INPUT_SOURCES,
OPTION_MAX_VOLUME,
OPTION_VOLUME_RESOLUTION,
)
@ -87,35 +88,6 @@ async def test_manual_invalid_host(hass: HomeAssistant, stub_mock_discovery) ->
assert host_result["errors"]["base"] == "cannot_connect"
async def test_ssdp_discovery_already_configured(
hass: HomeAssistant, default_mock_discovery
) -> None:
"""Test SSDP discovery with already configured device."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "192.168.1.100"},
unique_id="id1",
)
config_entry.add_to_hass(hass)
discovery_info = SsdpServiceInfo(
ssdp_location="http://192.168.1.100:8080",
upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"},
ssdp_usn="uuid:mock_usn",
ssdp_udn="uuid:00000000-0000-0000-0000-000000000000",
ssdp_st="mock_st",
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=discovery_info,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_manual_valid_host_unexpected_error(
hass: HomeAssistant, empty_mock_discovery
) -> None:
@ -262,6 +234,35 @@ async def test_ssdp_discovery_success(
assert select_result["result"].unique_id == "id1"
async def test_ssdp_discovery_already_configured(
hass: HomeAssistant, default_mock_discovery
) -> None:
"""Test SSDP discovery with already configured device."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "192.168.1.100"},
unique_id="id1",
)
config_entry.add_to_hass(hass)
discovery_info = SsdpServiceInfo(
ssdp_location="http://192.168.1.100:8080",
upnp={ATTR_UPNP_FRIENDLY_NAME: "Onkyo Receiver"},
ssdp_usn="uuid:mock_usn",
ssdp_udn="uuid:00000000-0000-0000-0000-000000000000",
ssdp_st="mock_st",
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_SSDP},
data=discovery_info,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_ssdp_discovery_host_info_error(hass: HomeAssistant) -> None:
"""Test SSDP discovery with host info error."""
discovery_info = SsdpServiceInfo(
@ -466,7 +467,7 @@ async def test_reconfigure(hass: HomeAssistant, default_mock_discovery) -> None:
await setup_integration(hass, config_entry, receiver_info)
old_host = config_entry.data[CONF_HOST]
old_max_volume = config_entry.options[OPTION_MAX_VOLUME]
old_options = config_entry.options
result = await config_entry.start_reconfigure_flow(hass)
@ -483,7 +484,7 @@ async def test_reconfigure(hass: HomeAssistant, default_mock_discovery) -> None:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
user_input={"volume_resolution": 200, "input_sources": ["TUNER"]},
user_input={OPTION_VOLUME_RESOLUTION: 200},
)
assert result3["type"] is FlowResultType.ABORT
@ -491,7 +492,10 @@ async def test_reconfigure(hass: HomeAssistant, default_mock_discovery) -> None:
assert config_entry.data[CONF_HOST] == old_host
assert config_entry.options[OPTION_VOLUME_RESOLUTION] == 200
assert config_entry.options[OPTION_MAX_VOLUME] == old_max_volume
for option, option_value in old_options.items():
if option == OPTION_VOLUME_RESOLUTION:
continue
assert config_entry.options[option] == option_value
async def test_reconfigure_new_device(hass: HomeAssistant) -> None:
@ -610,8 +614,8 @@ async def test_import_success(
"ignore_translations",
[
[ # The schema is dynamically created from input sources
"component.onkyo.options.step.init.data.TV",
"component.onkyo.options.step.init.data_description.TV",
"component.onkyo.options.step.names.sections.input_sources.data.TV",
"component.onkyo.options.step.names.sections.input_sources.data_description.TV",
]
],
)
@ -622,23 +626,43 @@ async def test_options_flow(hass: HomeAssistant, config_entry: MockConfigEntry)
config_entry = create_empty_config_entry()
await setup_integration(hass, config_entry, receiver_info)
old_volume_resolution = config_entry.options[OPTION_VOLUME_RESOLUTION]
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",
OPTION_MAX_VOLUME: 42,
OPTION_INPUT_SOURCES: [],
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"] == {OPTION_INPUT_SOURCES: "empty_input_source_list"}
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
OPTION_MAX_VOLUME: 42,
OPTION_INPUT_SOURCES: ["TV"],
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "names"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
OPTION_INPUT_SOURCES: {"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",
},
OPTION_VOLUME_RESOLUTION: old_volume_resolution,
OPTION_MAX_VOLUME: 42.0,
OPTION_INPUT_SOURCES: {"12": "television"},
}