#!/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()