import numpy as np import matplotlib.pyplot as plt import multiprocessing.shared_memory as shared_memory import time import zlib from matplotlib.animation import FuncAnimation import logging # 本文件用于演示如何从厂家提供的共享内存中持续读取压力矩阵， # 并实时可视化为热力图。代码仅做读取与显示，不负责写入共享内存。 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class SharedMemoryContext: """共享内存上下文管理器。 作用： - 进入 with 语句时连接共享内存。 - 退出 with 语句时自动关闭连接，避免句柄泄漏。 """ def __init__(self, name): self.name = name self.shm = None def __enter__(self): # 按名字连接已存在的共享内存（由厂家驱动/进程创建） self.shm = shared_memory.SharedMemory(name=self.name) logging.info(f"Connected to shared memory: {self.name}") return self.shm def __exit__(self, exc_type, exc_val, exc_tb): # 只关闭连接，不 unlink（不能销毁由外部创建的共享内存） if self.shm is not None: self.shm.close() logging.info(f"Closed shared memory: {self.name}") def read_shared_memory_stream(name="x2_pressure"): """生成器函数：持续从共享内存读取数据流。 共享内存协议（按当前厂家示例）： - 前 72 字节：头部（元数据，当前示例未解析） - 后续数据区：288 * 64 个 float32（4 字节），总计 73728 字节 - 数据矩阵形状：reshape 为 (288, 64) Yields: tuple(counter, event_tick, datas) - counter: 帧序号（从 0 开始） - event_tick: 相对启动时间（毫秒） - datas: ndarray，shape=(288,64)，dtype=float32 """ counter = 0 # 帧计数器 start_tick = int(time.time() * 1000) # 记录开始时间戳 while True: try: # 每次循环使用 with 连接共享内存，确保异常时也能释放句柄 with SharedMemoryContext(name) as shm: # 建立连接后，持续读取同一块共享内存 while True: # 以 uint8 视图读取原始字节流（零拷贝） buffer = np.frombuffer(shm.buf, dtype=np.uint8) # 验证缓冲区长度是否满足“头部 + 数据区”最小要求 if buffer.nbytes < 72 + 288 * 64 * 4: logging.warning("Data size is insufficient. Waiting for more data...") time.sleep(0.1) continue # 跳过 72 字节头部，仅解析数据区并转换为 float32 矩阵 header = bytes(buffer[:72]) datas = buffer[72:72 + 288 * 64 * 4].view(np.float32).reshape((288, 64)) event_tick = int(time.time() * 1000) - start_tick # 计算相对时间戳 # 通过生成器把一帧数据交给上层消费者（可视化或业务处理） yield counter, event_tick, datas, header counter += 1 except FileNotFoundError: # 共享内存尚未创建：通常是写端程序未启动 logging.warning("Waiting for shared memory...") time.sleep(1) except KeyboardInterrupt: # 手动中断（Ctrl+C）时优雅退出 break def visualize_data_stream( data_stream, update_interval=50, low_percentile=5, high_percentile=98, gamma=1.0, ema_alpha=0.1, ): """实时可视化数据流。 Args: data_stream: 来自 read_shared_memory_stream 的生成器 update_interval: 动画刷新间隔（毫秒） low_percentile: 低端分位数（用于抑制噪声地板） high_percentile: 高端分位数（用于抑制少量极值，提升主体对比度） gamma: 非线性增强系数，>1 时高压区域颜色更“深”更突出 ema_alpha: 分位数平滑系数，越小越稳定、越大越灵敏 """ fig, ax = plt.subplots() # 初始化一个 64x64 的空图，后续每帧覆盖更新 init_data = np.zeros((64, 64)) im = ax.imshow(init_data, cmap="jet", aspect='equal') #,interpolation='bicubic' plt.colorbar(im) # 添加颜色条 # 使用“分位数 + 指数平滑”的动态范围，兼顾灵敏度与稳定性 vmin, vmax = None, None last_crc = None last_change_ts = time.time() last_diag_ts = 0.0 frame_counter = 0 fps_window_start = time.time() prev_u64 = {} prev_u32 = {} def probe_header_fields(header: bytes): """解析72字节头部的候选计数器字段并返回变化摘要。""" candidates_u64 = {} candidates_u32 = {} for off in range(0, min(len(header), 72) - 7, 8): candidates_u64[off] = int.from_bytes(header[off:off + 8], byteorder="little", signed=False) for off in range(0, min(len(header), 72) - 3, 4): candidates_u32[off] = int.from_bytes(header[off:off + 4], byteorder="little", signed=False) changed_u64 = [] changed_u32 = [] for off, val in candidates_u64.items(): prev = prev_u64.get(off) if prev is not None and val != prev: delta = val - prev changed_u64.append((off, val, delta)) prev_u64[off] = val for off, val in candidates_u32.items(): prev = prev_u32.get(off) if prev is not None and val != prev: delta = val - prev changed_u32.append((off, val, delta)) prev_u32[off] = val return changed_u64, changed_u32 def update(frame): nonlocal vmin, vmax, last_crc, last_change_ts, last_diag_ts, frame_counter, fps_window_start try: counter, event_tick, data, header = next(data_stream) # 从生成器获取最新数据 # 示例展示策略：取前 64 行并转置，得到 64x64 画面 # 说明：这只是可视化截取方式，不代表业务计算必须这样切片 data = data[:64, :].T.astype(np.float64) # 取前 64 行，再转置 → 64×64 # 1) 使用分位数而不是绝对 min/max，避免少量尖峰值“拉扁”整体颜色层次 p_low = np.percentile(data, low_percentile) p_high = np.percentile(data, high_percentile) if p_high <= p_low: p_low, p_high = float(np.min(data)), float(np.max(data)) if p_high <= p_low: p_high = p_low + 1e-9 # 2) 用 EMA 平滑动态范围，减少每帧抖动导致的闪烁 if vmin is None or vmax is None: vmin, vmax = p_low, p_high else: vmin = (1 - ema_alpha) * vmin + ema_alpha * p_low vmax = (1 - ema_alpha) * vmax + ema_alpha * p_high if vmax <= vmin: vmax = vmin + 1e-9 # 3) 归一化后做 gamma 增强：gamma>1 可让高压区域更快进入深色高亮区 norm = np.clip((data - vmin) / (vmax - vmin), 0.0, 1.0) enhanced = np.power(norm, gamma) # 颜色范围固定在 [0,1]，增强后高值会更“深”更突出 im.set_data(enhanced) im.set_clim(vmin=0.0, vmax=1.0) # 每秒打印一次“帧新鲜度诊断”，用于判断写端是否在持续更新 crc = int(zlib.crc32(np.ascontiguousarray(data).tobytes())) now = time.time() if last_crc is None or crc != last_crc: last_crc = crc last_change_ts = now frame_counter += 1 elapsed = now - fps_window_start fps = (frame_counter / elapsed) if elapsed > 0 else 0.0 if now - last_diag_ts >= 1.0: frame_age_ms = int(max(0.0, now - last_change_ts) * 1000.0) changed_u64, changed_u32 = probe_header_fields(header) u64_msg = "none" u32_msg = "none" if changed_u64: # 取前3个变化字段，格式 off:value(delta) u64_msg = ", ".join([f"{off}:{val}({delta:+d})" for off, val, delta in changed_u64[:3]]) if changed_u32: u32_msg = ", ".join([f"{off}:{val}({delta:+d})" for off, val, delta in changed_u32[:3]]) logging.info( "VIS诊断 frame_age_ms=%d crc=%s fps=%.1f tick_ms=%d | hdr_u64_changed=%s | hdr_u32_changed=%s", frame_age_ms, crc, fps, event_tick, u64_msg, u32_msg ) last_diag_ts = now frame_counter = 0 fps_window_start = now return [im] except StopIteration: plt.close() ani = FuncAnimation(fig, update, interval=update_interval, blit=True, cache_frame_data=False) plt.show() if __name__ == "__main__": # 读取内存数据 data_stream = read_shared_memory_stream() # 启动可视化（20ms 间隔约等于 50Hz 刷新） visualize_data_stream(data_stream, update_interval=20) # 20Hz更新