diff --git a/.strict-typing b/.strict-typing
index b664fc3b886b22bdf8d8bb47f680130a774b7fa5..64fbcb9e82dccce7c735019d6de5a01bee7b61db 100644
--- a/.strict-typing
+++ b/.strict-typing
@@ -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.*
diff --git a/homeassistant/components/samsungtv/__init__.py b/homeassistant/components/samsungtv/__init__.py
index 773c340d7b9c45c277188c577538cccddc5d9f64..f55dc0639bafbbe3e4e2c6a8488b45315c8df008 100644
--- a/homeassistant/components/samsungtv/__init__.py
+++ b/homeassistant/components/samsungtv/__init__.py
@@ -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 .bridge import SamsungTVBridge, async_get_device_info, mac_from_device_info
+from homeassistant.helpers.typing import ConfigType
+
+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
diff --git a/homeassistant/components/samsungtv/bridge.py b/homeassistant/components/samsungtv/bridge.py
index 0d00a0cb94f3f3be594afd70a4c95b3ce626684f..262bf4ce67f09e7a8d25a2ce22122a4ee15e5f08 100644
--- a/homeassistant/components/samsungtv/bridge.py
+++ b/homeassistant/components/samsungtv/bridge.py
@@ -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
+
+ return None
- def _send_key(self, key):
+ 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()
diff --git a/homeassistant/components/samsungtv/config_flow.py b/homeassistant/components/samsungtv/config_flow.py
index da13d0fe70c6b41b30d359a14ac11c05b22bcd88..bcce5eec5ed17a72703498202172ece29f8b0596 100644
--- a/homeassistant/components/samsungtv/config_flow.py
+++ b/homeassistant/components/samsungtv/config_flow.py
@@ -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
-
- def _get_entry_from_bridge(self):
+ 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) -> 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]
diff --git a/homeassistant/components/samsungtv/media_player.py b/homeassistant/components/samsungtv/media_player.py
index 7efdcdcd43908dd025ad729bbfaf33dfbf444e3f..4e17c65b461dce7c72c7fbab23e51c4e24e2c999 100644
--- a/homeassistant/components/samsungtv/media_player.py
+++ b/homeassistant/components/samsungtv/media_player.py
@@ -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")
diff --git a/homeassistant/components/samsungtv/strings.json b/homeassistant/components/samsungtv/strings.json
index f92990e6163feb911369eba7ab17069b5ef48cc9..89ac85f85ebd2911c4c85ff04b69d1f45938f027 100644
--- a/homeassistant/components/samsungtv/strings.json
+++ b/homeassistant/components/samsungtv/strings.json
@@ -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."
}
}
}
\ No newline at end of file
diff --git a/homeassistant/components/samsungtv/translations/en.json b/homeassistant/components/samsungtv/translations/en.json
index 91576e76ee51b98fa81a77a8d0fc58f91ef2ce6a..fa5369012c098110a3ce123633a6550230356793 100644
--- a/homeassistant/components/samsungtv/translations/en.json
+++ b/homeassistant/components/samsungtv/translations/en.json
@@ -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."
diff --git a/mypy.ini b/mypy.ini
index e8524d236b38d1fe47a71b170c9a743422eb0530..084355c2beb25cd23f25beec77497a7ff0583d03 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -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