rt-4d/tools/tuning_read.py

#!/usr/bin/env python3
"""RT-4D: РИД-ОНЛИ снятие калибровки DMR-модуля протоколом штатного тюнера.

Реплика последовательности Radio Tuning Software (реверс, docs/tuning-software-protocol.md):
одна сессия COM @115200, шлём кадры и ловим ответы. ТОЛЬКО ЧТЕНИЕ — ни одной write/set-команды
калибровки (CmdWriteAll/Set* НЕ шлём). Рация в режиме «Обновление DMR чипа».

Под WSL COM через powershell.exe. Порядок:
  CmdConnect(34 52 05 10 9b) -> [опц. CmdOpenCal 68 b0..01] -> Read* (68 b?..) -> CmdColseCal -> CmdEnd
"""
import argparse, subprocess, sys, json

# --- команды (из реверса, research/tuning-re/command-templates.txt) ---
CMDS = {
    "Ping":        "68 84 01 01 00 00 00 01 00 10",
    "CmdConnect":  "34 52 05 10 9b",
    "CmdOpenCal":  "68 b0 01 01 00 00 00 01 01 10",
    "ReadDigVCCN": "68 b4 01 01 00 00 00 02 00 02 10",
    "ReadAnaVCCN": "68 b2 01 01 00 00 00 02 00 02 10",
    "ReadMod":     "68 b6 01 01 00 00 00 03 00 01 00 10",
    "ReadPWR":     "68 b8 01 01 00 00 00 04 00 01 01 00 10",
    "ReadSlot":    "68 b7 01 01 00 00 00 01 00 10",
    "CmdColseCal": "68 b0 01 01 00 00 00 01 00 10",
    "CmdEnd":      "34 52 05 ee 79",
}

def hx(name):
    if name.startswith("hex:"):
        return bytes.fromhex(name[4:].replace(" ", ""))
    return bytes.fromhex(CMDS[name].replace(" ", ""))

# powershell: open port, run a sequence of (write, wait) steps, capture each response
PS = r'''
$ErrorActionPreference = "Stop"
$p = New-Object System.IO.Ports.SerialPort "{port}",115200,"None",8,"one"
$p.ReadTimeout = 60
try {{ $p.Open() }} catch {{ Write-Output "ERR:$($_.Exception.Message)"; exit }}
$p.DiscardInBuffer(); $p.DiscardOutBuffer()
$steps = New-Object System.Collections.ArrayList
{steps}
foreach ($s in $steps) {{
  $name = $s[0]; $tx = [byte[]]$s[1]
  $p.DiscardInBuffer()
  $p.Write($tx,0,$tx.Length)
  Start-Sleep -Milliseconds {dwell}
  $buf = New-Object System.Collections.Generic.List[byte]
  $sw=[Diagnostics.Stopwatch]::StartNew(); $last=0
  while ($sw.ElapsedMilliseconds -lt {tmo}) {{
    try {{ $buf.Add($p.ReadByte()); $last=$sw.ElapsedMilliseconds }} catch {{}}
    if ($buf.Count -gt 0 -and ($sw.ElapsedMilliseconds-$last) -gt {idle}) {{ break }}
  }}
  $hex = ($buf | ForEach-Object {{ "{{0:x2}}" -f $_ }}) -join ""
  Write-Output ("RSP|"+$name+"|"+$hex)
}}
$p.Close()
Write-Output "DONE"
'''

def find_port():
    out = subprocess.run(["powershell.exe","-NoProfile","-Command",
        "[System.IO.Ports.SerialPort]::GetPortNames() -join ','"],
        capture_output=True, text=True).stdout.strip().replace("\r","")
    ports=[p for p in out.split(",") if p]
    if not ports: sys.exit("COM не найден.")
    return ports[0]

def ps_arr(b): return "[byte[]]("+",".join(str(x) for x in b)+")"

def run(port, seq, dwell=120, tmo=700, idle=200):
    steps = "\n".join(f"$null = $steps.Add(@('{n}',{ps_arr(hx(n))}))" for n in seq)
    script = PS.format(port=port, steps=steps, dwell=dwell, tmo=tmo, idle=idle)
    r = subprocess.run(["powershell.exe","-NoProfile","-Command",script],
                       capture_output=True, text=True)
    out=r.stdout.replace("\r","")
    res=[]
    for line in out.split("\n"):
        if line.startswith("ERR:"): sys.exit(line)
        if line.startswith("RSP|"):
            _,name,hexs=line.split("|",2)
            res.append((name, bytes.fromhex(hexs) if hexs else b""))
    return res

def parse68(rx):
    if len(rx)<9 or rx[0]!=0x68: return None
    cmd=rx[1]; ln=(rx[6]<<8)|rx[7]; pl=rx[8:8+ln]
    return dict(cmd=cmd, status=rx[3], length=ln, payload=pl)

def main():
    ap=argparse.ArgumentParser()
    ap.add_argument("--port")
    ap.add_argument("--connect-only", action="store_true", help="только CmdConnect (проверка отклика)")
    ap.add_argument("--direct", action="store_true", help="без 34.. Connect: 0x68 напрямую в модуль (режим Обновление DMR)")
    ap.add_argument("--seq", help="произвольная последовательность имён через запятую")
    a=ap.parse_args()
    port=a.port or find_port()
    print(f"[порт {port} @115200] РИД-ОНЛИ снятие калибровки\n")

    if a.seq:
        seq=[s.strip() for s in a.seq.split(",")]
    elif a.connect_only:
        seq=["CmdConnect"]
    elif a.direct:
        seq=["Ping","CmdOpenCal","ReadDigVCCN","ReadAnaVCCN","ReadMod","ReadPWR",
             "ReadSlot","CmdColseCal"]
    else:
        seq=["CmdConnect","CmdOpenCal","ReadDigVCCN","ReadAnaVCCN","ReadMod","ReadPWR",
             "ReadSlot","CmdColseCal","CmdEnd"]
    res=run(port, seq)
    if not res:
        print("Нет ответов от рации (powershell не вернул RSP). Порт/режим?")
        return
    for name,rx in res:
        tag=f"  {name:14s} TX={CMDS.get(name, name[4:] if name.startswith('hex:') else name)}"
        if not rx:
            print(tag+"  RX=(нет ответа)")
        else:
            p=parse68(rx)
            extra=f" [cmd=0x{p['cmd']:02X} st=0x{p['status']:02X} len={p['length']} pl={p['payload'].hex(' ')}]" if p else ""
            print(tag+f"  RX={rx.hex(' ')}"+extra)

if __name__=="__main__":
    main()