Velix-MPV-Core/benchmark/runner/run_emulator.py

#!/usr/bin/env python3
"""Repeatable Playback Lab emulator benchmark runner.

Only uses the standard library and the OpenHarmony hdc/uitest command line.
Metrics unavailable from the emulator are emitted as null, never inferred.
"""

from __future__ import annotations

import argparse
import json
import math
import os
import re
import shlex
import subprocess
import tempfile
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable


ROOT = Path(__file__).resolve().parents[2]
PACKAGE = "com.velix.playbacklab"
DEFAULT_SERIAL = "127.0.0.1:5555"
DEFAULT_HDC = "/Applications/DevEco-Studio.app/Contents/sdk/default/openharmony/toolchains/hdc"


def command(argv: list[str], timeout: float = 30.0) -> str:
    result = subprocess.run(argv, check=False, capture_output=True, text=True, timeout=timeout)
    if result.returncode != 0:
        raise RuntimeError(f"command failed ({result.returncode}): {shlex.join(argv)}\n{result.stderr.strip()}")
    return result.stdout.strip()


class Emulator:
    def __init__(self, serial: str, hdc: str) -> None:
        self.serial = serial
        self.hdc = hdc
        self.layout_counter = 0

    def hdc_shell(self, text: str, timeout: float = 30.0) -> str:
        return command([self.hdc, "-t", self.serial, "shell", text], timeout)

    def click(self, x: int, y: int) -> None:
        self.hdc_shell(f"uitest uiInput click {x} {y}", timeout=10)

    def layout(self) -> dict[str, Any]:
        self.layout_counter += 1
        marker = self.hdc_shell(f"uitest dumpLayout -b {PACKAGE} -m true -d 0", timeout=10)
        match = re.search(r"saved to:(\S+)", marker)
        if not match:
            raise RuntimeError(f"layout dump did not return a path: {marker}")
        remote = match.group(1)
        local = Path(tempfile.gettempdir()) / f"velix-layout-{os.getpid()}-{self.layout_counter}.json"
        command([self.hdc, "-t", self.serial, "file", "recv", remote, str(local)], timeout=10)
        with local.open(encoding="utf-8") as stream:
            return json.load(stream)

    def pid(self) -> int | None:
        output = self.hdc_shell(f"pidof {PACKAGE}", timeout=10)
        match = re.search(r"\b(\d+)\b", output)
        return int(match.group(1)) if match else None

    def process_sample(self) -> dict[str, Any]:
        pid = self.pid()
        sample: dict[str, Any] = {"pid": pid, "rssKb": None, "pssKb": None, "heapKb": None,
                                  "virtualMemoryKb": None,
                                  "threadCount": None, "fdCount": None, "cpuPermille": None,
                                  "cpuSupported": False}
        if pid is None:
            return sample
        status = self.hdc_shell(f"cat /proc/{pid}/status", timeout=10)
        for key, field in (("rssKb", "VmRSS"), ("virtualMemoryKb", "VmSize"), ("threadCount", "Threads")):
            match = re.search(rf"^{field}:\s+(\d+)", status, re.MULTILINE)
            if match:
                sample[key] = int(match.group(1))
        fd_output = self.hdc_shell(f"ls /proc/{pid}/fd 2>&1 | wc -l", timeout=10)
        if fd_output.isdigit() and fd_output != "0":
            sample["fdCount"] = int(fd_output)
        stat = self.hdc_shell(f"cat /proc/{pid}/stat", timeout=10)
        match = re.match(r"^\d+ \(.+\) (.*)$", stat)
        if match:
            fields = match.group(1).split()
            if len(fields) > 12:
                sample["cpuJiffies"] = int(fields[11]) + int(fields[12])
        return sample

    def product_version(self) -> str | None:
        try:
            value = self.hdc_shell("param get const.product.software.version", timeout=10)
        except RuntimeError:
            return None
        return value or None


def walk_nodes(node: dict[str, Any]) -> Iterable[dict[str, Any]]:
    yield node
    for child in node.get("children", []):
        if isinstance(child, dict):
            yield from walk_nodes(child)


def node_text(node: dict[str, Any]) -> str:
    attrs = node.get("attributes", {})
    return str(attrs.get("text") or attrs.get("originalText") or "")


def find_node(layout: dict[str, Any], text: str) -> dict[str, Any] | None:
    for node in walk_nodes(layout):
        if node_text(node) == text:
            return node
    return None


def center(bounds: str) -> tuple[int, int]:
    values = [int(value) for value in re.findall(r"\d+", bounds)]
    if len(values) != 4:
        raise ValueError(f"invalid bounds: {bounds}")
    return (values[0] + values[2]) // 2, (values[1] + values[3]) // 2


def click_text(emulator: Emulator, text: str) -> None:
    node = find_node(emulator.layout(), text)
    if node is None:
        raise RuntimeError(f"UI node not found: {text}")
    emulator.click(*center(node["attributes"]["bounds"]))


def all_texts(layout: dict[str, Any]) -> list[str]:
    return [node_text(node) for node in walk_nodes(layout) if node_text(node)]


def text_matching(layout: dict[str, Any], prefix: str) -> str | None:
    for value in all_texts(layout):
        if value.startswith(prefix):
            return value
    return None


def parse_int(text: str | None, pattern: str) -> int | None:
    if not text:
        return None
    match = re.search(pattern, text)
    return int(match.group(1)) if match else None


def parse_state(layout: dict[str, Any]) -> str | None:
    value = text_matching(layout, "state: ")
    return value.split(": ", 1)[1] if value and ": " in value else None


def parse_run(layout: dict[str, Any]) -> dict[str, Any]:
    startup = text_matching(layout, "startup=") or ""
    buffer_line = text_matching(layout, "buffer enter=") or ""
    napi_line = text_matching(layout, "napi calls=") or ""
    render_line = text_matching(layout, "render updates=") or napi_line
    error = text_matching(layout, "error=") or ""
    timeline: dict[str, Any] = {}
    for key, value in re.findall(r"(startup|firstSwap|seekRecovery|audioStart|audioCallback)=(-?\d+)", startup):
        timeline[{"startup": "startupTotalMs", "firstSwap": "firstSwapMs", "seekRecovery": "seekRecoveryMs",
                  "audioStart": "audioRendererStartMs", "audioCallback": "firstAudioCallbackMs"}[key]] = int(value)
    first_frame = parse_int(text_matching(layout, "dropped="), r"firstFrameMs=(-?\d+)")
    if first_frame is None or first_frame < 0:
        first_frame = timeline.get("firstSwapMs")
    if first_frame is not None and first_frame < 0:
        first_frame = None
    startup_ms = timeline.get("startupTotalMs")
    first_swap_ms = timeline.get("firstSwapMs")
    seek_recovery_ms = timeline.get("seekRecoveryMs")
    audio_start_ms = timeline.get("audioRendererStartMs")
    audio_callback_ms = timeline.get("firstAudioCallbackMs")
    if startup_ms is not None and startup_ms < 0:
        startup_ms = None
    if first_swap_ms is not None and first_swap_ms < 0:
        first_swap_ms = None
    if seek_recovery_ms is not None and seek_recovery_ms < 0:
        seek_recovery_ms = None
    if audio_start_ms is not None and audio_start_ms < 0:
        audio_start_ms = None
    if audio_callback_ms is not None and audio_callback_ms < 0:
        audio_callback_ms = None
    return {
        "state": parse_state(layout),
        "firstFrameMs": first_frame,
        "renderCount": parse_int(text_matching(layout, "dropped="), r"renderCount=(-?\d+)"),
        "startupTotalMs": startup_ms,
        "firstSwapMs": first_swap_ms,
        "seekRecoveryMs": seek_recovery_ms,
        "audioRendererStartMs": audio_start_ms,
        "firstAudioCallbackMs": audio_callback_ms,
        "bufferingEnterCount": parse_int(buffer_line, r"buffer enter=(-?\d+)"),
        "bufferingExitCount": parse_int(buffer_line, r"exit=(-?\d+)"),
        "napiCalls": parse_int(napi_line, r"napi calls=(-?\d+)"),
        "renderUpdateCalls": parse_int(render_line, r"updates=(-?\d+)/"),
        "renderFrameUpdates": parse_int(render_line, r"updates=-?\d+/(-?\d+)"),
        "coreSyncCalls": parse_int(render_line, r"sync=(-?\d+)/"),
        "coreSyncSkipped": parse_int(render_line, r"sync=-?\d+/(-?\d+)"),
        "error": error,
    }


def wait_for_first_frame(emulator: Emulator, timeout: float = 10.0) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    deadline = time.monotonic() + timeout
    samples: list[dict[str, Any]] = []
    latest: dict[str, Any] = {}
    next_sample = 0.0
    while time.monotonic() < deadline:
        latest = emulator.layout()
        now = time.monotonic()
        if now >= next_sample:
            samples.append({"atMs": int((timeout - max(0.0, deadline - now)) * 1000),
                            **emulator.process_sample()})
            next_sample = now + 1.0
        run = parse_run(latest)
        if run.get("startupTotalMs") is not None or run.get("state") == "ERROR":
            return latest, samples
        time.sleep(0.15)
    raise TimeoutError(f"playback did not expose first-frame timeline: {parse_run(latest)}")


def stats(values: list[float | int]) -> dict[str, float | int | None]:
    numbers = sorted(float(value) for value in values if value is not None and math.isfinite(float(value)))
    if not numbers:
        return {"count": 0, "min": None, "max": None, "avg": None, "median": None, "p90": None, "p95": None}
    def percentile(fraction: float) -> float:
        return numbers[max(0, min(len(numbers) - 1, math.ceil(fraction * len(numbers)) - 1))]
    return {"count": len(numbers), "min": numbers[0], "max": numbers[-1],
            "avg": sum(numbers) / len(numbers), "median": percentile(0.5),
            "p90": percentile(0.9), "p95": percentile(0.95)}


def git_revision(path: Path) -> str | None:
    try:
        return command(["git", "-C", str(path), "rev-parse", "HEAD"], timeout=10)
    except RuntimeError:
        return None


def load_local(emulator: Emulator) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    click_text(emulator, "Load Local Test")
    return wait_for_first_frame(emulator)


def startup_local(emulator: Emulator, repeat: int) -> list[dict[str, Any]]:
    runs: list[dict[str, Any]] = []
    for index in range(repeat):
        began = time.monotonic()
        click_text(emulator, "Load Local Test")
        layout, samples = wait_for_first_frame(emulator)
        run = parse_run(layout)
        run.update({"runIndex": index + 1, "cacheClass": "cold-ish" if index == 0 else "warm",
                    "hostFirstFrameMs": int((time.monotonic() - began) * 1000),
                    "processSamples": samples})
        runs.append(run)
    return runs


def wait_for_seek_recovery(emulator: Emulator, timeout: float = 5.0) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    deadline = time.monotonic() + timeout
    samples: list[dict[str, Any]] = []
    latest: dict[str, Any] = {}
    next_sample = 0.0
    while time.monotonic() < deadline:
        latest = emulator.layout()
        now = time.monotonic()
        if now >= next_sample:
            samples.append({"atMs": int((timeout - max(0.0, deadline - now)) * 1000),
                            **emulator.process_sample()})
            next_sample = now + 1.0
        run = parse_run(latest)
        if run.get("seekRecoveryMs") is not None and run.get("seekRecoveryMs") >= 0:
            return latest, samples
        if run.get("state") == "ERROR":
            return latest, samples
        time.sleep(0.1)
    raise TimeoutError(f"seek did not expose recovery timeline: {parse_run(latest)}")


def seek_local(emulator: Emulator, repeat: int) -> list[dict[str, Any]]:
    runs: list[dict[str, Any]] = []
    for index in range(repeat):
        click_text(emulator, "Load Local Test")
        # The bundled clip is intentionally short and may finish before a
        # multi-command text-input interaction completes.  The native +/-10s
        # action exercises the same seek API without that race.
        time.sleep(0.05)
        click_text(emulator, "+10s")
        latest, samples = wait_for_seek_recovery(emulator)
        run = parse_run(latest)
        run.update({"runIndex": index + 1, "mode": "relative", "offsetSeconds": 10,
                    "processSamples": samples})
        runs.append(run)
    return runs


def stress_lifecycle(emulator: Emulator, repeat: int) -> list[dict[str, Any]]:
    initial = emulator.layout()
    load_center = center(find_node(initial, "Load Local Test")["attributes"]["bounds"])
    stop_center = center(find_node(initial, "Stop")["attributes"]["bounds"])
    runs: list[dict[str, Any]] = []
    for index in range(repeat):
        emulator.click(*load_center)
        time.sleep(0.08)
        emulator.click(*stop_center)
        time.sleep(0.08)
        layout = emulator.layout()
        run = parse_run(layout)
        run.update({"runIndex": index + 1, "processSample": emulator.process_sample()})
        runs.append(run)
    return runs


def play_pause(emulator: Emulator, repeat: int) -> list[dict[str, Any]]:
    load_local(emulator)
    initial = emulator.layout()
    play_center = center(find_node(initial, "Play")["attributes"]["bounds"])
    pause_center = center(find_node(initial, "Pause")["attributes"]["bounds"])
    runs: list[dict[str, Any]] = []
    for index in range(repeat):
        emulator.click(*play_center)
        emulator.click(*pause_center)
        if index % 10 == 0:
            time.sleep(0.05)
        run = {"runIndex": index + 1, "processSample": emulator.process_sample()}
        if index % 10 == 0 or index + 1 == repeat:
            run.update(parse_run(emulator.layout()))
        runs.append(run)
    return runs


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("scenario", choices=["startup_local", "seek_local", "stress_lifecycle", "play_pause"])
    parser.add_argument("--repeat", type=int, default=None)
    parser.add_argument("--serial", default=os.environ.get("DEVECO_DEVICE", DEFAULT_SERIAL))
    parser.add_argument("--hdc", default=os.environ.get("HDC", DEFAULT_HDC))
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    defaults = {"startup_local": 20, "seek_local": 50, "stress_lifecycle": 100, "play_pause": 200}
    repeat = args.repeat if args.repeat is not None else defaults[args.scenario]
    emulator = Emulator(args.serial, args.hdc)
    if args.scenario == "startup_local":
        runs = startup_local(emulator, repeat)
    elif args.scenario == "seek_local":
        runs = seek_local(emulator, repeat)
    elif args.scenario == "stress_lifecycle":
        runs = stress_lifecycle(emulator, repeat)
    else:
        runs = play_pause(emulator, repeat)
    metric_names = ["firstFrameMs", "startupTotalMs", "firstSwapMs", "seekRecoveryMs", "audioRendererStartMs",
                    "firstAudioCallbackMs", "bufferingEnterCount", "bufferingExitCount", "hostFirstFrameMs", "hostTerminalMs"]
    output = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "gitCommit": git_revision(ROOT),
        "mpvCommit": git_revision(ROOT),
        "ffmpegCommit": git_revision(ROOT.parent / "Velix-FFmpeg-Harmony"),
        "deviceType": "emulator",
        "deviceSerial": args.serial,
        "harmonyVersion": emulator.product_version(),
        "testScenario": args.scenario,
        "mediaId": "sample.mp4",
        "protocol": "file",
        "codec": "H.264/AAC",
        "resolution": None,
        "audio": "ao_harmony",
        "subtitle": "none",
        "hwdec": "software",
        "repeat": repeat,
        "metrics": {name: stats([run.get(name) for run in runs]) for name in metric_names},
        "runs": runs,
        "errors": [run.get("error") for run in runs if run.get("error") and "error=NONE" not in run.get("error", "")],
        "sampling": {"pss": "unsupported unless exposed by device", "cpu": "raw /proc stat only when available"},
    }
    output_path = args.output or ROOT / "benchmark" / "results" / f"{args.scenario}-{int(time.time())}.json"
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(output, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(output_path)
    print(json.dumps(output["metrics"], ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())