K1Guitar/tools/k1_harness/suites/keys_tm1617.py

# -*- coding: utf-8 -*-
"""keys1617 套件：TM1617 段落/导航键注入（RTT `tm1617 key N`）.

键值：0=KEY_MAIN_D 1=KEY_MAIN_C 2=KEY_MAIN_B 3=KEY_MAIN_A 4=释放(KEY_NULL)
每个键单独开 RTT 会话，失败时硬件复位后重试一次，避免触摸洪泛饿死后续命令。
"""
from __future__ import annotations

import os
import subprocess
import sys
import time

from ..paths import TOOLS_DIR
from ..transports.rtt_session import open_session

SUITE = "keys1617"
KEY_NAMES = {0: "MAIN_D", 1: "MAIN_C", 2: "MAIN_B", 3: "MAIN_A"}


def _hw_reset(device: str) -> None:
    script = os.path.join(TOOLS_DIR, "rtt_reset.py")
    subprocess.run(
        [sys.executable, script, "--device", device, "--wait-boot", "5"],
        timeout=60,
        check=False,
    )


def _inject_once(sess, key: int) -> tuple[bool, str]:
    tag = f"key={key}"
    before = len(sess.lines)
    ok = sess.cmd_ack(f"tm1617 key {key}", "TM1617_KEY_OK", timeout_s=2.5, retries=3)
    if ok:
        return True, "TM1617_KEY_OK"
    recent = sess.lines[before:]
    if any("tm1617 inject" in l and tag in l for l in recent):
        return True, "inject log"
    if any("TM1617_KEY_OK" in l and tag in l for l in recent):
        return True, "TM1617_KEY_OK (late)"
    return False, "无 ACK/inject 日志"


def _inject_with_retry(report, device: str, key: int, name: str) -> None:
    try:
        with open_session(device) as sess:
            ok, detail = _inject_once(sess, key)
            if ok:
                # 尽快释放，减轻触摸洪泛
                if key != 4:
                    sess.cmd_ack("tm1617 key 4", "TM1617_KEY_OK", timeout_s=1.5, retries=2)
                    time.sleep(0.3)
                report.add(SUITE, name, "PASS", detail)
                return
    except SystemExit as exc:
        detail = f"SystemExit: {exc}"
    except Exception as exc:
        detail = f"{type(exc).__name__}: {exc}"

    print(f"!! {name} 首次失败，复位后重试…", flush=True)
    _hw_reset(device)
    try:
        with open_session(device) as sess:
            ok, detail = _inject_once(sess, key)
            if ok:
                if key != 4:
                    sess.cmd_ack("tm1617 key 4", "TM1617_KEY_OK", timeout_s=1.5, retries=2)
                report.add(SUITE, name, "PASS", detail + " (retry)")
                return
            report.add(SUITE, name, "FAIL", detail)
    except SystemExit as exc:
        report.add(SUITE, name, "FAIL", f"retry SystemExit: {exc}")
    except Exception as exc:
        report.add(SUITE, name, "FAIL", f"retry {type(exc).__name__}: {exc}")


def run(report, ctx) -> None:
    # 探测命令是否存在
    try:
        with open_session(ctx.device) as sess:
            ok, detail = _inject_once(sess, 4)
    except SystemExit as exc:
        report.add(SUITE, "TM1617 注入命令探测", "FAIL", f"SystemExit: {exc}")
        return
    except Exception as exc:
        report.add(SUITE, "TM1617 注入命令探测", "FAIL", f"{type(exc).__name__}: {exc}")
        return

    if not ok:
        report.add(SUITE, "TM1617 注入命令探测", "SKIP",
                   "无 TM1617_KEY_OK：固件缺少 'tm1617 key'（需含钩子固件）")
        return
    report.add(SUITE, "TM1617 注入命令探测", "PASS", detail)

    for key, label in KEY_NAMES.items():
        _inject_with_retry(report, ctx.device, key, f"tm1617 {label}")
        _hw_reset(ctx.device)  # 每键后复位，保证下一键 RTT 干净

    _inject_with_retry(report, ctx.device, 4, "tm1617 释放(KEY_NULL)")