Add PS Vita support to PlayStation Network integration (home-assistant#148186)

Author: tr4nt0r

homeassistant/components/playstation_network/__init__.py

@@ -6,7 +6,12 @@
 from homeassistant.core import HomeAssistant
 
 from .const import CONF_NPSSO
-from .coordinator import PlaystationNetworkConfigEntry, PlaystationNetworkCoordinator
+from .coordinator import (
+    PlaystationNetworkConfigEntry,
+    PlaystationNetworkRuntimeData,
+    PlaystationNetworkTrophyTitlesCoordinator,
+    PlaystationNetworkUserDataCoordinator,
+)
 from .helpers import PlaystationNetwork
 
 PLATFORMS: list[Platform] = [
@@ -23,9 +28,12 @@ async def async_setup_entry(
 
     psn = PlaystationNetwork(hass, entry.data[CONF_NPSSO])
 
-    coordinator = PlaystationNetworkCoordinator(hass, psn, entry)
+    coordinator = PlaystationNetworkUserDataCoordinator(hass, psn, entry)
     await coordinator.async_config_entry_first_refresh()
-    entry.runtime_data = coordinator
+
+    trophy_titles = PlaystationNetworkTrophyTitlesCoordinator(hass, psn, entry)
+
+    entry.runtime_data = PlaystationNetworkRuntimeData(coordinator, trophy_titles)
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     return True

homeassistant/components/playstation_network/binary_sensor.py

@@ -49,7 +49,7 @@ async def async_setup_entry(
     async_add_entities: AddConfigEntryEntitiesCallback,
 ) -> None:
     """Set up the binary sensor platform."""
-    coordinator = config_entry.runtime_data
+    coordinator = config_entry.runtime_data.user_data
     async_add_entities(
         PlaystationNetworkBinarySensorEntity(coordinator, description)
         for description in BINARY_SENSOR_DESCRIPTIONS

homeassistant/components/playstation_network/config_flow.py

@@ -10,7 +10,6 @@
     PSNAWPInvalidTokenError,
     PSNAWPNotFoundError,
 )
-from psnawp_api.models.user import User
 from psnawp_api.utils.misc import parse_npsso_token
 import voluptuous as vol
 
@@ -42,7 +41,7 @@ async def async_step_user(
             else:
                 psn = PlaystationNetwork(self.hass, npsso)
                 try:
-                    user: User = await psn.get_user()
+                    user = await psn.get_user()
                 except PSNAWPAuthenticationError:
                     errors["base"] = "invalid_auth"
                 except PSNAWPNotFoundError:
@@ -98,7 +97,7 @@ async def async_step_reauth_confirm(
             try:
                 npsso = parse_npsso_token(user_input[CONF_NPSSO])
                 psn = PlaystationNetwork(self.hass, npsso)
-                user: User = await psn.get_user()
+                user = await psn.get_user()
             except PSNAWPAuthenticationError:
                 errors["base"] = "invalid_auth"
             except (PSNAWPNotFoundError, PSNAWPInvalidTokenError):

homeassistant/components/playstation_network/const.py

@@ -8,9 +8,10 @@
 CONF_NPSSO: Final = "npsso"
 
 SUPPORTED_PLATFORMS = {
-    PlatformType.PS5,
-    PlatformType.PS4,
+    PlatformType.PS_VITA,
     PlatformType.PS3,
+    PlatformType.PS4,
+    PlatformType.PS5,
     PlatformType.PSPC,
 }
 

homeassistant/components/playstation_network/coordinator.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from abc import abstractmethod
+from dataclasses import dataclass
 from datetime import timedelta
 import logging
 
@@ -10,6 +12,7 @@
     PSNAWPClientError,
     PSNAWPServerError,
 )
+from psnawp_api.models.trophies import TrophyTitle
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -21,13 +24,22 @@
 
 _LOGGER = logging.getLogger(__name__)
 
-type PlaystationNetworkConfigEntry = ConfigEntry[PlaystationNetworkCoordinator]
+type PlaystationNetworkConfigEntry = ConfigEntry[PlaystationNetworkRuntimeData]
 
 
-class PlaystationNetworkCoordinator(DataUpdateCoordinator[PlaystationNetworkData]):
-    """Data update coordinator for PSN."""
+@dataclass
+class PlaystationNetworkRuntimeData:
+    """Dataclass holding PSN runtime data."""
+
+    user_data: PlaystationNetworkUserDataCoordinator
+    trophy_titles: PlaystationNetworkTrophyTitlesCoordinator
+
+
+class PlayStationNetworkBaseCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
+    """Base coordinator for PSN."""
 
     config_entry: PlaystationNetworkConfigEntry
+    _update_inverval: timedelta
 
     def __init__(
         self,
@@ -41,38 +53,70 @@ def __init__(
             name=DOMAIN,
             logger=_LOGGER,
             config_entry=config_entry,
-            update_interval=timedelta(seconds=30),
+            update_interval=self._update_interval,
         )
 
         self.psn = psn
 
-    async def _async_setup(self) -> None:
-        """Set up the coordinator."""
+    @abstractmethod
+    async def update_data(self) -> _DataT:
+        """Update coordinator data."""
 
+    async def _async_update_data(self) -> _DataT:
+        """Get the latest data from the PSN."""
         try:
-            await self.psn.get_user()
+            return await self.update_data()
         except PSNAWPAuthenticationError as error:
             raise ConfigEntryAuthFailed(
                 translation_domain=DOMAIN,
                 translation_key="not_ready",
             ) from error
         except (PSNAWPServerError, PSNAWPClientError) as error:
-            raise ConfigEntryNotReady(
+            raise UpdateFailed(
                 translation_domain=DOMAIN,
                 translation_key="update_failed",
             ) from error
 
-    async def _async_update_data(self) -> PlaystationNetworkData:
-        """Get the latest data from the PSN."""
+
+class PlaystationNetworkUserDataCoordinator(
+    PlayStationNetworkBaseCoordinator[PlaystationNetworkData]
+):
+    """Data update coordinator for PSN."""
+
+    _update_interval = timedelta(seconds=30)
+
+    async def _async_setup(self) -> None:
+        """Set up the coordinator."""
+
         try:
-            return await self.psn.get_data()
+            await self.psn.async_setup()
         except PSNAWPAuthenticationError as error:
             raise ConfigEntryAuthFailed(
                 translation_domain=DOMAIN,
                 translation_key="not_ready",
             ) from error
         except (PSNAWPServerError, PSNAWPClientError) as error:
-            raise UpdateFailed(
+            raise ConfigEntryNotReady(
                 translation_domain=DOMAIN,
                 translation_key="update_failed",
             ) from error
+
+    async def update_data(self) -> PlaystationNetworkData:
+        """Get the latest data from the PSN."""
+        return await self.psn.get_data()
+
+
+class PlaystationNetworkTrophyTitlesCoordinator(
+    PlayStationNetworkBaseCoordinator[list[TrophyTitle]]
+):
+    """Trophy titles data update coordinator for PSN."""
+
+    _update_interval = timedelta(days=1)
+
+    async def update_data(self) -> list[TrophyTitle]:
+        """Update trophy titles data."""
+        self.psn.trophy_titles = await self.hass.async_add_executor_job(
+            lambda: list(self.psn.user.trophy_titles())
+        )
+        await self.config_entry.runtime_data.user_data.async_request_refresh()
+        return self.psn.trophy_titles

homeassistant/components/playstation_network/diagnostics.py

@@ -10,7 +10,7 @@
 from homeassistant.components.diagnostics import async_redact_data
 from homeassistant.core import HomeAssistant
 
-from .coordinator import PlaystationNetworkConfigEntry, PlaystationNetworkCoordinator
+from .coordinator import PlaystationNetworkConfigEntry
 
 TO_REDACT = {
     "account_id",
@@ -27,12 +27,12 @@ async def async_get_config_entry_diagnostics(
     hass: HomeAssistant, entry: PlaystationNetworkConfigEntry
 ) -> dict[str, Any]:
     """Return diagnostics for a config entry."""
-    coordinator: PlaystationNetworkCoordinator = entry.runtime_data
+    coordinator = entry.runtime_data.user_data
 
     return {
         "data": async_redact_data(
             _serialize_platform_types(asdict(coordinator.data)), TO_REDACT
-        ),
+        )
     }
 
 
@@ -46,10 +46,12 @@ def _serialize_platform_types(data: Any) -> Any:
             for platform, record in data.items()
         }
     if isinstance(data, set):
-        return [
-            record.value if isinstance(record, PlatformType) else record
-            for record in data
-        ]
+        return sorted(
+            [
+                record.value if isinstance(record, PlatformType) else record
+                for record in data
+            ]
+        )
     if isinstance(data, PlatformType):
         return data.value
     return data

homeassistant/components/playstation_network/entity.py

@@ -7,17 +7,19 @@
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DOMAIN
-from .coordinator import PlaystationNetworkCoordinator
+from .coordinator import PlaystationNetworkUserDataCoordinator
 
 
-class PlaystationNetworkServiceEntity(CoordinatorEntity[PlaystationNetworkCoordinator]):
+class PlaystationNetworkServiceEntity(
+    CoordinatorEntity[PlaystationNetworkUserDataCoordinator]
+):
     """Common entity class for PlayStationNetwork Service entities."""
 
     _attr_has_entity_name = True
 
     def __init__(
         self,
-        coordinator: PlaystationNetworkCoordinator,
+        coordinator: PlaystationNetworkUserDataCoordinator,
         entity_description: EntityDescription,
     ) -> None:
         """Initialize PlayStation Network Service Entity."""

homeassistant/components/playstation_network/helpers.py

@@ -8,15 +8,15 @@
 
 from psnawp_api import PSNAWP
 from psnawp_api.models.client import Client
-from psnawp_api.models.trophies import PlatformType, TrophySummary
+from psnawp_api.models.trophies import PlatformType, TrophySummary, TrophyTitle
 from psnawp_api.models.user import User
 from pyrate_limiter import Duration, Rate
 
 from homeassistant.core import HomeAssistant
 
 from .const import SUPPORTED_PLATFORMS
 
-LEGACY_PLATFORMS = {PlatformType.PS3, PlatformType.PS4}
+LEGACY_PLATFORMS = {PlatformType.PS3, PlatformType.PS4, PlatformType.PS_VITA}
 
 
 @dataclass
@@ -52,10 +52,22 @@ def __init__(self, hass: HomeAssistant, npsso: str) -> None:
         """Initialize the class with the npsso token."""
         rate = Rate(300, Duration.MINUTE * 15)
         self.psn = PSNAWP(npsso, rate_limit=rate)
-        self.client: Client | None = None
+        self.client: Client
         self.hass = hass
         self.user: User
         self.legacy_profile: dict[str, Any] | None = None
+        self.trophy_titles: list[TrophyTitle] = []
+        self._title_icon_urls: dict[str, str] = {}
+
+    def _setup(self) -> None:
+        """Setup PSN."""
+        self.user = self.psn.user(online_id="me")
+        self.client = self.psn.me()
+        self.trophy_titles = list(self.user.trophy_titles())
+
+    async def async_setup(self) -> None:
+        """Setup PSN."""
+        await self.hass.async_add_executor_job(self._setup)
 
     async def get_user(self) -> User:
         """Get the user object from the PlayStation Network."""
@@ -68,9 +80,6 @@ def retrieve_psn_data(self) -> PlaystationNetworkData:
         """Bundle api calls to retrieve data from the PlayStation Network."""
         data = PlaystationNetworkData()
 
-        if not self.client:
-            self.client = self.psn.me()
-
         data.registered_platforms = {
             PlatformType(device["deviceType"])
             for device in self.client.get_account_devices()
@@ -123,7 +132,7 @@ async def get_data(self) -> PlaystationNetworkData:
             presence = self.legacy_profile["profile"].get("presences", [])
             if (game_title_info := presence[0] if presence else {}) and game_title_info[
                 "onlineStatus"
-            ] == "online":
+            ] != "offline":
                 platform = PlatformType(game_title_info["platform"])
 
                 if platform is PlatformType.PS4:
@@ -135,6 +144,10 @@ async def get_data(self) -> PlaystationNetworkData:
                         account_id="me",
                         np_communication_id="",
                     ).get_title_icon_url()
+                elif platform is PlatformType.PS_VITA and game_title_info.get(
+                    "npTitleId"
+                ):
+                    media_image_url = self.get_psvita_title_icon_url(game_title_info)
                 else:
                     media_image_url = None
 
@@ -147,3 +160,28 @@ async def get_data(self) -> PlaystationNetworkData:
                     status=game_title_info["onlineStatus"],
                 )
         return data
+
+    def get_psvita_title_icon_url(self, game_title_info: dict[str, Any]) -> str | None:
+        """Look up title_icon_url from trophy titles data."""
+
+        if url := self._title_icon_urls.get(game_title_info["npTitleId"]):
+            return url
+
+        url = next(
+            (
+                title.title_icon_url
+                for title in self.trophy_titles
+                if game_title_info["titleName"]
+                == normalize_title(title.title_name or "")
+                and next(iter(title.title_platform)) == PlatformType.PS_VITA
+            ),
+            None,
+        )
+        if url is not None:
+            self._title_icon_urls[game_title_info["npTitleId"]] = url
+        return url
+
+
+def normalize_title(name: str) -> str:
+    """Normalize trophy title."""
+    return name.removesuffix("Trophies").removesuffix("Trophy Set").strip()