char-fork/llama.cpp

git clone https://git.t4t.associates/char-fork/llama.cpp

Max Krasnyanskysnapdragon: ci updates to use new run script (#28293)d30500b83

char/aliceai-t5-moe
4.9 KiB152 linesraw
1"""Shared helpers for QDC on-device test runners."""
2
3from __future__ import annotations
4
5import logging
6import os
7import subprocess
8import sys
9import tempfile
10
11from appium.options.common import AppiumOptions
12
13log = logging.getLogger(__name__)
14
15# ---------------------------------------------------------------------------
16# On-device paths
17# ---------------------------------------------------------------------------
18
19BUNDLE_PATH = "/data/local/tmp/llama.cpp"
20BIN_PATH = f"{BUNDLE_PATH}/bin"
21LIB_PATH = f"{BUNDLE_PATH}/lib"
22QDC_LOGS_PATH = "/data/local/tmp/QDC_logs"
23SCRIPTS_DIR = "/qdc/appium"
24MODEL_NAME = "model.gguf"
25MODEL_DEVICE_PATH = "/data/local/tmp/gguf/model.gguf"
26PROMPT_DIR = "/data/local/tmp/scorecard_prompts"
27
28# ---------------------------------------------------------------------------
29# Appium session options
30# ---------------------------------------------------------------------------
31
32options = AppiumOptions()
33options.set_capability("automationName", "UiAutomator2")
34options.set_capability("platformName", "Android")
35options.set_capability("deviceName", os.getenv("ANDROID_DEVICE_VERSION"))
36
37# ---------------------------------------------------------------------------
38# Shell / process helpers
39# ---------------------------------------------------------------------------
40
41
42def write_qdc_log(filename: str, content: str) -> None:
43    """Write content as a log file for QDC log collection."""
44    subprocess.run(
45        ["adb", "shell", f"mkdir -p {QDC_LOGS_PATH}"],
46        stdout=subprocess.PIPE,
47        stderr=subprocess.STDOUT,
48    )
49    with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as f:
50        f.write(content)
51        tmp_path = f.name
52    try:
53        subprocess.run(
54            ["adb", "push", tmp_path, f"{QDC_LOGS_PATH}/{filename}"],
55            stdout=subprocess.PIPE,
56            stderr=subprocess.STDOUT,
57        )
58    finally:
59        os.unlink(tmp_path)
60
61
62def ensure_bundle(check_binary: str | None = None) -> None:
63    """Ensure the llama_cpp_bundle is available on the target device."""
64    push_bundle_if_needed(check_binary or f"{BIN_PATH}/llama-cli")
65
66
67# ---------------------------------------------------------------------------
68# Android / Linux host helpers
69# ---------------------------------------------------------------------------
70
71
72def run_adb_command(cmd: str, *, check: bool = True) -> subprocess.CompletedProcess:
73    """Run a command on-device via ``adb shell`` with exit-code sentinel."""
74    raw = subprocess.run(
75        ["adb", "shell", f"{cmd}; echo __RC__:$?"],
76        text=True,
77        stdout=subprocess.PIPE,
78        stderr=subprocess.STDOUT,
79    )
80    stdout = raw.stdout
81    returncode = raw.returncode
82    if stdout:
83        lines = stdout.rstrip("\n").split("\n")
84        if lines and lines[-1].startswith("__RC__:"):
85            try:
86                returncode = int(lines[-1][7:])
87                stdout = "\n".join(lines[:-1]) + "\n"
88            except ValueError:
89                pass
90    log.info(stdout)
91    result = subprocess.CompletedProcess(raw.args, returncode, stdout=stdout)
92    if check:
93        assert returncode == 0, f"Command failed (exit {returncode})"
94    return result
95
96
97def run_snapdragon(
98    cmd_args: list[str],
99    *,
100    device: str | None = None,
101    extra_run_args: list[str] | None = None,
102    extra_env: dict[str, str] | None = None,
103) -> subprocess.CompletedProcess:
104    """Run a tool via scripts/snapdragon/run.py targeting android."""
105    env = os.environ.copy()
106    env["GGML_HEXAGON_EXPERIMENTAL"] = "1"
107    if extra_env:
108        env.update(extra_env)
109    cmd = [sys.executable, f"{SCRIPTS_DIR}/run.py", "--target", "android"]
110    if device is not None:
111        cmd.extend(["-d", device])
112    if extra_run_args:
113        cmd.extend(extra_run_args)
114    cmd.append("--")
115    cmd.extend(cmd_args)
116    result = subprocess.run(
117        cmd, env=env,
118        text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
119    )
120    log.info(result.stdout)
121    return result
122
123
124def adb_shell(cmd: str) -> None:
125    """Run a command via adb shell (fire-and-forget, no error check)."""
126    subprocess.run(
127        ["adb", "shell", "sh", "-c", cmd],
128        capture_output=True, encoding="utf-8", errors="replace", check=False,
129    )
130
131
132def push_bundle_if_needed(check_binary: str) -> None:
133    """Push llama_cpp_bundle to the device if check_binary is not already present."""
134    result = subprocess.run(
135        ["adb", "shell", f"ls {check_binary}"],
136        text=True,
137        stdout=subprocess.PIPE,
138        stderr=subprocess.STDOUT,
139    )
140    if result.returncode != 0:
141        subprocess.run(
142            ["adb", "push", "/qdc/appium/llama_cpp_bundle/", BUNDLE_PATH],
143            text=True,
144            stdout=subprocess.PIPE,
145            stderr=subprocess.STDOUT,
146        )
147        subprocess.run(
148            ["adb", "shell", f"find {BUNDLE_PATH}/bin -type f -exec chmod 755 {{}} +"],
149            text=True,
150            stdout=subprocess.PIPE,
151            stderr=subprocess.STDOUT,
152        )