trdo_player.py

"""
Trdo Linux — GStreamer internet radio player.

SPDX-License-Identifier: MIT

Audio engine with three-layer buffering for glitch-free internet radio:

  Layer 1 — GStreamer preroll   (gates initial playback start)
  Layer 2 — 16 MB queue2 FIFO  (absorbs network variance, silent)
  Layer 3 — 400 ms audio sink   (PulseAudio / PipeWire hardware buffer)

No FLAG_BUFFERING is used — the pipeline never pauses during playback.
Station switches reuse the pipeline via READY → URI → PLAYING.
"""

import logging
import logging.handlers
import threading
import time
from enum import Enum, auto
from pathlib import Path

import gi
gi.require_version("Gst", "1.0")
gi.require_version("GLib", "2.0")
from gi.repository import Gst, GLib

Gst.init(None)

# ── Logging ───────────────────────────────────────────────

LOG_DIR = Path.home() / ".config" / "trdo"
LOG_DIR.mkdir(parents=True, exist_ok=True)
LOG_FILE = LOG_DIR / "trdo.log"

_log = logging.getLogger("trdo.player")
_log.setLevel(logging.DEBUG)
_fh = logging.handlers.RotatingFileHandler(
    LOG_FILE, maxBytes=2 * 1024 * 1024, backupCount=1, encoding="utf-8")
_fh.setFormatter(logging.Formatter(
    "%(asctime)s  %(levelname)-7s  %(message)s", datefmt="%Y-%m-%d %H:%M:%S"))
_log.addHandler(_fh)
_sh = logging.StreamHandler()
_sh.setFormatter(logging.Formatter("%(levelname)-7s  %(message)s"))
_sh.setLevel(logging.INFO)
_log.addHandler(_sh)

# ── Constants ─────────────────────────────────────────────

# Deep buffer (layer 2): byte-limited only, silent FIFO.
Q2_MAX_BYTES   = 16 * 1024 * 1024
Q2_MAX_TIME    = 0
Q2_MAX_BUFFERS = 0

# Audio sink (µs)
SINK_BUFFER_US  = 400_000
SINK_LATENCY_US = 50_000

# HTTP source
HTTP_TIMEOUT_S  = 15
HTTP_USER_AGENT = "Trdo-Linux/1.0 GStreamer"

# Watchdog
WATCHDOG_POLL_S    = 10
WATCHDOG_TIMEOUT_S = 60
MAX_RECONNECTS     = 5

# Diagnostics
Q_LOG_INTERVAL_S = 30

# GstPlayFlags — NO FLAG_BUFFERING (0x100).
FLAG_AUDIO       = 1 << 1
FLAG_SOFT_VOLUME = 1 << 4


class PlayerState(Enum):
    STOPPED    = auto()
    CONNECTING = auto()
    PLAYING    = auto()
    ERROR      = auto()


class RadioPlayer:

    def __init__(self):
        self._pipeline: Gst.Element | None = None
        self._audio_queue: Gst.Element | None = None
        self._uri: str = ""
        self._volume: float = 0.7
        self._state = PlayerState.STOPPED
        self._metadata: dict = {}
        self._station_name: str = ""
        self._target_state = Gst.State.NULL
        self._play_request_t: float = 0.0   # startup timing

        # Callbacks (set by UI)
        self.on_state_changed:     callable | None = None
        self.on_metadata_changed:  callable | None = None
        self.on_error:             callable | None = None
        self.on_buffering_percent: callable | None = None

        # Timers / threads
        self._q_log_timer:   int | None = None
        self._wd_active = False
        self._wd_thread: threading.Thread | None = None
        self._last_data_t: float = 0.0
        self._reconnect_n = 0

        _log.info("RadioPlayer initialised")

    # ────────────────────────────────────────────────────
    #  Pipeline construction
    # ────────────────────────────────────────────────────

    def _build_pipeline(self) -> bool:
        """Create playbin + custom audio-sink bin from scratch."""
        self._destroy_pipeline()

        pb = Gst.ElementFactory.make("playbin3", "radio")
        variant = "playbin3"
        if pb is None:
            pb = Gst.ElementFactory.make("playbin", "radio")
            variant = "playbin"
        if pb is None:
            self._emit_error("GStreamer playbin not available")
            return False

        pb.set_property("uri", self._uri)
        pb.set_property("volume", self._volume)

        # ── Flags: audio + soft-volume, NO BUFFERING ──
        try:
            flags = pb.get_property("flags")
            flags = (flags | FLAG_AUDIO | FLAG_SOFT_VOLUME) & ~(1 << 8)
            pb.set_property("flags", flags)
            _log.debug("flags=0x%03x on %s", flags, variant)
        except Exception as e:
            _log.warning("could not set flags: %s", e)

        # ── Custom audio-sink bin ──
        audio_bin = self._build_audio_sink_bin()
        if audio_bin:
            pb.set_property("audio-sink", audio_bin)

        # ── Signals ──
        pb.connect("source-setup", self._on_source_setup)
        bus = pb.get_bus()
        bus.add_signal_watch()
        bus.connect("message::eos",           self._on_eos)
        bus.connect("message::error",         self._on_bus_error)
        bus.connect("message::warning",       self._on_bus_warning)
        bus.connect("message::tag",           self._on_tag)
        bus.connect("message::state-changed", self._on_gst_state)
        bus.connect("message::clock-lost",    self._on_clock_lost)

        self._pipeline = pb
        _log.info("pipeline built (%s)  uri=%s", variant, self._uri[:100])
        return True

    def _build_audio_sink_bin(self) -> Gst.Element | None:
        """queue2 (16 MB silent FIFO) → audioconvert → audioresample → autoaudiosink"""
        try:
            abin  = Gst.Bin.new("audio-sink-bin")
            queue = Gst.ElementFactory.make("queue2", "deepbuf")
            conv  = Gst.ElementFactory.make("audioconvert", "aconv")
            resmp = Gst.ElementFactory.make("audioresample", "aresamp")
            sink  = Gst.ElementFactory.make("autoaudiosink", "asink")
            if not all([queue, conv, resmp, sink]):
                _log.error("audio-sink element creation failed")
                return None

            queue.set_property("max-size-bytes",   Q2_MAX_BYTES)
            queue.set_property("max-size-time",    Q2_MAX_TIME)
            queue.set_property("max-size-buffers", Q2_MAX_BUFFERS)
            queue.set_property("use-buffering",    False)

            def _child_added(_, child, name):
                for prop, val in [("buffer-time", SINK_BUFFER_US),
                                  ("latency-time", SINK_LATENCY_US)]:
                    if child.find_property(prop):
                        child.set_property(prop, val)
            sink.connect("child-added", _child_added)

            for el in [queue, conv, resmp, sink]:
                abin.add(el)
            queue.link(conv)
            conv.link(resmp)
            resmp.link(sink)
            abin.add_pad(Gst.GhostPad.new("sink", queue.get_static_pad("sink")))

            self._audio_queue = queue
            _log.info("queue2: %d MB, silent FIFO", Q2_MAX_BYTES // (1024*1024))
            return abin
        except Exception as e:
            _log.exception("_build_audio_sink_bin: %s", e)
            return None

    def _on_source_setup(self, _pb, source):
        name = source.get_factory().get_name() if source.get_factory() else "?"
        _log.info("source-setup: %s (%.0f ms after play)",
                  name, (time.monotonic() - self._play_request_t) * 1000)
        for k, v in [("timeout", HTTP_TIMEOUT_S), ("retries", 3),
                      ("user-agent", HTTP_USER_AGENT),
                      ("keep-alive", True), ("compress", True)]:
            if source.find_property(k):
                source.set_property(k, v)
        if source.find_property("extra-headers"):
            try:
                h = Gst.Structure.new_empty("extra-headers")
                h.set_value("Icy-MetaData", "1")
                source.set_property("extra-headers", h)
            except Exception:
                pass

    def _destroy_pipeline(self):
        self._stop_q_log()
        if self._pipeline:
            self._pipeline.set_state(Gst.State.NULL)
            self._pipeline = None
        self._audio_queue = None
        _log.debug("pipeline destroyed")

    # ────────────────────────────────────────────────────
    #  Public controls
    # ────────────────────────────────────────────────────

    def play(self, uri: str, station_name: str = ""):
        """
        Start playing a station.  Two paths:

        FAST PATH (station switch) — pipeline exists:
          READY → change URI → PLAYING
          Keeps elements alive, just reconnects HTTP.

        COLD PATH — no pipeline:
          Build from scratch → PLAYING.

        No PAUSED fill phase — GStreamer's preroll is the only gate.
        """
        _log.info("play() station=%r  uri=%s", station_name, uri[:100])
        self._play_request_t = time.monotonic()
        self._uri = uri
        self._station_name = station_name
        self._metadata = {}
        self._reconnect_n = 0
        self._target_state = Gst.State.PLAYING
        self._set_state(PlayerState.CONNECTING)

        if self._pipeline:
            # ── FAST PATH: reuse pipeline ──
            self._stop_q_log()
            self._pipeline.set_state(Gst.State.READY)
            self._pipeline.set_property("uri", uri)
            self._pipeline.set_state(Gst.State.PLAYING)
            _log.info("fast switch: READY → URI → PLAYING")
        else:
            # ── COLD PATH: build from scratch ──
            if not self._build_pipeline():
                return
            self._pipeline.set_state(Gst.State.PLAYING)
            _log.info("cold start: → PLAYING")

        self._start_watchdog()

    def stop(self):
        _log.info("stop()")
        self._stop_watchdog()
        self._stop_q_log()
        self._target_state = Gst.State.NULL
        self._destroy_pipeline()
        self._metadata = {}
        self._set_state(PlayerState.STOPPED)

    def toggle(self):
        if self._state in (PlayerState.PLAYING, PlayerState.CONNECTING):
            self.stop()
        elif self._state == PlayerState.STOPPED and self._uri:
            self.play(self._uri, self._station_name)

    def pause(self):
        _log.info("pause()")
        self._stop_watchdog()
        self._stop_q_log()
        self._target_state = Gst.State.PAUSED
        if self._pipeline:
            self._pipeline.set_state(Gst.State.PAUSED)
        self._set_state(PlayerState.STOPPED)

    @property
    def volume(self) -> float:
        return self._volume

    @volume.setter
    def volume(self, val: float):
        self._volume = max(0.0, min(1.0, val))
        if self._pipeline:
            self._pipeline.set_property("volume", self._volume)

    @property
    def state(self) -> PlayerState:
        return self._state

    @property
    def metadata(self) -> dict:
        return self._metadata.copy()

    @property
    def current_uri(self) -> str:
        return self._uri

    @property
    def station_name(self) -> str:
        return self._station_name

    # ────────────────────────────────────────────────────
    #  Bus handlers
    # ────────────────────────────────────────────────────

    def _on_eos(self, bus, msg):
        _log.warning("EOS — will reconnect")
        GLib.idle_add(self._attempt_reconnect)

    def _on_bus_error(self, bus, msg):
        err, dbg = msg.parse_error()
        _log.error("BUS ERROR: %s  [%s]", err.message, dbg)
        GLib.idle_add(self._handle_error, err.message)

    def _on_bus_warning(self, bus, msg):
        err, dbg = msg.parse_warning()
        _log.warning("BUS WARN: %s  [%s]", err.message, dbg)

    def _on_clock_lost(self, bus, msg):
        _log.info("clock-lost (no action)")

    def _on_tag(self, bus, msg):
        taglist = msg.parse_tag()
        changed = False
        for i in range(taglist.n_tags()):
            tag = taglist.nth_tag_name(i)
            ok, val = taglist.get_string(tag)
            if ok and val:
                key = _TAG_MAP.get(tag)
                if key and self._metadata.get(key) != val:
                    self._metadata[key] = val
                    changed = True
        if changed and self.on_metadata_changed:
            GLib.idle_add(self.on_metadata_changed, self._metadata.copy())

    def _on_gst_state(self, bus, msg):
        """
        The only startup gate: when the pipeline reaches PLAYING,
        audio is flowing.  Log the startup latency.
        """
        if msg.src != self._pipeline:
            return
        old, new, pending = msg.parse_state_changed()
        _log.debug("pipeline: %s → %s (pending %s)",
                   old.value_nick, new.value_nick, pending.value_nick)

        if new == Gst.State.PLAYING and self._target_state == Gst.State.PLAYING:
            elapsed = (time.monotonic() - self._play_request_t) * 1000
            _log.info("▶ PLAYING — startup %.0f ms", elapsed)
            self._last_data_t = time.monotonic()
            self._reconnect_n = 0
            self._set_state(PlayerState.PLAYING)
            self._start_q_log()

    # ────────────────────────────────────────────────────
    #  State / error
    # ────────────────────────────────────────────────────

    def _set_state(self, st: PlayerState):
        if self._state != st:
            _log.info("state: %s → %s", self._state.name, st.name)
            self._state = st
            if self.on_state_changed:
                GLib.idle_add(self.on_state_changed, st)

    def _emit_error(self, msg: str):
        _log.error("player error: %s", msg)
        self._set_state(PlayerState.ERROR)
        if self.on_error:
            GLib.idle_add(self.on_error, msg)

    def _handle_error(self, msg: str):
        if self._reconnect_n < MAX_RECONNECTS:
            self._attempt_reconnect()
        else:
            self._emit_error(msg)

    # ────────────────────────────────────────────────────
    #  Watchdog
    # ────────────────────────────────────────────────────

    def _start_watchdog(self):
        self._wd_active = True
        self._last_data_t = time.monotonic()
        if self._wd_thread is None or not self._wd_thread.is_alive():
            self._wd_thread = threading.Thread(
                target=self._wd_loop, daemon=True)
            self._wd_thread.start()

    def _stop_watchdog(self):
        self._wd_active = False

    def _wd_loop(self):
        while self._wd_active:
            time.sleep(WATCHDOG_POLL_S)
            if not self._wd_active:
                break
            q = self._audio_queue
            if q:
                try:
                    if q.get_property("current-level-bytes") > 0:
                        self._last_data_t = time.monotonic()
                        continue
                except Exception:
                    pass
            if self._state == PlayerState.PLAYING:
                gap = time.monotonic() - self._last_data_t
                if gap > WATCHDOG_TIMEOUT_S:
                    _log.warning("watchdog: %ds silence", int(gap))
                    GLib.idle_add(self._attempt_reconnect)

    def _attempt_reconnect(self):
        if not self._uri or self._reconnect_n >= MAX_RECONNECTS:
            self._emit_error("Stream unavailable after %d retries"
                             % self._reconnect_n)
            return
        self._reconnect_n += 1
        delay = min(2 * self._reconnect_n, 10)
        _log.info("reconnect #%d — backoff %ds", self._reconnect_n, delay)
        # Full rebuild for reconnects (safer after errors)
        self._destroy_pipeline()
        GLib.timeout_add_seconds(delay, self._do_reconnect)

    def _do_reconnect(self) -> bool:
        if self._build_pipeline():
            self._target_state = Gst.State.PLAYING
            self._play_request_t = time.monotonic()
            self._pipeline.set_state(Gst.State.PLAYING)
            self._set_state(PlayerState.CONNECTING)
        return False

    # ────────────────────────────────────────────────────
    #  Queue-level diagnostics
    # ────────────────────────────────────────────────────

    def _start_q_log(self):
        self._stop_q_log()
        self._q_log_timer = GLib.timeout_add_seconds(
            Q_LOG_INTERVAL_S, self._log_queue_level)

    def _stop_q_log(self):
        if self._q_log_timer is not None:
            GLib.source_remove(self._q_log_timer)
            self._q_log_timer = None

    def _log_queue_level(self) -> bool:
        q = self._audio_queue
        if q is None:
            return False
        try:
            cur_b = q.get_property("current-level-bytes")
            max_b = q.get_property("max-size-bytes")
            pct = (cur_b / max_b * 100) if max_b else 0
            _log.info("queue2: %d KB / %d MB (%.1f%%)  state=%s",
                      cur_b // 1024, max_b // (1024*1024), pct, self._state.name)
        except Exception:
            pass
        return True

    def cleanup(self):
        _log.info("cleanup()")
        self._stop_watchdog()
        self._destroy_pipeline()


_TAG_MAP = {
    "title": "title", "artist": "artist", "album": "album",
    "genre": "genre", "organization": "station",
    "location": "url", "description": "description",
}