File size: 5,048 Bytes
0938148 31dadc6 0938148 f2f6076 0938148 31dadc6 0938148 f2f6076 0938148 6ec1d3e 0938148 31dadc6 6ec1d3e 31dadc6 0938148 6ec1d3e 0938148 31dadc6 0938148 f2f6076 31dadc6 0938148 f2f6076 0938148 31dadc6 6ec1d3e 0938148 f2f6076 0938148 31dadc6 f2f6076 31dadc6 f2f6076 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | # ---------------------------------------------------------------------
# Copyright (c) 2025 Qualcomm Technologies, Inc. and/or its subsidiaries.
# SPDX-License-Identifier: BSD-3-Clause
# ---------------------------------------------------------------------
"""Shared helpers for QDC on-device test runners (Linux IoT)."""
import logging
import os
import shutil
import subprocess
log = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# On-device paths
# ---------------------------------------------------------------------------
BUNDLE_PATH = "/tmp/llama_cpp_bundle"
QDC_LOGS_PATH = "/tmp/QDC_logs"
LIB_PATH = f"{BUNDLE_PATH}/lib"
BIN_PATH = f"{BUNDLE_PATH}/bin"
ENV_PREFIX = (
f"export LD_LIBRARY_PATH={LIB_PATH}:$LD_LIBRARY_PATH && "
f"export ADSP_LIBRARY_PATH={LIB_PATH} && "
f"chmod +x {BIN_PATH}/* &&"
)
CMD_PREFIX = f"cd {BUNDLE_PATH} && {ENV_PREFIX}"
# ---------------------------------------------------------------------------
# Shell helpers
# ---------------------------------------------------------------------------
def run_shell_command(cmd: str, *, check: bool = True, timeout: int = 600) -> subprocess.CompletedProcess:
log.info("Running: %s", cmd[:200])
try:
result = subprocess.run(
["sh", "-c", cmd],
text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
timeout=timeout,
)
except subprocess.TimeoutExpired as e:
output = e.stdout or ""
print(output)
raise AssertionError(f"Command timed out after {timeout}s") from e
print(result.stdout)
if check:
assert result.returncode == 0, (
f"Command failed (exit {result.returncode}):\n"
f" cmd: {cmd[:200]}\n"
f" last 500 chars of output: {(result.stdout or '')[-500:]}"
)
return result
def write_qdc_log(filename: str, content: str) -> None:
"""Write content as a log file to QDC_LOGS_PATH for QDC log collection."""
os.makedirs(QDC_LOGS_PATH, exist_ok=True)
with open(f"{QDC_LOGS_PATH}/{filename}", "w") as f:
f.write(content)
def _find_bundle_source() -> str:
"""Locate the llama_cpp_bundle directory from the QDC extraction."""
candidates = [
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "llama_cpp_bundle"),
"/qdc/appium/llama_cpp_bundle",
"/qdc/bash/llama_cpp_bundle",
]
for c in candidates:
real = os.path.realpath(c)
if os.path.isdir(real):
log.info("Found bundle source at %s", real)
return real
# Debug: dump known directories to help diagnose
for check_dir in ["/qdc", "/qdc/appium", "/qdc/bash",
os.path.dirname(os.path.abspath(__file__)),
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")]:
if os.path.isdir(check_dir):
log.info("[debug] %s: %s", check_dir, os.listdir(check_dir))
else:
log.info("[debug] %s: does not exist", check_dir)
raise FileNotFoundError("llama_cpp_bundle not found in any known QDC extraction path")
def push_bundle_if_needed(check_binary: str) -> None:
"""Copy llama_cpp_bundle to /tmp if check_binary is not already present."""
if os.path.exists(check_binary):
log.info("Binary already present: %s", check_binary)
return
src = _find_bundle_source()
log.info("Copying bundle from %s to %s", src, BUNDLE_PATH)
shutil.copytree(src, BUNDLE_PATH, dirs_exist_ok=True)
subprocess.run(["chmod", "-R", "+x", BIN_PATH], check=False)
if not os.path.exists(check_binary):
available = os.listdir(BIN_PATH) if os.path.isdir(BIN_PATH) else []
raise FileNotFoundError(
f"Binary {check_binary} not found after bundle copy. "
f"Available binaries: {available}"
)
log.info("Bundle installed, verified %s exists", check_binary)
def log_environment() -> None:
"""Log environment info for debugging."""
script_dir = os.path.dirname(os.path.abspath(__file__))
parent_dir = os.path.dirname(script_dir)
lines = [
f"__file__: {os.path.abspath(__file__)}",
f"script_dir: {script_dir}",
f"parent_dir contents: {os.listdir(parent_dir) if os.path.isdir(parent_dir) else 'missing'}",
f"BUNDLE_PATH exists: {os.path.isdir(BUNDLE_PATH)}",
f"BIN_PATH contents: {os.listdir(BIN_PATH) if os.path.isdir(BIN_PATH) else 'missing'}",
f"LIB_PATH contents: {os.listdir(LIB_PATH) if os.path.isdir(LIB_PATH) else 'missing'}",
f"LD_LIBRARY_PATH: {os.environ.get('LD_LIBRARY_PATH', 'not set')}",
f"/qdc/appium/ contents: {os.listdir('/qdc/appium/') if os.path.isdir('/qdc/appium/') else 'missing'}",
f"/qdc/bash/ contents: {os.listdir('/qdc/bash/') if os.path.isdir('/qdc/bash/') else 'missing'}",
]
for line in lines:
log.info("[env] %s", line)
write_qdc_log("environment.log", "\n".join(lines) + "\n")
|