273 lines
11 KiB
Python
273 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
I2C-Inject — LT8619C 寄存器择机注入工具
|
|
================================================
|
|
配合 RP2040 注入器固件(mcu-rp2040/rp2040_inject.py, 刷入 BPI-Pico-RP2040)使用,
|
|
在 LT8619C demo 板的 I2C 总线上择机读写寄存器。
|
|
|
|
特性:
|
|
- 串口选择/连接 与 寄存器操作 并列第一排, 手动输入 bank/寄存器/值 即刻注入
|
|
- 择机机制: 等总线一轮活动结束 -> 总线静置 50ms -> 发射(切bank->写->读回校验),
|
|
失败自动重试 3 次, 完成后立即释放总线回监听
|
|
- 0x606E(位交换)/0x606D(整组顺序) 快捷按钮, 全部为 demo 板 LA 实测确认的语义
|
|
|
|
依赖: pyserial(py -m pip install pyserial) + tkinter 标准库
|
|
用法: python i2c_inject_gui.py (也可带默认串口参数: python i2c_inject_gui.py COM9)
|
|
打包: python build_exe.py (生成 I2C-Inject.exe)
|
|
"""
|
|
|
|
import serial
|
|
import serial.tools.list_ports
|
|
import os
|
|
import sys
|
|
import threading
|
|
import queue
|
|
import time
|
|
import tkinter as tk
|
|
from tkinter import ttk, messagebox
|
|
|
|
if getattr(sys, 'frozen', False):
|
|
BASE_DIR = os.path.dirname(os.path.abspath(sys.executable))
|
|
else:
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
def get_icon_path():
|
|
"""工具图标: 源码模式取 logos/ 下, 打包模式取 exe 内嵌资源(_MEIPASS)"""
|
|
if getattr(sys, 'frozen', False):
|
|
p = os.path.join(getattr(sys, '_MEIPASS', ''), 'logo_i2c_orange.ico')
|
|
else:
|
|
p = os.path.join(BASE_DIR, 'logos', 'logo_i2c_orange.ico')
|
|
return p if os.path.exists(p) else None
|
|
|
|
|
|
class Injector:
|
|
"""串口工作线程: 目标端口动态切换; 命令队列 -> RP2040, 回显 -> 消息队列"""
|
|
|
|
def __init__(self, on_line, on_status):
|
|
self.q = queue.Queue()
|
|
self.on_line, self.on_status = on_line, on_status
|
|
self._target = None # 想要连接的端口名(None=断开)
|
|
self.ser = None
|
|
self._lock = threading.Lock()
|
|
threading.Thread(target=self._run, daemon=True).start()
|
|
|
|
def connect(self, port):
|
|
with self._lock:
|
|
self._target = port
|
|
|
|
def disconnect(self):
|
|
with self._lock:
|
|
self._target = None
|
|
|
|
def send(self, cmd):
|
|
with self._lock:
|
|
if self.ser is None:
|
|
self.on_status('未连接串口, 命令被丢弃: ' + cmd)
|
|
return
|
|
self.q.put(cmd)
|
|
|
|
def is_connected(self):
|
|
with self._lock:
|
|
return self.ser is not None
|
|
|
|
def _run(self):
|
|
buf = b''
|
|
while True:
|
|
with self._lock:
|
|
target = self._target
|
|
if target != (self.ser.port if self.ser else None):
|
|
if self.ser is not None:
|
|
try:
|
|
self.ser.close()
|
|
except Exception:
|
|
pass
|
|
self.ser = None
|
|
self.on_status('已断开')
|
|
if target:
|
|
try:
|
|
self.ser = serial.Serial(target, timeout=0.2)
|
|
buf = b''
|
|
time.sleep(0.3) # 等 Pico 复位横幅打完
|
|
self.ser.reset_input_buffer()
|
|
self.on_status('已连接 ' + target)
|
|
except Exception as e:
|
|
self.ser = None
|
|
with self._lock:
|
|
self._target = None
|
|
self.on_status('连接 %s 失败: %s' % (target, e))
|
|
if self.ser is not None:
|
|
try:
|
|
cmd = None
|
|
try:
|
|
cmd = self.q.get_nowait()
|
|
except queue.Empty:
|
|
pass
|
|
if cmd is not None:
|
|
self.ser.write((cmd + '\n').encode())
|
|
buf += self.ser.read(256)
|
|
while b'\n' in buf:
|
|
ln, buf = buf.split(b'\n', 1)
|
|
txt = ln.decode('utf-8', 'replace').strip()
|
|
if txt:
|
|
self.on_line(txt)
|
|
except OSError:
|
|
self.ser = None
|
|
with self._lock:
|
|
self._target = None
|
|
self.on_status('串口丢失(设备拔出?), 请重新连接')
|
|
time.sleep(0.02)
|
|
|
|
|
|
def run_gui(default_port=None):
|
|
root = tk.Tk()
|
|
root.title('I2C-Inject — LT8619C 寄存器择机注入器 (静置50ms)')
|
|
root.geometry('860x520')
|
|
ic = get_icon_path()
|
|
if ic:
|
|
try:
|
|
root.iconbitmap(ic) # 窗口标题栏图标
|
|
except Exception:
|
|
pass
|
|
|
|
log_q = queue.Queue()
|
|
|
|
# ============ 第一排: 串口选择 + 寄存器操作(并列) ============
|
|
row1 = ttk.LabelFrame(root, text='连接与寄存器操作 (十六进制)', padding=(8, 4))
|
|
row1.pack(fill='x', padx=8, pady=(8, 4))
|
|
|
|
# --- 左半: 串口选择 ---
|
|
ttk.Label(row1, text='串口:').grid(row=0, column=0)
|
|
cbo_port = ttk.Combobox(row1, state='readonly', width=8, values=[])
|
|
cbo_port.grid(row=0, column=1, padx=(2, 6))
|
|
|
|
inj = Injector(on_line=lambda s: log_q.put((s, 'rx')),
|
|
on_status=lambda s: log_q.put((s, 'st')))
|
|
|
|
def refresh_ports():
|
|
ports = [p.device for p in serial.tools.list_ports.comports()]
|
|
cbo_port['values'] = ports
|
|
if ports:
|
|
cbo_port.current(0)
|
|
else:
|
|
cbo_port.set('')
|
|
inj.on_status('串口列表: %s' % (', '.join(ports) if ports else '无'))
|
|
|
|
def toggle_connect():
|
|
if inj.is_connected():
|
|
inj.disconnect()
|
|
btn_conn.config(text='连接')
|
|
else:
|
|
port = cbo_port.get().strip()
|
|
if not port:
|
|
messagebox.showerror('未选择串口', '请先选择串口(或点刷新扫描)')
|
|
return
|
|
inj.connect(port)
|
|
btn_conn.config(text='断开')
|
|
|
|
ttk.Button(row1, text='刷新', width=6, command=refresh_ports).grid(row=0, column=2, padx=(0, 4))
|
|
btn_conn = ttk.Button(row1, text='连接', width=6, command=toggle_connect)
|
|
btn_conn.grid(row=0, column=3, padx=(0, 14))
|
|
|
|
# --- 右半: 寄存器操作 ---
|
|
ttk.Label(row1, text='Bank:').grid(row=0, column=4)
|
|
e_bank = ttk.Entry(row1, width=5); e_bank.insert(0, '60'); e_bank.grid(row=0, column=5, padx=(2, 8))
|
|
ttk.Label(row1, text='寄存器:').grid(row=0, column=6)
|
|
e_reg = ttk.Entry(row1, width=5); e_reg.insert(0, '6E'); e_reg.grid(row=0, column=7, padx=(2, 8))
|
|
ttk.Label(row1, text='值:').grid(row=0, column=8)
|
|
e_val = ttk.Entry(row1, width=5); e_val.insert(0, '00'); e_val.grid(row=0, column=9, padx=(2, 8))
|
|
|
|
def _hex_ok(s):
|
|
try:
|
|
return 0 <= int(s, 16) <= 0xFF
|
|
except ValueError:
|
|
return False
|
|
|
|
def do_write():
|
|
b, r, v = e_bank.get().strip(), e_reg.get().strip(), e_val.get().strip()
|
|
if not (_hex_ok(b) and _hex_ok(r) and _hex_ok(v)):
|
|
messagebox.showerror('输入错误', 'bank/寄存器/值 都必须是 00-FF 的十六进制')
|
|
return
|
|
inj.send('w %s %s %s' % (b, r, v))
|
|
|
|
def do_read():
|
|
b, r = e_bank.get().strip(), e_reg.get().strip()
|
|
if not (_hex_ok(b) and _hex_ok(r)):
|
|
messagebox.showerror('输入错误', 'bank/寄存器 必须是 00-FF 的十六进制')
|
|
return
|
|
inj.send('r %s %s' % (b, r))
|
|
|
|
ttk.Button(row1, text='写入', width=8, command=do_write).grid(row=0, column=10, padx=2)
|
|
ttk.Button(row1, text='读取', width=8, command=do_read).grid(row=0, column=11, padx=2)
|
|
ttk.Label(row1, text='写=择机写入后自动读回校验; 失败自动重试3次',
|
|
foreground='#666666').grid(row=1, column=0, columnspan=12,
|
|
sticky='w', pady=(4, 0))
|
|
|
|
# ============ 0x606E 快捷块(位交换) ============
|
|
qk = ttk.LabelFrame(root, text='0x606E 快捷操作 (位交换)', padding=(8, 4))
|
|
qk.pack(fill='x', padx=8, pady=4)
|
|
for i, v in enumerate(('00', '20', '40', '80')):
|
|
ttk.Button(qk, text='写 ' + v, width=8,
|
|
command=lambda v=v: (e_val.delete(0, 'end'), e_val.insert(0, v), do_write())
|
|
).grid(row=0, column=i, padx=4)
|
|
ttk.Button(qk, text='读 0x606E', width=10,
|
|
command=do_read).grid(row=0, column=5, padx=4)
|
|
ttk.Label(qk, text='00=关交换 20=镜像D[7:0](LA实测) 40/80=另外两组(待自板单测)',
|
|
foreground='#666666').grid(row=1, column=0, columnspan=7,
|
|
sticky='w', pady=(3, 0))
|
|
|
|
# ============ 0x606D 快捷块(整组顺序) ============
|
|
qk2 = ttk.LabelFrame(root, text='0x606D 快捷操作 (RGB 整组顺序, 双色实测=手册naive读法)',
|
|
padding=(8, 4))
|
|
qk2.pack(fill='x', padx=8, pady=4)
|
|
for i, (code, name) in enumerate((('00', 'BGR'), ('03', 'BRG'), ('04', 'GBR'),
|
|
('05', 'GRB'), ('06', 'RBG'), ('07', 'RGB'))):
|
|
ttk.Button(qk2, text='%s %s' % (code, name), width=8,
|
|
command=lambda c=code: inj.send('w 60 6D ' + c)
|
|
).grid(row=0, column=i, padx=3)
|
|
ttk.Button(qk2, text='读 0x606D', width=10,
|
|
command=lambda: inj.send('r 60 6D')).grid(row=0, column=6, padx=3)
|
|
ttk.Label(qk2, text='字母串=D[23:16]/D[15:8]/D[7:0] 的分量顺序; 切换实时生效, 不影响HDMI锁定',
|
|
foreground='#666666').grid(row=1, column=0, columnspan=7,
|
|
sticky='w', pady=(3, 0))
|
|
|
|
# ============ 底部: 日志 ============
|
|
frm = ttk.LabelFrame(root, text='串口日志', padding=(6, 4))
|
|
frm.pack(fill='both', expand=True, padx=8, pady=(4, 8))
|
|
txt_log = tk.Text(frm, height=10, font=('Consolas', 9), state='disabled',
|
|
background='#101010', foreground='#C8C8C8')
|
|
txt_log.pack(side='left', fill='both', expand=True)
|
|
sb = ttk.Scrollbar(frm, orient='vertical', command=txt_log.yview)
|
|
txt_log.configure(yscrollcommand=sb.set)
|
|
sb.pack(side='right', fill='y')
|
|
for tag, color in (('rx', '#C8C8C8'), ('st', '#88AAFF'), ('ok', '#66CC66'),
|
|
('warn', '#E0A030')):
|
|
txt_log.tag_configure(tag, foreground=color)
|
|
|
|
def poll_log():
|
|
try:
|
|
while True:
|
|
ln, tag = log_q.get_nowait()
|
|
txt_log.configure(state='normal')
|
|
txt_log.insert('end', ln + '\n', tag)
|
|
if ln.startswith('='):
|
|
hot = 'warn' if ('FAIL' in ln or '丢弃' in ln) else 'ok'
|
|
txt_log.tag_add(hot, 'end-2l linestart', 'end-2l lineend')
|
|
txt_log.see('end')
|
|
txt_log.configure(state='disabled')
|
|
except queue.Empty:
|
|
pass
|
|
root.after(80, poll_log)
|
|
|
|
refresh_ports()
|
|
if default_port:
|
|
cbo_port.set(default_port)
|
|
root.after(80, poll_log)
|
|
root.protocol('WM_DELETE_WINDOW', root.destroy)
|
|
root.mainloop()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run_gui(sys.argv[1] if len(sys.argv) > 1 else None)
|