LeviandClaude Opus 4.7
0ad51879ae
Home Assistant custom integration for NVIDIA Shield TV that composes the Android TV Remote, Jellyfin, smart plug, and receiver entities into a single unified media player with ADB-based now-playing from any app. Part of the Media Zone ecosystem. Entities: - media_player: power (with smart plug fallback), transport, input, volume routed to receiver, rich stream info overlay when Jellyfin is active - sensor: DAG attributes and stream info - button: Pair ADB Config: - Two-step flow: basic setup + optional ADB pairing on the Shield - Configurable ADB port (default 5555) - Pick an existing Jellyfin media_player entity — stream info is read from the HA Jellyfin integration's own DataUpdateCoordinator, so no server URL / API key / extra config needed here - Privacy mode (adb_store_app_info=False) suppresses non-Jellyfin media details, showing only the app name Ecosystem: - Exposes media_dag_* attributes for ha-media-zone orchestration - Exposes media_dag_running_apps and async_bring_to_foreground for the Media Zone card's app switcher Shared ADB code lives in ltomes/ha-adb-common and is vendored at custom_components/nvidia_shield/ha_adb_common/ for HACS installs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
"""Jellyfin session stream-info reader.
|
|
|
|
Piggybacks on HA's built-in Jellyfin integration: its coordinator polls
|
|
/Sessions every 10s and exposes the parsed result on
|
|
``config_entry.runtime_data.data``. We reach across to that cached dict
|
|
by walking from the user-selected Jellyfin media_player entity back to
|
|
its owning Jellyfin config entry. Zero additional config, zero extra
|
|
HTTP.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers import entity_registry as er
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def _parse_streams(item: dict) -> dict | None:
|
|
"""Extract friendly video/audio labels from a Jellyfin NowPlayingItem."""
|
|
streams = item.get("MediaStreams", [])
|
|
result: dict = {}
|
|
|
|
for stream in streams:
|
|
stype = stream.get("Type", "")
|
|
|
|
if stype == "Video" and "video" not in result:
|
|
width = stream.get("Width", 0)
|
|
height = stream.get("Height", 0)
|
|
codec = (stream.get("Codec") or "").upper()
|
|
display = stream.get("DisplayTitle", "")
|
|
|
|
if height >= 2160:
|
|
res = "4K"
|
|
elif height >= 1080:
|
|
res = "1080p"
|
|
elif height >= 720:
|
|
res = "720p"
|
|
elif height > 0:
|
|
res = f"{height}p"
|
|
else:
|
|
res = None
|
|
|
|
result["video"] = {
|
|
"resolution": res,
|
|
"width": width,
|
|
"height": height,
|
|
"codec": codec,
|
|
"display": display,
|
|
"hdr": "HDR" in display or "DV" in display or "Dolby Vision" in display,
|
|
}
|
|
|
|
elif stype == "Audio" and "audio" not in result:
|
|
codec = (stream.get("Codec") or "").upper()
|
|
layout = stream.get("ChannelLayout", "")
|
|
display = stream.get("DisplayTitle", "")
|
|
|
|
if "atmos" in display.lower():
|
|
audio_label = "Atmos"
|
|
elif "truehd" in codec.lower():
|
|
audio_label = "TrueHD"
|
|
elif "dts" in codec.lower():
|
|
audio_label = "DTS-HD" if "hd" in display.lower() else "DTS"
|
|
elif "eac3" in codec.lower() or "e-ac-3" in codec.lower():
|
|
audio_label = "DD+"
|
|
elif "ac3" in codec.lower():
|
|
audio_label = "DD"
|
|
elif "aac" in codec.lower():
|
|
audio_label = "AAC"
|
|
elif "flac" in codec.lower():
|
|
audio_label = "FLAC"
|
|
else:
|
|
audio_label = codec
|
|
|
|
result["audio"] = {
|
|
"codec": codec,
|
|
"channels": layout,
|
|
"label": audio_label,
|
|
"display": display,
|
|
}
|
|
|
|
return result or None
|
|
|
|
|
|
def _get_jellyfin_coordinator(hass: HomeAssistant, jellyfin_entity_id: str) -> Any:
|
|
"""Resolve the Jellyfin integration's data-update coordinator for the
|
|
entity the user picked. Returns the coordinator, or None if the
|
|
entity is not registered / its config entry has no runtime_data yet."""
|
|
if not jellyfin_entity_id:
|
|
return None
|
|
|
|
registry = er.async_get(hass)
|
|
entity_entry = registry.async_get(jellyfin_entity_id)
|
|
if not entity_entry or not entity_entry.config_entry_id:
|
|
return None
|
|
|
|
config_entry = hass.config_entries.async_get_entry(entity_entry.config_entry_id)
|
|
if not config_entry:
|
|
return None
|
|
|
|
return getattr(config_entry, "runtime_data", None)
|
|
|
|
|
|
async def fetch_session_media_info(
|
|
hass: HomeAssistant,
|
|
jellyfin_entity_id: str,
|
|
device_name: str,
|
|
) -> dict | None:
|
|
"""Find stream info for the session whose DeviceName matches ``device_name``.
|
|
|
|
Reads from the already-authenticated Jellyfin coordinator — no new
|
|
HTTP calls, no credentials in this integration's config.
|
|
"""
|
|
coordinator = _get_jellyfin_coordinator(hass, jellyfin_entity_id)
|
|
if coordinator is None:
|
|
return None
|
|
|
|
sessions = getattr(coordinator, "data", None) or {}
|
|
|
|
device_lower = device_name.lower()
|
|
for sess in sessions.values():
|
|
sess_device = (sess.get("DeviceName") or "").lower()
|
|
if device_lower not in sess_device:
|
|
continue
|
|
|
|
item = sess.get("NowPlayingItem")
|
|
if not item:
|
|
continue
|
|
|
|
return _parse_streams(item)
|
|
|
|
return None
|