Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Migrate BraviaTV to new async backend (#75727)

pull/76565/head
Artem Draft 2022-08-10 14:11:49 +03:00 committed by GitHub
parent 4a938ec33e
commit 19295d33ba
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 385 additions and 363 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -139,6 +139,7 @@ omit =
homeassistant/components/bosch_shc/switch.py
homeassistant/components/braviatv/__init__.py
homeassistant/components/braviatv/const.py
homeassistant/components/braviatv/coordinator.py
homeassistant/components/braviatv/entity.py
homeassistant/components/braviatv/media_player.py
homeassistant/components/braviatv/remote.py

251
homeassistant/components/braviatv/__init__.py
View File

@ -1,27 +1,20 @@
"""The Bravia TV component."""
"""The Bravia TV integration."""
from __future__ import annotations
import asyncio
from collections.abc import Iterable
from datetime import timedelta
import logging
from typing import Final
from bravia_tv import BraviaRC
from bravia_tv.braviarc import NoIPControl
from aiohttp import CookieJar
from pybravia import BraviaTV
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import CLIENTID_PREFIX, CONF_IGNORED_SOURCES, DOMAIN, NICKNAME
_LOGGER = logging.getLogger(__name__)
from .const import CONF_IGNORED_SOURCES, DOMAIN
from .coordinator import BraviaTVCoordinator
PLATFORMS: Final[list[Platform]] = [Platform.MEDIA_PLAYER, Platform.REMOTE]
SCAN_INTERVAL: Final = timedelta(seconds=10)
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
@ -31,7 +24,11 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
pin = config_entry.data[CONF_PIN]
ignored_sources = config_entry.options.get(CONF_IGNORED_SOURCES, [])
coordinator = BraviaTVCoordinator(hass, host, mac, pin, ignored_sources)
session = async_create_clientsession(
hass, cookie_jar=CookieJar(unsafe=True, quote_cookie=False)
)
client = BraviaTV(host, mac, session=session)
coordinator = BraviaTVCoordinator(hass, client, pin, ignored_sources)
config_entry.async_on_unload(config_entry.add_update_listener(update_listener))
await coordinator.async_config_entry_first_refresh()
@ -59,229 +56,3 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(config_entry.entry_id)
class BraviaTVCoordinator(DataUpdateCoordinator[None]):
"""Representation of a Bravia TV Coordinator.
An instance is used per device to share the same power state between
several platforms.
"""
def __init__(
self,
hass: HomeAssistant,
host: str,
mac: str,
pin: str,
ignored_sources: list[str],
) -> None:
"""Initialize Bravia TV Client."""
self.braviarc = BraviaRC(host, mac)
self.pin = pin
self.ignored_sources = ignored_sources
self.muted: bool = False
self.channel_name: str | None = None
self.media_title: str | None = None
self.source: str | None = None
self.source_list: list[str] = []
self.original_content_list: list[str] = []
self.content_mapping: dict[str, str] = {}
self.duration: int | None = None
self.content_uri: str | None = None
self.program_media_type: str | None = None
self.audio_output: str | None = None
self.min_volume: int | None = None
self.max_volume: int | None = None
self.volume_level: float | None = None
self.is_on = False
# Assume that the TV is in Play mode
self.playing = True
self.state_lock = asyncio.Lock()
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
request_refresh_debouncer=Debouncer(
hass, _LOGGER, cooldown=1.0, immediate=False
),
)
def _send_command(self, command: Iterable[str], repeats: int = 1) -> None:
"""Send a command to the TV."""
for _ in range(repeats):
for cmd in command:
self.braviarc.send_command(cmd)
def _get_source(self) -> str | None:
"""Return the name of the source."""
for key, value in self.content_mapping.items():
if value == self.content_uri:
return key
return None
def _refresh_volume(self) -> bool:
"""Refresh volume information."""
volume_info = self.braviarc.get_volume_info(self.audio_output)
if volume_info is not None:
volume = volume_info.get("volume")
self.volume_level = volume / 100 if volume is not None else None
self.audio_output = volume_info.get("target")
self.min_volume = volume_info.get("minVolume")
self.max_volume = volume_info.get("maxVolume")
self.muted = volume_info.get("mute", False)
return True
return False
def _refresh_channels(self) -> bool:
"""Refresh source and channels list."""
if not self.source_list:
self.content_mapping = self.braviarc.load_source_list()
self.source_list = []
if not self.content_mapping:
return False
for key in self.content_mapping:
if key not in self.ignored_sources:
self.source_list.append(key)
return True
def _refresh_playing_info(self) -> None:
"""Refresh playing information."""
playing_info = self.braviarc.get_playing_info()
program_name = playing_info.get("programTitle")
self.channel_name = playing_info.get("title")
self.program_media_type = playing_info.get("programMediaType")
self.content_uri = playing_info.get("uri")
self.source = self._get_source()
self.duration = playing_info.get("durationSec")
if not playing_info:
self.channel_name = "App"
if self.channel_name is not None:
self.media_title = self.channel_name
if program_name is not None:
self.media_title = f"{self.media_title}: {program_name}"
else:
self.media_title = None
def _update_tv_data(self) -> None:
"""Connect and update TV info."""
power_status = self.braviarc.get_power_status()
if power_status != "off":
connected = self.braviarc.is_connected()
if not connected:
try:
connected = self.braviarc.connect(
self.pin, CLIENTID_PREFIX, NICKNAME
)
except NoIPControl:
_LOGGER.error("IP Control is disabled in the TV settings")
if not connected:
power_status = "off"
if power_status == "active":
self.is_on = True
if self._refresh_volume() and self._refresh_channels():
self._refresh_playing_info()
return
self.is_on = False
async def _async_update_data(self) -> None:
"""Fetch the latest data."""
if self.state_lock.locked():
return
await self.hass.async_add_executor_job(self._update_tv_data)
async def async_turn_on(self) -> None:
"""Turn the device on."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.turn_on)
await self.async_request_refresh()
async def async_turn_off(self) -> None:
"""Turn off device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.turn_off)
await self.async_request_refresh()
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
async with self.state_lock:
await self.hass.async_add_executor_job(
self.braviarc.set_volume_level, volume, self.audio_output
)
await self.async_request_refresh()
async def async_volume_up(self) -> None:
"""Send volume up command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(
self.braviarc.volume_up, self.audio_output
)
await self.async_request_refresh()
async def async_volume_down(self) -> None:
"""Send volume down command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(
self.braviarc.volume_down, self.audio_output
)
await self.async_request_refresh()
async def async_volume_mute(self, mute: bool) -> None:
"""Send mute command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.mute_volume, mute)
await self.async_request_refresh()
async def async_media_play(self) -> None:
"""Send play command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_play)
self.playing = True
await self.async_request_refresh()
async def async_media_pause(self) -> None:
"""Send pause command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_pause)
self.playing = False
await self.async_request_refresh()
async def async_media_stop(self) -> None:
"""Send stop command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_stop)
self.playing = False
await self.async_request_refresh()
async def async_media_next_track(self) -> None:
"""Send next track command."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_next_track)
await self.async_request_refresh()
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.media_previous_track)
await self.async_request_refresh()
async def async_select_source(self, source: str) -> None:
"""Set the input source."""
if source in self.content_mapping:
uri = self.content_mapping[source]
async with self.state_lock:
await self.hass.async_add_executor_job(self.braviarc.play_content, uri)
await self.async_request_refresh()
async def async_send_command(self, command: Iterable[str], repeats: int) -> None:
"""Send command to device."""
async with self.state_lock:
await self.hass.async_add_executor_job(self._send_command, command, repeats)
await self.async_request_refresh()

123
homeassistant/components/braviatv/config_flow.py
View File

@ -1,4 +1,4 @@
"""Adds config flow for Bravia TV integration."""
"""Config flow to configure the Bravia TV integration."""
from __future__ import annotations
from contextlib import suppress
@ -6,17 +6,19 @@ import ipaddress
import re
from typing import Any
from bravia_tv import BraviaRC
from bravia_tv.braviarc import NoIPControl
from aiohttp import CookieJar
from pybravia import BraviaTV, BraviaTVError, BraviaTVNotSupported
import voluptuous as vol
from homeassistant import config_entries, exceptions
from homeassistant import config_entries
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_create_clientsession
import homeassistant.helpers.config_validation as cv
from . import BraviaTVCoordinator
from .const import (
ATTR_CID,
ATTR_MAC,
@ -38,39 +40,15 @@ def host_valid(host: str) -> bool:
class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for BraviaTV integration."""
"""Handle a config flow for Bravia TV integration."""
VERSION = 1
client: BraviaTV
def __init__(self) -> None:
"""Initialize."""
self.braviarc: BraviaRC | None = None
self.host: str | None = None
self.title = ""
self.mac: str | None = None
async def init_device(self, pin: str) -> None:
"""Initialize Bravia TV device."""
assert self.braviarc is not None
await self.hass.async_add_executor_job(
self.braviarc.connect, pin, CLIENTID_PREFIX, NICKNAME
)
connected = await self.hass.async_add_executor_job(self.braviarc.is_connected)
if not connected:
raise CannotConnect()
system_info = await self.hass.async_add_executor_job(
self.braviarc.get_system_info
)
if not system_info:
raise ModelNotSupported()
await self.async_set_unique_id(system_info[ATTR_CID].lower())
self._abort_if_unique_id_configured()
self.title = system_info[ATTR_MODEL]
self.mac = system_info[ATTR_MAC]
"""Initialize config flow."""
self.device_config: dict[str, Any] = {}
@staticmethod
@callback
@ -78,6 +56,24 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Bravia TV options callback."""
return BraviaTVOptionsFlowHandler(config_entry)
async def async_init_device(self) -> FlowResult:
"""Initialize and create Bravia TV device from config."""
pin = self.device_config[CONF_PIN]
await self.client.connect(pin=pin, clientid=CLIENTID_PREFIX, nickname=NICKNAME)
await self.client.set_wol_mode(True)
system_info = await self.client.get_system_info()
cid = system_info[ATTR_CID].lower()
title = system_info[ATTR_MODEL]
self.device_config[CONF_MAC] = system_info[ATTR_MAC]
await self.async_set_unique_id(cid)
self._abort_if_unique_id_configured()
return self.async_create_entry(title=title, data=self.device_config)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -85,9 +81,14 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors: dict[str, str] = {}
if user_input is not None:
if host_valid(user_input[CONF_HOST]):
self.host = user_input[CONF_HOST]
self.braviarc = BraviaRC(self.host)
host = user_input[CONF_HOST]
if host_valid(host):
session = async_create_clientsession(
self.hass,
cookie_jar=CookieJar(unsafe=True, quote_cookie=False),
)
self.client = BraviaTV(host=host, session=session)
self.device_config[CONF_HOST] = host
return await self.async_step_authorize()
@ -106,23 +107,17 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors: dict[str, str] = {}
if user_input is not None:
self.device_config[CONF_PIN] = user_input[CONF_PIN]
try:
await self.init_device(user_input[CONF_PIN])
except CannotConnect:
errors["base"] = "cannot_connect"
except ModelNotSupported:
return await self.async_init_device()
except BraviaTVNotSupported:
errors["base"] = "unsupported_model"
else:
user_input[CONF_HOST] = self.host
user_input[CONF_MAC] = self.mac
return self.async_create_entry(title=self.title, data=user_input)
# Connecting with th PIN "0000" to start the pairing process on the TV.
except BraviaTVError:
errors["base"] = "cannot_connect"
try:
assert self.braviarc is not None
await self.hass.async_add_executor_job(
self.braviarc.connect, "0000", CLIENTID_PREFIX, NICKNAME
)
except NoIPControl:
await self.client.pair(CLIENTID_PREFIX, NICKNAME)
except BraviaTVError:
return self.async_abort(reason="no_ip_control")
return self.async_show_form(
@ -138,26 +133,20 @@ class BraviaTVOptionsFlowHandler(config_entries.OptionsFlow):
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize Bravia TV options flow."""
self.config_entry = config_entry
self.pin = config_entry.data[CONF_PIN]
self.ignored_sources = config_entry.options.get(CONF_IGNORED_SOURCES)
self.source_list: dict[str, str] = {}
self.source_list: list[str] = []
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options."""
coordinator = self.hass.data[DOMAIN][self.config_entry.entry_id]
braviarc = coordinator.braviarc
connected = await self.hass.async_add_executor_job(braviarc.is_connected)
if not connected:
await self.hass.async_add_executor_job(
braviarc.connect, self.pin, CLIENTID_PREFIX, NICKNAME
)
coordinator: BraviaTVCoordinator = self.hass.data[DOMAIN][
self.config_entry.entry_id
]
content_mapping = await self.hass.async_add_executor_job(
braviarc.load_source_list
)
self.source_list = {item: item for item in content_mapping}
await coordinator.async_update_sources()
sources = coordinator.source_map.values()
self.source_list = [item["title"] for item in sources]
return await self.async_step_user()
async def async_step_user(
@ -177,11 +166,3 @@ class BraviaTVOptionsFlowHandler(config_entries.OptionsFlow):
}
),
)
class CannotConnect(exceptions.HomeAssistantError):
"""Error to indicate we cannot connect."""
class ModelNotSupported(exceptions.HomeAssistantError):
"""Error to indicate not supported model."""

1
homeassistant/components/braviatv/const.py
View File

@ -10,7 +10,6 @@ ATTR_MODEL: Final = "model"
CONF_IGNORED_SOURCES: Final = "ignored_sources"
BRAVIA_CONFIG_FILE: Final = "bravia.conf"
CLIENTID_PREFIX: Final = "HomeAssistant"
DOMAIN: Final = "braviatv"
NICKNAME: Final = "Home Assistant"

258
homeassistant/components/braviatv/coordinator.py Normal file
View File

@ -0,0 +1,258 @@
"""Update coordinator for Bravia TV integration."""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine, Iterable
from datetime import timedelta
from functools import wraps
import logging
from typing import Any, Final, TypeVar
from pybravia import BraviaTV, BraviaTVError
from typing_extensions import Concatenate, ParamSpec
from homeassistant.components.media_player.const import (
MEDIA_TYPE_APP,
MEDIA_TYPE_CHANNEL,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CLIENTID_PREFIX, DOMAIN, NICKNAME
_BraviaTVCoordinatorT = TypeVar("_BraviaTVCoordinatorT", bound="BraviaTVCoordinator")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL: Final = timedelta(seconds=10)
def catch_braviatv_errors(
func: Callable[Concatenate[_BraviaTVCoordinatorT, _P], Awaitable[None]]
) -> Callable[Concatenate[_BraviaTVCoordinatorT, _P], Coroutine[Any, Any, None]]:
"""Catch BraviaTV errors."""
@wraps(func)
async def wrapper(
self: _BraviaTVCoordinatorT,
*args: _P.args,
**kwargs: _P.kwargs,
) -> None:
"""Catch BraviaTV errors and log message."""
try:
await func(self, *args, **kwargs)
except BraviaTVError as err:
_LOGGER.error("Command error: %s", err)
await self.async_request_refresh()
return wrapper
class BraviaTVCoordinator(DataUpdateCoordinator[None]):
"""Representation of a Bravia TV Coordinator."""
def __init__(
self,
hass: HomeAssistant,
client: BraviaTV,
pin: str,
ignored_sources: list[str],
) -> None:
"""Initialize Bravia TV Client."""
self.client = client
self.pin = pin
self.ignored_sources = ignored_sources
self.source: str | None = None
self.source_list: list[str] = []
self.source_map: dict[str, dict] = {}
self.media_title: str | None = None
self.media_content_id: str | None = None
self.media_content_type: str | None = None
self.media_uri: str | None = None
self.media_duration: int | None = None
self.volume_level: float | None = None
self.volume_target: str | None = None
self.volume_muted = False
self.is_on = False
self.is_channel = False
self.connected = False
# Assume that the TV is in Play mode
self.playing = True
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
request_refresh_debouncer=Debouncer(
hass, _LOGGER, cooldown=1.0, immediate=False
),
)
def _sources_extend(self, sources: list[dict], source_type: str) -> None:
"""Extend source map and source list."""
for item in sources:
item["type"] = source_type
title = item.get("title")
uri = item.get("uri")
if not title or not uri:
continue
self.source_map[uri] = item
if title not in self.ignored_sources:
self.source_list.append(title)
async def _async_update_data(self) -> None:
"""Connect and fetch data."""
try:
if not self.connected:
await self.client.connect(
pin=self.pin, clientid=CLIENTID_PREFIX, nickname=NICKNAME
)
self.connected = True
power_status = await self.client.get_power_status()
self.is_on = power_status == "active"
if self.is_on is False:
return
if not self.source_map:
await self.async_update_sources()
await self.async_update_volume()
await self.async_update_playing()
except BraviaTVError as err:
self.is_on = False
self.connected = False
raise UpdateFailed("Error communicating with device") from err
async def async_update_sources(self) -> None:
"""Update sources."""
self.source_list = []
self.source_map = {}
externals = await self.client.get_external_status()
self._sources_extend(externals, "input")
apps = await self.client.get_app_list()
self._sources_extend(apps, "app")
channels = await self.client.get_content_list_all("tv")
self._sources_extend(channels, "channel")
async def async_update_volume(self) -> None:
"""Update volume information."""
volume_info = await self.client.get_volume_info()
volume_level = volume_info.get("volume")
if volume_level is not None:
self.volume_level = volume_level / 100
self.volume_muted = volume_info.get("mute", False)
self.volume_target = volume_info.get("target")
async def async_update_playing(self) -> None:
"""Update current playing information."""
playing_info = await self.client.get_playing_info()
self.media_title = playing_info.get("title")
self.media_uri = playing_info.get("uri")
self.media_duration = playing_info.get("durationSec")
if program_title := playing_info.get("programTitle"):
self.media_title = f"{self.media_title}: {program_title}"
if self.media_uri:
source = self.source_map.get(self.media_uri, {})
self.source = source.get("title")
self.is_channel = self.media_uri[:2] == "tv"
if self.is_channel:
self.media_content_id = playing_info.get("dispNum")
self.media_content_type = MEDIA_TYPE_CHANNEL
else:
self.media_content_id = self.media_uri
self.media_content_type = None
else:
self.source = None
self.is_channel = False
self.media_content_id = None
self.media_content_type = None
if not playing_info:
self.media_title = "Smart TV"
self.media_content_type = MEDIA_TYPE_APP
@catch_braviatv_errors
async def async_turn_on(self) -> None:
"""Turn the device on."""
await self.client.turn_on()
@catch_braviatv_errors
async def async_turn_off(self) -> None:
"""Turn off device."""
await self.client.turn_off()
@catch_braviatv_errors
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
await self.client.volume_level(round(volume * 100))
@catch_braviatv_errors
async def async_volume_up(self) -> None:
"""Send volume up command to device."""
await self.client.volume_up()
@catch_braviatv_errors
async def async_volume_down(self) -> None:
"""Send volume down command to device."""
await self.client.volume_down()
@catch_braviatv_errors
async def async_volume_mute(self, mute: bool) -> None:
"""Send mute command to device."""
await self.client.volume_mute()
@catch_braviatv_errors
async def async_media_play(self) -> None:
"""Send play command to device."""
await self.client.play()
self.playing = True
@catch_braviatv_errors
async def async_media_pause(self) -> None:
"""Send pause command to device."""
await self.client.pause()
self.playing = False
@catch_braviatv_errors
async def async_media_stop(self) -> None:
"""Send stop command to device."""
await self.client.stop()
@catch_braviatv_errors
async def async_media_next_track(self) -> None:
"""Send next track command."""
if self.is_channel:
await self.client.channel_up()
else:
await self.client.next_track()
@catch_braviatv_errors
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
if self.is_channel:
await self.client.channel_down()
else:
await self.client.previous_track()
@catch_braviatv_errors
async def async_select_source(self, source: str) -> None:
"""Set the input source."""
for uri, item in self.source_map.items():
if item.get("title") == source:
if item.get("type") == "app":
await self.client.set_active_app(uri)
else:
await self.client.set_play_content(uri)
break
@catch_braviatv_errors
async def async_send_command(self, command: Iterable[str], repeats: int) -> None:
"""Send command to device."""
for _ in range(repeats):
for cmd in command:
await self.client.send_command(cmd)

2
homeassistant/components/braviatv/entity.py
View File

@ -1,4 +1,4 @@
"""A entity class for BraviaTV integration."""
"""A entity class for Bravia TV integration."""
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity

4
homeassistant/components/braviatv/manifest.json
View File

@ -2,9 +2,9 @@
"domain": "braviatv",
"name": "Sony Bravia TV",
"documentation": "https://www.home-assistant.io/integrations/braviatv",
"requirements": ["bravia-tv==1.0.11"],
"requirements": ["pybravia==0.2.0"],
"codeowners": ["@bieniu", "@Drafteed"],
"config_flow": true,
"iot_class": "local_polling",
"loggers": ["bravia_tv"]
"loggers": ["pybravia"]
}

13
homeassistant/components/braviatv/media_player.py
View File

@ -1,4 +1,4 @@
"""Support for interface with a Bravia TV."""
"""Media player support for Bravia TV integration."""
from __future__ import annotations
from homeassistant.components.media_player import (
@ -74,7 +74,7 @@ class BraviaTVMediaPlayer(BraviaTVEntity, MediaPlayerEntity):
@property
def is_volume_muted(self) -> bool:
"""Boolean if volume is currently muted."""
return self.coordinator.muted
return self.coordinator.volume_muted
@property
def media_title(self) -> str | None:
@ -84,12 +84,17 @@ class BraviaTVMediaPlayer(BraviaTVEntity, MediaPlayerEntity):
@property
def media_content_id(self) -> str | None:
"""Content ID of current playing media."""
return self.coordinator.channel_name
return self.coordinator.media_content_id
@property
def media_content_type(self) -> str | None:
"""Content type of current playing media."""
return self.coordinator.media_content_type
@property
def media_duration(self) -> int | None:
"""Duration of current playing media in seconds."""
return self.coordinator.duration
return self.coordinator.media_duration
async def async_turn_on(self) -> None:
"""Turn the device on."""

6
requirements_all.txt
View File

@ -436,9 +436,6 @@ boschshcpy==0.2.30
# homeassistant.components.route53
boto3==1.20.24
# homeassistant.components.braviatv
bravia-tv==1.0.11
# homeassistant.components.broadlink
broadlink==0.18.2
@ -1421,6 +1418,9 @@ pyblackbird==0.5
# homeassistant.components.neato
pybotvac==0.0.23
# homeassistant.components.braviatv
pybravia==0.2.0
# homeassistant.components.nissan_leaf
pycarwings2==2.13

6
requirements_test_all.txt
View File

@ -343,9 +343,6 @@ bond-async==0.1.22
# homeassistant.components.bosch_shc
boschshcpy==0.2.30
# homeassistant.components.braviatv
bravia-tv==1.0.11
# homeassistant.components.broadlink
broadlink==0.18.2
@ -994,6 +991,9 @@ pyblackbird==0.5
# homeassistant.components.neato
pybotvac==0.0.23
# homeassistant.components.braviatv
pybravia==0.2.0
# homeassistant.components.cloudflare
pycfdns==1.2.2

83
tests/components/braviatv/test_config_flow.py
View File

@ -1,7 +1,7 @@
"""Define tests for the Bravia TV config flow."""
from unittest.mock import patch
from bravia_tv.braviarc import NoIPControl
from pybravia import BraviaTVConnectionError, BraviaTVNotSupported
from homeassistant import data_entry_flow
from homeassistant.components.braviatv.const import CONF_IGNORED_SOURCES, DOMAIN
@ -23,13 +23,13 @@ BRAVIA_SYSTEM_INFO = {
"cid": "very_unique_string",
}
BRAVIA_SOURCE_LIST = {
"HDMI 1": "extInput:hdmi?port=1",
"HDMI 2": "extInput:hdmi?port=2",
"HDMI 3/ARC": "extInput:hdmi?port=3",
"HDMI 4": "extInput:hdmi?port=4",
"AV/Component": "extInput:component?port=1",
}
BRAVIA_SOURCES = [
{"title": "HDMI 1", "uri": "extInput:hdmi?port=1"},
{"title": "HDMI 2", "uri": "extInput:hdmi?port=2"},
{"title": "HDMI 3/ARC", "uri": "extInput:hdmi?port=3"},
{"title": "HDMI 4", "uri": "extInput:hdmi?port=4"},
{"title": "AV/Component", "uri": "extInput:component?port=1"},
]
async def test_show_form(hass):
@ -53,9 +53,10 @@ async def test_user_invalid_host(hass):
async def test_authorize_cannot_connect(hass):
"""Test that errors are shown when cannot connect to host at the authorize step."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=False
):
with patch(
"pybravia.BraviaTV.connect",
side_effect=BraviaTVConnectionError,
), patch("pybravia.BraviaTV.pair"):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
)
@ -68,12 +69,14 @@ async def test_authorize_cannot_connect(hass):
async def test_authorize_model_unsupported(hass):
"""Test that errors are shown when the TV is not supported at the authorize step."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch("bravia_tv.BraviaRC.get_system_info", return_value={}):
with patch(
"pybravia.BraviaTV.connect",
side_effect=BraviaTVNotSupported,
), patch("pybravia.BraviaTV.pair"):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "10.10.10.12"}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_PIN: "1234"}
)
@ -83,13 +86,12 @@ async def test_authorize_model_unsupported(hass):
async def test_authorize_no_ip_control(hass):
"""Test that errors are shown when IP Control is disabled on the TV."""
with patch("bravia_tv.BraviaRC.connect", side_effect=NoIPControl("No IP Control")):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "no_ip_control"
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "no_ip_control"
async def test_duplicate_error(hass):
@ -106,9 +108,12 @@ async def test_duplicate_error(hass):
)
config_entry.add_to_hass(hass)
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch("bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO):
with patch("pybravia.BraviaTV.connect"), patch("pybravia.BraviaTV.pair"), patch(
"pybravia.BraviaTV.set_wol_mode"
), patch(
"pybravia.BraviaTV.get_system_info",
return_value=BRAVIA_SYSTEM_INFO,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
@ -123,10 +128,11 @@ async def test_duplicate_error(hass):
async def test_create_entry(hass):
"""Test that the user step works."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
with patch("pybravia.BraviaTV.connect"), patch("pybravia.BraviaTV.pair"), patch(
"pybravia.BraviaTV.set_wol_mode"
), patch(
"bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO
"pybravia.BraviaTV.get_system_info",
return_value=BRAVIA_SYSTEM_INFO,
), patch(
"homeassistant.components.braviatv.async_setup_entry", return_value=True
):
@ -154,10 +160,11 @@ async def test_create_entry(hass):
async def test_create_entry_with_ipv6_address(hass):
"""Test that the user step works with device IPv6 address."""
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
with patch("pybravia.BraviaTV.connect"), patch("pybravia.BraviaTV.pair"), patch(
"pybravia.BraviaTV.set_wol_mode"
), patch(
"bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO
"pybravia.BraviaTV.get_system_info",
return_value=BRAVIA_SYSTEM_INFO,
), patch(
"homeassistant.components.braviatv.async_setup_entry", return_value=True
):
@ -199,19 +206,19 @@ async def test_options_flow(hass):
)
config_entry.add_to_hass(hass)
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=True
), patch("bravia_tv.BraviaRC.get_power_status"), patch(
"bravia_tv.BraviaRC.get_system_info", return_value=BRAVIA_SYSTEM_INFO
with patch("pybravia.BraviaTV.connect"), patch(
"pybravia.BraviaTV.get_power_status",
return_value="active",
), patch(
"pybravia.BraviaTV.get_external_status",
return_value=BRAVIA_SOURCES,
), patch(
"pybravia.BraviaTV.send_rest_req",
return_value={},
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
with patch("bravia_tv.BraviaRC.connect", return_value=True), patch(
"bravia_tv.BraviaRC.is_connected", return_value=False
), patch("bravia_tv.BraviaRC.get_power_status"), patch(
"bravia_tv.BraviaRC.load_source_list", return_value=BRAVIA_SOURCE_LIST
):
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM