matrix-screen-controller/核桃派软件源代码/scripts/dedicated_host.py

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Callable, Sequence


EXPECTED_PINCTRL = "allwinner,sun50i-h616-pinctrl"
DISABLED_SERVICES = (
    "lightdm.service",
    "bluetooth.service",
    "aw859-bluetooth.service",
    "gpioc-server.service",
    "map_device.service",
)


@dataclass(frozen=True)
class CommandResult:
    returncode: int
    stdout: str = ""
    stderr: str = ""


@dataclass(frozen=True)
class HostPaths:
    pinctrl_compatible: Path
    dev_mem: Path
    i2c_device: Path
    cpu3: Path
    native_library: Path

    @classmethod
    def detect(
        cls,
        root: Path = Path("/"),
        source_root: Path | None = None,
    ) -> "HostPaths":
        source_root = source_root or Path(__file__).resolve().parents[1]
        compatible = (
            root
            / "proc"
            / "device-tree"
            / "soc"
            / "pinctrl@300b000"
            / "compatible"
        )
        return cls(
            pinctrl_compatible=compatible,
            dev_mem=root / "dev" / "mem",
            i2c_device=root / "dev" / "i2c-1",
            cpu3=root / "sys" / "devices" / "system" / "cpu" / "cpu3",
            native_library=(
                source_root / "app" / "display" / "native" / "libh618_hub75.so"
            ),
        )


@dataclass(frozen=True)
class HostCheck:
    name: str
    ok: bool
    detail: str


Runner = Callable[[Sequence[str]], CommandResult]


def run_command(args: Sequence[str]) -> CommandResult:
    completed = subprocess.run(
        list(args),
        check=False,
        capture_output=True,
        text=True,
    )
    return CommandResult(completed.returncode, completed.stdout, completed.stderr)


def _compatible_values(path: Path) -> tuple[str, ...]:
    if not path.is_file():
        return ()
    return tuple(
        value.decode("ascii", errors="replace")
        for value in path.read_bytes().split(b"\0")
        if value
    )


def check_profile(
    paths: HostPaths,
    *,
    service_only: bool = False,
    runner: Runner = run_command,
) -> list[HostCheck]:
    compatible = _compatible_values(paths.pinctrl_compatible)
    architecture = runner(("uname", "-m"))
    network_manager = runner(("systemctl", "is-active", "NetworkManager.service"))

    checks = [
        HostCheck(
            "H618 PI pinctrl",
            EXPECTED_PINCTRL in compatible,
            ", ".join(compatible) if compatible else "compatible node missing",
        ),
        HostCheck(
            "/dev/mem accessible",
            paths.dev_mem.exists()
            and os.access(paths.dev_mem, os.R_OK | os.W_OK),
            str(paths.dev_mem),
        ),
        HostCheck(
            "/dev/i2c-1 present",
            paths.i2c_device.exists(),
            str(paths.i2c_device),
        ),
        HostCheck("CPU3 present", paths.cpu3.is_dir(), str(paths.cpu3)),
        HostCheck(
            "native HUB75 library",
            paths.native_library.is_file(),
            str(paths.native_library),
        ),
        HostCheck(
            "AArch64 userspace",
            architecture.returncode == 0 and architecture.stdout.strip() == "aarch64",
            architecture.stdout.strip() or architecture.stderr.strip(),
        ),
        HostCheck(
            "NetworkManager active",
            network_manager.returncode == 0
            and network_manager.stdout.strip() == "active",
            network_manager.stdout.strip() or network_manager.stderr.strip(),
        ),
    ]
    if not service_only:
        route = runner(("ip", "route", "show", "default"))
        checks.insert(
            -1,
            HostCheck(
                "default network route",
                route.returncode == 0 and bool(route.stdout.strip()),
                route.stdout.strip() or route.stderr.strip(),
            ),
        )
    for unit in DISABLED_SERVICES:
        state = runner(("systemctl", "is-active", unit))
        checks.append(
            HostCheck(
                f"{unit} inactive",
                state.stdout.strip() != "active",
                state.stdout.strip() or state.stderr.strip() or "inactive",
            )
        )
    return checks


def main(argv: Sequence[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Verify the dedicated WalnutPi ZeroW/H618 host profile."
    )
    parser.add_argument("command", choices=("check",))
    parser.add_argument("--service", action="store_true")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args(argv)

    checks = check_profile(HostPaths.detect(), service_only=args.service)
    if args.json:
        print(json.dumps([asdict(check) for check in checks], ensure_ascii=False))
    else:
        for check in checks:
            print(f"{'PASS' if check.ok else 'FAIL'}: {check.name}: {check.detail}")
    return 0 if all(check.ok for check in checks) else 1


if __name__ == "__main__":
    sys.exit(main())