coordinator.py

Go to the documentation of this file.

"""Data update coordinator for the Jellyfin integration."""
 
from __future__ import annotations
 
from datetime import timedelta
from typing import Any
 
from jellyfin_apiclient_python import JellyfinClient
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
from .const import CONF_CLIENT_DEVICE_ID, DOMAIN, LOGGER, USER_APP_NAME
 
 
class JellyfinDataUpdateCoordinator(DataUpdateCoordinator[dict[str, dict[str, Any]]]):
    """Data update coordinator for the Jellyfin integration."""
 
    config_entry: ConfigEntry
 
    def __init__(
        self,
        hass: HomeAssistant,
        api_client: JellyfinClient,
        system_info: dict[str, Any],
        user_id: str,
    ) -> None:
        """Initialize the coordinator."""
        super().__init__(
            hass=hass,
            logger=LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=10),
        )
        self.api_clientapi_client = api_client
        self.server_id: str = system_info["Id"]
        self.server_name: str = system_info["Name"]
        self.server_version: str | None = system_info.get("Version")
        self.client_device_id: str = self.config_entryconfig_entry.data[CONF_CLIENT_DEVICE_ID]
        self.user_id: str = user_id
 
        self.session_ids: set[str] = set()
        self.remote_session_ids: set[str] = set()
        self.device_idsdevice_ids: set[str] = set()
 
    async def _async_update_data(self) -> dict[str, dict[str, Any]]:
        """Get the latest data from Jellyfin."""
        sessions = await self.hasshass.async_add_executor_job(
            self.api_clientapi_client.jellyfin.sessions
        )
 
        sessions_by_id: dict[str, dict[str, Any]] = {
            session["Id"]: session
            for session in sessions
            if session["DeviceId"] != self.client_device_id
            and session["Client"] != USER_APP_NAME
        }
 
        self.device_idsdevice_ids = {session["DeviceId"] for session in sessions_by_id.values()}
 
        return sessions_by_id