python串口的stream数据mock

参考zynq的stream数据mock和fifo缓冲和同步.csdnMockWebSerialAdStream.pypip install pyserial#!/usr/bin/env python3# -*- coding: utf-8 -*- 模拟 Web Serial AD 二进制数据流。 这个脚本用于配合前端的 WebSerial AD Stream 测试。它会持续向指定串口写入二进制数据 前端按同样的协议解析后就能看到 mock 波形。 单个 sample 的二进制布局如下所有字段都是 little-endian u32 u32 seq # 递增序号只用于协议占位和排查丢包不算 AD 通道 u32 ad0 # 第 1 路 AD默认正弦波 u32 ad1 # 第 2 路 AD默认三角波 u32 ad2 # 第 3 路 AD默认方波 u32 ad3 # 第 4 路 AD默认常量 默认 4 路 ADseq 不计入通道数。前端页面里“通道数”应设置为 4。 默认每 100 ms 发送 1 个 packet。packet 内包含多少组 sample由下面公式决定 samples_per_packet sample_rate_hz * interval_ms / 1000 例如默认 sample_rate_hz1000、interval_ms100则每 100ms 发 100 组 sample。 每组 sample 是 5 个 u32也就是 20 字节所以默认每包 2000 字节。 from__future__importannotationsimportargparseimportmathimportstructimportsysimporttimefromtypingimportBinaryIO# 默认 AD 通道数。注意 seq 不算通道所以 4 表示 ad0~ad3 四路。DEFAULT_CHANNEL_COUNT4# 默认 100ms 发一次 packet。这个是“吐包周期”不是采样周期。DEFAULT_INTERVAL_MS100# 默认每通道 1000 Sa/s。配合 100ms 周期每包会有 100 组 sample正弦波会更光滑。DEFAULT_SAMPLE_RATE_HZ1000# 默认串口波特率。前端 Web Serial 里也要选同样的波特率。DEFAULT_BAUD_RATE921600# 波形振幅。最终 AD 值大致在 center /- amplitude 范围内。DEFAULT_AMPLITUDE1200# AD 中心值。这里模拟无符号 AD所以波形围绕 2048 上下变化。DEFAULT_CENTER2048# 所有动态波形默认 1Hz方便肉眼观察完整周期。DEFAULT_WAVE_FREQUENCY_HZ1.0defclamp(value:float,min_value:float,max_value:float)-float:把数值限制在指定范围内避免生成负数或超过 u32 上限。returnmin(max(value,min_value),max_value)deftriangle_wave(t:float,frequency_hz:float)-float: 生成 -1~1 的三角波。 phase 是当前时间在一个周期里的位置范围 0~1。 公式 1 - 4 * abs(phase - 0.5) 会得到 phase0.0 - -1 phase0.5 - 1 phase1.0 - -1 phase(t*frequency_hz)%1.0return1.0-4.0*abs(phase-0.5)defsquare_wave(t:float,frequency_hz:float)-float: 生成 -1 或 1 的方波。 一个周期前半段输出 1后半段输出 -1。 phase(t*frequency_hz)%1.0return1.0ifphase0.5else-1.0defcreate_sample_values(seq:int,sample_rate_hz:int,channel_count:int,amplitude:int,center:int)-list[int]: 根据 seq 生成当前 sample 的各路 AD 值。 seq 每增加 1就表示时间向后推进 1 / sample_rate_hz 秒。 这样即使每次发送多个 sample波形时间轴也是连续的。 tseq/sample_rate_hz values:list[int][]forchannel_indexinrange(channel_count):ifchannel_index0:# ad0正弦波范围大致是 center /- amplitude。valueround(centeramplitude*math.sin(2*math.pi*DEFAULT_WAVE_FREQUENCY_HZ*t))elifchannel_index1:# ad1三角波范围同样是 center /- amplitude。valueround(centeramplitude*triangle_wave(t,DEFAULT_WAVE_FREQUENCY_HZ))elifchannel_index2:# ad2方波只在 center amplitude 和 center - amplitude 之间跳变。valueround(centeramplitude*square_wave(t,DEFAULT_WAVE_FREQUENCY_HZ))else:# ad3 以及更多手动扩展的通道常量方便确认通道解析和数码管显示是否稳定。valuecenter# 协议使用 u32因此这里把值限制到 0~0xFFFFFFFF。values.append(int(clamp(value,0,0xFFFFFFFF)))returnvaluesdefcreate_packet(start_seq:int,samples_per_packet:int,sample_rate_hz:int,channel_count:int,amplitude:int,center:int,)-bytes: 创建一整个 packet 的二进制内容。 packet 里可以包含多组 sample。每组 sample 都按 seq, ad0, ad1, ad2, ad3... 的顺序追加到 words 数组最后一次性 struct.pack 成小端 u32 字节流。 words:list[int][]forsample_offsetinrange(samples_per_packet):# seq 使用 u32 回绕避免长时间运行后超过协议字段范围。seq(start_seqsample_offset)0xFFFFFFFF# 每一组 sample 的第一个字段永远是 seq前端解析时会跳过这个字段。words.append(seq)# 紧跟 seq 后面的是 channel_count 路 AD 值。words.extend(create_sample_values(start_seqsample_offset,sample_rate_hz,channel_count,amplitude,center))# 表示 little-endianI 表示 unsigned int 32-bit。returnstruct.pack(f{len(words)}I,*words)defopen_serial(port:str,baud_rate:int)-BinaryIO:打开串口。这里延迟导入 pyserial方便 --stdout 模式在没安装 pyserial 时也能用。try:importserialexceptImportErrorasexc:raiseRuntimeError(pyserial is required: pip install pyserial)fromexcreturnserial.Serial(portport,baudratebaud_rate,timeout0,write_timeout1)defparse_args()-argparse.Namespace:解析命令行参数便于测试时临时改串口、波特率、采样率和每包大小。parserargparse.ArgumentParser(description模拟串口 AD 二进制数据流。)parser.add_argument(--port,defaultCOM10,help要写入的串口例如 COM10。)parser.add_argument(--baud,typeint,defaultDEFAULT_BAUD_RATE,help串口波特率前端要使用同样的值。)parser.add_argument(--interval-ms,typeint,defaultDEFAULT_INTERVAL_MS,help吐包周期单位 ms默认 100ms 发一次。)parser.add_argument(--sample-rate,typeint,defaultDEFAULT_SAMPLE_RATE_HZ,help每通道采样率单位 Hz。)parser.add_argument(--samples-per-packet,typeint,default0,help强制指定每包 sample 数0 表示按 sample-rate 和 interval-ms 自动计算。,)parser.add_argument(--channels,typeint,defaultDEFAULT_CHANNEL_COUNT,helpAD 通道数不包含 seq。)parser.add_argument(--amplitude,typeint,defaultDEFAULT_AMPLITUDE,help动态波形振幅。)parser.add_argument(--center,typeint,defaultDEFAULT_CENTER,helpAD 中心值常量通道也使用这个值。)parser.add_argument(--stdout,actionstore_true,help把二进制 packet 写到 stdout不打开真实串口。)returnparser.parse_args()defmain()-int:argsparse_args()# 基础参数做最小值保护避免命令行传 0 或负数导致除零、空包等问题。channel_countmax(1,int(args.channels))sample_rate_hzmax(1,int(args.sample_rate))interval_msmax(1,int(args.interval_ms))# 如果用户指定 --samples-per-packet就优先使用指定值。# 否则按“采样率 * 吐包周期”自动推导。## 默认1000Hz * 100ms / 1000 100 samples/packet。samples_per_packet(max(1,int(args.samples_per_packet))ifargs.samples_per_packet0elsemax(1,round(sample_rate_hz*interval_ms/1000)))# 这些统计信息只用于启动时打印方便确认当前真实发送速率。packet_interval_secondsinterval_ms/1000bytes_per_sample4*(channel_count1)bytes_per_packetbytes_per_sample*samples_per_packet data_ratebytes_per_packet/packet_interval_seconds writer:BinaryIOifargs.stdout:# stdout 模式主要用于本地验证二进制格式不需要 pyserial。writersys.stdout.buffertargetstdoutelse:# 正常测试 Web Serial 时写入一个虚拟串口或真实串口。writeropen_serial(args.port,max(1,int(args.baud)))targetf{args.port}{args.baud}print(fmock serial AD stream -{target}, channels{channel_count}, fsample_rate{sample_rate_hz}Hz, interval{interval_ms}ms, fsamples_per_packet{samples_per_packet}, packet{bytes_per_packet}B, frate{data_rate:.1f}B/s,filesys.stderr,)seq0# 使用 perf_counter 做发送节拍精度比 time.time 更适合这种周期任务。next_send_attime.perf_counter()try:whileTrue:# 创建当前 packet。packet 内部包含 samples_per_packet 组连续 sample。packetcreate_packet(seq,samples_per_packet,sample_rate_hz,channel_count,args.amplitude,args.center,)# 直接写原始二进制字节。不要转 hex / base64 / 文本否则前端协议解析会错。writer.write(packet)writer.flush()# 推进 seq。因为每包可能有多组 sample所以一次增加 samples_per_packet。seq(seqsamples_per_packet)0xFFFFFFFF# 按固定周期发送。用累计 next_send_at 可以减少长期漂移。next_send_atpacket_interval_seconds sleep_secondsnext_send_at-time.perf_counter()ifsleep_seconds0:time.sleep(sleep_seconds)else:# 如果某次写串口阻塞太久已经错过下一次发送就从当前时间重新对齐。next_send_attime.perf_counter()exceptKeyboardInterrupt:print(\nstopped,filesys.stderr)finally:ifnotargs.stdout:# CtrlC 退出时关闭串口避免端口被占用。writer.close()return0if__name____main__:raiseSystemExit(main())测试python .\MockWebSerialAdStream.py--portCOM28# 每100ms 输出 100个包python .\MockWebSerialAdStream.py--portCOM28 --samples-per-packet10000 00 00 00 00 08 00 005003 00 00 B0 0C 00 00 00 08 00 00 01 00 00 00 08 08 00 005503 00 00 B0 0C 00 00 00 08 00 00 02 00 00 00 0F 08 00 00 5A 03 00 00 B0 0C 00 00 00 08 00 00 03 00 00 001708 00 00 5E 03 00 00 B0 0C 00 00 00 08 00 00 04 00 00 00 1E 08 00 006303 00 00 B0 0C 00 00 00 08 00 00 05 00 00 002608 00 006803 00 00 B0 0C 00 00 00 08 00 00