From e608b95bf1fe225915eaeafca2a933f687844733 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 11 Aug 2026 16:47:01 +0200 Subject: feat: Phase 10c — MSE video streaming (real-time playback) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace download-then-play VideoPlayer with MSE (MediaSource Extensions) streaming. Node remuxes to fMP4 via ffmpeg, probes codecs with ffprobe, and sends encrypted segments over DataChannel. Browser decrypts and appends to SourceBuffer — playback starts within seconds. Co-Authored-By: Claude Opus 4.6 --- .../src/meshbay_node/transport/webrtc_server.py | 133 +++++++++++++++++++++ 1 file changed, 133 insertions(+) (limited to 'packages/meshbay-node/src/meshbay_node') diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 1d10aa8..ab95550 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -46,6 +46,58 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 +STREAM_SEGMENT_SIZE = 256 * 1024 + +_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} + + +async def _probe_video(path: str) -> tuple[str | None, float]: + """Probe video file with ffprobe, return (MSE codec string, duration).""" + import json as _json + proc = await asyncio.create_subprocess_exec( + "ffprobe", "-v", "error", + "-show_entries", "stream=codec_name,profile,level,codec_type", + "-show_entries", "format=duration", + "-of", "json", path, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await proc.communicate() + info = _json.loads(stdout) + duration = float(info.get("format", {}).get("duration", 0)) + + v_codec = a_codec = "" + for s in info.get("streams", []): + if s.get("codec_type") == "video" and not v_codec: + cn = s.get("codec_name", "") + if cn == "h264": + p = _H264_PROFILES.get(s.get("profile", "High"), "64") + lvl = int(s.get("level", 40)) + v_codec = f"avc1.{p}00{lvl:02x}" + elif cn == "hevc": + v_codec = "hev1.1.6.L93.B0" + elif cn == "vp9": + v_codec = "vp09.00.10.08" + elif cn == "av1": + v_codec = "av01.0.01M.08" + elif s.get("codec_type") == "audio" and not a_codec: + cn = s.get("codec_name", "") + if cn == "aac": + a_codec = "mp4a.40.2" + elif cn in ("mp3", "mp2"): + a_codec = "mp4a.6b" + elif cn == "opus": + a_codec = "opus" + elif cn == "ac3": + a_codec = "ac-3" + elif cn == "flac": + a_codec = "flac" + + if not v_codec: + return None, duration + codec = f"{v_codec},{a_codec}" if a_codec else v_codec + return codec, duration + + def _pack(obj: dict) -> bytes: data = msgpack.packb(obj, use_bin_type=True) return struct.pack(">I", len(data)) + data @@ -118,6 +170,8 @@ class WebRTCPeerSession: self._do_file_upload(msg) elif mtype == MNP.FILE_DELETE: self._do_file_delete(msg) + elif mtype == MNP.STREAM_REQUEST: + asyncio.ensure_future(self._stream_video(msg)) else: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: @@ -417,6 +471,85 @@ class WebRTCPeerSession: "file_id": file_id, }) + async def _stream_video(self, msg: dict) -> None: + """Stream a video file as fMP4 segments via MSE-compatible output.""" + ctx = self._group_ctx() + file_id = msg.get("file_id", "") + entry = ctx["index"].get_entry(file_id) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + file_path = ctx["shared_root"] / entry.path / entry.name + if not file_path.exists(): + self._send({"type": "error", "detail": "File not on disk"}) + return + + gek = ctx.get("gek") + file_hash = bytes.fromhex(entry.id) + + try: + codec_str, duration = await _probe_video(str(file_path)) + except Exception as e: + self._send({"type": "error", "detail": f"Probe failed: {e}"}) + return + + if not codec_str: + self._send({"type": "error", "detail": "Unsupported video codec"}) + return + + proc = await asyncio.create_subprocess_exec( + "ffmpeg", "-hide_banner", "-loglevel", "error", + "-i", str(file_path), + "-c", "copy", + "-movflags", "frag_keyframe+empty_moov+default_base_moof", + "-f", "mp4", "pipe:1", + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + + self._send({ + "type": MNP.STREAM_INIT, + "v": MNP_VERSION, + "file_id": file_id, + "codec": codec_str, + "duration": duration, + }) + + index = 0 + try: + while True: + data = await proc.stdout.read(STREAM_SEGMENT_SIZE) + if not data: + break + ckey = chunk_key_aes(gek, file_hash, index) + nonce, ct = encrypt_chunk_aes(ckey, data) + self._send({ + "type": MNP.STREAM_DATA, + "v": MNP_VERSION, + "file_id": file_id, + "segment_index": index, + "nonce": nonce, + "ct": ct, + "plaintext_size": len(data), + }) + index += 1 + await asyncio.sleep(0) + except Exception as e: + log.error("Stream error: %s", e) + finally: + try: + proc.kill() + except ProcessLookupError: + pass + await proc.wait() + + self._send({ + "type": MNP.STREAM_END, + "v": MNP_VERSION, + "file_id": file_id, + }) + log.info("Streamed %s: %d segments", entry.name, index) + def _send(self, obj: dict) -> None: if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) -- cgit v1.2.3