From 93d363ea577beafe497b5d4152baa25f6a491b8a Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sat, 13 Jan 2024 22:37:04 -1000 Subject: [PATCH] Improve apple_tv typing (#107694) --- homeassistant/components/apple_tv/__init__.py | 51 +++++++++++-------- .../components/apple_tv/media_player.py | 20 +++++--- homeassistant/components/apple_tv/remote.py | 2 +- 3 files changed, 43 insertions(+), 30 deletions(-) diff --git a/homeassistant/components/apple_tv/__init__.py b/homeassistant/components/apple_tv/__init__.py index 818d27bcf77..8f52db13cfa 100644 --- a/homeassistant/components/apple_tv/__init__.py +++ b/homeassistant/components/apple_tv/__init__.py @@ -2,10 +2,13 @@ import asyncio import logging from random import randrange +from typing import TYPE_CHECKING, cast from pyatv import connect, exceptions, scan +from pyatv.conf import AppleTV from pyatv.const import DeviceModel, Protocol from pyatv.convert import model_str +from pyatv.interface import AppleTV as AppleTVInterface, DeviceListener from homeassistant.components import zeroconf from homeassistant.config_entries import ConfigEntry @@ -92,10 +95,14 @@ class AppleTVEntity(Entity): _attr_has_entity_name = True _attr_name = None - def __init__(self, name, identifier, manager): + def __init__( + self, name: str, identifier: str | None, manager: "AppleTVManager" + ) -> None: """Initialize device.""" - self.atv = None + self.atv: AppleTVInterface = None # type: ignore[assignment] self.manager = manager + if TYPE_CHECKING: + assert identifier is not None self._attr_unique_id = identifier self._attr_device_info = DeviceInfo( identifiers={(DOMAIN, identifier)}, @@ -143,7 +150,7 @@ class AppleTVEntity(Entity): """Handle when connection was lost to device.""" -class AppleTVManager: +class AppleTVManager(DeviceListener): """Connection and power manager for an Apple TV. An instance is used per device to share the same power state between @@ -151,11 +158,11 @@ class AppleTVManager: in case of problems. """ - def __init__(self, hass, config_entry): + def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None: """Initialize power manager.""" self.config_entry = config_entry self.hass = hass - self.atv = None + self.atv: AppleTVInterface | None = None self.is_on = not config_entry.options.get(CONF_START_OFF, False) self._connection_attempts = 0 self._connection_was_lost = False @@ -220,7 +227,7 @@ class AppleTVManager: "Not starting connect loop (%s, %s)", self.atv is None, self.is_on ) - async def connect_once(self, raise_missing_credentials): + async def connect_once(self, raise_missing_credentials: bool) -> None: """Try to connect once.""" try: if conf := await self._scan(): @@ -264,49 +271,51 @@ class AppleTVManager: _LOGGER.debug("Connect loop ended") self._task = None - async def _scan(self): + async def _scan(self) -> AppleTV | None: """Try to find device by scanning for it.""" - identifiers = set( - self.config_entry.data.get(CONF_IDENTIFIERS, [self.config_entry.unique_id]) + config_entry = self.config_entry + identifiers: set[str] = set( + config_entry.data.get(CONF_IDENTIFIERS, [config_entry.unique_id]) ) - address = self.config_entry.data[CONF_ADDRESS] + address: str = config_entry.data[CONF_ADDRESS] + hass = self.hass # Only scan for and set up protocols that was successfully paired protocols = { - Protocol(int(protocol)) - for protocol in self.config_entry.data[CONF_CREDENTIALS] + Protocol(int(protocol)) for protocol in config_entry.data[CONF_CREDENTIALS] } - _LOGGER.debug("Discovering device %s", self.config_entry.title) - aiozc = await zeroconf.async_get_async_instance(self.hass) + _LOGGER.debug("Discovering device %s", config_entry.title) + aiozc = await zeroconf.async_get_async_instance(hass) atvs = await scan( - self.hass.loop, + hass.loop, identifier=identifiers, protocol=protocols, hosts=[address], aiozc=aiozc, ) if atvs: - return atvs[0] + return cast(AppleTV, atvs[0]) _LOGGER.debug( "Failed to find device %s with address %s", - self.config_entry.title, + config_entry.title, address, ) # We no longer multicast scan for the device since as soon as async_step_zeroconf runs, # it will update the address and reload the config entry when the device is found. return None - async def _connect(self, conf, raise_missing_credentials): + async def _connect(self, conf: AppleTV, raise_missing_credentials: bool) -> None: """Connect to device.""" - credentials = self.config_entry.data[CONF_CREDENTIALS] - name = self.config_entry.data[CONF_NAME] + config_entry = self.config_entry + credentials: dict[int, str | None] = config_entry.data[CONF_CREDENTIALS] + name: str = config_entry.data[CONF_NAME] missing_protocols = [] for protocol_int, creds in credentials.items(): protocol = Protocol(int(protocol_int)) if conf.get_service(protocol) is not None: - conf.set_credentials(protocol, creds) + conf.set_credentials(protocol, creds) # type: ignore[arg-type] else: missing_protocols.append(protocol.name) diff --git a/homeassistant/components/apple_tv/media_player.py b/homeassistant/components/apple_tv/media_player.py index dd1f554919e..e8fd9d5acfc 100644 --- a/homeassistant/components/apple_tv/media_player.py +++ b/homeassistant/components/apple_tv/media_player.py @@ -154,9 +154,9 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity): _LOGGER.exception("Failed to update app list") else: self._app_list = { - app.name: app.identifier - for app in sorted(apps, key=lambda app: app.name.lower()) - if app.name is not None + app_name: app.identifier + for app in sorted(apps, key=lambda app: app_name.lower()) + if (app_name := app.name) is not None } self.async_write_ha_state() @@ -214,15 +214,19 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity): @property def app_id(self) -> str | None: """ID of the current running app.""" - if self._is_feature_available(FeatureName.App): - return self.atv.metadata.app.identifier + if self._is_feature_available(FeatureName.App) and ( + app := self.atv.metadata.app + ): + return app.identifier return None @property def app_name(self) -> str | None: """Name of the current running app.""" - if self._is_feature_available(FeatureName.App): - return self.atv.metadata.app.name + if self._is_feature_available(FeatureName.App) and ( + app := self.atv.metadata.app + ): + return app.name return None @property @@ -479,7 +483,7 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity): async def async_media_seek(self, position: float) -> None: """Send seek command.""" if self.atv: - await self.atv.remote_control.set_position(position) + await self.atv.remote_control.set_position(round(position)) async def async_volume_up(self) -> None: """Turn volume up for media player.""" diff --git a/homeassistant/components/apple_tv/remote.py b/homeassistant/components/apple_tv/remote.py index bab3421c58d..24d2ef68ed4 100644 --- a/homeassistant/components/apple_tv/remote.py +++ b/homeassistant/components/apple_tv/remote.py @@ -81,5 +81,5 @@ class AppleTVRemote(AppleTVEntity, RemoteEntity): raise ValueError("Command not found. Exiting sequence") _LOGGER.info("Sending command %s", single_command) - await attr_value() + await attr_value() # type: ignore[operator] await asyncio.sleep(delay)