self-hosted/home-assistant-core
self-hosted/home-assistant-core
1
0
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-01 18:09:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Strict typing for SamsungTV (#53585)

Browse source 
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Simone Chemelli 2021-09-18 06:51:07 +02:00 committed by GitHub
parent f5dd71d1e0
commit 4160a5ee3b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 234 additions and 137 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -87,6 +87,7 @@ homeassistant.components.recorder.statistics
homeassistant.components.remote.*
homeassistant.components.renault.*
homeassistant.components.rituals_perfume_genie.*
homeassistant.components.samsungtv.*
homeassistant.components.scene.*
homeassistant.components.select.*
homeassistant.components.sensor.*

36
homeassistant/components/samsungtv/__init__.py
View file

@ -1,13 +1,16 @@
"""The Samsung TV integration."""
from __future__ import annotations
from functools import partial
import socket
from typing import Any
import getmac
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components.media_player.const import DOMAIN as MP_DOMAIN
from homeassistant.config_entries import ConfigEntryNotReady
from homeassistant.config_entries import ConfigEntry, ConfigEntryNotReady
from homeassistant.const import (
CONF_HOST,
CONF_MAC,
@ -17,10 +20,17 @@ from homeassistant.const import (
CONF_TOKEN,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import callback
from homeassistant.core import Event, HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .bridge import SamsungTVBridge, async_get_device_info, mac_from_device_info
from .bridge import (
SamsungTVBridge,
SamsungTVLegacyBridge,
SamsungTVWSBridge,
async_get_device_info,
mac_from_device_info,
)
from .const import (
CONF_ON_ACTION,
DEFAULT_NAME,
@ -32,7 +42,7 @@ from .const import (
)
def ensure_unique_hosts(value):
def ensure_unique_hosts(value: dict[Any, Any]) -> dict[Any, Any]:
"""Validate that all configs have a unique host."""
vol.Schema(vol.Unique("duplicate host entries found"))(
[entry[CONF_HOST] for entry in value]
@ -64,7 +74,7 @@ CONFIG_SCHEMA = vol.Schema(
)
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Samsung TV integration."""
hass.data[DOMAIN] = {}
if DOMAIN not in config:
@ -88,7 +98,9 @@ async def async_setup(hass, config):
@callback
def _async_get_device_bridge(data):
def _async_get_device_bridge(
data: dict[str, Any]
) -> SamsungTVLegacyBridge | SamsungTVWSBridge:
"""Get device bridge."""
return SamsungTVBridge.get_bridge(
data[CONF_METHOD],
@ -98,13 +110,13 @@ def _async_get_device_bridge(data):
)
async def async_setup_entry(hass, entry):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up the Samsung TV platform."""
# Initialize bridge
bridge = await _async_create_bridge_with_updated_data(hass, entry)
def stop_bridge(event):
def stop_bridge(event: Event) -> None:
"""Stop SamsungTV bridge connection."""
bridge.stop()
@ -117,7 +129,9 @@ async def async_setup_entry(hass, entry):
return True
async def _async_create_bridge_with_updated_data(hass, entry):
async def _async_create_bridge_with_updated_data(
hass: HomeAssistant, entry: ConfigEntry
) -> SamsungTVLegacyBridge | SamsungTVWSBridge:
"""Create a bridge object and update any missing data in the config entry."""
updated_data = {}
host = entry.data[CONF_HOST]
@ -163,7 +177,7 @@ async def _async_create_bridge_with_updated_data(hass, entry):
return bridge
async def async_unload_entry(hass, entry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
@ -171,7 +185,7 @@ async def async_unload_entry(hass, entry):
return unload_ok
async def async_migrate_entry(hass, config_entry):
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate old entry."""
version = config_entry.version

95
homeassistant/components/samsungtv/bridge.py
View file

@ -1,6 +1,9 @@
"""samsungctl and samsungtvws bridge classes."""
from __future__ import annotations
from abc import ABC, abstractmethod
import contextlib
from typing import Any
from samsungctl import Remote
from samsungctl.exceptions import AccessDenied, ConnectionClosed, UnhandledResponse
@ -17,6 +20,7 @@ from homeassistant.const import (
CONF_TIMEOUT,
CONF_TOKEN,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers.device_registry import format_mac
from .const import (
@ -37,7 +41,7 @@ from .const import (
)
def mac_from_device_info(info):
def mac_from_device_info(info: dict[str, Any]) -> str | None:
"""Extract the mac address from the device info."""
dev_info = info.get("device", {})
if dev_info.get("networkType") == "wireless" and dev_info.get("wifiMac"):
@ -45,12 +49,18 @@ def mac_from_device_info(info):
return None
async def async_get_device_info(hass, bridge, host):
async def async_get_device_info(
hass: HomeAssistant,
bridge: SamsungTVWSBridge | SamsungTVLegacyBridge | None,
host: str,
) -> tuple[int | None, str | None, dict[str, Any] | None]:
"""Fetch the port, method, and device info."""
return await hass.async_add_executor_job(_get_device_info, bridge, host)
def _get_device_info(bridge, host):
def _get_device_info(
bridge: SamsungTVWSBridge | SamsungTVLegacyBridge, host: str
) -> tuple[int | None, str | None, dict[str, Any] | None]:
"""Fetch the port, method, and device info."""
if bridge and bridge.port:
return bridge.port, bridge.method, bridge.device_info()
@ -72,40 +82,42 @@ class SamsungTVBridge(ABC):
"""The Base Bridge abstract class."""
@staticmethod
def get_bridge(method, host, port=None, token=None):
def get_bridge(
method: str, host: str, port: int | None = None, token: str | None = None
) -> SamsungTVLegacyBridge | SamsungTVWSBridge:
"""Get Bridge instance."""
if method == METHOD_LEGACY or port == LEGACY_PORT:
return SamsungTVLegacyBridge(method, host, port)
return SamsungTVWSBridge(method, host, port, token)
def __init__(self, method, host, port):
def __init__(self, method: str, host: str, port: int | None = None) -> None:
"""Initialize Bridge."""
self.port = port
self.method = method
self.host = host
self.token = None
self._remote = None
self._callback = None
self.token: str | None = None
self._remote: Remote | None = None
self._callback: CALLBACK_TYPE | None = None
def register_reauth_callback(self, func):
def register_reauth_callback(self, func: CALLBACK_TYPE) -> None:
"""Register a callback function."""
self._callback = func
@abstractmethod
def try_connect(self):
def try_connect(self) -> str | None:
"""Try to connect to the TV."""
@abstractmethod
def device_info(self):
def device_info(self) -> dict[str, Any] | None:
"""Try to gather infos of this TV."""
@abstractmethod
def mac_from_device(self):
def mac_from_device(self) -> str | None:
"""Try to fetch the mac address of the TV."""
def is_on(self):
def is_on(self) -> bool:
"""Tells if the TV is on."""
if self._remote:
if self._remote is not None:
self.close_remote()
try:
@ -121,7 +133,7 @@ class SamsungTVBridge(ABC):
# Different reasons, e.g. hostname not resolveable
return False
def send_key(self, key):
def send_key(self, key: str) -> None:
"""Send a key to the tv and handles exceptions."""
try:
# recreate connection if connection was dead
@ -146,14 +158,14 @@ class SamsungTVBridge(ABC):
pass
@abstractmethod
def _send_key(self, key):
def _send_key(self, key: str) -> None:
"""Send the key."""
@abstractmethod
def _get_remote(self, avoid_open: bool = False):
def _get_remote(self, avoid_open: bool = False) -> Remote:
"""Get Remote object."""
def close_remote(self):
def close_remote(self) -> None:
"""Close remote object."""
try:
if self._remote is not None:
@ -163,16 +175,16 @@ class SamsungTVBridge(ABC):
except OSError:
LOGGER.debug("Could not establish connection")
def _notify_callback(self):
def _notify_callback(self) -> None:
"""Notify access denied callback."""
if self._callback:
if self._callback is not None:
self._callback()
class SamsungTVLegacyBridge(SamsungTVBridge):
"""The Bridge for Legacy TVs."""
def __init__(self, method, host, port):
def __init__(self, method: str, host: str, port: int | None) -> None:
"""Initialize Bridge."""
super().__init__(method, host, LEGACY_PORT)
self.config = {
@ -185,11 +197,11 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
CONF_TIMEOUT: 1,
}
def mac_from_device(self):
def mac_from_device(self) -> None:
"""Try to fetch the mac address of the TV."""
return None
def try_connect(self):
def try_connect(self) -> str:
"""Try to connect to the Legacy TV."""
config = {
CONF_NAME: VALUE_CONF_NAME,
@ -216,11 +228,11 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
LOGGER.debug("Failing config: %s, error: %s", config, err)
return RESULT_CANNOT_CONNECT
def device_info(self):
def device_info(self) -> None:
"""Try to gather infos of this device."""
return None
def _get_remote(self, avoid_open: bool = False):
def _get_remote(self, avoid_open: bool = False) -> Remote:
"""Create or return a remote control instance."""
if self._remote is None:
# We need to create a new instance to reconnect.
@ -238,12 +250,12 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
pass
return self._remote
def _send_key(self, key):
def _send_key(self, key: str) -> None:
"""Send the key using legacy protocol."""
if remote := self._get_remote():
remote.control(key)
def stop(self):
def stop(self) -> None:
"""Stop Bridge."""
LOGGER.debug("Stopping SamsungTVLegacyBridge")
self.close_remote()
@ -252,17 +264,19 @@ class SamsungTVLegacyBridge(SamsungTVBridge):
class SamsungTVWSBridge(SamsungTVBridge):
"""The Bridge for WebSocket TVs."""
def __init__(self, method, host, port, token=None):
def __init__(
self, method: str, host: str, port: int | None = None, token: str | None = None
) -> None:
"""Initialize Bridge."""
super().__init__(method, host, port)
self.token = token
def mac_from_device(self):
def mac_from_device(self) -> str | None:
"""Try to fetch the mac address of the TV."""
info = self.device_info()
return mac_from_device_info(info) if info else None
def try_connect(self):
def try_connect(self) -> str:
"""Try to connect to the Websocket TV."""
for self.port in WEBSOCKET_PORTS:
config = {
@ -286,7 +300,7 @@ class SamsungTVWSBridge(SamsungTVBridge):
) as remote:
remote.open()
self.token = remote.token
if self.token:
if self.token is None:
config[CONF_TOKEN] = "*****"
LOGGER.debug("Working config: %s", config)
return RESULT_SUCCESS
@ -304,22 +318,23 @@ class SamsungTVWSBridge(SamsungTVBridge):
return RESULT_CANNOT_CONNECT
def device_info(self):
def device_info(self) -> dict[str, Any] | None:
"""Try to gather infos of this TV."""
remote = self._get_remote(avoid_open=True)
if not remote:
return None
with contextlib.suppress(HttpApiError):
return remote.rest_device_info()
if remote := self._get_remote(avoid_open=True):
with contextlib.suppress(HttpApiError):
device_info: dict[str, Any] = remote.rest_device_info()
return device_info
def _send_key(self, key):
return None
def _send_key(self, key: str) -> None:
"""Send the key using websocket protocol."""
if key == "KEY_POWEROFF":
key = "KEY_POWER"
if remote := self._get_remote():
remote.send_key(key)
def _get_remote(self, avoid_open: bool = False):
def _get_remote(self, avoid_open: bool = False) -> Remote:
"""Create or return a remote control instance."""
if self._remote is None:
# We need to create a new instance to reconnect.
@ -344,7 +359,7 @@ class SamsungTVWSBridge(SamsungTVBridge):
self._remote = None
return self._remote
def stop(self):
def stop(self) -> None:
"""Stop Bridge."""
LOGGER.debug("Stopping SamsungTVWSBridge")
self.close_remote()

111
homeassistant/components/samsungtv/config_flow.py
View file

@ -1,5 +1,9 @@
"""Config flow for Samsung TV."""
from __future__ import annotations
import socket
from types import MappingProxyType
from typing import Any
from urllib.parse import urlparse
import getmac
@ -25,7 +29,13 @@ from homeassistant.core import callback
from homeassistant.helpers.device_registry import format_mac
from homeassistant.helpers.typing import DiscoveryInfoType
from .bridge import SamsungTVBridge, async_get_device_info, mac_from_device_info
from .bridge import (
SamsungTVBridge,
SamsungTVLegacyBridge,
SamsungTVWSBridge,
async_get_device_info,
mac_from_device_info,
)
from .const import (
ATTR_PROPERTIES,
CONF_MANUFACTURER,
@ -48,11 +58,11 @@ DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str, vol.Required(CONF_NAME):
SUPPORTED_METHODS = [METHOD_LEGACY, METHOD_WEBSOCKET]
def _strip_uuid(udn):
def _strip_uuid(udn: str) -> str:
return udn[5:] if udn.startswith("uuid:") else udn
def _entry_is_complete(entry):
def _entry_is_complete(entry: config_entries.ConfigEntry) -> bool:
"""Return True if the config entry information is complete."""
return bool(entry.unique_id and entry.data.get(CONF_MAC))
@ -62,22 +72,24 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 2
def __init__(self):
def __init__(self) -> None:
"""Initialize flow."""
self._reauth_entry = None
self._host = None
self._mac = None
self._udn = None
self._manufacturer = None
self._model = None
self._name = None
self._title = None
self._id = None
self._bridge = None
self._device_info = None
self._reauth_entry: config_entries.ConfigEntry | None = None
self._host: str = ""
self._mac: str | None = None
self._udn: str | None = None
self._manufacturer: str | None = None
self._model: str | None = None
self._name: str | None = None
self._title: str = ""
self._id: int | None = None
self._bridge: SamsungTVLegacyBridge | SamsungTVWSBridge | None = None
self._device_info: dict[str, Any] | None = None
def _get_entry_from_bridge(self):
def _get_entry_from_bridge(self) -> data_entry_flow.FlowResult:
"""Get device entry."""
assert self._bridge
data = {
CONF_HOST: self._host,
CONF_MAC: self._mac,
@ -94,14 +106,16 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data=data,
)
async def _async_set_device_unique_id(self, raise_on_progress=True):
async def _async_set_device_unique_id(self, raise_on_progress: bool = True) -> None:
"""Set device unique_id."""
if not await self._async_get_and_check_device_info():
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED)
await self._async_set_unique_id_from_udn(raise_on_progress)
self._async_update_and_abort_for_matching_unique_id()
async def _async_set_unique_id_from_udn(self, raise_on_progress=True):
async def _async_set_unique_id_from_udn(
self, raise_on_progress: bool = True
) -> None:
"""Set the unique id from the udn."""
assert self._host is not None
await self.async_set_unique_id(self._udn, raise_on_progress=raise_on_progress)
@ -110,14 +124,14 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
):
raise data_entry_flow.AbortFlow("already_configured")
def _async_update_and_abort_for_matching_unique_id(self):
def _async_update_and_abort_for_matching_unique_id(self) -> None:
"""Abort and update host and mac if we have it."""
updates = {CONF_HOST: self._host}
if self._mac:
updates[CONF_MAC] = self._mac
self._abort_if_unique_id_configured(updates=updates)
def _try_connect(self):
def _try_connect(self) -> None:
"""Try to connect and check auth."""
for method in SUPPORTED_METHODS:
self._bridge = SamsungTVBridge.get_bridge(method, self._host)
@ -129,7 +143,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
LOGGER.debug("No working config found")
raise data_entry_flow.AbortFlow(RESULT_CANNOT_CONNECT)
async def _async_get_and_check_device_info(self):
async def _async_get_and_check_device_info(self) -> bool:
"""Try to get the device info."""
_port, _method, info = await async_get_device_info(
self.hass, self._bridge, self._host
@ -160,7 +174,9 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._device_info = info
return True
async def async_step_import(self, user_input=None):
async def async_step_import(
self, user_input: dict[str, Any]
) -> data_entry_flow.FlowResult:
"""Handle configuration by yaml file."""
# We need to import even if we cannot validate
# since the TV may be off at startup
@ -177,21 +193,24 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
data=user_input,
)
async def _async_set_name_host_from_input(self, user_input):
async def _async_set_name_host_from_input(self, user_input: dict[str, Any]) -> None:
try:
self._host = await self.hass.async_add_executor_job(
socket.gethostbyname, user_input[CONF_HOST]
)
except socket.gaierror as err:
raise data_entry_flow.AbortFlow(RESULT_UNKNOWN_HOST) from err
self._name = user_input.get(CONF_NAME, self._host)
self._name = user_input.get(CONF_NAME, self._host) or ""
self._title = self._name
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult:
"""Handle a flow initialized by the user."""
if user_input is not None:
await self._async_set_name_host_from_input(user_input)
await self.hass.async_add_executor_job(self._try_connect)
assert self._bridge
self._async_abort_entries_match({CONF_HOST: self._host})
if self._bridge.method != METHOD_LEGACY:
# Legacy bridge does not provide device info
@ -201,7 +220,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(step_id="user", data_schema=DATA_SCHEMA)
@callback
def _async_update_existing_host_entry(self):
def _async_update_existing_host_entry(self) -> config_entries.ConfigEntry | None:
"""Check existing entries and update them.
Returns the existing entry if it was updated.
@ -209,7 +228,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
for entry in self._async_current_entries(include_ignore=False):
if entry.data[CONF_HOST] != self._host:
continue
entry_kw_args = {}
entry_kw_args: dict = {}
if self.unique_id and entry.unique_id is None:
entry_kw_args["unique_id"] = self.unique_id
if self._mac and not entry.data.get(CONF_MAC):
@ -222,7 +241,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return entry
return None
async def _async_start_discovery_with_mac_address(self):
async def _async_start_discovery_with_mac_address(self) -> None:
"""Start discovery."""
assert self._host is not None
if (entry := self._async_update_existing_host_entry()) and entry.unique_id:
@ -232,25 +251,28 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._async_abort_if_host_already_in_progress()
@callback
def _async_abort_if_host_already_in_progress(self):
def _async_abort_if_host_already_in_progress(self) -> None:
self.context[CONF_HOST] = self._host
for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == self._host:
raise data_entry_flow.AbortFlow("already_in_progress")
@callback
def _abort_if_manufacturer_is_not_samsung(self):
def _abort_if_manufacturer_is_not_samsung(self) -> None:
if not self._manufacturer or not self._manufacturer.lower().startswith(
"samsung"
):
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED)
async def async_step_ssdp(self, discovery_info: DiscoveryInfoType):
async def async_step_ssdp(
self, discovery_info: DiscoveryInfoType
) -> data_entry_flow.FlowResult:
"""Handle a flow initialized by ssdp discovery."""
LOGGER.debug("Samsung device found via SSDP: %s", discovery_info)
model_name = discovery_info.get(ATTR_UPNP_MODEL_NAME)
model_name: str = discovery_info.get(ATTR_UPNP_MODEL_NAME) or ""
self._udn = _strip_uuid(discovery_info[ATTR_UPNP_UDN])
self._host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
if hostname := urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname:
self._host = hostname
await self._async_set_unique_id_from_udn()
self._manufacturer = discovery_info[ATTR_UPNP_MANUFACTURER]
self._abort_if_manufacturer_is_not_samsung()
@ -263,7 +285,9 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.context["title_placeholders"] = {"device": self._title}
return await self.async_step_confirm()
async def async_step_dhcp(self, discovery_info: DiscoveryInfoType):
async def async_step_dhcp(
self, discovery_info: DiscoveryInfoType
) -> data_entry_flow.FlowResult:
"""Handle a flow initialized by dhcp discovery."""
LOGGER.debug("Samsung device found via DHCP: %s", discovery_info)
self._mac = discovery_info[MAC_ADDRESS]
@ -273,7 +297,9 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.context["title_placeholders"] = {"device": self._title}
return await self.async_step_confirm()
async def async_step_zeroconf(self, discovery_info: DiscoveryInfoType):
async def async_step_zeroconf(
self, discovery_info: DiscoveryInfoType
) -> data_entry_flow.FlowResult:
"""Handle a flow initialized by zeroconf discovery."""
LOGGER.debug("Samsung device found via ZEROCONF: %s", discovery_info)
self._mac = format_mac(discovery_info[ATTR_PROPERTIES]["deviceid"])
@ -283,11 +309,14 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.context["title_placeholders"] = {"device": self._title}
return await self.async_step_confirm()
async def async_step_confirm(self, user_input=None):
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult:
"""Handle user-confirmation of discovered node."""
if user_input is not None:
await self.hass.async_add_executor_job(self._try_connect)
assert self._bridge
return self._get_entry_from_bridge()
self._set_confirm_only()
@ -295,11 +324,14 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="confirm", description_placeholders={"device": self._title}
)
async def async_step_reauth(self, data):
async def async_step_reauth(
self, data: MappingProxyType[str, Any]
) -> data_entry_flow.FlowResult:
"""Handle configuration by re-auth."""
self._reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
assert self._reauth_entry
data = self._reauth_entry.data
if data.get(CONF_MODEL) and data.get(CONF_NAME):
self._title = f"{data[CONF_NAME]} ({data[CONF_MODEL]})"
@ -307,9 +339,12 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._title = data.get(CONF_NAME) or data[CONF_HOST]
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(self, user_input=None):
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult:
"""Confirm reauth."""
errors = {}
assert self._reauth_entry
if user_input is not None:
bridge = SamsungTVBridge.get_bridge(
self._reauth_entry.data[CONF_METHOD], self._reauth_entry.data[CONF_HOST]

108
homeassistant/components/samsungtv/media_player.py
View file

@ -1,6 +1,9 @@
"""Support for interface with an Samsung TV."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from datetime import datetime, timedelta
from typing import Any
import voluptuous as vol
from wakeonlan import send_magic_packet
@ -19,11 +22,18 @@ from homeassistant.components.media_player.const import (
SUPPORT_VOLUME_MUTE,
SUPPORT_VOLUME_STEP,
)
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.components.samsungtv.bridge import (
SamsungTVLegacyBridge,
SamsungTVWSBridge,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntry
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_NAME, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_component
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.script import Script
from homeassistant.util import dt as dt_util
@ -59,7 +69,9 @@ SCAN_INTERVAL_PLUS_OFF_TIME = entity_component.DEFAULT_SCAN_INTERVAL + timedelta
)
async def async_setup_entry(hass, entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the Samsung TV from a config entry."""
bridge = hass.data[DOMAIN][entry.entry_id]
@ -77,33 +89,38 @@ async def async_setup_entry(hass, entry, async_add_entities):
class SamsungTVDevice(MediaPlayerEntity):
"""Representation of a Samsung TV."""
def __init__(self, bridge, config_entry, on_script):
def __init__(
self,
bridge: SamsungTVLegacyBridge | SamsungTVWSBridge,
config_entry: ConfigEntry,
on_script: Script | None,
) -> None:
"""Initialize the Samsung device."""
self._config_entry = config_entry
self._host = config_entry.data[CONF_HOST]
self._mac = config_entry.data.get(CONF_MAC)
self._manufacturer = config_entry.data.get(CONF_MANUFACTURER)
self._model = config_entry.data.get(CONF_MODEL)
self._name = config_entry.data.get(CONF_NAME)
self._host: str | None = config_entry.data[CONF_HOST]
self._mac: str | None = config_entry.data.get(CONF_MAC)
self._manufacturer: str | None = config_entry.data.get(CONF_MANUFACTURER)
self._model: str | None = config_entry.data.get(CONF_MODEL)
self._name: str | None = config_entry.data.get(CONF_NAME)
self._on_script = on_script
self._uuid = config_entry.unique_id
# Assume that the TV is not muted
self._muted = False
self._muted: bool = False
# Assume that the TV is in Play mode
self._playing = True
self._state = None
self._playing: bool = True
self._state: str | None = None
# Mark the end of a shutdown command (need to wait 15 seconds before
# sending the next command to avoid turning the TV back ON).
self._end_of_power_off = None
self._end_of_power_off: datetime | None = None
self._bridge = bridge
self._auth_failed = False
self._bridge.register_reauth_callback(self.access_denied)
def access_denied(self):
def access_denied(self) -> None:
"""Access denied callback."""
LOGGER.debug("Access denied in getting remote object")
self._auth_failed = True
self.hass.add_job(
self.hass.async_create_task(
self.hass.config_entries.flow.async_init(
DOMAIN,
context={
@ -114,7 +131,7 @@ class SamsungTVDevice(MediaPlayerEntity):
)
)
def update(self):
def update(self) -> None:
"""Update state of device."""
if self._auth_failed:
return
@ -123,82 +140,83 @@ class SamsungTVDevice(MediaPlayerEntity):
else:
self._state = STATE_ON if self._bridge.is_on() else STATE_OFF
def send_key(self, key):
def send_key(self, key: str) -> None:
"""Send a key to the tv and handles exceptions."""
if self._power_off_in_progress() and key != "KEY_POWEROFF":
LOGGER.info("TV is powering off, not sending command: %s", key)
return
self._bridge.send_key(key)
def _power_off_in_progress(self):
def _power_off_in_progress(self) -> bool:
return (
self._end_of_power_off is not None
and self._end_of_power_off > dt_util.utcnow()
)
@property
def unique_id(self) -> str:
def unique_id(self) -> str | None:
"""Return the unique ID of the device."""
return self._uuid
@property
def name(self):
def name(self) -> str | None:
"""Return the name of the device."""
return self._name
@property
def state(self):
def state(self) -> str | None:
"""Return the state of the device."""
return self._state
@property
def available(self):
def available(self) -> bool:
"""Return the availability of the device."""
if self._auth_failed:
return False
return (
self._state == STATE_ON
or self._on_script
or self._mac
or self._on_script is not None
or self._mac is not None
or self._power_off_in_progress()
)
@property
def device_info(self):
def device_info(self) -> DeviceInfo | None:
"""Return device specific attributes."""
info = {
info: DeviceInfo = {
"name": self.name,
"identifiers": {(DOMAIN, self.unique_id)},
"manufacturer": self._manufacturer,
"model": self._model,
}
if self.unique_id:
info["identifiers"] = {(DOMAIN, self.unique_id)}
if self._mac:
info["connections"] = {(CONNECTION_NETWORK_MAC, self._mac)}
return info
@property
def is_volume_muted(self):
def is_volume_muted(self) -> bool:
"""Boolean if volume is currently muted."""
return self._muted
@property
def source_list(self):
def source_list(self) -> list:
"""List of available input sources."""
return list(SOURCES)
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag media player features that are supported."""
if self._on_script or self._mac:
return SUPPORT_SAMSUNGTV | SUPPORT_TURN_ON
return SUPPORT_SAMSUNGTV
@property
def device_class(self):
def device_class(self) -> str:
"""Set the device class to TV."""
return DEVICE_CLASS_TV
def turn_off(self):
def turn_off(self) -> None:
"""Turn off media player."""
self._end_of_power_off = dt_util.utcnow() + SCAN_INTERVAL_PLUS_OFF_TIME
@ -206,44 +224,46 @@ class SamsungTVDevice(MediaPlayerEntity):
# Force closing of remote session to provide instant UI feedback
self._bridge.close_remote()
def volume_up(self):
def volume_up(self) -> None:
"""Volume up the media player."""
self.send_key("KEY_VOLUP")
def volume_down(self):
def volume_down(self) -> None:
"""Volume down media player."""
self.send_key("KEY_VOLDOWN")
def mute_volume(self, mute):
def mute_volume(self, mute: bool) -> None:
"""Send mute command."""
self.send_key("KEY_MUTE")
def media_play_pause(self):
def media_play_pause(self) -> None:
"""Simulate play pause media player."""
if self._playing:
self.media_pause()
else:
self.media_play()
def media_play(self):
def media_play(self) -> None:
"""Send play command."""
self._playing = True
self.send_key("KEY_PLAY")
def media_pause(self):
def media_pause(self) -> None:
"""Send media pause command to media player."""
self._playing = False
self.send_key("KEY_PAUSE")
def media_next_track(self):
def media_next_track(self) -> None:
"""Send next track command."""
self.send_key("KEY_CHUP")
def media_previous_track(self):
def media_previous_track(self) -> None:
"""Send the previous track command."""
self.send_key("KEY_CHDOWN")
async def async_play_media(self, media_type, media_id, **kwargs):
async def async_play_media(
self, media_type: str, media_id: str, **kwargs: Any
) -> None:
"""Support changing a channel."""
if media_type != MEDIA_TYPE_CHANNEL:
LOGGER.error("Unsupported media type")
@ -261,21 +281,21 @@ class SamsungTVDevice(MediaPlayerEntity):
await asyncio.sleep(KEY_PRESS_TIMEOUT, self.hass.loop)
await self.hass.async_add_executor_job(self.send_key, "KEY_ENTER")
def _wake_on_lan(self):
def _wake_on_lan(self) -> None:
"""Wake the device via wake on lan."""
send_magic_packet(self._mac, ip_address=self._host)
# If the ip address changed since we last saw the device
# broadcast a packet as well
send_magic_packet(self._mac)
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn the media player on."""
if self._on_script:
await self._on_script.async_run(context=self._context)
elif self._mac:
await self.hass.async_add_executor_job(self._wake_on_lan)
def select_source(self, source):
def select_source(self, source: str) -> None:
"""Select input source."""
if source not in SOURCES:
LOGGER.error("Unsupported source")

5
homeassistant/components/samsungtv/strings.json
View file

@ -14,7 +14,7 @@
},
"reauth_confirm": {
"description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds."
}
}
},
"error": {
"auth_missing": "[%key:component::samsungtv::config::abort::auth_missing%]"
@ -27,7 +27,8 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"not_supported": "This Samsung device is currently not supported.",
"unknown": "[%key:common::config_flow::error::unknown%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"missing_config_entry": "This Samsung device doesn't have a configuration entry."
}
}
}

4
homeassistant/components/samsungtv/translations/en.json
View file

@ -6,6 +6,7 @@
"auth_missing": "Home Assistant is not authorized to connect to this Samsung TV. Check your TV's External Device Manager settings to authorize Home Assistant.",
"cannot_connect": "Failed to connect",
"id_missing": "This Samsung device doesn't have a SerialNumber.",
"missing_config_entry": "This Samsung device doesn't have a configuration entry.",
"not_supported": "This Samsung device is currently not supported.",
"reauth_successful": "Re-authentication was successful",
"unknown": "Unexpected error"
@ -16,8 +17,7 @@
"flow_title": "{device}",
"step": {
"confirm": {
"description": "Do you want to set up {device}? If you never connected Home Assistant before you should see a popup on your TV asking for authorization.",
"title": "Samsung TV"
"description": "Do you want to set up {device}? If you never connected Home Assistant before you should see a popup on your TV asking for authorization."
},
"reauth_confirm": {
"description": "After submitting, accept the the popup on {device} requesting authorization within 30 seconds."

11
mypy.ini
View file

@ -968,6 +968,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.samsungtv.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.scene.*]
check_untyped_defs = true
disallow_incomplete_defs = true