Enable basic typing for roku (#52478)

* enable basic typing for roku

* Update mypy.ini

* Update media_player.py

* Create coordinator.py

* Update __init__.py

* Update media_player.py

* Update remote.py

* Update media_player.py

* Update coordinator.py

* Update coordinator.py

* Update remote.py

* Update entity.py

* Update coordinator.py

* Update config_flow.py

* Update entity.py

* Update const.py

* Update const.py

* Update const.py

* Update entity.py

* Update entity.py

* Update entity.py

* Update test_media_player.py

* Update test_remote.py

homeassistant/components/roku/__init__.py

@@ -1,11 +1,9 @@
 """Support for Roku."""
 from __future__ import annotations
 
-from datetime import timedelta
 import logging
 
-from rokuecp import Roku, RokuConnectionError, RokuError
+from rokuecp import RokuConnectionError, RokuError
-from rokuecp.models import Device
 
 from homeassistant.components.media_player import DOMAIN as MEDIA_PLAYER_DOMAIN
 from homeassistant.components.remote import DOMAIN as REMOTE_DOMAIN
@@ -13,16 +11,13 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_HOST
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import config_validation as cv
-from homeassistant.helpers.aiohttp_client import async_get_clientsession
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
-from homeassistant.util.dt import utcnow
 
 from .const import DOMAIN
+from .coordinator import RokuDataUpdateCoordinator
 
 CONFIG_SCHEMA = cv.deprecated(DOMAIN)
 
 PLATFORMS = [MEDIA_PLAYER_DOMAIN, REMOTE_DOMAIN]
-SCAN_INTERVAL = timedelta(seconds=15)
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -63,42 +58,3 @@ def roku_exception_handler(func):
                 _LOGGER.error("Invalid response from API: %s", error)
 
     return handler
-
-
-class RokuDataUpdateCoordinator(DataUpdateCoordinator[Device]):
-    """Class to manage fetching Roku data."""
-
-    def __init__(
-        self,
-        hass: HomeAssistant,
-        *,
-        host: str,
-    ) -> None:
-        """Initialize global Roku data updater."""
-        self.roku = Roku(host=host, session=async_get_clientsession(hass))
-
-        self.full_update_interval = timedelta(minutes=15)
-        self.last_full_update = None
-
-        super().__init__(
-            hass,
-            _LOGGER,
-            name=DOMAIN,
-            update_interval=SCAN_INTERVAL,
-        )
-
-    async def _async_update_data(self) -> Device:
-        """Fetch data from Roku."""
-        full_update = self.last_full_update is None or utcnow() >= (
-            self.last_full_update + self.full_update_interval
-        )
-
-        try:
-            data = await self.roku.update(full_update=full_update)
-
-            if full_update:
-                self.last_full_update = utcnow()
-
-            return data
-        except RokuError as error:
-            raise UpdateFailed(f"Invalid response from API: {error}") from error

homeassistant/components/roku/config_flow.py

@@ -17,6 +17,7 @@ from homeassistant.const import CONF_HOST, CONF_NAME
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.data_entry_flow import FlowResult
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.typing import DiscoveryInfoType
 
 from .const import DOMAIN
 
@@ -111,7 +112,7 @@ class RokuConfigFlow(ConfigFlow, domain=DOMAIN):
 
         return await self.async_step_discovery_confirm()
 
-    async def async_step_ssdp(self, discovery_info: dict | None = None) -> FlowResult:
+    async def async_step_ssdp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
         """Handle a flow initialized by discovery."""
         host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
         name = discovery_info[ATTR_UPNP_FRIENDLY_NAME]

homeassistant/components/roku/const.py

@@ -2,12 +2,7 @@
 DOMAIN = "roku"
 
 # Attributes
-ATTR_IDENTIFIERS = "identifiers"
 ATTR_KEYWORD = "keyword"
-ATTR_MANUFACTURER = "manufacturer"
-ATTR_MODEL = "model"
-ATTR_SOFTWARE_VERSION = "sw_version"
-ATTR_SUGGESTED_AREA = "suggested_area"
 
 # Default Values
 DEFAULT_PORT = 8060

homeassistant/components/roku/coordinator.py

@@ -0,0 +1,60 @@
+"""Coordinator for Roku."""
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+import logging
+
+from rokuecp import Roku, RokuError
+from rokuecp.models import Device
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+from homeassistant.util.dt import utcnow
+
+from .const import DOMAIN
+
+SCAN_INTERVAL = timedelta(seconds=15)
+_LOGGER = logging.getLogger(__name__)
+
+
+class RokuDataUpdateCoordinator(DataUpdateCoordinator[Device]):
+    """Class to manage fetching Roku data."""
+
+    last_full_update: datetime | None
+    roku: Roku
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        *,
+        host: str,
+    ) -> None:
+        """Initialize global Roku data updater."""
+        self.roku = Roku(host=host, session=async_get_clientsession(hass))
+
+        self.full_update_interval = timedelta(minutes=15)
+        self.last_full_update = None
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=SCAN_INTERVAL,
+        )
+
+    async def _async_update_data(self) -> Device:
+        """Fetch data from Roku."""
+        full_update = self.last_full_update is None or utcnow() >= (
+            self.last_full_update + self.full_update_interval
+        )
+
+        try:
+            data = await self.roku.update(full_update=full_update)
+
+            if full_update:
+                self.last_full_update = utcnow()
+
+            return data
+        except RokuError as error:
+            raise UpdateFailed(f"Invalid response from API: {error}") from error

homeassistant/components/roku/entity.py

@@ -1,24 +1,25 @@
 """Base Entity for Roku."""
 from __future__ import annotations
 
-from homeassistant.const import ATTR_NAME
+from homeassistant.const import (
+    ATTR_IDENTIFIERS,
+    ATTR_MANUFACTURER,
+    ATTR_MODEL,
+    ATTR_NAME,
+    ATTR_SW_VERSION,
+)
 from homeassistant.helpers.entity import DeviceInfo
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from . import RokuDataUpdateCoordinator
-from .const import (
+from .const import DOMAIN
-    ATTR_IDENTIFIERS,
-    ATTR_MANUFACTURER,
-    ATTR_MODEL,
-    ATTR_SOFTWARE_VERSION,
-    ATTR_SUGGESTED_AREA,
-    DOMAIN,
-)
 
 
 class RokuEntity(CoordinatorEntity):
     """Defines a base Roku entity."""
 
+    coordinator: RokuDataUpdateCoordinator
+
     def __init__(
         self, *, device_id: str, coordinator: RokuDataUpdateCoordinator
     ) -> None:
@@ -34,9 +35,9 @@ class RokuEntity(CoordinatorEntity):
 
         return {
             ATTR_IDENTIFIERS: {(DOMAIN, self._device_id)},
-            ATTR_NAME: self.name,
+            ATTR_NAME: self.coordinator.data.info.name,
             ATTR_MANUFACTURER: self.coordinator.data.info.brand,
             ATTR_MODEL: self.coordinator.data.info.model_name,
-            ATTR_SOFTWARE_VERSION: self.coordinator.data.info.version,
+            ATTR_SW_VERSION: self.coordinator.data.info.version,
-            ATTR_SUGGESTED_AREA: self.coordinator.data.info.device_location,
+            "suggested_area": self.coordinator.data.info.device_location,
         }

homeassistant/components/roku/media_player.py

@@ -1,6 +1,7 @@
 """Support for the Roku media player."""
 from __future__ import annotations
 
+import datetime as dt
 import logging
 
 import voluptuous as vol
@@ -8,6 +9,7 @@ import voluptuous as vol
 from homeassistant.components.media_player import (
     DEVICE_CLASS_RECEIVER,
     DEVICE_CLASS_TV,
+    BrowseMedia,
     MediaPlayerEntity,
 )
 from homeassistant.components.media_player.const import (
@@ -37,9 +39,10 @@ from homeassistant.const import (
 from homeassistant.helpers import entity_platform
 from homeassistant.helpers.network import is_internal_request
 
-from . import RokuDataUpdateCoordinator, roku_exception_handler
+from . import roku_exception_handler
 from .browse_media import build_item_response, library_payload
 from .const import ATTR_KEYWORD, DOMAIN, SERVICE_SEARCH
+from .coordinator import RokuDataUpdateCoordinator
 from .entity import RokuEntity
 
 _LOGGER = logging.getLogger(__name__)
@@ -63,7 +66,7 @@ SEARCH_SCHEMA = {vol.Required(ATTR_KEYWORD): str}
 
 async def async_setup_entry(hass, entry, async_add_entities):
     """Set up the Roku config entry."""
-    coordinator = hass.data[DOMAIN][entry.entry_id]
+    coordinator: RokuDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
     unique_id = coordinator.data.info.serial_number
     async_add_entities([RokuMediaPlayer(unique_id, coordinator)], True)
 
@@ -88,6 +91,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
 
         self._attr_name = coordinator.data.info.name
         self._attr_unique_id = unique_id
+        self._attr_supported_features = SUPPORT_ROKU
 
     def _media_playback_trackable(self) -> bool:
         """Detect if we have enough media data to track playback."""
@@ -105,7 +109,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return DEVICE_CLASS_RECEIVER
 
     @property
-    def state(self) -> str:
+    def state(self) -> str | None:
         """Return the state of the device."""
         if self.coordinator.data.state.standby:
             return STATE_STANDBY
@@ -133,12 +137,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return None
 
     @property
-    def supported_features(self):
+    def media_content_type(self) -> str | None:
-        """Flag media player features that are supported."""
-        return SUPPORT_ROKU
-
-    @property
-    def media_content_type(self) -> str:
         """Content type of current playing media."""
         if self.app_id is None or self.app_name in ("Power Saver", "Roku"):
             return None
@@ -149,7 +148,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return MEDIA_TYPE_APP
 
     @property
-    def media_image_url(self) -> str:
+    def media_image_url(self) -> str | None:
         """Image url of current playing media."""
         if self.app_id is None or self.app_name in ("Power Saver", "Roku"):
             return None
@@ -157,7 +156,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return self.coordinator.roku.app_icon_url(self.app_id)
 
     @property
-    def app_name(self) -> str:
+    def app_name(self) -> str | None:
         """Name of the current running app."""
         if self.coordinator.data.app is not None:
             return self.coordinator.data.app.name
@@ -165,7 +164,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return None
 
     @property
-    def app_id(self) -> str:
+    def app_id(self) -> str | None:
         """Return the ID of the current running app."""
         if self.coordinator.data.app is not None:
             return self.coordinator.data.app.app_id
@@ -173,7 +172,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return None
 
     @property
-    def media_channel(self):
+    def media_channel(self) -> str | None:
         """Return the TV channel currently tuned."""
         if self.app_id != "tvinput.dtv" or self.coordinator.data.channel is None:
             return None
@@ -184,7 +183,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return self.coordinator.data.channel.number
 
     @property
-    def media_title(self):
+    def media_title(self) -> str | None:
         """Return the title of current playing media."""
         if self.app_id != "tvinput.dtv" or self.coordinator.data.channel is None:
             return None
@@ -195,7 +194,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return None
 
     @property
-    def media_duration(self):
+    def media_duration(self) -> int | None:
         """Duration of current playing media in seconds."""
         if self._media_playback_trackable():
             return self.coordinator.data.media.duration
@@ -203,7 +202,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return None
 
     @property
-    def media_position(self):
+    def media_position(self) -> int | None:
         """Position of current playing media in seconds."""
         if self._media_playback_trackable():
             return self.coordinator.data.media.position
@@ -211,7 +210,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return None
 
     @property
-    def media_position_updated_at(self):
+    def media_position_updated_at(self) -> dt.datetime | None:
         """When was the position of the current playing media valid."""
         if self._media_playback_trackable():
             return self.coordinator.data.media.at
@@ -219,7 +218,7 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         return None
 
     @property
-    def source(self) -> str:
+    def source(self) -> str | None:
         """Return the current input source."""
         if self.coordinator.data.app is not None:
             return self.coordinator.data.app.name
@@ -237,8 +236,11 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
         await self.coordinator.roku.search(keyword)
 
     async def async_get_browse_image(
-        self, media_content_type, media_content_id, media_image_id=None
+        self,
-    ):
+        media_content_type: str,
+        media_content_id: str,
+        media_image_id: str | None = None,
+    ) -> tuple[str | None, str | None]:
         """Fetch media browser image to serve via proxy."""
         if media_content_type == MEDIA_TYPE_APP and media_content_id:
             image_url = self.coordinator.roku.app_icon_url(media_content_id)
@@ -246,7 +248,11 @@ class RokuMediaPlayer(RokuEntity, MediaPlayerEntity):
 
         return (None, None)
 
-    async def async_browse_media(self, media_content_type=None, media_content_id=None):
+    async def async_browse_media(
+        self,
+        media_content_type: str | None = None,
+        media_content_id: str | None = None,
+    ) -> BrowseMedia:
         """Implement the websocket media browsing helper."""
         is_internal = is_internal_request(self.hass)
 

homeassistant/components/roku/remote.py

@@ -6,8 +6,9 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
-from . import RokuDataUpdateCoordinator, roku_exception_handler
+from . import roku_exception_handler
 from .const import DOMAIN
+from .coordinator import RokuDataUpdateCoordinator
 from .entity import RokuEntity
 
 
@@ -15,7 +16,7 @@ async def async_setup_entry(
     hass: HomeAssistant,
     entry: ConfigEntry,
     async_add_entities: AddEntitiesCallback,
-) -> bool:
+) -> None:
     """Load Roku remote based on a config entry."""
     coordinator = hass.data[DOMAIN][entry.entry_id]
     unique_id = coordinator.data.info.serial_number

mypy.ini

@@ -1465,9 +1465,6 @@ ignore_errors = true
 [mypy-homeassistant.components.ring.*]
 ignore_errors = true
 
-[mypy-homeassistant.components.roku.*]
-ignore_errors = true
-
 [mypy-homeassistant.components.rpi_power.*]
 ignore_errors = true
 

script/hassfest/mypy_config.py

@@ -165,7 +165,6 @@ IGNORED_MODULES: Final[list[str]] = [
     "homeassistant.components.recorder.*",
     "homeassistant.components.reddit.*",
     "homeassistant.components.ring.*",
-    "homeassistant.components.roku.*",
     "homeassistant.components.rpi_power.*",
     "homeassistant.components.ruckus_unleashed.*",
     "homeassistant.components.sabnzbd.*",

tests/components/roku/test_media_player.py

@@ -136,7 +136,7 @@ async def test_availability(
         await setup_integration(hass, aioclient_mock)
 
     with patch(
-        "homeassistant.components.roku.Roku.update", side_effect=RokuError
+        "homeassistant.components.roku.coordinator.Roku.update", side_effect=RokuError
     ), patch("homeassistant.util.dt.utcnow", return_value=future):
         async_fire_time_changed(hass, future)
         await hass.async_block_till_done()
@@ -336,21 +336,21 @@ async def test_services(
     """Test the different media player services."""
     await setup_integration(hass, aioclient_mock)
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: MAIN_ENTITY_ID}, blocking=True
         )
 
         remote_mock.assert_called_once_with("poweroff")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: MAIN_ENTITY_ID}, blocking=True
         )
 
         remote_mock.assert_called_once_with("poweron")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_MEDIA_PAUSE,
@@ -360,7 +360,7 @@ async def test_services(
 
         remote_mock.assert_called_once_with("play")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_MEDIA_PLAY,
@@ -370,7 +370,7 @@ async def test_services(
 
         remote_mock.assert_called_once_with("play")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_MEDIA_PLAY_PAUSE,
@@ -380,7 +380,7 @@ async def test_services(
 
         remote_mock.assert_called_once_with("play")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_MEDIA_NEXT_TRACK,
@@ -390,7 +390,7 @@ async def test_services(
 
         remote_mock.assert_called_once_with("forward")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_MEDIA_PREVIOUS_TRACK,
@@ -400,7 +400,7 @@ async def test_services(
 
         remote_mock.assert_called_once_with("reverse")
 
-    with patch("homeassistant.components.roku.Roku.launch") as launch_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.launch") as launch_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_PLAY_MEDIA,
@@ -414,7 +414,7 @@ async def test_services(
 
         launch_mock.assert_called_once_with("11")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_SELECT_SOURCE,
@@ -424,7 +424,7 @@ async def test_services(
 
         remote_mock.assert_called_once_with("home")
 
-    with patch("homeassistant.components.roku.Roku.launch") as launch_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.launch") as launch_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_SELECT_SOURCE,
@@ -434,7 +434,7 @@ async def test_services(
 
         launch_mock.assert_called_once_with("12")
 
-    with patch("homeassistant.components.roku.Roku.launch") as launch_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.launch") as launch_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_SELECT_SOURCE,
@@ -458,14 +458,14 @@ async def test_tv_services(
         unique_id=TV_SERIAL,
     )
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: TV_ENTITY_ID}, blocking=True
         )
 
         remote_mock.assert_called_once_with("volume_up")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_VOLUME_DOWN,
@@ -475,7 +475,7 @@ async def test_tv_services(
 
         remote_mock.assert_called_once_with("volume_down")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_VOLUME_MUTE,
@@ -485,7 +485,7 @@ async def test_tv_services(
 
         remote_mock.assert_called_once_with("volume_mute")
 
-    with patch("homeassistant.components.roku.Roku.tune") as tune_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.tune") as tune_mock:
         await hass.services.async_call(
             MP_DOMAIN,
             SERVICE_PLAY_MEDIA,
@@ -694,7 +694,7 @@ async def test_integration_services(
     """Test integration services."""
     await setup_integration(hass, aioclient_mock)
 
-    with patch("homeassistant.components.roku.Roku.search") as search_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.search") as search_mock:
         await hass.services.async_call(
             DOMAIN,
             SERVICE_SEARCH,

tests/components/roku/test_remote.py

@@ -42,7 +42,7 @@ async def test_main_services(
     """Test platform services."""
     await setup_integration(hass, aioclient_mock)
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             REMOTE_DOMAIN,
             SERVICE_TURN_OFF,
@@ -51,7 +51,7 @@ async def test_main_services(
         )
         remote_mock.assert_called_once_with("poweroff")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             REMOTE_DOMAIN,
             SERVICE_TURN_ON,
@@ -60,7 +60,7 @@ async def test_main_services(
         )
         remote_mock.assert_called_once_with("poweron")
 
-    with patch("homeassistant.components.roku.Roku.remote") as remote_mock:
+    with patch("homeassistant.components.roku.coordinator.Roku.remote") as remote_mock:
         await hass.services.async_call(
             REMOTE_DOMAIN,
             SERVICE_SEND_COMMAND,