Add PlayStation Network Integration

homeassistant/brands/sony.json

@@ -1,5 +1,11 @@
 {
   "domain": "sony",
   "name": "Sony",
-  "integrations": ["braviatv", "ps4", "sony_projector", "songpal"]
+  "integrations": [
+    "braviatv",
+    "ps4",
+    "sony_projector",
+    "songpal",
+    "playstation_network"
+  ]
 }

homeassistant/components/playstation_network/__init__.py

@@ -0,0 +1,34 @@
+"""The PlayStation Network integration."""
+
+from __future__ import annotations
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import CONF_NPSSO
+from .coordinator import PlaystationNetworkConfigEntry, PlaystationNetworkCoordinator
+from .helpers import PlaystationNetwork
+
+PLATFORMS: list[Platform] = [Platform.MEDIA_PLAYER]
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: PlaystationNetworkConfigEntry
+) -> bool:
+    """Set up Playstation Network from a config entry."""
+
+    psn = PlaystationNetwork(hass, entry.data[CONF_NPSSO])
+
+    coordinator = PlaystationNetworkCoordinator(hass, psn, entry)
+    await coordinator.async_config_entry_first_refresh()
+    entry.runtime_data = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: PlaystationNetworkConfigEntry
+) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/playstation_network/config_flow.py

@@ -0,0 +1,70 @@
+"""Config flow for the PlayStation Network integration."""
+
+import logging
+from typing import Any
+
+from psnawp_api.core.psnawp_exceptions import (
+    PSNAWPAuthenticationError,
+    PSNAWPError,
+    PSNAWPInvalidTokenError,
+    PSNAWPNotFoundError,
+)
+from psnawp_api.models.user import User
+from psnawp_api.utils.misc import parse_npsso_token
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+
+from .const import CONF_NPSSO, DOMAIN
+from .helpers import PlaystationNetwork
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema({vol.Required(CONF_NPSSO): str})
+
+
+class PlaystationNetworkConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Playstation Network."""
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        npsso: str | None = None
+        if user_input is not None:
+            try:
+                npsso = parse_npsso_token(user_input[CONF_NPSSO])
+            except PSNAWPInvalidTokenError:
+                errors["base"] = "invalid_account"
+
+            if npsso:
+                psn = PlaystationNetwork(self.hass, npsso)
+                try:
+                    user: User = await psn.get_user()
+                except PSNAWPAuthenticationError:
+                    errors["base"] = "invalid_auth"
+                except PSNAWPNotFoundError:
+                    errors["base"] = "invalid_account"
+                except PSNAWPError:
+                    errors["base"] = "cannot_connect"
+                except Exception:
+                    _LOGGER.exception("Unexpected exception")
+                    errors["base"] = "unknown"
+                else:
+                    await self.async_set_unique_id(user.account_id)
+                    self._abort_if_unique_id_configured()
+                    return self.async_create_entry(
+                        title=user.online_id,
+                        data={CONF_NPSSO: npsso},
+                    )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=STEP_USER_DATA_SCHEMA,
+            errors=errors,
+            description_placeholders={
+                "npsso_link": "https://ca.account.sony.com/api/v1/ssocookie",
+                "psn_link": "https://playstation.com",
+            },
+        )

homeassistant/components/playstation_network/const.py

@@ -0,0 +1,15 @@
+"""Constants for the Playstation Network integration."""
+
+from typing import Final
+
+from psnawp_api.models.trophies import PlatformType
+
+DOMAIN = "playstation_network"
+CONF_NPSSO: Final = "npsso"
+
+SUPPORTED_PLATFORMS = {
+    PlatformType.PS5,
+    PlatformType.PS4,
+    PlatformType.PS3,
+    PlatformType.PSPC,
+}

homeassistant/components/playstation_network/coordinator.py

@@ -0,0 +1,69 @@
+"""Coordinator for the PlayStation Network Integration."""
+
+from __future__ import annotations
+
+from datetime import timedelta
+import logging
+
+from psnawp_api.core.psnawp_exceptions import (
+    PSNAWPAuthenticationError,
+    PSNAWPServerError,
+)
+from psnawp_api.models.user import User
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+from .helpers import PlaystationNetwork, PlaystationNetworkData
+
+_LOGGER = logging.getLogger(__name__)
+
+type PlaystationNetworkConfigEntry = ConfigEntry[PlaystationNetworkCoordinator]
+
+
+class PlaystationNetworkCoordinator(DataUpdateCoordinator[PlaystationNetworkData]):
+    """Data update coordinator for PSN."""
+
+    config_entry: PlaystationNetworkConfigEntry
+    user: User
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        psn: PlaystationNetwork,
+        config_entry: PlaystationNetworkConfigEntry,
+    ) -> None:
+        """Initialize the Coordinator."""
+        super().__init__(
+            hass,
+            name=DOMAIN,
+            logger=_LOGGER,
+            config_entry=config_entry,
+            update_interval=timedelta(seconds=30),
+        )
+
+        self.psn = psn
+
+    async def _async_setup(self) -> None:
+        """Set up the coordinator."""
+
+        try:
+            self.user = await self.psn.get_user()
+        except PSNAWPAuthenticationError as error:
+            raise ConfigEntryNotReady(
+                translation_domain=DOMAIN,
+                translation_key="not_ready",
+            ) from error
+
+    async def _async_update_data(self) -> PlaystationNetworkData:
+        """Get the latest data from the PSN."""
+        try:
+            return await self.psn.get_data()
+        except (PSNAWPAuthenticationError, PSNAWPServerError) as error:
+            raise UpdateFailed(
+                translation_domain=DOMAIN,
+                translation_key="update_failed",
+            ) from error

homeassistant/components/playstation_network/helpers.py

@@ -0,0 +1,151 @@
+"""Helper methods for common PlayStation Network integration operations."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from functools import partial
+from typing import Any
+
+from psnawp_api import PSNAWP
+from psnawp_api.core.psnawp_exceptions import PSNAWPNotFoundError
+from psnawp_api.models.client import Client
+from psnawp_api.models.trophies import PlatformType
+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}
+
+
+@dataclass
+class SessionData:
+    """Dataclass representing console session data."""
+
+    platform: PlatformType = PlatformType.UNKNOWN
+    title_id: str | None = None
+    title_name: str | None = None
+    format: PlatformType | None = None
+    media_image_url: str | None = None
+    status: str = ""
+
+
+@dataclass
+class PlaystationNetworkData:
+    """Dataclass representing data retrieved from the Playstation Network api."""
+
+    presence: dict[str, Any] = field(default_factory=dict)
+    username: str = ""
+    account_id: str = ""
+    available: bool = False
+    active_sessions: dict[PlatformType, SessionData] = field(default_factory=dict)
+    registered_platforms: set[PlatformType] = field(default_factory=set)
+
+
+class PlaystationNetwork:
+    """Helper Class to return playstation network data in an easy to use structure."""
+
+    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.hass = hass
+        self.user: User
+        self.legacy_profile: dict[str, Any] | None = None
+
+    async def get_user(self) -> User:
+        """Get the user object from the PlayStation Network."""
+        self.user = await self.hass.async_add_executor_job(
+            partial(self.psn.user, online_id="me")
+        )
+        return self.user
+
+    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()
+        } & SUPPORTED_PLATFORMS
+
+        data.presence = self.user.get_presence()
+
+        # check legacy platforms if owned
+        if LEGACY_PLATFORMS & data.registered_platforms:
+            self.legacy_profile = self.client.get_profile_legacy()
+        return data
+
+    async def get_data(self) -> PlaystationNetworkData:
+        """Get title data from the PlayStation Network."""
+        data = await self.hass.async_add_executor_job(self.retrieve_psn_data)
+        data.username = self.user.online_id
+        data.account_id = self.user.account_id
+
+        data.available = (
+            data.presence.get("basicPresence", {}).get("availability")
+            == "availableToPlay"
+        )
+
+        session = SessionData()
+        session.platform = PlatformType(
+            data.presence["basicPresence"]["primaryPlatformInfo"]["platform"]
+        )
+
+        if session.platform in SUPPORTED_PLATFORMS:
+            session.status = data.presence.get("basicPresence", {}).get(
+                "primaryPlatformInfo"
+            )["onlineStatus"]
+
+            game_title_info = data.presence.get("basicPresence", {}).get(
+                "gameTitleInfoList"
+            )
+
+            if game_title_info:
+                session.title_id = game_title_info[0]["npTitleId"]
+                session.title_name = game_title_info[0]["titleName"]
+                session.format = PlatformType(game_title_info[0]["format"])
+                if session.format in {PlatformType.PS5, PlatformType.PSPC}:
+                    session.media_image_url = game_title_info[0]["conceptIconUrl"]
+                else:
+                    session.media_image_url = game_title_info[0]["npTitleIconUrl"]
+
+            data.active_sessions[session.platform] = session
+
+        if self.legacy_profile:
+            presence = self.legacy_profile["profile"].get("presences", [])
+            game_title_info = presence[0] if presence else {}
+            session = SessionData()
+
+            # If primary console isn't online, check legacy platforms for status
+            if not data.available:
+                data.available = game_title_info["onlineStatus"] == "online"
+
+            if "npTitleId" in game_title_info:
+                session.title_id = game_title_info["npTitleId"]
+                session.title_name = game_title_info["titleName"]
+                session.format = game_title_info["platform"]
+                session.platform = game_title_info["platform"]
+                session.status = game_title_info["onlineStatus"]
+                if PlatformType(session.format) is PlatformType.PS4:
+                    session.media_image_url = game_title_info["npTitleIconUrl"]
+                elif PlatformType(session.format) is PlatformType.PS3:
+                    try:
+                        title = self.psn.game_title(
+                            session.title_id, platform=PlatformType.PS3, account_id="me"
+                        )
+                    except PSNAWPNotFoundError:
+                        session.media_image_url = None
+
+                    if title:
+                        session.media_image_url = title.get_title_icon_url()
+
+            if game_title_info["onlineStatus"] == "online":
+                data.active_sessions[session.platform] = session
+        return data

homeassistant/components/playstation_network/icons.json

@@ -0,0 +1,9 @@
+{
+  "entity": {
+    "media_player": {
+      "playstation": {
+        "default": "mdi:sony-playstation"
+      }
+    }
+  }
+}

homeassistant/components/playstation_network/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "playstation_network",
+  "name": "PlayStation Network",
+  "codeowners": ["@jackjpowell"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/playstation_network",
+  "integration_type": "service",
+  "iot_class": "cloud_polling",
+  "quality_scale": "bronze",
+  "requirements": ["PSNAWP==3.0.0", "pyrate-limiter==3.7.0"]
+}