| |
| |
| |
| |
| """Shared helpers for QDC on-device test runners (Linux IoT).""" |
|
|
| import logging |
| import os |
| import shutil |
| import subprocess |
|
|
| log = logging.getLogger(__name__) |
|
|
| |
| |
| |
|
|
| 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}" |
|
|
| |
| |
| |
|
|
| 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 |
| |
| 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") |
|
|
