Introduce Jellyfin client/server base entities (#127572)

homeassistant/components/jellyfin/entity.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
-from homeassistant.helpers.entity import EntityDescription
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import DEFAULT_NAME, DOMAIN
@@ -15,20 +16,60 @@ class JellyfinEntity(CoordinatorEntity[JellyfinDataUpdateCoordinator]):
 
     _attr_has_entity_name = True
 
+
+class JellyfinServerEntity(JellyfinEntity):
+    """Defines a base Jellyfin server entity."""
+
+    def __init__(self, coordinator: JellyfinDataUpdateCoordinator) -> None:
+        """Initialize the Jellyfin entity."""
+        super().__init__(coordinator)
+        self._attr_device_info = DeviceInfo(
+            entry_type=DeviceEntryType.SERVICE,
+            identifiers={(DOMAIN, coordinator.server_id)},
+            manufacturer=DEFAULT_NAME,
+            name=coordinator.server_name,
+            sw_version=coordinator.server_version,
+        )
+
+
+class JellyfinClientEntity(JellyfinEntity):
+    """Defines a base Jellyfin client entity."""
+
     def __init__(
         self,
         coordinator: JellyfinDataUpdateCoordinator,
-        description: EntityDescription,
+        session_id: str,
     ) -> None:
         """Initialize the Jellyfin entity."""
         super().__init__(coordinator)
-        self.coordinator = coordinator
+        self.session_id = session_id
-        self.entity_description = description
+        self.device_id: str = self.session_data["DeviceId"]
-        self._attr_unique_id = f"{coordinator.server_id}-{description.key}"
+        self.device_name: str = self.session_data["DeviceName"]
+        self.client_name: str = self.session_data["Client"]
+        self.app_version: str = self.session_data["ApplicationVersion"]
+        self.capabilities: dict[str, Any] = self.session_data["Capabilities"]
+
+        if self.capabilities.get("SupportsPersistentIdentifier", False):
             self._attr_device_info = DeviceInfo(
-            entry_type=DeviceEntryType.SERVICE,
+                identifiers={(DOMAIN, self.device_id)},
-            identifiers={(DOMAIN, self.coordinator.server_id)},
+                manufacturer="Jellyfin",
-            manufacturer=DEFAULT_NAME,
+                model=self.client_name,
-            name=self.coordinator.server_name,
+                name=self.device_name,
-            sw_version=self.coordinator.server_version,
+                sw_version=self.app_version,
+                via_device=(DOMAIN, coordinator.server_id),
             )
+            self._attr_name = None
+        else:
+            self._attr_device_info = None
+            self._attr_has_entity_name = False
+            self._attr_name = self.device_name
+
+    @property
+    def session_data(self) -> dict[str, Any]:
+        """Return the session data."""
+        return self.coordinator.data[self.session_id]
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        return super().available and self.session_id in self.coordinator.data

homeassistant/components/jellyfin/media_player.py

@@ -7,22 +7,20 @@ from typing import Any
 from homeassistant.components.media_player import (
     BrowseMedia,
     MediaPlayerEntity,
-    MediaPlayerEntityDescription,
     MediaPlayerEntityFeature,
     MediaPlayerState,
     MediaType,
 )
 from homeassistant.core import HomeAssistant, callback
-from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.util.dt import parse_datetime
 
 from . import JellyfinConfigEntry
 from .browse_media import build_item_response, build_root_response
 from .client_wrapper import get_artwork_url
-from .const import CONTENT_TYPE_MAP, DOMAIN, LOGGER
+from .const import CONTENT_TYPE_MAP, LOGGER
 from .coordinator import JellyfinDataUpdateCoordinator
-from .entity import JellyfinEntity
+from .entity import JellyfinClientEntity
 
 
 async def async_setup_entry(
@@ -37,11 +35,9 @@ async def async_setup_entry(
     def handle_coordinator_update() -> None:
         """Add media player per session."""
         entities: list[MediaPlayerEntity] = []
-        for session_id, session_data in coordinator.data.items():
+        for session_id in coordinator.data:
             if session_id not in coordinator.session_ids:
-                entity: MediaPlayerEntity = JellyfinMediaPlayer(
+                entity: MediaPlayerEntity = JellyfinMediaPlayer(coordinator, session_id)
-                    coordinator, session_id, session_data
-                )
                 LOGGER.debug("Creating media player for session: %s", session_id)
                 coordinator.session_ids.add(session_id)
                 entities.append(entity)
@@ -52,60 +48,28 @@ async def async_setup_entry(
     entry.async_on_unload(coordinator.async_add_listener(handle_coordinator_update))
 
 
-class JellyfinMediaPlayer(JellyfinEntity, MediaPlayerEntity):
+class JellyfinMediaPlayer(JellyfinClientEntity, MediaPlayerEntity):
     """Represents a Jellyfin Player device."""
 
     def __init__(
         self,
         coordinator: JellyfinDataUpdateCoordinator,
         session_id: str,
-        session_data: dict[str, Any],
     ) -> None:
         """Initialize the Jellyfin Media Player entity."""
-        super().__init__(
+        super().__init__(coordinator, session_id)
-            coordinator,
+        self._attr_unique_id = f"{coordinator.server_id}-{session_id}"
-            MediaPlayerEntityDescription(
+
-                key=session_id,
+        self.now_playing: dict[str, Any] | None = self.session_data.get(
-            ),
+            "NowPlayingItem"
         )
-
+        self.play_state: dict[str, Any] | None = self.session_data.get("PlayState")
-        self.session_id = session_id
-        self.session_data: dict[str, Any] | None = session_data
-        self.device_id: str = session_data["DeviceId"]
-        self.device_name: str = session_data["DeviceName"]
-        self.client_name: str = session_data["Client"]
-        self.app_version: str = session_data["ApplicationVersion"]
-
-        self.capabilities: dict[str, Any] = session_data["Capabilities"]
-        self.now_playing: dict[str, Any] | None = session_data.get("NowPlayingItem")
-        self.play_state: dict[str, Any] | None = session_data.get("PlayState")
-
-        if self.capabilities.get("SupportsPersistentIdentifier", False):
-            self._attr_device_info = DeviceInfo(
-                identifiers={(DOMAIN, self.device_id)},
-                manufacturer="Jellyfin",
-                model=self.client_name,
-                name=self.device_name,
-                sw_version=self.app_version,
-                via_device=(DOMAIN, coordinator.server_id),
-            )
-            self._attr_name = None
-        else:
-            self._attr_device_info = None
-            self._attr_has_entity_name = False
-            self._attr_name = self.device_name
 
         self._update_from_session_data()
 
     @callback
     def _handle_coordinator_update(self) -> None:
-        self.session_data = (
+        if self.available:
-            self.coordinator.data.get(self.session_id)
-            if self.coordinator.data is not None
-            else None
-        )
-
-        if self.session_data is not None:
             self.now_playing = self.session_data.get("NowPlayingItem")
             self.play_state = self.session_data.get("PlayState")
         else:
@@ -135,7 +99,7 @@ class JellyfinMediaPlayer(JellyfinEntity, MediaPlayerEntity):
         volume_muted = False
         volume_level = None
 
-        if self.session_data is not None:
+        if self.available:
             state = MediaPlayerState.IDLE
             media_position_updated = (
                 parse_datetime(self.session_data["LastPlaybackCheckIn"])
@@ -233,11 +197,6 @@ class JellyfinMediaPlayer(JellyfinEntity, MediaPlayerEntity):
 
         return features
 
-    @property
-    def available(self) -> bool:
-        """Return if entity is available."""
-        return self.coordinator.last_update_success and self.session_data is not None
-
     def media_seek(self, position: float) -> None:
         """Send seek command."""
         self.coordinator.api_client.jellyfin.remote_seek(

homeassistant/components/jellyfin/sensor.py

@@ -11,8 +11,8 @@ from homeassistant.core import HomeAssistant
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.typing import StateType
 
-from . import JellyfinConfigEntry
+from . import JellyfinConfigEntry, JellyfinDataUpdateCoordinator
-from .entity import JellyfinEntity
+from .entity import JellyfinServerEntity
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -50,15 +50,25 @@ async def async_setup_entry(
     coordinator = entry.runtime_data
 
     async_add_entities(
-        JellyfinSensor(coordinator, description) for description in SENSOR_TYPES
+        JellyfinServerSensor(coordinator, description) for description in SENSOR_TYPES
     )
 
 
-class JellyfinSensor(JellyfinEntity, SensorEntity):
+class JellyfinServerSensor(JellyfinServerEntity, SensorEntity):
     """Defines a Jellyfin sensor entity."""
 
     entity_description: JellyfinSensorEntityDescription
 
+    def __init__(
+        self,
+        coordinator: JellyfinDataUpdateCoordinator,
+        description: JellyfinSensorEntityDescription,
+    ) -> None:
+        """Initialize Jellyfin sensor."""
+        super().__init__(coordinator)
+        self.entity_description = description
+        self._attr_unique_id = f"{coordinator.server_id}-{description.key}"
+
     @property
     def native_value(self) -> StateType:
         """Return the state of the sensor."""