R36SHack/tests/test_video_playback_device.py

#!/usr/bin/env python3
"""
Video playback diagnostic for R36S / ArkOS.

Tests GStreamer availability, codec coverage, and a short live-playback loop
using the same pipeline the app uses (playbin → appsink, BGRA frames).

Run directly on the device:
    /home/ark/miniconda3/envs/r36s-dlna-browser/bin/python \
        /home/ark/R36SHack/tests/test_video_playback_device.py

Accepts an optional URL/path to test real playback:
    ... test_video_playback_device.py  http://server/video.mkv
"""

from __future__ import annotations

import sys
import time
import textwrap

# ── pretty output helpers ───────────────────────────────────────────────────

def _ok(msg: str) -> None:
    print(f"  [OK]  {msg}")

def _warn(msg: str) -> None:
    print(f"  [WRN] {msg}")

def _fail(msg: str) -> None:
    print(f"  [ERR] {msg}")

def _section(title: str) -> None:
    print(f"\n{'='*60}")
    print(f"  {title}")
    print(f"{'='*60}")


# ── 1. Python env ───────────────────────────────────────────────────────────

_section("1. Python environment")
import platform
print(f"  Python {sys.version}")
print(f"  Platform: {platform.machine()} / {platform.system()}")


# ── 2. GI / GStreamer core ──────────────────────────────────────────────────

_section("2. GStreamer core")

try:
    import gi
    _ok(f"PyGObject (gi) {gi.__version__}")
except ImportError as exc:
    _fail(f"PyGObject not found: {exc}")
    sys.exit(1)

try:
    gi.require_version("Gst", "1.0")
    from gi.repository import Gst
    Gst.init(None)
    v = Gst.version()
    _ok(f"GStreamer {v.major}.{v.minor}.{v.micro}.{v.nano}")
except Exception as exc:
    _fail(f"GStreamer init failed: {exc}")
    sys.exit(1)

try:
    gi.require_version("GstApp", "1.0")
    gi.require_version("GstVideo", "1.0")
    from gi.repository import GstApp, GstVideo
    _ok("GstApp + GstVideo bindings present")
except Exception as exc:
    _fail(f"GstApp/GstVideo bindings missing: {exc}")
    sys.exit(1)


# ── 3. Plugin registry ──────────────────────────────────────────────────────

_section("3. Plugin registry")

REQUIRED_ELEMENTS = {
    "playbin":      "core orchestrator",
    "appsink":      "app-side frame sink",
    "videoconvert": "pixel format conversion",
    "videoscale":   "frame scaling",
    "typefind":     "format detection",
    "decodebin":    "auto demux/decode",
    "uridecodebin": "URI decode helper",
}

CODEC_ELEMENTS = {
    # demuxers
    "matroskademux": "MKV/WebM demuxer",
    "qtdemux":       "MP4/MOV demuxer",
    "flvdemux":      "FLV demuxer",
    "tsdemux":       "MPEG-TS demuxer",
    # video decoders
    "avdec_h264":    "H.264 software decoder (libav)",
    "avdec_hevc":    "H.265/HEVC software decoder (libav)",
    "avdec_mpeg2video": "MPEG-2 video decoder",
    "avdec_vp8":     "VP8 decoder",
    "avdec_vp9":     "VP9 decoder",
    "v4l2h264dec":   "H.264 V4L2 HW decoder",
    "v4l2h265dec":   "H.265 V4L2 HW decoder",
    # audio decoders
    "avdec_aac":     "AAC decoder (libav)",
    "avdec_mp3":     "MP3 decoder (libav)",
    "avdec_ac3":     "AC-3 decoder (libav)",
    "vorbisdec":     "Vorbis decoder",
    "opusdec":       "Opus decoder",
    # audio output
    "autoaudiosink": "Auto audio sink",
    "alsasink":      "ALSA sink",
    "pulsesink":     "PulseAudio sink",
}

missing_required = []
registry = Gst.Registry.get()

for element, desc in REQUIRED_ELEMENTS.items():
    feat = registry.find_feature(element, Gst.ElementFactory.__gtype__)
    if feat:
        _ok(f"{element:20s}  ({desc})")
    else:
        missing_required.append(element)
        _fail(f"{element:20s}  ({desc})  ← MISSING")

print()

missing_codecs = []
present_codecs = []
for element, desc in CODEC_ELEMENTS.items():
    feat = registry.find_feature(element, Gst.ElementFactory.__gtype__)
    if feat:
        present_codecs.append(element)
        _ok(f"{element:20s}  ({desc})")
    else:
        missing_codecs.append(element)
        _warn(f"{element:20s}  ({desc})  ← not found")

if missing_required:
    print(f"\n  CRITICAL: {len(missing_required)} required element(s) missing: {missing_required}")

print(f"\n  Codecs present: {len(present_codecs)} / {len(CODEC_ELEMENTS)}")


# ── 4. Caps negotiation: can we build a BGRA appsink pipeline? ──────────────

_section("4. BGRA appsink pipeline negotiation")

TEST_PIPE = (
    "videotestsrc num-buffers=5 ! "
    "videoconvert ! "
    "video/x-raw,format=BGRA,width=64,height=64 ! "
    "appsink name=sink emit-signals=true max-buffers=1 drop=true"
)

frames_received = 0
negotiation_ok = False

try:
    pipe = Gst.parse_launch(TEST_PIPE)
    sink = pipe.get_by_name("sink")

    def _on_sample(sink, *_):
        global frames_received
        sample = sink.emit("pull-sample")
        if sample:
            frames_received += 1
        return Gst.FlowReturn.OK

    sink.connect("new-sample", _on_sample)
    pipe.set_state(Gst.State.PLAYING)
    time.sleep(1.5)
    pipe.set_state(Gst.State.NULL)

    if frames_received > 0:
        negotiation_ok = True
        _ok(f"BGRA appsink pipeline: received {frames_received} frame(s)")
    else:
        _fail("BGRA appsink pipeline ran but produced 0 frames")

except Exception as exc:
    _fail(f"Could not build BGRA appsink pipeline: {exc}")


# ── 5. Audio output probe ───────────────────────────────────────────────────

_section("5. Audio output probe")

AUDIO_PIPE = "audiotestsrc num-buffers=10 freq=440 ! autoaudiosink"
audio_ok = False
try:
    p = Gst.parse_launch(AUDIO_PIPE)
    p.set_state(Gst.State.PLAYING)
    time.sleep(0.8)
    p.set_state(Gst.State.NULL)
    audio_ok = True
    _ok("autoaudiosink: played 440 Hz tone without error")
except Exception as exc:
    _fail(f"autoaudiosink failed: {exc}")

if not audio_ok:
    ALSA_PIPE = "audiotestsrc num-buffers=10 freq=440 ! alsasink device=hw:0"
    try:
        p = Gst.parse_launch(ALSA_PIPE)
        p.set_state(Gst.State.PLAYING)
        time.sleep(0.8)
        p.set_state(Gst.State.NULL)
        audio_ok = True
        _ok("alsasink hw:0: played 440 Hz tone")
    except Exception as exc:
        _warn(f"alsasink also failed: {exc}")


# ── 6. Live URL playback (optional) ────────────────────────────────────────

test_url = sys.argv[1] if len(sys.argv) > 1 else None

_section("6. Live URL / file playback")

if not test_url:
    _warn("No URL provided — skipping live playback test.")
    print("       Pass a media URL or path as the first argument to test real decode.")
else:
    print(f"  URL: {test_url}")
    live_frames = 0
    live_error = None

    LIVE_PIPE = (
        f"playbin uri=\"{test_url}\" "
        f"video-sink=\"videoconvert ! video/x-raw,format=BGRA ! appsink name=vsink emit-signals=true max-buffers=2 drop=true\""
    )

    try:
        pipe = Gst.parse_launch(LIVE_PIPE)
        vsink = pipe.get_by_name("vsink")

        def _on_live_sample(sink, *_):
            global live_frames
            sample = sink.emit("pull-sample")
            if sample:
                buf = sample.get_buffer()
                info = buf.map(Gst.MapFlags.READ)
                if info.size > 0:
                    live_frames += 1
                buf.unmap(info)
            return Gst.FlowReturn.OK

        vsink.connect("new-sample", _on_live_sample)

        bus = pipe.get_bus()
        pipe.set_state(Gst.State.PLAYING)

        deadline = time.monotonic() + 15
        while time.monotonic() < deadline:
            msg = bus.timed_pop_filtered(
                200 * Gst.MSECOND,
                Gst.MessageType.ERROR | Gst.MessageType.WARNING | Gst.MessageType.EOS,
            )
            if msg:
                if msg.type == Gst.MessageType.ERROR:
                    err, debug = msg.parse_error()
                    live_error = f"{err.message} | debug: {debug}"
                    break
                if msg.type == Gst.MessageType.WARNING:
                    w, d = msg.parse_warning()
                    _warn(f"GStreamer warning: {w.message}")
                if msg.type == Gst.MessageType.EOS:
                    break
            if live_frames >= 10:
                break

        pipe.set_state(Gst.State.NULL)

        if live_error:
            _fail(f"Playback error: {live_error}")
        elif live_frames == 0:
            _fail("Playback ran but decoded 0 video frames (audio-only or decode failure)")
        else:
            _ok(f"Decoded {live_frames} video frame(s) successfully")

    except Exception as exc:
        _fail(f"Could not start live playback pipeline: {exc}")


# ── 7. Summary ─────────────────────────────────────────────────────────────

_section("7. Summary")

issues = []
if missing_required:
    issues.append(f"Missing required elements: {', '.join(missing_required)}")
if not negotiation_ok:
    issues.append("BGRA appsink pipeline negotiation failed")
if not audio_ok:
    issues.append("No working audio sink found")

key_missing = [e for e in ("avdec_h264", "avdec_aac", "matroskademux", "qtdemux") if e in missing_codecs]
if key_missing:
    issues.append(f"Key codecs missing (install gst-libav): {', '.join(key_missing)}")

if not issues:
    print("  All checks passed — video playback should work.")
else:
    print("  Issues found:")
    for issue in issues:
        print(f"    • {issue}")
    print()
    print("  Suggested fix:")
    print(textwrap.dedent("""\
        /home/ark/miniconda3/bin/conda install -n r36s-dlna-browser \\
            -c conda-forge gst-libav gst-plugins-good gst-plugins-bad gst-plugins-ugly
    """))

print()