128 lines
4.7 KiB
Python
128 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Read a USB CDC tty on macOS using only Python's standard library."""
|
||
import argparse
|
||
import codecs
|
||
from datetime import datetime
|
||
import glob
|
||
import os
|
||
from pathlib import Path
|
||
import select
|
||
import sys
|
||
import termios
|
||
import time
|
||
import tty
|
||
|
||
|
||
def available_ports():
|
||
return sorted(set(glob.glob('/dev/cu.usbmodem*') + glob.glob('/dev/cu.usbserial*')))
|
||
|
||
|
||
def open_port(path):
|
||
fd = os.open(path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
|
||
try:
|
||
saved = termios.tcgetattr(fd)
|
||
tty.setraw(fd, termios.TCSANOW)
|
||
config = termios.tcgetattr(fd)
|
||
config[2] |= termios.CLOCAL | termios.CREAD
|
||
config[2] &= ~(termios.PARENB | termios.CSTOPB | termios.CSIZE)
|
||
config[2] |= termios.CS8
|
||
if hasattr(termios, 'CRTSCTS'):
|
||
config[2] &= ~termios.CRTSCTS
|
||
config[4] = config[5] = termios.B115200
|
||
termios.tcsetattr(fd, termios.TCSANOW, config)
|
||
return fd, saved
|
||
except BaseException:
|
||
os.close(fd)
|
||
raise
|
||
|
||
|
||
def read_session(path, output, diag_only=False):
|
||
fd, saved = open_port(path)
|
||
decoder = codecs.getincrementaldecoder('utf-8')(errors='replace')
|
||
pending = ''
|
||
|
||
def emit(line):
|
||
line = line.rstrip('\r')
|
||
if not line or (diag_only and '[现场诊断]' not in line and '[USB日志]' not in line):
|
||
return
|
||
stamped = f'{datetime.now().astimezone().isoformat(timespec="milliseconds")} {line}\n'
|
||
output.write(stamped)
|
||
output.flush()
|
||
sys.stdout.write(stamped)
|
||
sys.stdout.flush()
|
||
|
||
print(f'已连接 {path}，UTF-8 / 115200 8N1；Ctrl+C 结束。', file=sys.stderr)
|
||
try:
|
||
while True:
|
||
readable, _, _ = select.select([fd], [], [], 1)
|
||
if not readable:
|
||
continue
|
||
try:
|
||
block = os.read(fd, 4096)
|
||
except BlockingIOError:
|
||
continue
|
||
if not block:
|
||
raise OSError('USB端口已关闭')
|
||
pending += decoder.decode(block)
|
||
while '\n' in pending:
|
||
line, pending = pending.split('\n', 1)
|
||
emit(line)
|
||
if len(pending) > 8192:
|
||
emit(pending + ' [未换行，分段保存]')
|
||
pending = ''
|
||
finally:
|
||
pending += decoder.decode(b'', final=True)
|
||
if pending:
|
||
emit(pending)
|
||
try:
|
||
termios.tcsetattr(fd, termios.TCSANOW, saved)
|
||
except termios.error:
|
||
pass
|
||
os.close(fd)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('--list', action='store_true', help='列出 USB 串口，不打开设备')
|
||
parser.add_argument('--port', help='例如 /dev/cu.usbmodemXXXX；仅一个USB串口时可省略')
|
||
parser.add_argument('--output', type=Path, help='日志路径；默认 usb-logs/banban-时间.log')
|
||
parser.add_argument('--diag-only', action='store_true', help='只保存现场诊断与USB状态行')
|
||
parser.add_argument('--reconnect', action='store_true', help='拔插后等待同名端口重连；端口改名需重新指定')
|
||
args = parser.parse_args()
|
||
ports = available_ports()
|
||
if args.list:
|
||
print('\n'.join(ports) if ports else '未发现 USB 串口；检查数据线和设备枚举。')
|
||
return 0
|
||
path = args.port
|
||
if not path:
|
||
if len(ports) != 1:
|
||
print('请用 --port 指定设备 CDC 口；先运行 --list 并对比拔插前后的列表。', file=sys.stderr)
|
||
for port in ports:
|
||
print(port, file=sys.stderr)
|
||
return 2
|
||
path = ports[0]
|
||
destination = args.output or Path('usb-logs') / f'banban-{datetime.now():%Y%m%d-%H%M%S}.log'
|
||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
print(f'日志保存到 {destination.resolve()}；时间戳是电脑接收时间。', file=sys.stderr)
|
||
try:
|
||
with destination.open('a', encoding='utf-8') as output:
|
||
while True:
|
||
try:
|
||
read_session(path, output, args.diag_only)
|
||
except (OSError, termios.error) as error:
|
||
notice = f'[电脑读取] {datetime.now().astimezone().isoformat(timespec="milliseconds")} 连接中断：{error}'
|
||
print(notice, file=sys.stderr)
|
||
output.write(notice + '\n')
|
||
output.flush()
|
||
if not args.reconnect:
|
||
return 1
|
||
print('等待同名USB端口重连……', file=sys.stderr)
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
print('\n已停止，日志已保存。', file=sys.stderr)
|
||
return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
raise SystemExit(main())
|