mirror of
https://github.com/home-assistant/core.git
synced 2026-04-02 00:20:30 +01:00
Refactor Vizio integration to use DataUpdateCoordinator (#162188)
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -2,20 +2,34 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from pyvizio import VizioAsync
|
||||
|
||||
from homeassistant.components.media_player import MediaPlayerDeviceClass
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_DEVICE_CLASS, Platform
|
||||
from homeassistant.const import (
|
||||
CONF_ACCESS_TOKEN,
|
||||
CONF_DEVICE_CLASS,
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.storage import Store
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
from homeassistant.util.hass_dict import HassKey
|
||||
|
||||
from .const import CONF_APPS, DOMAIN
|
||||
from .coordinator import VizioAppsDataUpdateCoordinator
|
||||
from .const import DEFAULT_TIMEOUT, DEVICE_ID, DOMAIN, VIZIO_DEVICE_CLASSES
|
||||
from .coordinator import (
|
||||
VizioAppsDataUpdateCoordinator,
|
||||
VizioConfigEntry,
|
||||
VizioDeviceCoordinator,
|
||||
VizioRuntimeData,
|
||||
)
|
||||
from .services import async_setup_services
|
||||
|
||||
DATA_APPS: HassKey[VizioAppsDataUpdateCoordinator] = HassKey(f"{DOMAIN}_apps")
|
||||
|
||||
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
|
||||
PLATFORMS = [Platform.MEDIA_PLAYER]
|
||||
|
||||
@@ -26,38 +40,54 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: VizioConfigEntry) -> bool:
|
||||
"""Load the saved entities."""
|
||||
host = entry.data[CONF_HOST]
|
||||
token = entry.data.get(CONF_ACCESS_TOKEN)
|
||||
device_class = entry.data[CONF_DEVICE_CLASS]
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
if (
|
||||
CONF_APPS not in hass.data[DOMAIN]
|
||||
and entry.data[CONF_DEVICE_CLASS] == MediaPlayerDeviceClass.TV
|
||||
):
|
||||
store: Store[list[dict[str, Any]]] = Store(hass, 1, DOMAIN)
|
||||
coordinator = VizioAppsDataUpdateCoordinator(hass, store)
|
||||
await coordinator.async_setup()
|
||||
hass.data[DOMAIN][CONF_APPS] = coordinator
|
||||
await coordinator.async_refresh()
|
||||
# Create device
|
||||
device = VizioAsync(
|
||||
DEVICE_ID,
|
||||
host,
|
||||
entry.data[CONF_NAME],
|
||||
auth_token=token,
|
||||
device_type=VIZIO_DEVICE_CLASSES[device_class],
|
||||
session=async_get_clientsession(hass, False),
|
||||
timeout=DEFAULT_TIMEOUT,
|
||||
)
|
||||
|
||||
# Create device coordinator
|
||||
device_coordinator = VizioDeviceCoordinator(hass, entry, device)
|
||||
await device_coordinator.async_config_entry_first_refresh()
|
||||
|
||||
# Create apps coordinator for TVs (shared across entries)
|
||||
if device_class == MediaPlayerDeviceClass.TV and DATA_APPS not in hass.data:
|
||||
apps_coordinator = VizioAppsDataUpdateCoordinator(hass, Store(hass, 1, DOMAIN))
|
||||
await apps_coordinator.async_setup()
|
||||
hass.data[DATA_APPS] = apps_coordinator
|
||||
await apps_coordinator.async_refresh()
|
||||
|
||||
entry.runtime_data = VizioRuntimeData(
|
||||
device_coordinator=device_coordinator,
|
||||
)
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: VizioConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(
|
||||
config_entry, PLATFORMS
|
||||
)
|
||||
if not any(
|
||||
entry.data[CONF_DEVICE_CLASS] == MediaPlayerDeviceClass.TV
|
||||
for entry in hass.config_entries.async_loaded_entries(DOMAIN)
|
||||
):
|
||||
if coordinator := hass.data[DOMAIN].pop(CONF_APPS, None):
|
||||
await coordinator.async_shutdown()
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
if not hass.data[DOMAIN]:
|
||||
hass.data.pop(DOMAIN)
|
||||
# Clean up apps coordinator if no TV entries remain
|
||||
if unload_ok and not any(
|
||||
e.data[CONF_DEVICE_CLASS] == MediaPlayerDeviceClass.TV
|
||||
for e in hass.config_entries.async_loaded_entries(DOMAIN)
|
||||
if e.entry_id != entry.entry_id
|
||||
):
|
||||
if apps_coordinator := hass.data.pop(DATA_APPS, None):
|
||||
await apps_coordinator.async_shutdown()
|
||||
|
||||
return unload_ok
|
||||
|
||||
@@ -8,13 +8,12 @@ import socket
|
||||
from typing import Any
|
||||
|
||||
from pyvizio import VizioAsync, async_guess_device_type
|
||||
from pyvizio.const import APP_HOME
|
||||
from pyvizio.const import APP_HOME, APPS
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.media_player import MediaPlayerDeviceClass
|
||||
from homeassistant.config_entries import (
|
||||
SOURCE_ZEROCONF,
|
||||
ConfigEntry,
|
||||
ConfigFlow,
|
||||
ConfigFlowResult,
|
||||
OptionsFlow,
|
||||
@@ -34,6 +33,7 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
|
||||
from homeassistant.util.network import is_ip_address
|
||||
|
||||
from . import DATA_APPS
|
||||
from .const import (
|
||||
CONF_APPS,
|
||||
CONF_APPS_TO_INCLUDE_OR_EXCLUDE,
|
||||
@@ -45,6 +45,7 @@ from .const import (
|
||||
DEVICE_ID,
|
||||
DOMAIN,
|
||||
)
|
||||
from .coordinator import VizioConfigEntry
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@@ -106,6 +107,14 @@ def _host_is_same(host1: str, host2: str) -> bool:
|
||||
class VizioOptionsConfigFlow(OptionsFlow):
|
||||
"""Handle Vizio options."""
|
||||
|
||||
def _get_app_list(self) -> list[dict[str, Any]]:
|
||||
"""Return the current apps list, falling back to defaults."""
|
||||
if (
|
||||
apps_coordinator := self.hass.data.get(DATA_APPS)
|
||||
) and apps_coordinator.data:
|
||||
return apps_coordinator.data
|
||||
return APPS
|
||||
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
@@ -157,10 +166,7 @@ class VizioOptionsConfigFlow(OptionsFlow):
|
||||
): cv.multi_select(
|
||||
[
|
||||
APP_HOME["name"],
|
||||
*(
|
||||
app["name"]
|
||||
for app in self.hass.data[DOMAIN][CONF_APPS].data
|
||||
),
|
||||
*(app["name"] for app in self._get_app_list()),
|
||||
]
|
||||
),
|
||||
}
|
||||
@@ -176,7 +182,9 @@ class VizioConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(config_entry: ConfigEntry) -> VizioOptionsConfigFlow:
|
||||
def async_get_options_flow(
|
||||
config_entry: VizioConfigEntry,
|
||||
) -> VizioOptionsConfigFlow:
|
||||
"""Get the options flow for this handler."""
|
||||
return VizioOptionsConfigFlow()
|
||||
|
||||
|
||||
@@ -2,22 +2,150 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from pyvizio.const import APPS
|
||||
from pyvizio import VizioAsync
|
||||
from pyvizio.api.apps import AppConfig
|
||||
from pyvizio.api.input import InputItem
|
||||
from pyvizio.const import APPS, INPUT_APPS
|
||||
from pyvizio.util import gen_apps_list_from_url
|
||||
|
||||
from homeassistant.components.media_player import MediaPlayerDeviceClass
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_DEVICE_CLASS, CONF_HOST, CONF_NAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.storage import Store
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DOMAIN
|
||||
from .const import DOMAIN, VIZIO_AUDIO_SETTINGS, VIZIO_SOUND_MODE
|
||||
|
||||
type VizioConfigEntry = ConfigEntry[VizioRuntimeData]
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=30)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class VizioRuntimeData:
|
||||
"""Runtime data for Vizio integration."""
|
||||
|
||||
device_coordinator: VizioDeviceCoordinator
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class VizioDeviceData:
|
||||
"""Raw data fetched from Vizio device."""
|
||||
|
||||
# Power state
|
||||
is_on: bool
|
||||
|
||||
# Audio settings from get_all_settings("audio")
|
||||
audio_settings: dict[str, Any] | None = None
|
||||
|
||||
# Sound mode options from get_setting_options("audio", "eq")
|
||||
sound_mode_list: list[str] | None = None
|
||||
|
||||
# Current input from get_current_input()
|
||||
current_input: str | None = None
|
||||
|
||||
# Available inputs from get_inputs_list()
|
||||
input_list: list[InputItem] | None = None
|
||||
|
||||
# Current app config from get_current_app_config() (TVs only)
|
||||
current_app_config: AppConfig | None = None
|
||||
|
||||
|
||||
class VizioDeviceCoordinator(DataUpdateCoordinator[VizioDeviceData]):
|
||||
"""Coordinator for Vizio device data."""
|
||||
|
||||
config_entry: VizioConfigEntry
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
config_entry: VizioConfigEntry,
|
||||
device: VizioAsync,
|
||||
) -> None:
|
||||
"""Initialize the coordinator."""
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
config_entry=config_entry,
|
||||
name=DOMAIN,
|
||||
update_interval=SCAN_INTERVAL,
|
||||
)
|
||||
self.device = device
|
||||
|
||||
async def _async_setup(self) -> None:
|
||||
"""Fetch device info and update device registry."""
|
||||
model = await self.device.get_model_name(log_api_exception=False)
|
||||
version = await self.device.get_version(log_api_exception=False)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
assert self.config_entry.unique_id
|
||||
|
||||
device_registry = dr.async_get(self.hass)
|
||||
device_registry.async_get_or_create(
|
||||
config_entry_id=self.config_entry.entry_id,
|
||||
identifiers={(DOMAIN, self.config_entry.unique_id)},
|
||||
manufacturer="VIZIO",
|
||||
name=self.config_entry.data[CONF_NAME],
|
||||
model=model,
|
||||
sw_version=version,
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> VizioDeviceData:
|
||||
"""Fetch all device data."""
|
||||
is_on = await self.device.get_power_state(log_api_exception=False)
|
||||
|
||||
if is_on is None:
|
||||
raise UpdateFailed(
|
||||
f"Unable to connect to {self.config_entry.data[CONF_HOST]}"
|
||||
)
|
||||
|
||||
if not is_on:
|
||||
return VizioDeviceData(is_on=False)
|
||||
|
||||
# Device is on - fetch all data
|
||||
audio_settings = await self.device.get_all_settings(
|
||||
VIZIO_AUDIO_SETTINGS, log_api_exception=False
|
||||
)
|
||||
|
||||
sound_mode_list = None
|
||||
if audio_settings and VIZIO_SOUND_MODE in audio_settings:
|
||||
sound_mode_list = await self.device.get_setting_options(
|
||||
VIZIO_AUDIO_SETTINGS, VIZIO_SOUND_MODE, log_api_exception=False
|
||||
)
|
||||
|
||||
current_input = await self.device.get_current_input(log_api_exception=False)
|
||||
input_list = await self.device.get_inputs_list(log_api_exception=False)
|
||||
|
||||
current_app_config = None
|
||||
# Only attempt to fetch app config if the device is a TV and supports apps
|
||||
if (
|
||||
self.config_entry.data[CONF_DEVICE_CLASS] == MediaPlayerDeviceClass.TV
|
||||
and input_list
|
||||
and any(input_item.name in INPUT_APPS for input_item in input_list)
|
||||
):
|
||||
current_app_config = await self.device.get_current_app_config(
|
||||
log_api_exception=False
|
||||
)
|
||||
|
||||
return VizioDeviceData(
|
||||
is_on=True,
|
||||
audio_settings=audio_settings,
|
||||
sound_mode_list=sound_mode_list,
|
||||
current_input=current_input,
|
||||
input_list=input_list,
|
||||
current_app_config=current_app_config,
|
||||
)
|
||||
|
||||
|
||||
class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]):
|
||||
"""Define an object to hold Vizio app config data."""
|
||||
|
||||
@@ -2,11 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from pyvizio import AppConfig, VizioAsync
|
||||
from pyvizio.api.apps import find_app_name
|
||||
from pyvizio.api.apps import AppConfig, find_app_name
|
||||
from pyvizio.const import APP_HOME, INPUT_APPS, NO_APP_RUNNING, UNKNOWN_APP
|
||||
|
||||
from homeassistant.components.media_player import (
|
||||
@@ -15,58 +11,45 @@ from homeassistant.components.media_player import (
|
||||
MediaPlayerEntityFeature,
|
||||
MediaPlayerState,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_ACCESS_TOKEN,
|
||||
CONF_DEVICE_CLASS,
|
||||
CONF_EXCLUDE,
|
||||
CONF_HOST,
|
||||
CONF_INCLUDE,
|
||||
CONF_NAME,
|
||||
)
|
||||
from homeassistant.const import CONF_DEVICE_CLASS, CONF_EXCLUDE, CONF_INCLUDE
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.dispatcher import (
|
||||
async_dispatcher_connect,
|
||||
async_dispatcher_send,
|
||||
)
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import DATA_APPS
|
||||
from .const import (
|
||||
CONF_ADDITIONAL_CONFIGS,
|
||||
CONF_APPS,
|
||||
CONF_VOLUME_STEP,
|
||||
DEFAULT_TIMEOUT,
|
||||
DEFAULT_VOLUME_STEP,
|
||||
DEVICE_ID,
|
||||
DOMAIN,
|
||||
SUPPORTED_COMMANDS,
|
||||
VIZIO_AUDIO_SETTINGS,
|
||||
VIZIO_DEVICE_CLASSES,
|
||||
VIZIO_MUTE,
|
||||
VIZIO_MUTE_ON,
|
||||
VIZIO_SOUND_MODE,
|
||||
VIZIO_VOLUME,
|
||||
)
|
||||
from .coordinator import VizioAppsDataUpdateCoordinator
|
||||
from .coordinator import (
|
||||
VizioAppsDataUpdateCoordinator,
|
||||
VizioConfigEntry,
|
||||
VizioDeviceCoordinator,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=30)
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
config_entry: VizioConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up a Vizio media player entry."""
|
||||
host = config_entry.data[CONF_HOST]
|
||||
token = config_entry.data.get(CONF_ACCESS_TOKEN)
|
||||
name = config_entry.data[CONF_NAME]
|
||||
device_class = config_entry.data[CONF_DEVICE_CLASS]
|
||||
|
||||
# If config entry options not set up, set them up,
|
||||
@@ -105,59 +88,51 @@ async def async_setup_entry(
|
||||
**params, # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
device = VizioAsync(
|
||||
DEVICE_ID,
|
||||
host,
|
||||
name,
|
||||
auth_token=token,
|
||||
device_type=VIZIO_DEVICE_CLASSES[device_class],
|
||||
session=async_get_clientsession(hass, False),
|
||||
timeout=DEFAULT_TIMEOUT,
|
||||
entity = VizioDevice(
|
||||
config_entry,
|
||||
device_class,
|
||||
config_entry.runtime_data.device_coordinator,
|
||||
hass.data.get(DATA_APPS) if device_class == MediaPlayerDeviceClass.TV else None,
|
||||
)
|
||||
|
||||
apps_coordinator = hass.data[DOMAIN].get(CONF_APPS)
|
||||
|
||||
entity = VizioDevice(config_entry, device, name, device_class, apps_coordinator)
|
||||
|
||||
async_add_entities([entity], update_before_add=True)
|
||||
async_add_entities([entity])
|
||||
|
||||
|
||||
class VizioDevice(MediaPlayerEntity):
|
||||
class VizioDevice(CoordinatorEntity[VizioDeviceCoordinator], MediaPlayerEntity):
|
||||
"""Media Player implementation which performs REST requests to device."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
_attr_name = None
|
||||
_received_device_info = False
|
||||
_current_input: str | None = None
|
||||
_current_app_config: AppConfig | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config_entry: ConfigEntry,
|
||||
device: VizioAsync,
|
||||
name: str,
|
||||
config_entry: VizioConfigEntry,
|
||||
device_class: MediaPlayerDeviceClass,
|
||||
coordinator: VizioDeviceCoordinator,
|
||||
apps_coordinator: VizioAppsDataUpdateCoordinator | None,
|
||||
) -> None:
|
||||
"""Initialize Vizio device."""
|
||||
super().__init__(coordinator)
|
||||
|
||||
self._config_entry = config_entry
|
||||
self._apps_coordinator = apps_coordinator
|
||||
|
||||
self._volume_step = config_entry.options[CONF_VOLUME_STEP]
|
||||
self._current_input: str | None = None
|
||||
self._current_app_config: AppConfig | None = None
|
||||
self._attr_sound_mode_list = []
|
||||
self._available_inputs: list[str] = []
|
||||
self._available_apps: list[str] = []
|
||||
|
||||
self._volume_step = config_entry.options[CONF_VOLUME_STEP]
|
||||
self._all_apps = apps_coordinator.data if apps_coordinator else None
|
||||
self._conf_apps = config_entry.options.get(CONF_APPS, {})
|
||||
self._additional_app_configs = config_entry.data.get(CONF_APPS, {}).get(
|
||||
CONF_ADDITIONAL_CONFIGS, []
|
||||
)
|
||||
self._device = device
|
||||
self._max_volume = float(device.get_max_volume())
|
||||
self._attr_assumed_state = True
|
||||
self._device = coordinator.device
|
||||
self._max_volume = float(coordinator.device.get_max_volume())
|
||||
|
||||
# Entity class attributes that will change with each update (we only include
|
||||
# the ones that are initialized differently from the defaults)
|
||||
self._attr_sound_mode_list = []
|
||||
self._attr_supported_features = SUPPORTED_COMMANDS[device_class]
|
||||
|
||||
# Entity class attributes that will not change
|
||||
@@ -165,11 +140,7 @@ class VizioDevice(MediaPlayerEntity):
|
||||
assert unique_id
|
||||
self._attr_unique_id = unique_id
|
||||
self._attr_device_class = device_class
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, unique_id)},
|
||||
manufacturer="VIZIO",
|
||||
name=name,
|
||||
)
|
||||
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, unique_id)})
|
||||
|
||||
def _apps_list(self, apps: list[str]) -> list[str]:
|
||||
"""Return process apps list based on configured filters."""
|
||||
@@ -181,112 +152,72 @@ class VizioDevice(MediaPlayerEntity):
|
||||
|
||||
return apps
|
||||
|
||||
async def async_update(self) -> None:
|
||||
"""Retrieve latest state of the device."""
|
||||
if (
|
||||
is_on := await self._device.get_power_state(log_api_exception=False)
|
||||
) is None:
|
||||
if self._attr_available:
|
||||
_LOGGER.warning(
|
||||
"Lost connection to %s", self._config_entry.data[CONF_HOST]
|
||||
)
|
||||
self._attr_available = False
|
||||
return
|
||||
@callback
|
||||
def _handle_coordinator_update(self) -> None:
|
||||
"""Handle updated data from the coordinator."""
|
||||
data = self.coordinator.data
|
||||
|
||||
if not self._attr_available:
|
||||
_LOGGER.warning(
|
||||
"Restored connection to %s", self._config_entry.data[CONF_HOST]
|
||||
)
|
||||
self._attr_available = True
|
||||
|
||||
if not self._received_device_info:
|
||||
device_reg = dr.async_get(self.hass)
|
||||
assert self._config_entry.unique_id
|
||||
device = device_reg.async_get_device(
|
||||
identifiers={(DOMAIN, self._config_entry.unique_id)}
|
||||
)
|
||||
if device:
|
||||
device_reg.async_update_device(
|
||||
device.id,
|
||||
model=await self._device.get_model_name(log_api_exception=False),
|
||||
sw_version=await self._device.get_version(log_api_exception=False),
|
||||
)
|
||||
self._received_device_info = True
|
||||
|
||||
if not is_on:
|
||||
# Handle device off
|
||||
if not data.is_on:
|
||||
self._attr_state = MediaPlayerState.OFF
|
||||
self._attr_volume_level = None
|
||||
self._attr_is_volume_muted = None
|
||||
self._current_input = None
|
||||
self._attr_app_name = None
|
||||
self._current_app_config = None
|
||||
self._attr_sound_mode = None
|
||||
self._attr_app_name = None
|
||||
self._current_input = None
|
||||
self._current_app_config = None
|
||||
super()._handle_coordinator_update()
|
||||
return
|
||||
|
||||
# Device is on - apply coordinator data
|
||||
self._attr_state = MediaPlayerState.ON
|
||||
|
||||
if audio_settings := await self._device.get_all_settings(
|
||||
VIZIO_AUDIO_SETTINGS, log_api_exception=False
|
||||
):
|
||||
# Audio settings
|
||||
if data.audio_settings:
|
||||
self._attr_volume_level = (
|
||||
float(audio_settings[VIZIO_VOLUME]) / self._max_volume
|
||||
float(data.audio_settings[VIZIO_VOLUME]) / self._max_volume
|
||||
)
|
||||
if VIZIO_MUTE in audio_settings:
|
||||
if VIZIO_MUTE in data.audio_settings:
|
||||
self._attr_is_volume_muted = (
|
||||
audio_settings[VIZIO_MUTE].lower() == VIZIO_MUTE_ON
|
||||
data.audio_settings[VIZIO_MUTE].lower() == VIZIO_MUTE_ON
|
||||
)
|
||||
else:
|
||||
self._attr_is_volume_muted = None
|
||||
|
||||
if VIZIO_SOUND_MODE in audio_settings:
|
||||
if VIZIO_SOUND_MODE in data.audio_settings:
|
||||
self._attr_supported_features |= (
|
||||
MediaPlayerEntityFeature.SELECT_SOUND_MODE
|
||||
)
|
||||
self._attr_sound_mode = audio_settings[VIZIO_SOUND_MODE]
|
||||
self._attr_sound_mode = data.audio_settings[VIZIO_SOUND_MODE]
|
||||
if not self._attr_sound_mode_list:
|
||||
self._attr_sound_mode_list = await self._device.get_setting_options(
|
||||
VIZIO_AUDIO_SETTINGS,
|
||||
VIZIO_SOUND_MODE,
|
||||
log_api_exception=False,
|
||||
)
|
||||
self._attr_sound_mode_list = data.sound_mode_list or []
|
||||
else:
|
||||
# Explicitly remove MediaPlayerEntityFeature.SELECT_SOUND_MODE from supported features
|
||||
self._attr_supported_features &= (
|
||||
~MediaPlayerEntityFeature.SELECT_SOUND_MODE
|
||||
)
|
||||
|
||||
if input_ := await self._device.get_current_input(log_api_exception=False):
|
||||
self._current_input = input_
|
||||
# Input state
|
||||
if data.current_input:
|
||||
self._current_input = data.current_input
|
||||
if data.input_list:
|
||||
self._available_inputs = [i.name for i in data.input_list]
|
||||
|
||||
# If no inputs returned, end update
|
||||
if not (inputs := await self._device.get_inputs_list(log_api_exception=False)):
|
||||
return
|
||||
|
||||
self._available_inputs = [input_.name for input_ in inputs]
|
||||
|
||||
# Return before setting app variables if INPUT_APPS isn't in available inputs
|
||||
if self._attr_device_class == MediaPlayerDeviceClass.SPEAKER or not any(
|
||||
app for app in INPUT_APPS if app in self._available_inputs
|
||||
# App state (TV only) - check if device supports apps
|
||||
if (
|
||||
self._attr_device_class == MediaPlayerDeviceClass.TV
|
||||
and self._available_inputs
|
||||
and any(app in self._available_inputs for app in INPUT_APPS)
|
||||
):
|
||||
return
|
||||
all_apps = self._all_apps or ()
|
||||
self._available_apps = self._apps_list([app["name"] for app in all_apps])
|
||||
self._current_app_config = data.current_app_config
|
||||
self._attr_app_name = find_app_name(
|
||||
self._current_app_config,
|
||||
[APP_HOME, *all_apps, *self._additional_app_configs],
|
||||
)
|
||||
if self._attr_app_name == NO_APP_RUNNING:
|
||||
self._attr_app_name = None
|
||||
|
||||
# Create list of available known apps from known app list after
|
||||
# filtering by CONF_INCLUDE/CONF_EXCLUDE
|
||||
self._available_apps = self._apps_list(
|
||||
[app["name"] for app in self._all_apps or ()]
|
||||
)
|
||||
|
||||
self._current_app_config = await self._device.get_current_app_config(
|
||||
log_api_exception=False
|
||||
)
|
||||
|
||||
self._attr_app_name = find_app_name(
|
||||
self._current_app_config,
|
||||
[APP_HOME, *(self._all_apps or ()), *self._additional_app_configs],
|
||||
)
|
||||
|
||||
if self._attr_app_name == NO_APP_RUNNING:
|
||||
self._attr_app_name = None
|
||||
super()._handle_coordinator_update()
|
||||
|
||||
def _get_additional_app_names(self) -> list[str]:
|
||||
"""Return list of additional apps that were included in configuration.yaml."""
|
||||
@@ -296,7 +227,7 @@ class VizioDevice(MediaPlayerEntity):
|
||||
|
||||
@staticmethod
|
||||
async def _async_send_update_options_signal(
|
||||
hass: HomeAssistant, config_entry: ConfigEntry
|
||||
hass: HomeAssistant, config_entry: VizioConfigEntry
|
||||
) -> None:
|
||||
"""Send update event when Vizio config entry is updated."""
|
||||
# Move this method to component level if another entity ever gets added for a
|
||||
@@ -304,7 +235,7 @@ class VizioDevice(MediaPlayerEntity):
|
||||
# See here: https://github.com/home-assistant/core/pull/30653#discussion_r366426121
|
||||
async_dispatcher_send(hass, config_entry.entry_id, config_entry)
|
||||
|
||||
async def _async_update_options(self, config_entry: ConfigEntry) -> None:
|
||||
async def _async_update_options(self, config_entry: VizioConfigEntry) -> None:
|
||||
"""Update options if the update signal comes from this entity."""
|
||||
self._volume_step = config_entry.options[CONF_VOLUME_STEP]
|
||||
# Update so that CONF_ADDITIONAL_CONFIGS gets retained for imports
|
||||
@@ -323,6 +254,11 @@ class VizioDevice(MediaPlayerEntity):
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Register callbacks when entity is added."""
|
||||
await super().async_added_to_hass()
|
||||
|
||||
# Process initial coordinator data
|
||||
self._handle_coordinator_update()
|
||||
|
||||
# Register callback for when config entry is updated.
|
||||
self.async_on_remove(
|
||||
self._config_entry.add_update_listener(
|
||||
@@ -337,21 +273,17 @@ class VizioDevice(MediaPlayerEntity):
|
||||
)
|
||||
)
|
||||
|
||||
if not self._apps_coordinator:
|
||||
if not (apps_coordinator := self._apps_coordinator):
|
||||
return
|
||||
|
||||
# Register callback for app list updates if device is a TV
|
||||
@callback
|
||||
def apps_list_update() -> None:
|
||||
"""Update list of all apps."""
|
||||
if not self._apps_coordinator:
|
||||
return
|
||||
self._all_apps = self._apps_coordinator.data
|
||||
self._all_apps = apps_coordinator.data
|
||||
self.async_write_ha_state()
|
||||
|
||||
self.async_on_remove(
|
||||
self._apps_coordinator.async_add_listener(apps_list_update)
|
||||
)
|
||||
self.async_on_remove(apps_coordinator.async_add_listener(apps_list_update))
|
||||
|
||||
@property
|
||||
def source(self) -> str | None:
|
||||
|
||||
@@ -142,13 +142,36 @@ def vizio_bypass_setup_fixture() -> Generator[None]:
|
||||
|
||||
@pytest.fixture(name="vizio_bypass_update")
|
||||
def vizio_bypass_update_fixture() -> Generator[None]:
|
||||
"""Mock component update."""
|
||||
"""Mock component update with minimal data."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.can_connect_with_auth_check",
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=True,
|
||||
),
|
||||
patch("homeassistant.components.vizio.media_player.VizioDevice.async_update"),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_all_settings",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_input",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_inputs_list",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_model_name",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_version",
|
||||
return_value=None,
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
@@ -172,7 +195,15 @@ def vizio_cant_connect_fixture() -> Generator[None]:
|
||||
AsyncMock(return_value=False),
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_power_state",
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_model_name",
|
||||
return_value=None,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_version",
|
||||
return_value=None,
|
||||
),
|
||||
):
|
||||
@@ -184,11 +215,7 @@ def vizio_update_fixture() -> Generator[None]:
|
||||
"""Mock valid updates to vizio device."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.can_connect_with_auth_check",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_all_settings",
|
||||
"homeassistant.components.vizio.VizioAsync.get_all_settings",
|
||||
return_value={
|
||||
"volume": int(MAX_VOLUME[DEVICE_CLASS_SPEAKER] / 2),
|
||||
"eq": CURRENT_EQ,
|
||||
@@ -196,29 +223,33 @@ def vizio_update_fixture() -> Generator[None]:
|
||||
},
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_setting_options",
|
||||
"homeassistant.components.vizio.VizioAsync.get_setting_options",
|
||||
return_value=EQ_LIST,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_current_input",
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_input",
|
||||
return_value=CURRENT_INPUT,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_inputs_list",
|
||||
"homeassistant.components.vizio.VizioAsync.get_inputs_list",
|
||||
return_value=get_mock_inputs(INPUT_LIST),
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_power_state",
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_model_name",
|
||||
"homeassistant.components.vizio.VizioAsync.get_model_name",
|
||||
return_value=MODEL,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_version",
|
||||
"homeassistant.components.vizio.VizioAsync.get_version",
|
||||
return_value=VERSION,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
|
||||
return_value=None,
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
@@ -228,15 +259,15 @@ def vizio_update_with_apps_fixture(vizio_update: None) -> Generator[None]:
|
||||
"""Mock valid updates to vizio device that supports apps."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_inputs_list",
|
||||
"homeassistant.components.vizio.VizioAsync.get_inputs_list",
|
||||
return_value=get_mock_inputs(INPUT_LIST_WITH_APPS),
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_current_input",
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_input",
|
||||
return_value="CAST",
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_current_app_config",
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
|
||||
return_value=AppConfig(**CURRENT_APP_CONFIG),
|
||||
),
|
||||
):
|
||||
@@ -248,15 +279,15 @@ def vizio_update_with_apps_on_input_fixture(vizio_update: None) -> Generator[Non
|
||||
"""Mock valid updates to vizio device that supports apps but is on a TV input."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_inputs_list",
|
||||
"homeassistant.components.vizio.VizioAsync.get_inputs_list",
|
||||
return_value=get_mock_inputs(INPUT_LIST_WITH_APPS),
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_current_input",
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_input",
|
||||
return_value=CURRENT_INPUT,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_current_app_config",
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
|
||||
return_value=AppConfig("unknown", 1, "app"),
|
||||
),
|
||||
):
|
||||
|
||||
@@ -5,6 +5,7 @@ import dataclasses
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.media_player import MediaPlayerDeviceClass
|
||||
from homeassistant.components.vizio import DATA_APPS
|
||||
from homeassistant.components.vizio.const import (
|
||||
CONF_APPS,
|
||||
CONF_APPS_TO_INCLUDE_OR_EXCLUDE,
|
||||
@@ -142,6 +143,36 @@ async def test_tv_options_flow_no_apps(hass: HomeAssistant) -> None:
|
||||
assert CONF_APPS not in result["data"]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_bypass_update")
|
||||
async def test_tv_options_flow_apps_fallback(hass: HomeAssistant) -> None:
|
||||
"""Test options config flow falls back to default APPS when coordinator absent."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}, data=MOCK_USER_VALID_TV_CONFIG
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
entry = result["result"]
|
||||
|
||||
# Remove apps coordinator to simulate it being unavailable
|
||||
hass.data.pop(DATA_APPS)
|
||||
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id, data=None)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
# Completing the flow should still work with the APPS fallback
|
||||
options = {CONF_VOLUME_STEP: VOLUME_STEP}
|
||||
options.update(MOCK_INCLUDE_APPS)
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"], user_input=options
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_APPS] == {CONF_INCLUDE: [CURRENT_APP]}
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_bypass_update")
|
||||
async def test_tv_options_flow_with_apps(hass: HomeAssistant) -> None:
|
||||
"""Test options config flow for TV with providing apps option."""
|
||||
|
||||
@@ -7,6 +7,7 @@ from freezegun.api import FrozenDateTimeFactory
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.media_player import MediaPlayerDeviceClass
|
||||
from homeassistant.components.vizio import DATA_APPS
|
||||
from homeassistant.components.vizio.const import DOMAIN
|
||||
from homeassistant.const import (
|
||||
CONF_ACCESS_TOKEN,
|
||||
@@ -17,14 +18,17 @@ from homeassistant.const import (
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
|
||||
from .const import (
|
||||
APP_LIST,
|
||||
HOST2,
|
||||
MOCK_SPEAKER_CONFIG,
|
||||
MOCK_USER_VALID_TV_CONFIG,
|
||||
MODEL,
|
||||
NAME2,
|
||||
UNIQUE_ID,
|
||||
VERSION,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
@@ -40,7 +44,7 @@ async def test_tv_load_and_unload(hass: HomeAssistant) -> None:
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert len(hass.states.async_entity_ids(Platform.MEDIA_PLAYER)) == 1
|
||||
assert DOMAIN in hass.data
|
||||
assert DATA_APPS in hass.data
|
||||
|
||||
assert await hass.config_entries.async_unload(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
@@ -48,7 +52,7 @@ async def test_tv_load_and_unload(hass: HomeAssistant) -> None:
|
||||
assert len(entities) == 1
|
||||
for entity in entities:
|
||||
assert hass.states.get(entity).state == STATE_UNAVAILABLE
|
||||
assert DOMAIN not in hass.data
|
||||
assert DATA_APPS not in hass.data
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
@@ -61,7 +65,6 @@ async def test_speaker_load_and_unload(hass: HomeAssistant) -> None:
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert len(hass.states.async_entity_ids(Platform.MEDIA_PLAYER)) == 1
|
||||
assert DOMAIN in hass.data
|
||||
|
||||
assert await hass.config_entries.async_unload(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
@@ -69,7 +72,6 @@ async def test_speaker_load_and_unload(hass: HomeAssistant) -> None:
|
||||
assert len(entities) == 1
|
||||
for entity in entities:
|
||||
assert hass.states.get(entity).state == STATE_UNAVAILABLE
|
||||
assert DOMAIN not in hass.data
|
||||
|
||||
|
||||
@pytest.mark.usefixtures(
|
||||
@@ -88,6 +90,7 @@ async def test_coordinator_update_failure(
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert len(hass.states.async_entity_ids(Platform.MEDIA_PLAYER)) == 1
|
||||
assert DATA_APPS in hass.data
|
||||
|
||||
# Failing 25 days in a row should result in a single log message
|
||||
# (first one after 10 days, next one would be at 30 days)
|
||||
@@ -152,3 +155,41 @@ async def test_apps_coordinator_persists_until_last_tv_unloads(
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_fetch.call_count == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_device_registry_model_and_version(
|
||||
hass: HomeAssistant, device_registry: dr.DeviceRegistry
|
||||
) -> None:
|
||||
"""Test that coordinator populates device registry with model and version."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN, data=MOCK_USER_VALID_TV_CONFIG, unique_id=UNIQUE_ID
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, UNIQUE_ID)})
|
||||
assert device is not None
|
||||
assert device.model == MODEL
|
||||
assert device.sw_version == VERSION
|
||||
assert device.manufacturer == "VIZIO"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_bypass_update")
|
||||
async def test_device_registry_without_model_or_version(
|
||||
hass: HomeAssistant, device_registry: dr.DeviceRegistry
|
||||
) -> None:
|
||||
"""Test device registry when model and version are unavailable."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN, data=MOCK_USER_VALID_TV_CONFIG, unique_id=UNIQUE_ID
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, UNIQUE_ID)})
|
||||
assert device is not None
|
||||
assert device.model is None
|
||||
assert device.sw_version is None
|
||||
assert device.manufacturer == "VIZIO"
|
||||
|
||||
@@ -8,7 +8,7 @@ from datetime import timedelta
|
||||
from typing import Any
|
||||
from unittest.mock import call, patch
|
||||
|
||||
from freezegun import freeze_time
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
import pytest
|
||||
from pyvizio.api.apps import AppConfig
|
||||
from pyvizio.const import (
|
||||
@@ -40,6 +40,7 @@ from homeassistant.components.media_player import (
|
||||
SERVICE_VOLUME_SET,
|
||||
SERVICE_VOLUME_UP,
|
||||
MediaPlayerDeviceClass,
|
||||
MediaPlayerEntityFeature,
|
||||
)
|
||||
from homeassistant.components.vizio.const import (
|
||||
CONF_ADDITIONAL_CONFIGS,
|
||||
@@ -49,6 +50,7 @@ from homeassistant.components.vizio.const import (
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.components.vizio.services import SERVICE_UPDATE_SETTING
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, STATE_UNAVAILABLE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.util import dt as dt_util
|
||||
@@ -88,15 +90,12 @@ async def _add_config_entry_to_hass(
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
def _get_ha_power_state(vizio_power_state: bool | None) -> str:
|
||||
def _get_ha_power_state(vizio_power_state: bool) -> str:
|
||||
"""Return HA power state given Vizio power state."""
|
||||
if vizio_power_state:
|
||||
return STATE_ON
|
||||
|
||||
if vizio_power_state is False:
|
||||
return STATE_OFF
|
||||
|
||||
return STATE_UNAVAILABLE
|
||||
return STATE_OFF
|
||||
|
||||
|
||||
def _assert_sources_and_volume(attr: dict[str, Any], vizio_device_class: str) -> None:
|
||||
@@ -124,27 +123,27 @@ def _get_attr_and_assert_base_attr(
|
||||
|
||||
@asynccontextmanager
|
||||
async def _cm_for_test_setup_without_apps(
|
||||
all_settings: dict[str, Any], vizio_power_state: bool | None
|
||||
all_settings: dict[str, Any], vizio_power_state: bool
|
||||
) -> AsyncIterator[None]:
|
||||
"""Context manager to setup test for Vizio devices without including app specific patches."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_all_settings",
|
||||
"homeassistant.components.vizio.VizioAsync.get_all_settings",
|
||||
return_value=all_settings,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_setting_options",
|
||||
"homeassistant.components.vizio.VizioAsync.get_setting_options",
|
||||
return_value=EQ_LIST,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_power_state",
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=vizio_power_state,
|
||||
),
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
async def _test_setup_tv(hass: HomeAssistant, vizio_power_state: bool | None) -> None:
|
||||
async def _test_setup_tv(hass: HomeAssistant, vizio_power_state: bool) -> None:
|
||||
"""Test Vizio TV entity setup."""
|
||||
ha_power_state = _get_ha_power_state(vizio_power_state)
|
||||
|
||||
@@ -155,7 +154,11 @@ async def _test_setup_tv(hass: HomeAssistant, vizio_power_state: bool | None) ->
|
||||
)
|
||||
|
||||
async with _cm_for_test_setup_without_apps(
|
||||
{"volume": int(MAX_VOLUME[VIZIO_DEVICE_CLASS_TV] / 2), "mute": "Off"},
|
||||
{
|
||||
"volume": int(MAX_VOLUME[VIZIO_DEVICE_CLASS_TV] / 2),
|
||||
"mute": "Off",
|
||||
"eq": CURRENT_EQ,
|
||||
},
|
||||
vizio_power_state,
|
||||
):
|
||||
await _add_config_entry_to_hass(hass, config_entry)
|
||||
@@ -165,12 +168,10 @@ async def _test_setup_tv(hass: HomeAssistant, vizio_power_state: bool | None) ->
|
||||
)
|
||||
if ha_power_state == STATE_ON:
|
||||
_assert_sources_and_volume(attr, VIZIO_DEVICE_CLASS_TV)
|
||||
assert "sound_mode" not in attr
|
||||
assert attr[ATTR_SOUND_MODE] == CURRENT_EQ
|
||||
|
||||
|
||||
async def _test_setup_speaker(
|
||||
hass: HomeAssistant, vizio_power_state: bool | None
|
||||
) -> None:
|
||||
async def _test_setup_speaker(hass: HomeAssistant, vizio_power_state: bool) -> None:
|
||||
"""Test Vizio Speaker entity setup."""
|
||||
ha_power_state = _get_ha_power_state(vizio_power_state)
|
||||
|
||||
@@ -190,18 +191,14 @@ async def _test_setup_speaker(
|
||||
audio_settings,
|
||||
vizio_power_state,
|
||||
):
|
||||
with patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_current_app_config",
|
||||
) as service_call:
|
||||
await _add_config_entry_to_hass(hass, config_entry)
|
||||
await _add_config_entry_to_hass(hass, config_entry)
|
||||
|
||||
attr = _get_attr_and_assert_base_attr(
|
||||
hass, MediaPlayerDeviceClass.SPEAKER, ha_power_state
|
||||
)
|
||||
if ha_power_state == STATE_ON:
|
||||
_assert_sources_and_volume(attr, VIZIO_DEVICE_CLASS_SPEAKER)
|
||||
assert not service_call.called
|
||||
assert "sound_mode" in attr
|
||||
attr = _get_attr_and_assert_base_attr(
|
||||
hass, MediaPlayerDeviceClass.SPEAKER, ha_power_state
|
||||
)
|
||||
if ha_power_state == STATE_ON:
|
||||
_assert_sources_and_volume(attr, VIZIO_DEVICE_CLASS_SPEAKER)
|
||||
assert "sound_mode" in attr
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
@@ -218,7 +215,7 @@ async def _cm_for_test_setup_tv_with_apps(
|
||||
True,
|
||||
):
|
||||
with patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_current_app_config",
|
||||
"homeassistant.components.vizio.VizioAsync.get_current_app_config",
|
||||
return_value=AppConfig(**app_config),
|
||||
):
|
||||
await _add_config_entry_to_hass(hass, config_entry)
|
||||
@@ -262,7 +259,7 @@ async def _test_service(
|
||||
service_data.update(additional_service_data)
|
||||
|
||||
with patch(
|
||||
f"homeassistant.components.vizio.media_player.VizioAsync.{vizio_func_name}"
|
||||
f"homeassistant.components.vizio.VizioAsync.{vizio_func_name}"
|
||||
) as service_call:
|
||||
await hass.services.async_call(
|
||||
domain,
|
||||
@@ -288,14 +285,6 @@ async def test_speaker_off(hass: HomeAssistant) -> None:
|
||||
await _test_setup_speaker(hass, False)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_speaker_unavailable(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test Vizio Speaker entity setup when unavailable."""
|
||||
await _test_setup_speaker(hass, None)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_init_tv_on(hass: HomeAssistant) -> None:
|
||||
"""Test Vizio TV entity setup when on."""
|
||||
@@ -308,32 +297,28 @@ async def test_init_tv_off(hass: HomeAssistant) -> None:
|
||||
await _test_setup_tv(hass, False)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_init_tv_unavailable(hass: HomeAssistant) -> None:
|
||||
"""Test Vizio TV entity setup when unavailable."""
|
||||
await _test_setup_tv(hass, None)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_cant_connect")
|
||||
async def test_setup_unavailable_speaker(hass: HomeAssistant) -> None:
|
||||
"""Test speaker entity sets up as unavailable."""
|
||||
"""Test speaker config entry retries setup when device is unavailable."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN, data=MOCK_SPEAKER_CONFIG, unique_id=UNIQUE_ID
|
||||
)
|
||||
await _add_config_entry_to_hass(hass, config_entry)
|
||||
assert len(hass.states.async_entity_ids(MP_DOMAIN)) == 1
|
||||
assert hass.states.get("media_player.vizio").state == STATE_UNAVAILABLE
|
||||
config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_cant_connect")
|
||||
async def test_setup_unavailable_tv(hass: HomeAssistant) -> None:
|
||||
"""Test TV entity sets up as unavailable."""
|
||||
"""Test TV config entry retries setup when device is unavailable."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN, data=MOCK_USER_VALID_TV_CONFIG, unique_id=UNIQUE_ID
|
||||
)
|
||||
await _add_config_entry_to_hass(hass, config_entry)
|
||||
assert len(hass.states.async_entity_ids(MP_DOMAIN)) == 1
|
||||
assert hass.states.get("media_player.vizio").state == STATE_UNAVAILABLE
|
||||
config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
@@ -377,7 +362,7 @@ async def test_services(hass: HomeAssistant) -> None:
|
||||
"vol_up",
|
||||
SERVICE_VOLUME_SET,
|
||||
{ATTR_MEDIA_VOLUME_LEVEL: 1},
|
||||
num=(100 - 15),
|
||||
num=50, # From 50% to 100% = 50 steps (TV max volume 100, starting at 50)
|
||||
)
|
||||
await _test_service(
|
||||
hass,
|
||||
@@ -385,7 +370,7 @@ async def test_services(hass: HomeAssistant) -> None:
|
||||
"vol_down",
|
||||
SERVICE_VOLUME_SET,
|
||||
{ATTR_MEDIA_VOLUME_LEVEL: 0},
|
||||
num=(15 - 0),
|
||||
num=100, # From 100% (after previous vol_up) to 0% = 100 steps
|
||||
)
|
||||
await _test_service(hass, MP_DOMAIN, "ch_up", SERVICE_MEDIA_NEXT_TRACK, None)
|
||||
await _test_service(hass, MP_DOMAIN, "ch_down", SERVICE_MEDIA_PREVIOUS_TRACK, None)
|
||||
@@ -444,66 +429,52 @@ async def test_options_update(hass: HomeAssistant) -> None:
|
||||
)
|
||||
|
||||
|
||||
async def _test_update_availability_switch(
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_update_available_to_unavailable(
|
||||
hass: HomeAssistant,
|
||||
initial_power_state: bool | None,
|
||||
final_power_state: bool | None,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
now = dt_util.utcnow()
|
||||
future_interval = timedelta(minutes=1)
|
||||
"""Test device becomes unavailable after being available."""
|
||||
await _test_setup_speaker(hass, True)
|
||||
|
||||
# Setup device as if time is right now
|
||||
with freeze_time(now):
|
||||
await _test_setup_speaker(hass, initial_power_state)
|
||||
|
||||
# Clear captured logs so that only availability state changes are captured for
|
||||
# future assertion
|
||||
caplog.clear()
|
||||
|
||||
# Fast forward time to future twice to trigger update and assert vizio log message
|
||||
for i in range(1, 3):
|
||||
future = now + (future_interval * i)
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.get_power_state",
|
||||
return_value=final_power_state,
|
||||
),
|
||||
freeze_time(future),
|
||||
):
|
||||
async_fire_time_changed(hass, future)
|
||||
await hass.async_block_till_done()
|
||||
if final_power_state is None:
|
||||
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
|
||||
else:
|
||||
assert hass.states.get(ENTITY_ID).state != STATE_UNAVAILABLE
|
||||
|
||||
# Ensure connection status messages from vizio.media_player appear exactly once
|
||||
# (on availability state change)
|
||||
vizio_log_list = [
|
||||
log
|
||||
for log in caplog.records
|
||||
if log.name == "homeassistant.components.vizio.media_player"
|
||||
]
|
||||
assert len(vizio_log_list) == 1
|
||||
# Simulate device becoming unreachable
|
||||
with patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=None,
|
||||
):
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_update_unavailable_to_available(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test device becomes available after being unavailable."""
|
||||
await _test_update_availability_switch(hass, None, True, caplog)
|
||||
await _test_setup_speaker(hass, True)
|
||||
|
||||
# First, make device unavailable
|
||||
with patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=None,
|
||||
):
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_update_available_to_unavailable(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test device becomes unavailable after being available."""
|
||||
await _test_update_availability_switch(hass, True, None, caplog)
|
||||
# Then, make device available again
|
||||
with patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=True,
|
||||
):
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(ENTITY_ID).state != STATE_UNAVAILABLE
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update_with_apps")
|
||||
@@ -619,11 +590,9 @@ async def test_setup_with_apps_additional_apps_config(
|
||||
|
||||
# Test that invalid app does nothing
|
||||
with (
|
||||
patch("homeassistant.components.vizio.VizioAsync.launch_app") as service_call1,
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.launch_app"
|
||||
) as service_call1,
|
||||
patch(
|
||||
"homeassistant.components.vizio.media_player.VizioAsync.launch_app_config"
|
||||
"homeassistant.components.vizio.VizioAsync.launch_app_config"
|
||||
) as service_call2,
|
||||
):
|
||||
await hass.services.async_call(
|
||||
@@ -679,7 +648,7 @@ async def test_setup_tv_without_mute(hass: HomeAssistant) -> None:
|
||||
|
||||
async with _cm_for_test_setup_without_apps(
|
||||
{"volume": int(MAX_VOLUME[VIZIO_DEVICE_CLASS_TV] / 2)},
|
||||
STATE_ON,
|
||||
True,
|
||||
):
|
||||
await _add_config_entry_to_hass(hass, config_entry)
|
||||
|
||||
@@ -735,3 +704,122 @@ async def test_vizio_update_with_apps_on_input(hass: HomeAssistant) -> None:
|
||||
attr = _get_attr_and_assert_base_attr(hass, MediaPlayerDeviceClass.TV, STATE_ON)
|
||||
# app ID should not be in the attributes
|
||||
assert "app_id" not in attr
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_coordinator_update_on_to_off(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test device transitions from on to off during coordinator refresh."""
|
||||
await _test_setup_speaker(hass, True)
|
||||
attr = _get_attr_and_assert_base_attr(
|
||||
hass, MediaPlayerDeviceClass.SPEAKER, STATE_ON
|
||||
)
|
||||
assert attr[ATTR_MEDIA_VOLUME_LEVEL] is not None
|
||||
assert ATTR_SOUND_MODE in attr
|
||||
|
||||
# Device turns off
|
||||
with patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=False,
|
||||
):
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get(ENTITY_ID).state == STATE_OFF
|
||||
attr = hass.states.get(ENTITY_ID).attributes
|
||||
assert attr.get(ATTR_MEDIA_VOLUME_LEVEL) is None
|
||||
assert attr.get(ATTR_MEDIA_VOLUME_MUTED) is None
|
||||
assert attr.get(ATTR_SOUND_MODE) is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_coordinator_update_off_to_on(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test device transitions from off to on during coordinator refresh."""
|
||||
await _test_setup_speaker(hass, False)
|
||||
assert hass.states.get(ENTITY_ID).state == STATE_OFF
|
||||
|
||||
# Device turns on
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get(ENTITY_ID).state == STATE_ON
|
||||
attr = hass.states.get(ENTITY_ID).attributes
|
||||
assert attr[ATTR_MEDIA_VOLUME_LEVEL] is not None
|
||||
assert ATTR_SOUND_MODE in attr
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_sound_mode_feature_toggling(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test sound mode feature is added when present and removed when absent."""
|
||||
await _test_setup_speaker(hass, True)
|
||||
attr = _get_attr_and_assert_base_attr(
|
||||
hass, MediaPlayerDeviceClass.SPEAKER, STATE_ON
|
||||
)
|
||||
assert ATTR_SOUND_MODE in attr
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert (
|
||||
state.attributes["supported_features"]
|
||||
& MediaPlayerEntityFeature.SELECT_SOUND_MODE
|
||||
)
|
||||
|
||||
# Update with audio settings that have no sound mode
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_all_settings",
|
||||
return_value={"volume": 50, "mute": "Off"},
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=True,
|
||||
),
|
||||
):
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_ON
|
||||
assert not (
|
||||
state.attributes["supported_features"]
|
||||
& MediaPlayerEntityFeature.SELECT_SOUND_MODE
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("vizio_connect", "vizio_update")
|
||||
async def test_sound_mode_list_cached(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test sound mode list is cached after first retrieval."""
|
||||
await _test_setup_speaker(hass, True)
|
||||
attr = hass.states.get(ENTITY_ID).attributes
|
||||
assert attr["sound_mode_list"] == EQ_LIST
|
||||
|
||||
# Update with different sound mode options — cached list should persist
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_setting_options",
|
||||
return_value=["Different1", "Different2"],
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.vizio.VizioAsync.get_power_state",
|
||||
return_value=True,
|
||||
),
|
||||
):
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
attr = hass.states.get(ENTITY_ID).attributes
|
||||
# Sound mode list should still be the original cached list
|
||||
assert attr["sound_mode_list"] == EQ_LIST
|
||||
|
||||
Reference in New Issue
Block a user