Add type hints to ps4 media player (#89236)

This commit is contained in:
epenet 2023-03-06 15:56:34 +01:00 committed by GitHub
parent 5ee383456f
commit ee6f969c2a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -2,6 +2,7 @@
import asyncio import asyncio
from contextlib import suppress from contextlib import suppress
import logging import logging
from typing import Any, cast
from pyps4_2ndscreen.errors import NotReady, PSDataIncomplete from pyps4_2ndscreen.errors import NotReady, PSDataIncomplete
from pyps4_2ndscreen.media_art import TYPE_APP as PS_TYPE_APP from pyps4_2ndscreen.media_art import TYPE_APP as PS_TYPE_APP
@ -27,6 +28,7 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.json import JsonObjectType
from . import format_unique_id, load_games, save_games from . import format_unique_id, load_games, save_games
from .const import ( from .const import (
@ -52,12 +54,12 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up PS4 from a config entry.""" """Set up PS4 from a config entry."""
config = config_entry config = config_entry
creds = config.data[CONF_TOKEN] creds: str = config.data[CONF_TOKEN]
device_list = [] device_list = []
for device in config.data["devices"]: for device in config.data["devices"]:
host = device[CONF_HOST] host: str = device[CONF_HOST]
region = device[CONF_REGION] region: str = device[CONF_REGION]
name = device[CONF_NAME] name: str = device[CONF_NAME]
ps4 = pyps4.Ps4Async(host, creds, device_name=DEFAULT_ALIAS) ps4 = pyps4.Ps4Async(host, creds, device_name=DEFAULT_ALIAS)
device_list.append(PS4Device(config, name, host, region, ps4, creds)) device_list.append(PS4Device(config, name, host, region, ps4, creds))
async_add_entities(device_list, update_before_add=True) async_add_entities(device_list, update_before_add=True)
@ -75,7 +77,15 @@ class PS4Device(MediaPlayerEntity):
| MediaPlayerEntityFeature.SELECT_SOURCE | MediaPlayerEntityFeature.SELECT_SOURCE
) )
def __init__(self, config, name, host, region, ps4, creds): def __init__(
self,
config: ConfigEntry,
name: str,
host: str,
region: str,
ps4: pyps4.Ps4Async,
creds: str,
) -> None:
"""Initialize the ps4 device.""" """Initialize the ps4 device."""
self._entry_id = config.entry_id self._entry_id = config.entry_id
self._ps4 = ps4 self._ps4 = ps4
@ -83,30 +93,30 @@ class PS4Device(MediaPlayerEntity):
self._attr_name = name self._attr_name = name
self._region = region self._region = region
self._creds = creds self._creds = creds
self._media_image = None self._media_image: str | None = None
self._games = {} self._games: JsonObjectType = {}
self._retry = 0 self._retry = 0
self._disconnected = False self._disconnected = False
@callback @callback
def status_callback(self): def status_callback(self) -> None:
"""Handle status callback. Parse status.""" """Handle status callback. Parse status."""
self._parse_status() self._parse_status()
self.async_write_ha_state() self.async_write_ha_state()
@callback @callback
def subscribe_to_protocol(self): def subscribe_to_protocol(self) -> None:
"""Notify protocol to callback with update changes.""" """Notify protocol to callback with update changes."""
self.hass.data[PS4_DATA].protocol.add_callback(self._ps4, self.status_callback) self.hass.data[PS4_DATA].protocol.add_callback(self._ps4, self.status_callback)
@callback @callback
def unsubscribe_to_protocol(self): def unsubscribe_to_protocol(self) -> None:
"""Notify protocol to remove callback.""" """Notify protocol to remove callback."""
self.hass.data[PS4_DATA].protocol.remove_callback( self.hass.data[PS4_DATA].protocol.remove_callback(
self._ps4, self.status_callback self._ps4, self.status_callback
) )
def check_region(self): def check_region(self) -> None:
"""Display logger msg if region is deprecated.""" """Display logger msg if region is deprecated."""
# Non-Breaking although data returned may be inaccurate. # Non-Breaking although data returned may be inaccurate.
if self._region in deprecated_regions: if self._region in deprecated_regions:
@ -151,10 +161,11 @@ class PS4Device(MediaPlayerEntity):
self._parse_status() self._parse_status()
def _parse_status(self): def _parse_status(self) -> None:
"""Parse status.""" """Parse status."""
if (status := self._ps4.status) is not None: status: dict[str, Any] | None = self._ps4.status
self._games = load_games(self.hass, self.unique_id) if status is not None:
self._games = load_games(self.hass, cast(str, self.unique_id))
if self._games: if self._games:
self.get_source_list() self.get_source_list()
@ -193,28 +204,30 @@ class PS4Device(MediaPlayerEntity):
def _use_saved(self) -> bool: def _use_saved(self) -> bool:
"""Return True, Set media attrs if data is locked.""" """Return True, Set media attrs if data is locked."""
if self.media_content_id in self._games: if self.media_content_id in self._games:
store = self._games[self.media_content_id] store = cast(JsonObjectType, self._games[self.media_content_id])
# If locked get attributes from file. # If locked get attributes from file.
if store.get(ATTR_LOCKED): if store.get(ATTR_LOCKED):
self._attr_media_title = store.get(ATTR_MEDIA_TITLE) self._attr_media_title = cast(str | None, store.get(ATTR_MEDIA_TITLE))
self._attr_source = self._attr_media_title self._attr_source = self._attr_media_title
self._media_image = store.get(ATTR_MEDIA_IMAGE_URL) self._media_image = cast(str | None, store.get(ATTR_MEDIA_IMAGE_URL))
self._attr_media_content_type = store.get(ATTR_MEDIA_CONTENT_TYPE) self._attr_media_content_type = cast(
str | None, store.get(ATTR_MEDIA_CONTENT_TYPE)
)
return True return True
return False return False
def idle(self): def idle(self) -> None:
"""Set states for state idle.""" """Set states for state idle."""
self.reset_title() self.reset_title()
self._attr_state = MediaPlayerState.IDLE self._attr_state = MediaPlayerState.IDLE
def state_standby(self): def state_standby(self) -> None:
"""Set states for state standby.""" """Set states for state standby."""
self.reset_title() self.reset_title()
self._attr_state = MediaPlayerState.STANDBY self._attr_state = MediaPlayerState.STANDBY
def state_unknown(self): def state_unknown(self) -> None:
"""Set states for state unknown.""" """Set states for state unknown."""
self.reset_title() self.reset_title()
self._attr_state = None self._attr_state = None
@ -223,14 +236,14 @@ class PS4Device(MediaPlayerEntity):
self._disconnected = True self._disconnected = True
self._retry = 0 self._retry = 0
def reset_title(self): def reset_title(self) -> None:
"""Update if there is no title.""" """Update if there is no title."""
self._attr_media_title = None self._attr_media_title = None
self._attr_media_content_id = None self._attr_media_content_id = None
self._attr_media_content_type = None self._attr_media_content_type = None
self._attr_source = None self._attr_source = None
async def async_get_title_data(self, title_id, name): async def async_get_title_data(self, title_id: str, name: str) -> None:
"""Get PS Store Data.""" """Get PS Store Data."""
app_name = None app_name = None
@ -272,10 +285,10 @@ class PS4Device(MediaPlayerEntity):
await self.hass.async_add_executor_job(self.update_list) await self.hass.async_add_executor_job(self.update_list)
self.async_write_ha_state() self.async_write_ha_state()
def update_list(self): def update_list(self) -> None:
"""Update Game List, Correct data if different.""" """Update Game List, Correct data if different."""
if self.media_content_id in self._games: if self.media_content_id in self._games:
store = self._games[self.media_content_id] store = cast(JsonObjectType, self._games[self.media_content_id])
if ( if (
store.get(ATTR_MEDIA_TITLE) != self.media_title store.get(ATTR_MEDIA_TITLE) != self.media_title
@ -290,7 +303,7 @@ class PS4Device(MediaPlayerEntity):
self._media_image, self._media_image,
self._attr_media_content_type, self._attr_media_content_type,
) )
self._games = load_games(self.hass, self.unique_id) self._games = load_games(self.hass, cast(str, self.unique_id))
self.get_source_list() self.get_source_list()
@ -298,14 +311,22 @@ class PS4Device(MediaPlayerEntity):
"""Parse data entry and update source list.""" """Parse data entry and update source list."""
games = [] games = []
for data in self._games.values(): for data in self._games.values():
games.append(data[ATTR_MEDIA_TITLE]) data = cast(JsonObjectType, data)
games.append(cast(str, data[ATTR_MEDIA_TITLE]))
self._attr_source_list = sorted(games) self._attr_source_list = sorted(games)
def add_games(self, title_id, app_name, image, g_type, is_locked=False): def add_games(
self,
title_id: str | None,
app_name: str | None,
image: str | None,
g_type: str | None,
is_locked: bool = False,
) -> None:
"""Add games to list.""" """Add games to list."""
games = self._games games = self._games
if title_id is not None and title_id not in games: if title_id is not None and title_id not in games:
game = { game: JsonObjectType = {
title_id: { title_id: {
ATTR_MEDIA_TITLE: app_name, ATTR_MEDIA_TITLE: app_name,
ATTR_MEDIA_IMAGE_URL: image, ATTR_MEDIA_IMAGE_URL: image,
@ -314,9 +335,9 @@ class PS4Device(MediaPlayerEntity):
} }
} }
games.update(game) games.update(game)
save_games(self.hass, games, self.unique_id) save_games(self.hass, games, cast(str, self.unique_id))
async def async_get_device_info(self, status): async def async_get_device_info(self, status: dict[str, Any] | None) -> None:
"""Set device info for registry.""" """Set device info for registry."""
# If cannot get status on startup, assume info from registry. # If cannot get status on startup, assume info from registry.
if status is None: if status is None:
@ -362,7 +383,7 @@ class PS4Device(MediaPlayerEntity):
self.hass.data[PS4_DATA].devices.remove(self) self.hass.data[PS4_DATA].devices.remove(self)
@property @property
def entity_picture(self): def entity_picture(self) -> str | None:
"""Return picture.""" """Return picture."""
if ( if (
self.state == MediaPlayerState.PLAYING self.state == MediaPlayerState.PLAYING
@ -376,7 +397,7 @@ class PS4Device(MediaPlayerEntity):
return None return None
@property @property
def media_image_url(self): def media_image_url(self) -> str | None:
"""Image url of current playing media.""" """Image url of current playing media."""
if self.media_content_id is None: if self.media_content_id is None:
return None return None
@ -405,7 +426,8 @@ class PS4Device(MediaPlayerEntity):
async def async_select_source(self, source: str) -> None: async def async_select_source(self, source: str) -> None:
"""Select input source.""" """Select input source."""
for title_id, data in self._games.items(): for title_id, data in self._games.items():
game = data[ATTR_MEDIA_TITLE] data = cast(JsonObjectType, data)
game = cast(str, data[ATTR_MEDIA_TITLE])
if ( if (
source.lower().encode(encoding="utf-8") source.lower().encode(encoding="utf-8")
== game.lower().encode(encoding="utf-8") == game.lower().encode(encoding="utf-8")
@ -421,10 +443,10 @@ class PS4Device(MediaPlayerEntity):
_LOGGER.warning("Could not start title. '%s' is not in source list", source) _LOGGER.warning("Could not start title. '%s' is not in source list", source)
return return
async def async_send_command(self, command): async def async_send_command(self, command: str) -> None:
"""Send Button Command.""" """Send Button Command."""
await self.async_send_remote_control(command) await self.async_send_remote_control(command)
async def async_send_remote_control(self, command): async def async_send_remote_control(self, command: str) -> None:
"""Send RC command.""" """Send RC command."""
await self._ps4.remote_control(command) await self._ps4.remote_control(command)