Busch-Radio-Spotify/custom_components/busch_radio_spotify/media_player.py

"""Media-Player-Entity für den Busch-Radio Spotify Bridge Stream.

Pollt den Icecast-Admin-Endpoint alle 30 Sekunden und zeigt:
  - Ob der Stream aktiv ist (Quelle verbunden)
  - Anzahl der aktuellen Hörer
  - Die Stream-URL zum Eintragen ins Radio

Der Entity-Status entspricht dem HA MediaPlayerState:
  - PLAYING  → mindestens ein Hörer verbunden
  - IDLE     → Quelle aktiv, aber kein Hörer
  - OFF      → Icecast nicht erreichbar / Add-on gestoppt
"""

from __future__ import annotations

import logging
import xml.etree.ElementTree as ET
from datetime import timedelta

import aiohttp
from homeassistant.components.media_player import (
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    MediaPlayerState,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from . import CONF_HOST, CONF_MOUNT, CONF_PASSWORD, CONF_PORT, DOMAIN

_LOGGER = logging.getLogger(__name__)

SCAN_INTERVAL = timedelta(seconds=30)
REQUEST_TIMEOUT = aiohttp.ClientTimeout(total=5)


async def async_setup_entry(
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Media-Player-Entity aus Config-Entry anlegen."""
    async_add_entities([BuschRadioMediaPlayer(hass, entry)], update_before_add=True)


class BuschRadioMediaPlayer(MediaPlayerEntity):
    """Repräsentiert den Busch-Radio Spotify Bridge Stream.

    Zeigt den Icecast-Stream-Status in der HA-Oberfläche an.
    Die eigentliche Wiedergabe wird über die Spotify-App gesteuert.
    """

    _attr_icon = "mdi:radio"
    _attr_has_entity_name = False
    _attr_should_poll = True

    # Dieser Player hat keine eigenen Steuer-Features (Spotify-App übernimmt das),
    # aber wir exportieren PLAY/PAUSE als Dummy damit HA ihn als Media-Player erkennt.
    _attr_supported_features = MediaPlayerEntityFeature(0)

    def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
        self._hass = hass
        self._entry = entry
        self._host: str = entry.data[CONF_HOST]
        self._port: int = entry.data[CONF_PORT]
        self._mount: str = entry.data[CONF_MOUNT]
        self._password: str = entry.data[CONF_PASSWORD]

        self._listeners: int = 0
        self._source_connected: bool = False
        self._available: bool = False
        self._bitrate: str | None = None
        self._audio_codec: str | None = None

        self._attr_name = f"Busch-Radio ({self._host})"
        self._attr_unique_id = f"busch_radio_{self._host}_{self._port}"

    # ── Properties ───────────────────────────────────────────────────────────

    @property
    def stream_url(self) -> str:
        """Öffentliche Stream-URL."""
        return f"http://{self._host}:{self._port}{self._mount}"

    @property
    def state(self) -> MediaPlayerState:
        if not self._available:
            return MediaPlayerState.OFF
        if not self._source_connected:
            return MediaPlayerState.IDLE
        if self._listeners > 0:
            return MediaPlayerState.PLAYING
        return MediaPlayerState.IDLE

    @property
    def available(self) -> bool:
        return True  # Entity immer verfügbar; Status spiegelt Icecast-Zustand

    @property
    def extra_state_attributes(self) -> dict:
        return {
            "stream_url": self.stream_url,
            "listeners": self._listeners,
            "source_connected": self._source_connected,
            "bitrate_kbps": self._bitrate,
            "codec": self._audio_codec,
            "icecast_status_url": f"http://{self._host}:{self._port}",
        }

    @property
    def media_title(self) -> str | None:
        if self._source_connected:
            return "Spotify"
        return None

    @property
    def media_artist(self) -> str | None:
        if self._listeners > 0:
            return f"{self._listeners} Hörer"
        return None

    # ── Update ───────────────────────────────────────────────────────────────

    async def async_update(self) -> None:
        """Icecast-Stats abfragen und Entity-Status aktualisieren."""
        session = async_get_clientsession(self._hass)
        stats_url = f"http://{self._host}:{self._port}/admin/stats"
        auth = aiohttp.BasicAuth("admin", self._password)

        try:
            async with session.get(
                stats_url, auth=auth, timeout=REQUEST_TIMEOUT
            ) as resp:
                if resp.status == 200:
                    text = await resp.text()
                    self._parse_icecast_stats(text)
                    self._available = True
                elif resp.status == 401:
                    _LOGGER.warning(
                        "Icecast: Authentifizierung fehlgeschlagen. "
                        "Passwort in der Integration prüfen."
                    )
                    self._available = False
                else:
                    _LOGGER.debug(
                        "Icecast-Stats: HTTP %d von %s", resp.status, stats_url
                    )
                    self._available = False

        except aiohttp.ClientConnectorError:
            _LOGGER.debug(
                "Icecast nicht erreichbar unter %s:%d (Add-on läuft?)",
                self._host,
                self._port,
            )
            self._available = False
            self._source_connected = False
            self._listeners = 0

        except Exception as exc:
            _LOGGER.warning("Fehler beim Abrufen der Icecast-Stats: %s", exc)
            self._available = False

    def _parse_icecast_stats(self, xml_text: str) -> None:
        """Icecast-XML-Statistiken parsen."""
        try:
            root = ET.fromstring(xml_text)
        except ET.ParseError as exc:
            _LOGGER.warning("Icecast-XML konnte nicht geparst werden: %s", exc)
            return

        # Suche nach unserem Mount-Point
        self._source_connected = False
        self._listeners = 0
        self._bitrate = None
        self._audio_codec = None

        for source in root.findall("source"):
            mount = source.get("mount", "")
            if mount != self._mount:
                continue

            self._source_connected = True

            listeners_el = source.find("listeners")
            if listeners_el is not None and listeners_el.text:
                try:
                    self._listeners = int(listeners_el.text)
                except ValueError:
                    pass

            bitrate_el = source.find("bitrate")
            if bitrate_el is not None:
                self._bitrate = bitrate_el.text

            codec_el = source.find("audio_codecname")
            if codec_el is not None:
                self._audio_codec = codec_el.text

            break