mirror of https://github.com/home-assistant/core
Refactor Remote class in panasonic_viera (#34911)
This commit is contained in:
João Gabriel
GitHub
parent
bd72ddda3c
commit
4e55fa6c5c
|
|
@ -531,7 +531,6 @@ omit =
|
|||
homeassistant/components/osramlightify/light.py
|
||||
homeassistant/components/otp/sensor.py
|
||||
homeassistant/components/panasonic_bluray/media_player.py
|
||||
homeassistant/components/panasonic_viera/__init__.py
|
||||
homeassistant/components/panasonic_viera/media_player.py
|
||||
homeassistant/components/pandora/media_player.py
|
||||
homeassistant/components/pcal9535a/*
|
||||
|
|
|
|||
|
|
@ -1,13 +1,29 @@
|
|||
"""The Panasonic Viera integration."""
|
||||
import asyncio
|
||||
from functools import partial
|
||||
import logging
|
||||
from urllib.request import URLError
|
||||
|
||||
from panasonic_viera import EncryptionRequired, Keys, RemoteControl, SOAPError
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.media_player.const import DOMAIN as MEDIA_PLAYER_DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, STATE_OFF, STATE_ON
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.script import Script
|
||||
|
||||
from .const import CONF_ON_ACTION, DEFAULT_NAME, DEFAULT_PORT, DOMAIN
|
||||
from .const import (
|
||||
ATTR_REMOTE,
|
||||
CONF_APP_ID,
|
||||
CONF_ENCRYPTION_KEY,
|
||||
CONF_ON_ACTION,
|
||||
DEFAULT_NAME,
|
||||
DEFAULT_PORT,
|
||||
DOMAIN,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONFIG_SCHEMA = vol.Schema(
|
||||
{
|
||||
|
|
@ -28,7 +44,7 @@ CONFIG_SCHEMA = vol.Schema(
|
|||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
PLATFORMS = ["media_player"]
|
||||
PLATFORMS = [MEDIA_PLAYER_DOMAIN]
|
||||
|
||||
|
||||
async def async_setup(hass, config):
|
||||
|
|
@ -49,6 +65,27 @@ async def async_setup(hass, config):
|
|||
async def async_setup_entry(hass, config_entry):
|
||||
"""Set up Panasonic Viera from a config entry."""
|
||||
|
||||
panasonic_viera_data = hass.data.setdefault(DOMAIN, {})
|
||||
|
||||
config = config_entry.data
|
||||
|
||||
host = config[CONF_HOST]
|
||||
port = config[CONF_PORT]
|
||||
|
||||
on_action = config[CONF_ON_ACTION]
|
||||
if on_action is not None:
|
||||
on_action = Script(hass, on_action)
|
||||
|
||||
params = {}
|
||||
if CONF_APP_ID in config and CONF_ENCRYPTION_KEY in config:
|
||||
params["app_id"] = config[CONF_APP_ID]
|
||||
params["encryption_key"] = config[CONF_ENCRYPTION_KEY]
|
||||
|
||||
remote = Remote(hass, host, port, on_action, **params)
|
||||
await remote.async_create_remote_control(during_setup=True)
|
||||
|
||||
panasonic_viera_data[config_entry.entry_id] = {ATTR_REMOTE: remote}
|
||||
|
||||
for component in PLATFORMS:
|
||||
hass.async_create_task(
|
||||
hass.config_entries.async_forward_entry_setup(config_entry, component)
|
||||
|
|
@ -59,7 +96,7 @@ async def async_setup_entry(hass, config_entry):
|
|||
|
||||
async def async_unload_entry(hass, config_entry):
|
||||
"""Unload a config entry."""
|
||||
return all(
|
||||
unload_ok = all(
|
||||
await asyncio.gather(
|
||||
*[
|
||||
hass.config_entries.async_forward_entry_unload(config_entry, component)
|
||||
|
|
@ -67,3 +104,135 @@ async def async_unload_entry(hass, config_entry):
|
|||
]
|
||||
)
|
||||
)
|
||||
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(config_entry.entry_id)
|
||||
|
||||
return unload_ok
|
||||
|
||||
|
||||
class Remote:
|
||||
"""The Remote class. It stores the TV properties and the remote control connection itself."""
|
||||
|
||||
def __init__(
|
||||
self, hass, host, port, on_action=None, app_id=None, encryption_key=None,
|
||||
):
|
||||
"""Initialize the Remote class."""
|
||||
self._hass = hass
|
||||
|
||||
self._host = host
|
||||
self._port = port
|
||||
|
||||
self._on_action = on_action
|
||||
|
||||
self._app_id = app_id
|
||||
self._encryption_key = encryption_key
|
||||
|
||||
self.state = None
|
||||
self.available = False
|
||||
self.volume = 0
|
||||
self.muted = False
|
||||
self.playing = True
|
||||
|
||||
self._control = None
|
||||
|
||||
async def async_create_remote_control(self, during_setup=False):
|
||||
"""Create remote control."""
|
||||
control_existed = self._control is not None
|
||||
try:
|
||||
params = {}
|
||||
if self._app_id and self._encryption_key:
|
||||
params["app_id"] = self._app_id
|
||||
params["encryption_key"] = self._encryption_key
|
||||
|
||||
self._control = await self._hass.async_add_executor_job(
|
||||
partial(RemoteControl, self._host, self._port, **params)
|
||||
)
|
||||
|
||||
self.state = STATE_ON
|
||||
self.available = True
|
||||
except (TimeoutError, URLError, SOAPError, OSError) as err:
|
||||
if control_existed or during_setup:
|
||||
_LOGGER.debug("Could not establish remote connection: %s", err)
|
||||
|
||||
self._control = None
|
||||
self.state = STATE_OFF
|
||||
self.available = self._on_action is not None
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
if control_existed or during_setup:
|
||||
_LOGGER.exception("An unknown error occurred: %s", err)
|
||||
self._control = None
|
||||
self.state = STATE_OFF
|
||||
self.available = self._on_action is not None
|
||||
|
||||
async def async_update(self):
|
||||
"""Update device data."""
|
||||
if self._control is None:
|
||||
await self.async_create_remote_control()
|
||||
return
|
||||
|
||||
await self._handle_errors(self._update)
|
||||
|
||||
def _update(self):
|
||||
"""Retrieve the latest data."""
|
||||
self.muted = self._control.get_mute()
|
||||
self.volume = self._control.get_volume() / 100
|
||||
|
||||
self.state = STATE_ON
|
||||
self.available = True
|
||||
|
||||
async def async_send_key(self, key):
|
||||
"""Send a key to the TV and handle exceptions."""
|
||||
try:
|
||||
key = getattr(Keys, key)
|
||||
except (AttributeError, TypeError):
|
||||
key = getattr(key, "value", key)
|
||||
|
||||
await self._handle_errors(self._control.send_key, key)
|
||||
|
||||
async def async_turn_on(self):
|
||||
"""Turn on the TV."""
|
||||
if self._on_action is not None:
|
||||
await self._on_action.async_run()
|
||||
self.state = STATE_ON
|
||||
elif self.state != STATE_ON:
|
||||
await self.async_send_key(Keys.power)
|
||||
self.state = STATE_ON
|
||||
|
||||
async def async_turn_off(self):
|
||||
"""Turn off the TV."""
|
||||
if self.state != STATE_OFF:
|
||||
await self.async_send_key(Keys.power)
|
||||
self.state = STATE_OFF
|
||||
await self.async_update()
|
||||
|
||||
async def async_set_mute(self, enable):
|
||||
"""Set mute based on 'enable'."""
|
||||
await self._handle_errors(self._control.set_mute, enable)
|
||||
|
||||
async def async_set_volume(self, volume):
|
||||
"""Set volume level, range 0..1."""
|
||||
volume = int(volume * 100)
|
||||
await self._handle_errors(self._control.set_volume, volume)
|
||||
|
||||
async def async_play_media(self, media_type, media_id):
|
||||
"""Play media."""
|
||||
_LOGGER.debug("Play media: %s (%s)", media_id, media_type)
|
||||
await self._handle_errors(self._control.open_webpage, media_id)
|
||||
|
||||
async def _handle_errors(self, func, *args):
|
||||
"""Handle errors from func, set available and reconnect if needed."""
|
||||
try:
|
||||
return await self._hass.async_add_executor_job(func, *args)
|
||||
except EncryptionRequired:
|
||||
_LOGGER.error(
|
||||
"The connection couldn't be encrypted. Please reconfigure your TV"
|
||||
)
|
||||
except (TimeoutError, URLError, SOAPError, OSError):
|
||||
self.state = STATE_OFF
|
||||
self.available = self._on_action is not None
|
||||
await self.async_create_remote_control()
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.exception("An unknown error occurred: %s", err)
|
||||
self.state = STATE_OFF
|
||||
self.available = self._on_action is not None
|
||||
|
|
|
|||
|
|
@ -10,6 +10,8 @@ CONF_ENCRYPTION_KEY = "encryption_key"
|
|||
DEFAULT_NAME = "Panasonic Viera TV"
|
||||
DEFAULT_PORT = 55000
|
||||
|
||||
ATTR_REMOTE = "remote"
|
||||
|
||||
ERROR_NOT_CONNECTED = "not_connected"
|
||||
ERROR_INVALID_PIN_CODE = "invalid_pin_code"
|
||||
|
||||
|
|
|
|||
|
|
@ -1,9 +1,7 @@
|
|||
"""Support for interface with a Panasonic Viera TV."""
|
||||
from functools import partial
|
||||
"""Media player support for Panasonic Viera TV."""
|
||||
import logging
|
||||
from urllib.request import URLError
|
||||
|
||||
from panasonic_viera import EncryptionRequired, Keys, RemoteControl, SOAPError
|
||||
from panasonic_viera import Keys
|
||||
|
||||
from homeassistant.components.media_player import MediaPlayerEntity
|
||||
from homeassistant.components.media_player.const import (
|
||||
|
|
@ -20,10 +18,9 @@ from homeassistant.components.media_player.const import (
|
|||
SUPPORT_VOLUME_SET,
|
||||
SUPPORT_VOLUME_STEP,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, STATE_OFF, STATE_ON
|
||||
from homeassistant.helpers.script import Script
|
||||
from homeassistant.const import CONF_NAME
|
||||
|
||||
from .const import CONF_APP_ID, CONF_ENCRYPTION_KEY, CONF_ON_ACTION
|
||||
from .const import ATTR_REMOTE, DOMAIN
|
||||
|
||||
SUPPORT_VIERATV = (
|
||||
SUPPORT_PAUSE
|
||||
|
|
@ -47,42 +44,25 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
|
|||
|
||||
config = config_entry.data
|
||||
|
||||
host = config[CONF_HOST]
|
||||
port = config[CONF_PORT]
|
||||
remote = hass.data[DOMAIN][config_entry.entry_id][ATTR_REMOTE]
|
||||
name = config[CONF_NAME]
|
||||
|
||||
on_action = config[CONF_ON_ACTION]
|
||||
if on_action is not None:
|
||||
on_action = Script(hass, on_action)
|
||||
|
||||
params = {}
|
||||
if CONF_APP_ID in config and CONF_ENCRYPTION_KEY in config:
|
||||
params["app_id"] = config[CONF_APP_ID]
|
||||
params["encryption_key"] = config[CONF_ENCRYPTION_KEY]
|
||||
|
||||
remote = Remote(hass, host, port, on_action, **params)
|
||||
await remote.async_create_remote_control(during_setup=True)
|
||||
|
||||
tv_device = PanasonicVieraTVDevice(remote, name)
|
||||
|
||||
tv_device = PanasonicVieraTVEntity(remote, name)
|
||||
async_add_entities([tv_device])
|
||||
|
||||
|
||||
class PanasonicVieraTVDevice(MediaPlayerEntity):
|
||||
class PanasonicVieraTVEntity(MediaPlayerEntity):
|
||||
"""Representation of a Panasonic Viera TV."""
|
||||
|
||||
def __init__(
|
||||
self, remote, name, uuid=None,
|
||||
):
|
||||
"""Initialize the Panasonic device."""
|
||||
# Save a reference to the imported class
|
||||
def __init__(self, remote, name, uuid=None):
|
||||
"""Initialize the entity."""
|
||||
self._remote = remote
|
||||
self._name = name
|
||||
self._uuid = uuid
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return the unique ID of this Viera TV."""
|
||||
def unique_id(self):
|
||||
"""Return the unique ID of the device."""
|
||||
return self._uuid
|
||||
|
||||
@property
|
||||
|
|
@ -97,7 +77,7 @@ class PanasonicVieraTVDevice(MediaPlayerEntity):
|
|||
|
||||
@property
|
||||
def available(self):
|
||||
"""Return if True the device is available."""
|
||||
"""Return True if the device is available."""
|
||||
return self._remote.available
|
||||
|
||||
@property
|
||||
|
|
@ -176,125 +156,8 @@ class PanasonicVieraTVDevice(MediaPlayerEntity):
|
|||
|
||||
async def async_play_media(self, media_type, media_id, **kwargs):
|
||||
"""Play media."""
|
||||
await self._remote.async_play_media(media_type, media_id)
|
||||
|
||||
|
||||
class Remote:
|
||||
"""The Remote class. It stores the TV properties and the remote control connection itself."""
|
||||
|
||||
def __init__(
|
||||
self, hass, host, port, on_action=None, app_id=None, encryption_key=None,
|
||||
):
|
||||
"""Initialize the Remote class."""
|
||||
self._hass = hass
|
||||
|
||||
self._host = host
|
||||
self._port = port
|
||||
|
||||
self._on_action = on_action
|
||||
|
||||
self._app_id = app_id
|
||||
self._encryption_key = encryption_key
|
||||
|
||||
self.state = None
|
||||
self.available = False
|
||||
self.volume = 0
|
||||
self.muted = False
|
||||
self.playing = True
|
||||
|
||||
self._control = None
|
||||
|
||||
async def async_create_remote_control(self, during_setup=False):
|
||||
"""Create remote control."""
|
||||
control_existed = self._control is not None
|
||||
try:
|
||||
params = {}
|
||||
if self._app_id and self._encryption_key:
|
||||
params["app_id"] = self._app_id
|
||||
params["encryption_key"] = self._encryption_key
|
||||
|
||||
self._control = await self._hass.async_add_executor_job(
|
||||
partial(RemoteControl, self._host, self._port, **params)
|
||||
)
|
||||
|
||||
self.state = STATE_ON
|
||||
self.available = True
|
||||
except (TimeoutError, URLError, SOAPError, OSError) as err:
|
||||
if control_existed or during_setup:
|
||||
_LOGGER.error("Could not establish remote connection: %s", err)
|
||||
|
||||
self._control = None
|
||||
self.state = STATE_OFF
|
||||
self.available = self._on_action is not None
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
if control_existed or during_setup:
|
||||
_LOGGER.exception("An unknown error occurred: %s", err)
|
||||
self._control = None
|
||||
self.state = STATE_OFF
|
||||
self.available = self._on_action is not None
|
||||
|
||||
async def async_update(self):
|
||||
"""Update device data."""
|
||||
if self._control is None:
|
||||
await self.async_create_remote_control()
|
||||
return
|
||||
|
||||
await self._handle_errors(self._update)
|
||||
|
||||
async def _update(self):
|
||||
"""Retrieve the latest data."""
|
||||
self.muted = self._control.get_mute()
|
||||
self.volume = self._control.get_volume() / 100
|
||||
|
||||
self.state = STATE_ON
|
||||
self.available = True
|
||||
|
||||
async def async_send_key(self, key):
|
||||
"""Send a key to the TV and handle exceptions."""
|
||||
await self._handle_errors(self._control.send_key, key)
|
||||
|
||||
async def async_turn_on(self):
|
||||
"""Turn on the TV."""
|
||||
if self._on_action is not None:
|
||||
await self._on_action.async_run()
|
||||
self.state = STATE_ON
|
||||
elif self.state != STATE_ON:
|
||||
await self.async_send_key(Keys.power)
|
||||
self.state = STATE_ON
|
||||
|
||||
async def async_turn_off(self):
|
||||
"""Turn off the TV."""
|
||||
if self.state != STATE_OFF:
|
||||
await self.async_send_key(Keys.power)
|
||||
self.state = STATE_OFF
|
||||
await self.async_update()
|
||||
|
||||
async def async_set_mute(self, enable):
|
||||
"""Set mute based on 'enable'."""
|
||||
await self._handle_errors(self._control.set_mute, enable)
|
||||
|
||||
async def async_set_volume(self, volume):
|
||||
"""Set volume level, range 0..1."""
|
||||
volume = int(volume * 100)
|
||||
await self._handle_errors(self._control.set_volume, volume)
|
||||
|
||||
async def async_play_media(self, media_type, media_id):
|
||||
"""Play media."""
|
||||
_LOGGER.debug("Play media: %s (%s)", media_id, media_type)
|
||||
|
||||
if media_type != MEDIA_TYPE_URL:
|
||||
_LOGGER.warning("Unsupported media_type: %s", media_type)
|
||||
return
|
||||
|
||||
await self._handle_errors(self._control.open_webpage, media_id)
|
||||
|
||||
async def _handle_errors(self, func, *args):
|
||||
"""Handle errors from func, set available and reconnect if needed."""
|
||||
try:
|
||||
await self._hass.async_add_executor_job(func, *args)
|
||||
except EncryptionRequired:
|
||||
_LOGGER.error("The connection couldn't be encrypted")
|
||||
except (TimeoutError, URLError, SOAPError, OSError):
|
||||
self.state = STATE_OFF
|
||||
self.available = self._on_action is not None
|
||||
await self.async_create_remote_control()
|
||||
await self._remote.async_play_media(media_type, media_id)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,123 @@
|
|||
"""Test the Panasonic Viera setup process."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from asynctest import patch
|
||||
|
||||
from homeassistant.components.panasonic_viera.const import (
|
||||
CONF_APP_ID,
|
||||
CONF_ENCRYPTION_KEY,
|
||||
CONF_ON_ACTION,
|
||||
DEFAULT_NAME,
|
||||
DEFAULT_PORT,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.config_entries import ENTRY_STATE_NOT_LOADED
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
MOCK_CONFIG_DATA = {
|
||||
CONF_HOST: "0.0.0.0",
|
||||
CONF_NAME: DEFAULT_NAME,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
CONF_ON_ACTION: None,
|
||||
}
|
||||
|
||||
MOCK_ENCRYPTION_DATA = {
|
||||
CONF_APP_ID: "mock-app-id",
|
||||
CONF_ENCRYPTION_KEY: "mock-encryption-key",
|
||||
}
|
||||
|
||||
|
||||
def get_mock_remote():
|
||||
"""Return a mock remote."""
|
||||
mock_remote = Mock()
|
||||
|
||||
async def async_create_remote_control(during_setup=False):
|
||||
return
|
||||
|
||||
mock_remote.async_create_remote_control = async_create_remote_control
|
||||
|
||||
return mock_remote
|
||||
|
||||
|
||||
async def test_setup_entry_encrypted(hass):
|
||||
"""Test setup with encrypted config entry."""
|
||||
mock_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
unique_id=MOCK_CONFIG_DATA[CONF_HOST],
|
||||
data={**MOCK_CONFIG_DATA, **MOCK_ENCRYPTION_DATA},
|
||||
)
|
||||
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
mock_remote = get_mock_remote()
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.panasonic_viera.Remote", return_value=mock_remote,
|
||||
):
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("media_player.panasonic_viera_tv")
|
||||
|
||||
assert state
|
||||
assert state.name == DEFAULT_NAME
|
||||
|
||||
|
||||
async def test_setup_entry_unencrypted(hass):
|
||||
"""Test setup with unencrypted config entry."""
|
||||
mock_entry = MockConfigEntry(
|
||||
domain=DOMAIN, unique_id=MOCK_CONFIG_DATA[CONF_HOST], data=MOCK_CONFIG_DATA,
|
||||
)
|
||||
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
mock_remote = get_mock_remote()
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.panasonic_viera.Remote", return_value=mock_remote,
|
||||
):
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("media_player.panasonic_viera_tv")
|
||||
|
||||
assert state
|
||||
assert state.name == DEFAULT_NAME
|
||||
|
||||
|
||||
async def test_setup_config_flow_initiated(hass):
|
||||
"""Test if config flow is initiated in setup."""
|
||||
assert (
|
||||
await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_HOST: "0.0.0.0"}},)
|
||||
is True
|
||||
)
|
||||
|
||||
assert len(hass.config_entries.flow.async_progress()) == 1
|
||||
|
||||
|
||||
async def test_setup_unload_entry(hass):
|
||||
"""Test if config entry is unloaded."""
|
||||
mock_entry = MockConfigEntry(
|
||||
domain=DOMAIN, unique_id=MOCK_CONFIG_DATA[CONF_HOST], data=MOCK_CONFIG_DATA
|
||||
)
|
||||
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
mock_remote = get_mock_remote()
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.panasonic_viera.Remote", return_value=mock_remote,
|
||||
):
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await hass.config_entries.async_unload(mock_entry.entry_id)
|
||||
|
||||
assert mock_entry.state == ENTRY_STATE_NOT_LOADED
|
||||
|
||||
state = hass.states.get("media_player.panasonic_viera_tv")
|
||||
|
||||
assert state is None
|
||||
Loading…
Reference in New Issue