文章目录1. 先写成 JSON2. 先说结论3. 事件字段与上一篇对齐4. 写入事件前先做过滤5. 核心模块event_writer.py6. 先用假数据把事件写入跑通7. 接到现有 YOLO / TensorRT 循环7.1 最小插入方式7.2 视频循环里还要注意7.3 和 detect_trt.py 的分工8. JSON 稳定后再接解释层9. 常见坑9.1 每帧都写 JSON9.2 类别名混乱9.3 先接大模型事件还不稳定9.4 截图路径写了但文件不存在9.5 把 EventWriter 写死进引擎代码深处9.6 多目标同一帧全部写成事件9.7 事件目录与检测工作目录不一致10. 验收清单11. 小结12. 相关阅读摘要上一篇把端侧应用封装成「检测 → 事件 → 小模型解释」。卡点通常不在解释脚本而在检测结果仍停在画框上下游拿不到稳定输入。本文写如何把现有 YOLO / TensorRT 检测输出自动写成events/*.json字段约定、阈值过滤、冷却去重、截图保存以及怎样接到已有推理循环。适合已经跑通 Orin 检测闭环准备接大模型解释层的人。可与 Orin 上目标检测 大模型做端侧应用、Orin 上目标检测最小闭环 一起看。建议目录可与上一篇共用mkdir-p~/orin-edge-app/{events,scripts,out,logs,snapshots}cd~/orin-edge-app# 本文新增event_writer.py / demo_from_detections.py / watch_and_explain.sh文件作用scripts/event_writer.py过滤、冷却、写 JSON、可选存截图scripts/demo_from_detections.py用一组假检测结果验证事件写入逻辑scripts/watch_and_explain.sh监听到新事件后调用解释脚本可选1. 先写成 JSON很多检测程序到这一步就停了屏幕上有框终端打印了检出 N 个目标下一帧继续画对值班系统和大模型解释层来说这还不够。它们需要的是可消费的记录什么时间、哪路相机、什么类别、置信度多少、要不要重复报。图1. 画框是给人看的events/*.json才是给下游系统看的。所以本篇目标很具体不重写整个检测工程只在“得到 boxes / scores / cls_ids”之后多写一步把结果保存成事件文件。2. 先说结论步骤做什么1复用上一篇的事件字段保证解释层不用改2只对白名单类别 高置信度目标写事件3同一相机、同一类别加冷却避免短时间重复告警4把结果写成 JSON同时可选保存截图路径5JSON 稳定后再接explain_event.py事件层是检测和大模型之间的稳定接口。先过滤和冷却再谈解释否则语言模型会被误报灌爆。先用假检测数据把事件写入跑通再挂到真实detect_trt.py循环。图2. 检测输出 → 过滤 → 冷却 → 写成 JSON。3. 事件字段与上一篇对齐继续使用这套字段与上一篇一致便于直接喂给解释脚本字段来源event_id相机 时间 序号ts告警时刻camera_id/line_id配置写入defect类别名由 cls_id 映射confidence检测分数bbox[x1,y1,x2,y2]取整即可repeat_count_1min冷却窗口内计数snapshot_path可选截图kb_hint可选先留空或写固定提示示例{event_id:cam01-20260804-093000-001,ts:2026-08-04T09:30:00,camera_id:cam01,line_id:line-A,defect:scratch,confidence:0.91,bbox:[100,120,240,260],repeat_count_1min:2,snapshot_path:snapshots/cam01_20260804_093000.jpg,kb_hint:}4. 写入事件前先做过滤图3. 白名单、阈值、冷却决定事件质量。规则建议默认说明类别白名单只保留缺陷类COCO 通用模型先自己映射关心的类置信度阈值如0.50起可比可视化阈值更高减少垃圾事件冷却时间如同相机同类1030s防止同一缺陷每帧都写文件没有这三个规则大模型解释层会变成“误报扩音器”。5. 核心模块event_writer.py保存为scripts/event_writer.py#!/usr/bin/env python3把检测结果写成 events/*.json。 输入约定 boxes: Nx4, xyxy scores: N cls_ids: N from__future__importannotationsimportjsonimporttimefromdataclassesimportdataclass,fieldfromdatetimeimportdatetimefrompathlibimportPathfromtypingimportDict,Iterable,List,Optional,Sequence,Tupleimportcv2importnumpyasnpdataclassclassEventWriterConfig:events_dir:PathPath(events)snapshots_dir:PathPath(snapshots)camera_id:strcam01line_id:strline-Aclass_names:Dict[int,str]field(default_factorydict)allow_classes:Optional[set]None# 空全开建议填缺陷类名集合conf_thres:float0.5cooldown_sec:float15.0save_snapshot:boolTruekb_hint:strclassEventWriter:def__init__(self,cfg:EventWriterConfig):self.cfgcfg self.cfg.events_dir.mkdir(parentsTrue,exist_okTrue)self.cfg.snapshots_dir.mkdir(parentsTrue,exist_okTrue)self._last_fire:Dict[Tuple[str,str],float]{}self._repeat_1min:Dict[Tuple[str,str],List[float]]{}self._seq0def_classname(self,cls_id:int)-str:returnself.cfg.class_names.get(int(cls_id),str(int(cls_id)))def_allowed(self,name:str)-bool:ifnotself.cfg.allow_classes:returnTruereturnnameinself.cfg.allow_classesdef_repeat_count(self,key:Tuple[str,str],now:float)-int:arr[tfortinself._repeat_1min.get(key,[])ifnow-t60.0]arr.append(now)self._repeat_1min[key]arrreturnlen(arr)defwrite_detections(self,boxes:Sequence[Sequence[float]],scores:Sequence[float],cls_ids:Sequence[int],frame_bgr:Optional[np.ndarray]None,)-List[Path]:返回本帧新写入的事件文件路径。written:List[Path][]nowtime.time()tsdatetime.now().strftime(%Y-%m-%dT%H:%M:%S)stampdatetime.now().strftime(%Y%m%d_%H%M%S)forbox,score,cls_idinzip(boxes,scores,cls_ids):iffloat(score)self.cfg.conf_thres:continuenameself._classname(int(cls_id))ifnotself._allowed(name):continuekey(self.cfg.camera_id,name)lastself._last_fire.get(key,0.0)ifnow-lastself.cfg.cooldown_sec:# 冷却期内仍计入 1 分钟重复次数但先不写新文件self._repeat_count(key,now)continueself._last_fire[key]now repeatself._repeat_count(key,now)self._seq1event_idf{self.cfg.camera_id}-{stamp}-{self._seq:03d}snapshot_pathifself.cfg.save_snapshotandframe_bgrisnotNone:snapself.cfg.snapshots_dir/f{self.cfg.camera_id}_{stamp}_{self._seq:03d}.jpgcv2.imwrite(str(snap),frame_bgr)snapshot_pathstr(snap)x1,y1,x2,y2[int(round(float(v)))forvinbox]event{event_id:event_id,ts:ts,camera_id:self.cfg.camera_id,line_id:self.cfg.line_id,defect:name,confidence:round(float(score),4),bbox:[x1,y1,x2,y2],repeat_count_1min:repeat,snapshot_path:snapshot_path,kb_hint:self.cfg.kb_hint,}outself.cfg.events_dir/f{event_id}.jsonout.write_text(json.dumps(event,ensure_asciiFalse,indent2)\n,encodingutf-8)written.append(out)returnwrittendefdemo_boxes():无引擎时的本地自测数据。boxesnp.array([[100,120,240,260],[300,80,400,180]],dtypenp.float32)scoresnp.array([0.91,0.32],dtypenp.float32)cls_idsnp.array([0,2],dtypenp.int32)returnboxes,scores,cls_ids这段代码保持独立不绑定某一版 TensorRT 封装只要提供boxes/scores/cls_ids就能用。6. 先用假数据把事件写入跑通保存scripts/demo_from_detections.py#!/usr/bin/env python3不依赖引擎先验证 EventWriter 是否正常写出事件。frompathlibimportPathimportsys sys.path.append(str(Path(__file__).resolve().parent))fromevent_writerimportEventWriter,EventWriterConfig,demo_boxes# noqa: E402defmain()-None:cfgEventWriterConfig(events_dirPath(events),snapshots_dirPath(snapshots),camera_idcam01,line_idline-A,class_names{0:scratch,1:dent,2:person},allow_classes{scratch,dent},# person 会被过滤conf_thres0.5,cooldown_sec5.0,save_snapshotFalse,kb_hint历史相似导轨毛刺可能导致周期性划痕,)writerEventWriter(cfg)boxes,scores,cls_idsdemo_boxes()foriinrange(3):pathswriter.write_detections(boxes,scores,cls_ids,frame_bgrNone)print(fround{i1}written{[str(p)forpinpaths]})if__name____main__:main()cd~/orin-edge-app python3 scripts/demo_from_detections.pylsevents/catevents/*.json|head预期第 1 轮写入scratchperson因白名单被滤低分目标因阈值被滤第 2、3 轮若在冷却期内不再重复写同相机同类文件确认这一点后再接到真实检测循环排障会轻松很多。7. 接到现有 YOLO / TensorRT 循环图4. 在得到 boxes 之后插入 EventWriter不必重写引擎加载。以 Orin 上目标检测最小闭环 里的流程为例你已经有preprocess → infer → postprocess → boxes/scores/cls_ids → draw只需在draw前后加上事件写入。关键不是改 TensorRT 初始化而是确认一件事引擎负责推理事件模块负责“哪些结果值得留下”。7.1 最小插入方式# 放在检测主程序同目录或把 scripts 加进 PYTHONPATHfrompathlibimportPathfromevent_writerimportEventWriter,EventWriterConfig writerEventWriter(EventWriterConfig(events_dirPath(events),snapshots_dirPath(snapshots),camera_idcam01,line_idline-A,# 按你的类别映射改若是自训缺陷模型这里写缺陷名class_names{0:scratch,1:dent},allow_classes{scratch,dent},conf_thres0.5,cooldown_sec15.0,save_snapshotTrue,))# ... 推理得到 boxes, scores, cls_ids, frame ...pathswriter.write_detections(boxes,scores,cls_ids,frame_bgrframe)forpinpaths:print(f[event]{p})7.2 视频循环里还要注意若你用的是视频或相机循环下面几条比“会不会写 JSON”更容易踩坑每帧都可能有框但不代表每帧都该写事件冷却就是为这个场景准备的。没有冷却解释队列和磁盘都会先被打满。截图保存必须限频1080p 连续存 JPEG几小时就能把小容量系统盘写满。事件可以留截图可先关或只在成功写出事件时存。类别映射一定要显式写清不要默认把cls_id当缺陷名。自训模型还好办若临时用 COCO 权重做联调更要在配置里写明“演示映射”避免后面误当成业务类别。告警阈值可以高于可视化阈值画面上0.25能看到框不代表就要写事件。事件阈值建议更严例如0.5起先保证下游干净。先统计“每分钟事件数”接大模型前先看真实运行 10 分钟会写出多少 JSON。若已经上百条先调白名单和冷却不要先开解释。7.3 和detect_trt.py的分工更稳的结构是两个文件文件职责detect_trt.py/ 相机循环读帧、推理、画框、展示event_writer.py过滤、冷却、写出事件不要把事件逻辑塞进 TensorRT 封装类深处。下次换导出方式、换后处理事件层还能原样复用。COCO 预训练权重做质检 demo 时常见做法是优先改用自训缺陷模型或临时把某个类映射成演示缺陷名但配置和日志里必须写清楚这是演示映射8. JSON 稳定后再接解释层图5. 事件接口稳了上一篇的explain_event.py才能真正进工作流。可选监听到新文件就解释先串行别一上来高并发。保存scripts/watch_and_explain.sh#!/usr/bin/env bashset-euopipefailEVENTS_DIR${1:-events}MODEL${MODEL:-qwen2.5:7b}PROMPT${PROMPT:-prompts/explain_alarm.txt}OUT_DIR${OUT_DIR:-out}mkdir-p$OUT_DIR# 需要 inotifywaitsudo apt-get install -y inotify-toolsinotifywait-m-eclose_write--format%w%f$EVENTS_DIR|whileread-rf;docase$fin*.json)base$(basename$f.json)echo[explain]$fpython3 scripts/explain_event.py\--event$f\--prompt$PROMPT\--model$MODEL\--out$OUT_DIR/${base}.txt||echo[explain failed]$f;;esacdonechmodx scripts/watch_and_explain.sh# 先确保 explain_event.py 与 prompts/explain_alarm.txt 已按上一篇就位# ./scripts/watch_and_explain.sh events没有inotify-tools时也可以简单轮询python3 -PY from pathlib import Path import time, subprocess seenset() while True: for p in sorted(Path(events).glob(*.json)): if p in seen: continue seen.add(p) outPath(out)/(p.stem.txt) print(explain, p) subprocess.run([ python3,scripts/explain_event.py, --event,str(p),--out,str(out) ], checkFalse) time.sleep(1) PY9. 常见坑9.1 每帧都写 JSON结果磁盘爆、解释队列堵、值班端重复告警过多。处理冷却 更高告警阈值。先看ls events | wc -l不要凭感觉。9.2 类别名混乱cls_id0直接当缺陷名解释层会不知所云。处理显式class_names映射并在日志里打印映射后的defect。9.3 先接大模型事件还不稳定模型会把垃圾输入组织成很像样的废话。处理先ls events cat人工看 20 条确认字段和类别都对再接解释。9.4 截图路径写了但文件不存在解释或复盘时对不上。处理检查cv2.imwrite是否成功磁盘满时先关截图只留 JSON。9.5 把 EventWriter 写死进引擎代码深处后期换 Ultralytics / TensorRT 版本会很痛。处理保持“推理后钩子”这种薄接入。9.6 多目标同一帧全部写成事件一帧里 8 个同类框可能对应同一物理缺陷的多次响应。处理同一帧内可先按类别取最高分或提高冷却不要无脑全写。9.7 事件目录与检测工作目录不一致检测在~/orin-detect跑事件写到别处解释脚本找不到。处理在配置里写绝对路径或统一到~/orin-edge-app/events。10. 验收清单检查项通过标准假数据脚本能写出至少 1 个 JSON字段齐全白名单不关心的类不会写成事件冷却短时间重复框不会狂写文件真检测接入有框且超阈值时出现新 JSON解释衔接任选一个 JSON 能跑通explain_event.py# 快速抽查最近事件ls-ltevents|headpython3 -PY import json,glob filessorted(glob.glob(events/*.json)) print(count,len(files)) if files: print(json.load(open(files[-1],encodingutf-8))) PY11. 小结把端侧 YOLO 检测结果接到大模型告警本质是补上检测与大模型之间的接口层统一字段阈值 白名单过滤冷却去重写成 JSON并可选取景再交给解释脚本先让检测结果变成可消费事件端侧大模型才接得上否则再强的模型也只是对着空气聊天。12. 相关阅读Orin 上目标检测 大模型做端侧应用Orin 上目标检测最小闭环Orin 上跑本地大模型适合什么场景本地大模型跑通了为什么还是不好用相关链接Ultralytics YOLOv8Ollama OpenAI compatibility如果这篇对你有帮助欢迎点赞、收藏也欢迎关注后续更新。