#!/usr/bin/env python3 """Measure the companion's HCS1 video stream; no screenshots/files are retained. Development tool, not the plugin's production player. Requires official HDC, the signed companion on the phone, and ffmpeg (or ffplay with --view). """ import argparse import json import os from pathlib import Path import re import secrets import select import shutil import socket import struct import subprocess import threading import time BUNDLE = "ink.render.trove.harmonyconnector" MAX_PACKET = 2 * 1024 * 1024 def read_exact(sock, size, deadline): result = bytearray() while len(result) < size: remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("Video stream timed out") sock.settimeout(min(remaining, 0.2)) try: part = sock.recv(size - len(result)) except socket.timeout: continue if not part: raise EOFError("Screen sharing stopped or USB disconnected") result.extend(part) return bytes(result) def read_hello(sock, deadline): magic, width, height, display_width, display_height, fps = struct.unpack( ">4s5I", read_exact(sock, 24, deadline)) if (magic != b"HCS1" or not 1 <= fps <= 60 or not all(2 <= d <= 8192 for d in (width, height, display_width, display_height)) or width % 2 or height % 2 or max(width, height) > 1280): raise ValueError("Unsupported stream header") return {"width": width, "height": height, "displayWidth": display_width, "displayHeight": display_height, "configuredFps": fps} def read_packet(sock, deadline): size, flags, pts = struct.unpack( ">IIq", read_exact(sock, 16, deadline)) if not 0 < size <= MAX_PACKET or flags & ~0x7f or pts < 0: raise ValueError("Invalid video packet") data = read_exact(sock, size, deadline) if not data.startswith((b"\0\0\1", b"\0\0\0\1")): raise ValueError("Encoder did not return Annex B H.264") return flags, pts, data class Decoder: def __init__(self, view): name = "ffplay" if view else "ffmpeg" executable = shutil.which(name) if not executable: raise RuntimeError(f"Development measurement requires {name} on PATH") args = [executable, "-hide_banner", "-loglevel", "error", "-flags", "low_delay", "-probesize", "32", "-analyzeduration", "0", "-f", "h264", "-i", "pipe:0"] if view: args += ["-window_title", "Harmony Connector · H.264 preview", "-autoexit"] else: args += ["-progress", "pipe:1", "-stats_period", "0.25", "-f", "null", "-"] self.frames = 0 self.error = bytearray() self.process = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) os.set_blocking(self.process.stdin.fileno(), False) self.threads = [threading.Thread(target=self.progress, daemon=True), threading.Thread(target=self.errors, daemon=True)] for thread in self.threads: thread.start() def progress(self): for line in self.process.stdout: if line.startswith(b"frame="): self.frames = int(line.split(b"=", 1)[1]) def errors(self): for line in self.process.stderr: if len(self.error) < 8192: self.error.extend(line[:8192 - len(self.error)]) def write(self, data): deadline = time.monotonic() + 1 view = memoryview(data) while view: if self.process.poll() is not None: raise RuntimeError("Video decoder exited: " + self.error.decode(errors="replace")) if time.monotonic() >= deadline: raise TimeoutError("Decoder is falling behind; stopped instead of queuing video") if not select.select([], [self.process.stdin], [], 0.05)[1]: continue try: count = os.write(self.process.stdin.fileno(), view) view = view[count:] except BlockingIOError: continue def close(self): self.process.stdin.close() try: self.process.wait(timeout=3) except subprocess.TimeoutExpired: self.process.terminate() try: self.process.wait(timeout=1) except subprocess.TimeoutExpired: self.process.kill() self.process.wait() for thread in self.threads: thread.join(timeout=1) self.process.stdout.close() self.process.stderr.close() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--hdc", required=True, type=Path) parser.add_argument("--target") parser.add_argument("--seconds", type=int, default=20) parser.add_argument("--view", action="store_true") args = parser.parse_args() if not 5 <= args.seconds <= 120 or not args.hdc.is_file(): parser.error("Use an existing HDC executable and 5–120 measurement seconds") def hdc(*command): try: result = subprocess.run([str(args.hdc.resolve()), *command], capture_output=True, text=True, timeout=10, check=False) except subprocess.TimeoutExpired: raise RuntimeError("HDC command timed out") from None if result.returncode: raise RuntimeError("HDC command did not complete successfully") if re.search(r"\b(fail(?:ed)?|error)\b", result.stdout + result.stderr, re.I): raise RuntimeError("HDC command failed: " + result.stdout + result.stderr) return result.stdout targets = [line.split()[0] for line in hdc("list", "targets", "-v").splitlines() if "Connected" in line.split()] target = args.target or (targets[0] if len(targets) == 1 else None) if target not in targets: raise RuntimeError("Select one connected, authorized phone with --target") installed = hdc("-t", target, "shell", f"bm dump -n {BUNDLE}") if BUNDLE not in installed: raise RuntimeError("Install the signed Harmony Connector companion before measuring") decoder = Decoder(args.view) sock = None forward = False with socket.socket() as reservation: reservation.bind(("127.0.0.1", 0)) local_port = reservation.getsockname()[1] remote_port = 20000 + secrets.randbelow(40000) local, remote = f"tcp:{local_port}", f"tcp:{remote_port}" token = secrets.token_hex(32) frames = total_bytes = 0 started = first_frame = None try: hdc("-t", target, "fport", local, remote) forward = True hdc("-t", target, "shell", f"aa start -b {BUNDLE} -a EntryAbility -m entry " f"--ps token {token} --pi port {remote_port}") print("请在手机点击「共享屏幕」，并同意系统授权。", flush=True) deadline = time.monotonic() + 90 while time.monotonic() < deadline: try: sock = socket.create_connection(("127.0.0.1", local_port), timeout=1) sock.sendall(token.encode("ascii")) hello = read_hello(sock, min(deadline, time.monotonic() + 2)) break except (OSError, EOFError, TimeoutError): if sock: sock.close() sock = None time.sleep(0.2) if not sock: raise TimeoutError("Phone did not start screen sharing") print(json.dumps({"stage": "connected", **hello}), flush=True) started = time.monotonic() last_report = started while first_frame is None or time.monotonic() - first_frame < args.seconds: flags, _, data = read_packet(sock, time.monotonic() + (45 if first_frame is None else 3)) decoder.write(data) total_bytes += len(data) now = time.monotonic() if not flags & 8: frames += 1 if first_frame is None: first_frame = now print(json.dumps({"stage": "first-frame", "waitMs": round((now - started) * 1000)}), flush=True) if first_frame and now - last_report >= 1: elapsed = now - first_frame print(json.dumps({"stage": "receiving", "frames": frames, "receivedFps": round((frames - 1) / elapsed, 1), "decodedFrames": decoder.frames, "mbps": round(total_bytes * 8 / elapsed / 1000000, 2)}), flush=True) last_report = now elapsed = time.monotonic() - first_frame finally: if sock: try: sock.sendall(b"\0") except OSError: pass sock.close() try: if forward: hdc("-t", target, "fport", "rm", local, remote) finally: decoder.close() passed = (frames - 1) / elapsed >= 20 and (args.view or decoder.frames >= frames * 0.95) print(json.dumps({"passed": passed, "frames": frames, "decodedFrames": decoder.frames, "receivedFps": round((frames - 1) / elapsed, 1), "durationSeconds": round(elapsed, 2), "endToEndLatencyMeasured": False, "inputSent": False, "screenFilesSaved": False}), flush=True) if not passed: raise SystemExit(1) if __name__ == "__main__": main()