diff --git a/README.md b/README.md index e76c497..511c654 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,11 @@ # FlowEngine -> Lightweight middleware for autonomous driving & robotics — C kernel + C++20 coroutine shell. +> Simulation-driven middleware framework for autonomous driving & robotics — C kernel + C++20 coroutine shell. +> +> **Scope:** FlowEngine is a *simulation-first, reproducible experiment platform*. It deliberately does **not** +> target real-vehicle deployment (no automotive mass production, no real ECU/CAN integration, no hard real-time +> or functional-safety certification). Everything — perception, fusion, planning, control, learning — is +> exercised, observed, tested, replayed and scored **entirely in simulation**. [![CI](https://github.com/caixuf/FlowEngine/actions/workflows/ci.yml/badge.svg)](https://github.com/caixuf/FlowEngine/actions) ![License](https://img.shields.io/badge/license-MIT-blue) @@ -11,7 +16,8 @@ ## What is FlowEngine? -A from-scratch middleware framework providing the core abstractions of CyberRT in a lightweight, embeddable package: +A from-scratch middleware framework providing the core abstractions of CyberRT in a lightweight, embeddable package. +It is built to be **organizable, observable, testable, replayable and scoreable — all inside simulation**: | Layer | Modules | |-------|---------| @@ -22,7 +28,7 @@ A from-scratch middleware framework providing the core abstractions of CyberRT i | **Data** | Type-safe Serialization (IDL + codegen), Bag v2 Record/Replay, Data Fusion, Schema Validation | | **QoS** | Per-topic QoS (depth + drop policy), Topic Stats (frequency, latency p50/p99, subscribers) | | **Operations** | Unified Logger (ms timestamps), flowctl CLI, FlowBoard Dashboard, flowmond Monitor Daemon, Stats Bridge (cross-process IPC stats), CI/CD | -| **Learning** | Vehicle-side learning loop: data recorder → offline trainer → on-vehicle tiny-MLP inference (shadow mode). See [docs/LEARNING_LOOP.md](docs/LEARNING_LOOP.md) | +| **Learning** | In-sim learning loop: data recorder → offline trainer → shadow-mode tiny-MLP inference (evaluated against the rule-based controller, never actuated). See [docs/LEARNING_LOOP.md](docs/LEARNING_LOOP.md) | ## Quick Start diff --git a/docs/EVOLUTION_ROADMAP.md b/docs/EVOLUTION_ROADMAP.md index c9e42f5..fd102b6 100644 --- a/docs/EVOLUTION_ROADMAP.md +++ b/docs/EVOLUTION_ROADMAP.md @@ -1,12 +1,17 @@ # FlowEngine 进化路线图 > 日期：2026-07-04（更新 2026-07-10） -> 当前定位：自动驾驶/机器人中间件内核原型 -> 目标方向：从”功能原型”进化为”可组织、可观察、可测试、可部署的框架” +> 当前定位：**仿真驱动的自动驾驶中间件框架 + 可复现实验平台** +> 目标方向：从"功能原型"进化为"可组织、可观察、可测试、可回放、可评估的仿真框架" +> +> **定位说明（2026-07-10 重制）：** FlowEngine 明确**不做实车方向**——不追车规量产、 +> 不接真实 ECU/CAN、不追硬实时/功能安全认证。全部能力（感知/融合/规划/控制/学习） +> 都在**仿真内闭环**验证。发展重心为两条主线：软件框架质量（主线 A）+ 仿真能力（主线 B）。 > > **实现状态更新 (2026-07-10)：** Phase 1-3 已完成，Phase 4-7 大部分完成。 > FlowRegistry、ParamRegistry、flow_launcher、flowctl 主要命令、QoS 系统均已实现。 -> 未完成项：跨机 TCP bridge、schema-aware bag 全功能、多进程 IPC 验证。 +> 未完成项：跨进程 topic bridge、schema-aware bag 全功能、多进程仿真部署验证。 +> 落地细化见 [IMPLEMENTATION_GUIDE.md](IMPLEMENTATION_GUIDE.md)。 ## 1. 总体方向 @@ -478,11 +483,12 @@ flowctl param set control_node.max_speed 20 - 参数可以查询。 - 支持部分参数热更新。 -## 12. Phase 10：真实多进程验证 +## 12. Phase 10：多进程仿真部署验证 ### 12.1 目标 验证 FlowEngine 是否真的具备框架能力，而不是只在单进程 demo 中工作。 +**注意：这里的"部署"指多进程仿真部署，不涉及任何实车 / ECU / CAN。** ### 12.2 推荐 demo diff --git a/docs/IMPLEMENTATION_GUIDE.md b/docs/IMPLEMENTATION_GUIDE.md new file mode 100644 index 0000000..0265ac6 --- /dev/null +++ b/docs/IMPLEMENTATION_GUIDE.md @@ -0,0 +1,209 @@ +# FlowEngine 落地实施指南（Implementation Guide） + +> 配套文档：[EVOLUTION_ROADMAP.md](EVOLUTION_ROADMAP.md)（方向）· [PROJECT_REVIEW.md](PROJECT_REVIEW.md)（现状评估） +> +> **本文用途：** 把「发展计划（软件 + 仿真双主线）」拆成一个个**接口已定义、契约已写清**的小任务， +> 让后续实现者（包括能力较弱的模型）**只需补齐实现**即可推进，无需再做架构决策。 +> +> **定位红线：** FlowEngine 不做实车。所有"部署 / 验证"均指**仿真内**（多进程仿真属于软件目标）。 +> 不要引入车规 / ECU / CAN / 硬实时 / 功能安全相关代码。 + +## 怎么用这份指南 + +每个任务卡片都遵循同一结构： + +- **目标** —— 做完后系统多了什么能力。 +- **现状** —— 已有哪些零件可复用。 +- **接口/契约** —— 已经写好的头文件或函数签名（你**只填实现**）。 +- **待实现** —— 具体要写的文件 / 函数 / stub。 +- **验收** —— 用哪条命令 / 测试证明做完了。 + +优先级顺序见文末「排期」。**先做 P0，再做 P1**；每做完一张卡片跑一次对应验收命令。 + +--- + +## 主线 A：软件框架质量 + +### A1. 工程健康度收尾 + +**目标：** CI 能自动跑 e2e 冒烟 + 长稳（ASAN/TSAN）轨道，单测覆盖 QoS/param/registry。 + +**现状：** `ctest` 全绿；`build-tsan/` 已存在；`flow_e2e` 支持带时长运行；`test_modules` 已覆盖 serializer/statem/scheduler/fusion。 + +**接口/契约：** +- `flow_e2e` 已可接收运行时长参数（见 `src/e2e_demo.c`）。需新增 `--smoke` 短时自检模式（几秒内退出、返回非零即失败）。 +- 单测框架沿用 `test/`、`tests/` 现有风格（见 `tests/` 里的 `new_module_tests`）。 + +**待实现：** +1. `src/e2e_demo.c`：解析 `--smoke`，跑 ~3s 最小链路后自检关键 topic 频率>0 并退出（返回非零表示失败）。 +2. `CMakeLists.txt`：新增 `add_test(NAME e2e_smoke_cli COMMAND flow_e2e --smoke)` 并设 `TIMEOUT 30`。 +3. `tests/`：给 `message_bus` QoS（depth + drop policy）、`param_registry` 范围校验、`flow_registry_export_json()` 各加 1~2 个断言用例。 +4. CI：新增可选 job 调用 `build-tsan` 跑 1 小时压力（可用现有 `benchmark` / `flow_bus --test` 循环）。 + +**验收：** `ctest --test-dir build --output-on-failure` 全绿，且包含 `e2e_smoke_cli`。 + +--- + +### A2. 统一元信息与内省收口 + +**目标：** flowctl / dashboard / 评估器**只从一个数据源**取元信息：`flow_registry_export_json()`。 + +**现状：** `include/flow_registry.h` 已定义 Task/Topic/Type/Plugin/Schema/Param 全部 Meta 与 +`flow_registry_export_json()`；`flowctl` 已有 `list/graph/schema/param/bag info`（见 `src/flowctl.c`）。 + +**接口/契约：** 无需新增接口 —— `flow_registry.h` 已是唯一契约。工作是**收敛调用点**。 + +**待实现：** +1. 审计 `src/flowctl.c`：确保每个 `list_*` 子命令都走 `flow_registry_list_*()` 而非各模块私有表。 +2. 让 monitor / dashboard JSON 也复用 `flow_registry_export_json()`（对比 `src/core/monitor_server.c` 当前拼 JSON 的方式）。 +3. 补 `flowctl list types` 与 `flowctl topic stats `（若缺）——数据源同上。 + +**验收：** `flowctl list tasks|topics|types|plugins|params` 全部有输出，且与 dashboard 显示一致。 + +--- + +### A3. 跨进程 Topic 桥接（多进程仿真部署） + +**目标：** 把单进程 demo 拆成多进程仍能互通，验证"框架"而非"单进程 demo"。 + +**现状：** `include/transport.h` 已有自动路由（local/IPC/TCP）；`include/ipc_channel.h` 有 SHM 通道。 + +**接口/契约：** ✅ 已写好 —— [`include/topic_bridge.h`](../include/topic_bridge.h)。 +定义了 `topic_bridge_create / add_topic / start / stop / destroy / get_stats`，方向 PUB/SUB/BIDIR。 + +**待实现：** +1. 新建 `src/core/topic_bridge.c` 实现 `topic_bridge.h` 全部函数： + - PUB 侧：`message_bus_subscribe` 本地 topic → `serializer_serialize` → `ipc_channel` 写。 + - SUB 侧：`ipc_channel` 读 → `serializer_deserialize` → `message_bus_publish` 本地。 + - 维护 `TopicBridgeStats`（forwarded/dropped/bytes/last_ts_us）。 +2. `CMakeLists.txt`：把 `topic_bridge.c` 加入 `flowengine_core` 源列表。 +3. 新增 `examples/` 或 `tests/` 双进程 demo：进程 A 发 `sensor/lidar`，进程 B 收到并计数。 + +**验收：** 双进程 demo 中 SUB 侧收到 PUB 侧的消息数 > 0；`topic_bridge_get_stats` 的 `forwarded` 递增。 + +--- + +### A4. 数据资产化（Schema + Bag） + +**目标：** 每次仿真可录制成可回放、可比对的数据资产。 + +**现状（大部分已完成）：** `include/bag.h` 已支持写入 `type_id|schema_ver|endian`， +`bag_reader_info / get_topics / get_type_info / play_filtered`；`flowctl bag info` 已实现（`src/flowctl.c`）。 + +**接口/契约：** 见 `include/bag.h`。剩余是**补齐字段级 schema 与 remap**。 + +**待实现：** +1. `bag.h` + `src/core/bag.c`：在 bag 头写入 field-level schema 摘要（复用 `msg_schema.h` 的 `struct_size` 与 + `serializer.h` 的 `schema_hash`），并让 `bag_reader_get_type_info` 回填 schema hash/version。 +2. Topic remap：给 `bag_reader_play_filtered` 增加"源 topic → 目标 topic"映射参数，或新增 + `bag_reader_play_remap(r, bus, speed, const char* from[], const char* to[], int n)`。 +3. `flowctl bag info` 输出补 `frequency`（用 count / duration 估算）。 + +**验收：** `flowctl bag info demo.bag` 输出每 topic 的 type / schema_version / count / frequency； +remap 回放后目标 topic 在 bus 上出现。 + +--- + +## 主线 B：仿真能力 + +### B1. 仿真世界保真度 + +**目标：** 更真实的车辆动力学 + 传感器模型 + 多 NPC 交互行为。 + +**现状：** `modules/adas_nodes/sim_world_node.c`（车辆/ NPC 更新）、`sensor_model_node.c`（FOV/遮挡/噪声）、 +`include/scenario_loader.h`（actor/ego JSON 契约，已支持 car/pedestrian/truck）。 + +**接口/契约：** `scenario_loader.h` 的 `ScenarioActor/ScenarioEgo/ScenarioCriteria` 已是稳定契约。 +**新增场景无需改代码**，只加 JSON（见 `scenarios/*.json`）。改保真度才需动 C。 + +**待实现（按需，从易到难）：** +1. **纯数据（最简单，弱模型友好）**：继续往 `scenarios/` 加场景 JSON，并登记进 `scenarios/suite.json`。 +2. 传感器模型：在 `sensor_model_node.c` 内细化噪声/遮挡（如按距离衰减、雨雾降 SNR）——保持既有 topic 契约不变。 +3. NPC 行为：在 `sim_world_node.c` 给 actor 增加简单跟车/让行策略（读 `ScenarioActor` 扩展字段）。 + 如需新字段，先在 `scenario_loader.h` 的 struct 里加，再在 `scenario_load()` 解析。 + +**验收：** 新场景能被 `flow_launcher` 加载并跑完；`demo_evaluator.py --scenario <新场景>` 有评分输出。 + +--- + +### B2. 仿真即测试（回归评估体系）✅ 框架已就绪 + +**目标：** 一条命令跑通「多场景批量仿真 → 自动评分 → 回归对比报告」。 + +**现状（本次已交付）：** +- `tools/demo_evaluator.py`：新增 `--scenario `（临时覆盖 `sim_world.scenario_file`）与 + `--json-out `（机器可读结果）。 +- `tools/scenario_regression.py`：读取 `scenarios/suite.json`，逐场景跑分、聚合矩阵、与 baseline 对比。 +- `scenarios/suite.json`：场景清单 + `baseline_tolerances`（数值回归阈值）。 + +**接口/契约（扩展点，弱模型只碰这两处）：** +- 加场景：编辑 `scenarios/suite.json` 的 `scenarios[]`（无需改代码）。 +- 调回归阈值：编辑 `scenarios/suite.json` 的 `baseline_tolerances`（`min_ratio` / `max_abs_increase`）。 +- 改评分口径：`demo_evaluator.py::score()`；改回归判定：`scenario_regression.py::compare_summary()`。 + +**待实现（收尾）：** +1. 首次录制基线：`python3 tools/scenario_regression.py --update-baseline`（产物落 `scenarios/baseline/`）。 +2. 把 `python3 tools/scenario_regression.py --baseline` 接入 CI（可选、耗时任务）。 + +**验收：** +- `python3 tools/scenario_regression.py --dry-run` 列出全部场景。 +- `python3 tools/scenario_regression.py` 产出 Regression Matrix 报告；全 PASS 时退出码 0。 + +--- + +### B3. 三层仿真体系打通 + +**目标：** Layer 1（Bag 回放）/ Layer 2（2D 模拟器）/ Layer 3（内置 3D 场景）共用同一数据契约， +同一场景可在三层间切换。 + +**现状：** 数据契约见 [FLOWBOARD_CONTRACT.md](FLOWBOARD_CONTRACT.md)（`/tmp/flow_topology.json` + `scene`）； +可视化见 `tools/flowboard.html`、`tools/foxglove_bridge.py`；仿真设计见 [E2E_SIMULATION_DESIGN.md](E2E_SIMULATION_DESIGN.md)。 + +**接口/契约：** `/tmp/flow_topology.json` 的 `scene` 字段是唯一契约。三层都读/写它。 + +**待实现：** +1. 确认 Bag 回放路径也能产出等价 `scene`（复用 `bag_reader_play` + monitor 写 JSON）。 +2. 文档化"同一 `scenarios/*.json` 如何分别喂给三层"，补到 [SIMULATION_GUIDE.md](SIMULATION_GUIDE.md)。 + +**验收：** 同一场景在 2D 与 3D 面板显示一致的 ego / actor 位置。 + +--- + +### B4. 车端学习闭环（在仿真中闭合） + +**目标：** 采集 → 离线训练 → 影子推理 → 评估，全在仿真内闭环，**影子模式绝不接管控制**。 + +**现状：** `modules/adas_nodes/data_recorder_node.c`（Stage 0 采样）、`tools/train/train.py`（离线训练）、 +`modules/adas_nodes/inference_node.c` + `tiny_mlp.h`（Stage 2 影子推理）；架构见 [LEARNING_LOOP.md](LEARNING_LOOP.md)。 + +**接口/契约：** `tiny_mlp.h` 的推理内核与 `train.py` 导出的权重格式是两端契约（保持一致）。 + +**待实现：** +1. 跑通链路脚本：录样本 → `train.py` 产出权重 → `inference_node` 加载 → 与规则控制器输出对比。 +2. 在 `demo_evaluator.py` 的 `summary` 里增加"影子推理 vs 规则控制"偏差指标（只读、不干预控制）。 +3. 用 `scenario_regression.py` 跨场景评估影子模型，纳入报告。 + +**验收：** 一条脚本从采集跑到"影子偏差指标"入回归报告；控制链路输出不受影子推理影响。 + +--- + +## 排期（去实车后的新顺序） + +| 优先级 | 卡片 | 一句话 | +|---|---|---| +| P0 | A1 | e2e `--smoke` 入 CI + 单测补覆盖 | +| P0 | B2（收尾） | 录基线 + 回归入 CI（框架已就绪） | +| P1 | A2 | 内省收口到 `flow_registry_export_json()` 单一数据源 | +| P1 | B1（纯数据） | 加场景 JSON + 登记 suite | +| P2 | A4 | schema-aware bag remap / frequency | +| P2 | B1（保真度） | 传感器/动力学/NPC 行为细化 | +| P3 | A3 | 实现 `topic_bridge.c` + 双进程仿真 demo | +| P3 | B3 | 三层仿真数据契约打通 | +| P3 | B4 | 学习闭环全链路 + 影子偏差入报告 | + +## 总体验收（计划级） + +- [ ] 一条命令跑通「多场景批量仿真 → 自动评分 → 回归对比报告」——`tools/scenario_regression.py`（框架已就绪，待录基线）。 +- [ ] 任一算法改动都能在纯仿真下验证，无任何实车依赖。 +- [ ] task/topic/type/param/schema/bag/state 均可经 `flowctl` 内省。 +- [ ] 文档与代码定位一致，对外统一表述为「仿真驱动的自动驾驶中间件框架」。 diff --git a/docs/PROJECT_REVIEW.md b/docs/PROJECT_REVIEW.md index af67af7..c0462eb 100644 --- a/docs/PROJECT_REVIEW.md +++ b/docs/PROJECT_REVIEW.md @@ -1,8 +1,11 @@ # FlowEngine 项目完善度评估 -> 评估日期：2026-07-04 +> 评估日期：2026-07-04（定位更新 2026-07-10） > 评估对象：FlowEngine 当前工作区版本 -> 评估定位：轻量级任务调度、消息总线、协程、IPC、Bag、状态机、发现与融合能力的中间件框架原型 +> 评估定位：**仿真驱动的自动驾驶中间件框架 + 可复现实验平台**（轻量级任务调度、消息总线、协程、IPC、Bag、状态机、发现与融合能力） +> +> **定位说明：** FlowEngine 明确不做实车方向（不追车规量产、不接真实 ECU/CAN、不追硬实时/功能安全认证）。 +> 下文中凡涉及"生产级智能驾驶框架"的历史措辞，均应理解为"仿真实验平台目标"。 ## 1. 总体结论 @@ -23,14 +26,14 @@ FlowEngine 现在已经不是单纯的“玩具 demo”，而是一个功能面 - 多传感器融合框架 - ADAS 假感知/假控制链路 -但它距离“生产级智能驾驶框架”还有明显差距，主要缺在： +但它距离"完善的仿真实验平台"还有明显差距，主要缺在： - 工程化 launch 系统 - 统一元信息/反射注册中心 - QoS 与通信可靠性策略 - 参数系统 - 监控诊断工具链 -- 多进程真实部署验证 +- 多进程仿真部署验证 - 长时间压力测试 - API/ABI 稳定性 @@ -53,11 +56,11 @@ FlowEngine 现在已经不是单纯的“玩具 demo”，而是一个功能面 当前 FlowEngine 最准确的定位是： -> 自动驾驶/机器人中间件的教学型内核 + 可运行原型。 +> 仿真驱动的自动驾驶中间件框架 + 可复现实验平台（兼具教学价值）。 它不是简单 demo，因为核心模块已经比较完整，并且可以编译、链接、运行、测试。 -但它也还不是生产级框架，因为生产级框架还需要完整工具链、可靠通信、部署系统、监控体系、参数管理、schema 兼容、真实业务验证和长期稳定性测试。 +但它也还不是完善的仿真实验平台，因为还需要更完整的工具链、可靠通信、多进程仿真部署、监控体系、参数管理、schema 兼容、场景回归验证和长期稳定性测试。 ## 4. 构建与测试状态 @@ -344,7 +347,11 @@ README 对以下新模块体现不足： 否则模块继续增加后，每个模块都会有自己的注册表，系统会变散。 -## 7. 与生产级智能驾驶框架的差距 +## 7. 与仿真实验平台目标的差距 + +> 本节原为"与生产级智能驾驶框架的差距"。因项目明确**不做实车方向**，重制为 +> "与仿真实验平台目标的差距"——衡量标准是能否在仿真里把「组织 / 观察 / 测试 / 回放 / 评估」闭环做扎实， +> 而非是否满足车规量产 / 硬实时 / 功能安全。 | 能力 | 当前状态 | |---|---| @@ -365,9 +372,9 @@ README 对以下新模块体现不足： | 安全/容错 | 初级 | | 工程发布 | 初级 | -因此，当前 FlowEngine 更适合称为： +因此，当前 FlowEngine 最准确的定位是： -> 自动驾驶中间件内核原型。 +> 仿真驱动的自动驾驶中间件框架 + 可复现实验平台。 ## 8. 下一步优先级 @@ -444,7 +451,7 @@ README 对以下新模块体现不足： | 阶段 3 | 协程 / IPC / Bag / Clock | 基本完成 | | 阶段 4 | IDL / 状态机 / Discovery / Fusion | 雏形完成 | | 阶段 5 | 工程化 launch + introspection | 待完成 | -| 阶段 6 | 生产级自动驾驶框架 | 还早 | +| 阶段 6 | 完善的仿真实验平台（多进程仿真 + 场景回归 + 学习闭环） | 还早 | ## 10. 最终评价 diff --git a/docs/README.md b/docs/README.md index 15836c8..776495f 100644 --- a/docs/README.md +++ b/docs/README.md @@ -19,7 +19,8 @@ ## 项目评估 - [项目完善度评估](PROJECT_REVIEW.md) - 当前完成度、模块成熟度、短板与下一步路线 -- [项目进化路线图](EVOLUTION_ROADMAP.md) - 从中间件原型进化到可部署框架的阶段规划 +- [项目进化路线图](EVOLUTION_ROADMAP.md) - 从中间件原型进化到仿真驱动框架的阶段规划 +- [落地实施指南](IMPLEMENTATION_GUIDE.md) - 把发展计划拆成"接口已定义、只需补实现"的小任务卡片 ## 可视化 (FlowBoard) diff --git a/docs/SIMULATION_GUIDE.md b/docs/SIMULATION_GUIDE.md index f9ade2b..fdfbf9d 100644 --- a/docs/SIMULATION_GUIDE.md +++ b/docs/SIMULATION_GUIDE.md @@ -110,3 +110,37 @@ open http://localhost:8800 [VISUALIZATION_ARCHITECTURE.md](VISUALIZATION_ARCHITECTURE.md#scene-数据结构真实-3d-仿真)。 完整的离线问题根因分析与鲁棒性设计详见 [E2E_SIMULATION_DESIGN.md](E2E_SIMULATION_DESIGN.md)。 + +## 场景矩阵回归（仿真即测试） + +把「场景库 × 评估指标」做成一条命令的批量回归，用于替代实车路测的"验证"职能。 + +场景清单与回归阈值集中在 `scenarios/suite.json`（加场景只需改这个 JSON，无需改代码）。 + +```bash +# 列出套件里将运行的全部场景（不启动 demo） +python3 tools/scenario_regression.py --dry-run + +# 跑整个套件，输出 PASS/FAIL 矩阵报告 +python3 tools/scenario_regression.py + +# 首次录制回归基线（落到 scenarios/baseline/） +python3 tools/scenario_regression.py --update-baseline + +# 跑套件并与基线做数值对比，退化即 FAIL（用于回归门禁） +python3 tools/scenario_regression.py --baseline + +# 只跑单个场景（按文件名去后缀匹配） +python3 tools/scenario_regression.py --only ghost_pedestrian +``` + +底层由 `tools/demo_evaluator.py` 逐场景执行；后者新增两个可组合参数： + +- `--scenario `：临时把 `config/pipeline.json` 的 `sim_world.scenario_file` + 指向该场景（运行后自动还原）。 +- `--json-out `：把 `{scenario, result, failures, warnings, summary}` + 写成机器可读 JSON，供回归矩阵聚合。 + +数值回归阈值支持两种门：`min_ratio`（当前值 ≥ 基线 × 比例）与 +`max_abs_increase`（当前值 ≤ 基线 + 增量）。判定逻辑见 +`tools/scenario_regression.py::compare_summary()`。 diff --git a/include/topic_bridge.h b/include/topic_bridge.h new file mode 100644 index 0000000..3e6beba --- /dev/null +++ b/include/topic_bridge.h @@ -0,0 +1,143 @@ +#ifndef TOPIC_BRIDGE_H +#define TOPIC_BRIDGE_H + +/** + * @file topic_bridge.h + * @brief 跨进程 Topic 桥接 — 显式把一组 topic 在两个独立进程间对拷 + * + * ── 定位 ────────────────────────────────────────────────── + * 这是一个 **接口契约（scaffold）**，用于 EVOLUTION_ROADMAP 的 + * Phase 10「多进程仿真部署验证」（plan 主线 A3）。 + * + * 与 `transport.h` 的区别： + * - `Transport` 是**自动路由**层：调用方不关心数据在哪，框架自选 local/IPC/TCP。 + * - `TopicBridge` 是**显式桥接**层：把本进程 bus 上指定的一组 topic， + * 单向/双向镜像到另一个进程的 bus。用于"多进程仿真部署"时， + * 把 sim_world / perception / planning 拆到不同进程后仍能互通。 + * + * 典型用法（进程 A 发布 sensor/lidar，进程 B 消费）： + * + * 进程 A: + * TopicBridge* b = topic_bridge_create(busA, "flowengine.sim", TOPIC_BRIDGE_PUB); + * topic_bridge_add_topic(b, "sensor/lidar", LIDARFRAME_TYPE_ID); + * topic_bridge_start(b); + * + * 进程 B: + * TopicBridge* b = topic_bridge_create(busB, "flowengine.sim", TOPIC_BRIDGE_SUB); + * topic_bridge_add_topic(b, "sensor/lidar", LIDARFRAME_TYPE_ID); + * topic_bridge_start(b); // sensor/lidar 现在会出现在 busB 上 + * + * ── 实现指引（给后续实现者） ──────────────────────────────── + * 建议实现文件：src/core/topic_bridge.c，并加入 CMakeLists 的 flowengine_core。 + * 复用已有能力： + * - `ipc_channel.h` —— 共享内存传输（同机跨进程）。 + * - `serializer.h` —— 出站 message_bus_publish 前序列化 / 入站反序列化。 + * - `message_bus.h` —— 本进程收发（PUB 侧订阅本地 topic；SUB 侧向本地发布）。 + * 所有函数当前**未实现**，返回 ERR_OK/句柄即视为契约占位；实现前请勿在生产链路依赖。 + * 详见 docs/IMPLEMENTATION_GUIDE.md（A3 小节）与 skills/ 下 IPC 教程。 + */ + +#include "message_bus.h" +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +#define TOPIC_BRIDGE_CHANNEL_LEN 64 +#define TOPIC_BRIDGE_MAX_TOPICS 64 + +/* ── 桥接方向 ─────────────────────────────────────────────── */ + +typedef enum { + TOPIC_BRIDGE_PUB = 0, /**< 本地 bus → IPC（导出本进程 topic 给别的进程） */ + TOPIC_BRIDGE_SUB = 1, /**< IPC → 本地 bus（把别的进程 topic 注入本进程） */ + TOPIC_BRIDGE_BIDIR = 2 /**< 双向镜像（谨慎使用，注意回环去重） */ +} TopicBridgeDirection; + +/* ── 运行时统计（可用于 flowctl / dashboard 边标注） ───────── */ + +typedef struct { + uint64_t forwarded; /**< 成功跨进程转发的消息数 */ + uint64_t dropped; /**< 因队列满 / 序列化失败丢弃数 */ + uint64_t bytes; /**< 累计转发字节数 */ + uint64_t last_ts_us; /**< 最近一次转发的时间戳（微秒） */ +} TopicBridgeStats; + +/* ── 不透明句柄 ──────────────────────────────────────────── */ + +typedef struct TopicBridge TopicBridge; + +/* ══════════════════════════════════════════════════════════ */ +/* 生命周期 */ +/* ══════════════════════════════════════════════════════════ */ + +/** + * 创建一个 topic 桥接器（尚未启动）。 + * + * @param bus 本进程消息总线（不可为 NULL）。 + * @param channel_name IPC 通道名，两个进程必须一致（如 "flowengine.sim"）。 + * @param direction 桥接方向，见 ::TopicBridgeDirection。 + * @return 句柄；失败返回 NULL。调用者负责 topic_bridge_destroy()。 + * 当 bus 或 channel_name 为 NULL、或 direction 非法（不在 ::TopicBridgeDirection 内）时返回 NULL。 + * + * @note 契约占位：实现前返回 NULL 亦属合法。 + */ +TopicBridge* topic_bridge_create(MessageBus* bus, const char* channel_name, + TopicBridgeDirection direction); + +/** + * 注册一条需要桥接的 topic。可在 start 前多次调用。 + * + * @param bridge 桥接句柄。 + * @param topic topic 名（如 "sensor/lidar"）。 + * @param type_id FNV-1a 类型 ID（用于两端 schema 一致性校验）。 + * @return ERR_OK 成功；ERR_OVERFLOW 超过 TOPIC_BRIDGE_MAX_TOPICS； + * ERR_INVALID_PARAM 参数非法。 + */ +int topic_bridge_add_topic(TopicBridge* bridge, const char* topic, uint32_t type_id); + +/** + * 启动桥接：PUB 侧开始订阅本地 topic 并向 IPC 转发； + * SUB 侧开始从 IPC 收包并向本地 bus 发布。 + * + * @return ERR_OK 成功；ERR_IO IPC 通道建立失败；ERR_BUSY 已启动。 + */ +int topic_bridge_start(TopicBridge* bridge); + +/** + * 停止桥接（可再次 start）。幂等。 + */ +int topic_bridge_stop(TopicBridge* bridge); + +/** + * 销毁桥接器并释放所有资源（内部会先 stop）。 + */ +void topic_bridge_destroy(TopicBridge* bridge); + +/* ══════════════════════════════════════════════════════════ */ +/* 内省 */ +/* ══════════════════════════════════════════════════════════ */ + +/** + * 读取某条已桥接 topic 的运行时统计。 + * + * @param bridge 桥接句柄。 + * @param topic topic 名。 + * @param out 输出统计（不可为 NULL）。 + * @return ERR_OK 成功；ERR_NOT_FOUND topic 未注册。 + */ +int topic_bridge_get_stats(const TopicBridge* bridge, const char* topic, + TopicBridgeStats* out); + +/** + * 已注册桥接 topic 的数量。 + */ +int topic_bridge_topic_count(const TopicBridge* bridge); + +#ifdef __cplusplus +} +#endif + +#endif /* TOPIC_BRIDGE_H */ diff --git a/scenarios/congestion_follow.json b/scenarios/congestion_follow.json new file mode 100644 index 0000000..6080331 --- /dev/null +++ b/scenarios/congestion_follow.json @@ -0,0 +1,45 @@ +{ + "name": "congestion_follow", + "description": "拥堵跟车：前方车队走走停停，考验纵向跟车平顺性与防追尾", + "random_seed": 91, + "duration_s": 70.0, + "ego": { + "x": 0.0, + "y": -1.75, + "heading": 0.0, + "init_speed": 3.0, + "target_speed": 8.0 + }, + "actors": [ + { + "id": 0, + "type": "car", + "x": 12.0, "y": -1.75, + "vx": 2.5, "vy": 0.0, + "len": 4.6, "wid": 2.0, + "_comment": "紧前车，低速蠕行" + }, + { + "id": 1, + "type": "car", + "x": 24.0, "y": -1.75, + "vx": 2.0, "vy": 0.0, + "len": 4.6, "wid": 2.0, + "_comment": "第二辆，走停" + }, + { + "id": 2, + "type": "car", + "x": 38.0, "y": -1.75, + "vx": 3.0, "vy": 0.0, + "len": 8.0, "wid": 2.5, + "_comment": "前方大货车，限制视距" + } + ], + "pass_criteria": { + "no_collision": true, + "max_duration_s": 70.0, + "min_avg_speed_mps": 1.5, + "min_distance_m": 40.0 + } +} diff --git a/scenarios/ghost_pedestrian.json b/scenarios/ghost_pedestrian.json new file mode 100644 index 0000000..61f2a40 --- /dev/null +++ b/scenarios/ghost_pedestrian.json @@ -0,0 +1,37 @@ +{ + "name": "ghost_pedestrian", + "description": "鬼探头：行人从路侧遮挡（停放货车）后突然横穿，考验极限制动/避让", + "random_seed": 53, + "duration_s": 45.0, + "ego": { + "x": 0.0, + "y": -1.75, + "heading": 0.0, + "init_speed": 9.0, + "target_speed": 12.0 + }, + "actors": [ + { + "id": 0, + "type": "truck", + "x": 42.0, "y": -4.6, + "vx": 0.0, "vy": 0.0, + "len": 8.0, "wid": 2.5, + "_comment": "路侧停放货车，遮挡视线" + }, + { + "id": 1, + "type": "pedestrian", + "x": 50.0, "y": -4.0, + "vx": 0.0, "vy": 1.6, + "len": 0.6, "wid": 0.6, + "_comment": "从货车后方鬼探头横穿的行人" + } + ], + "pass_criteria": { + "no_collision": true, + "max_duration_s": 45.0, + "min_avg_speed_mps": 2.0, + "min_distance_m": 45.0 + } +} diff --git a/scenarios/intersection_unprotected.json b/scenarios/intersection_unprotected.json new file mode 100644 index 0000000..f292bfe --- /dev/null +++ b/scenarios/intersection_unprotected.json @@ -0,0 +1,45 @@ +{ + "name": "intersection_unprotected", + "description": "无保护路口左转：横向来车流不断，ego 需在安全间隙内通过", + "random_seed": 71, + "duration_s": 60.0, + "ego": { + "x": 0.0, + "y": -1.75, + "heading": 0.0, + "init_speed": 6.0, + "target_speed": 10.0 + }, + "actors": [ + { + "id": 0, + "type": "car", + "x": 55.0, "y": -1.75, + "vx": 4.0, "vy": 0.0, + "len": 4.6, "wid": 2.0, + "_comment": "路口前的慢速前车，制造跟停/起步" + }, + { + "id": 1, + "type": "car", + "x": 70.0, "y": 6.0, + "vx": 0.0, "vy": -6.0, + "len": 4.6, "wid": 2.0, + "_comment": "横向来车 A（自上而下穿越路口）" + }, + { + "id": 2, + "type": "car", + "x": 78.0, "y": -8.0, + "vx": 0.0, "vy": 6.5, + "len": 4.6, "wid": 2.0, + "_comment": "横向来车 B（自下而上穿越路口）" + } + ], + "pass_criteria": { + "no_collision": true, + "max_duration_s": 60.0, + "min_avg_speed_mps": 3.0, + "min_distance_m": 60.0 + } +} diff --git a/scenarios/suite.json b/scenarios/suite.json new file mode 100644 index 0000000..e55e1b2 --- /dev/null +++ b/scenarios/suite.json @@ -0,0 +1,19 @@ +{ + "name": "flowengine-regression", + "description": "仿真即测试：场景矩阵回归套件。scenario_regression.py 逐场景跑分并与 baseline 对比。", + "default_duration_s": 30, + "scenarios": [ + { "file": "scenarios/pedestrian_crossing.json", "duration_s": 30, "enabled": true }, + { "file": "scenarios/highway_overtake.json", "duration_s": 45, "enabled": true }, + { "file": "scenarios/intersection_unprotected.json", "duration_s": 35, "enabled": true }, + { "file": "scenarios/congestion_follow.json", "duration_s": 40, "enabled": true }, + { "file": "scenarios/ghost_pedestrian.json", "duration_s": 25, "enabled": true } + ], + "baseline_tolerances": { + "_doc": "对 baseline summary 数值的允许漂移。缺省字段不做数值回归（仅 PASS/FAIL 回归）。", + "avg_speed_mps": { "min_ratio": 0.80 }, + "x_delta_m": { "min_ratio": 0.80 }, + "max_lane_error_m": { "max_abs_increase": 0.50 }, + "stagnation_duration_s": { "max_abs_increase": 2.0 } + } +} diff --git a/tools/demo_evaluator.py b/tools/demo_evaluator.py index f8b8928..ebb39d8 100644 --- a/tools/demo_evaluator.py +++ b/tools/demo_evaluator.py @@ -9,6 +9,7 @@ from __future__ import annotations import argparse +import contextlib import json import math import os @@ -23,6 +24,7 @@ ROOT = Path(__file__).resolve().parents[1] DEFAULT_JSON = Path("/tmp/flow_topology.json") LAUNCHER_STDERR = Path("/tmp/flow_launcher_stderr.txt") +PIPELINE_JSON = ROOT / "config" / "pipeline.json" REQUIRED_EDGES = [ ("sim_world", "vehicle/state", "perception"), @@ -48,6 +50,20 @@ } +def _pipeline_nodes(pipeline: dict) -> list: + """Return the node list of a launcher config. + + The launcher schema uses ``processes``; older/simpler configs use ``nodes``. + Accept either so scenario tooling works across both. + """ + if not isinstance(pipeline, dict): + return [] + nodes = pipeline.get("processes") + if not isinstance(nodes, list): + nodes = pipeline.get("nodes") + return nodes if isinstance(nodes, list) else [] + + def load_scenario_criteria_from_pipeline() -> tuple[dict, str | None]: """Load pass_criteria from config/pipeline.json -> sim_world.params.scenario_file. @@ -55,7 +71,7 @@ def load_scenario_criteria_from_pipeline() -> tuple[dict, str | None]: (criteria_dict, scenario_name) """ pipeline = load_json(ROOT / "config" / "pipeline.json") or {} - nodes = pipeline.get("nodes", []) if isinstance(pipeline, dict) else [] + nodes = _pipeline_nodes(pipeline) scenario_file = None for node in nodes: if not isinstance(node, dict): @@ -63,6 +79,13 @@ def load_scenario_criteria_from_pipeline() -> tuple[dict, str | None]: if node.get("name") != "sim_world": continue params = node.get("params", {}) + if isinstance(params, str): + try: + params = json.loads(params) + except json.JSONDecodeError: + print("warning: sim_world.params is not valid JSON; skipping scenario_file lookup", + file=sys.stderr) + params = {} if isinstance(params, dict): scenario_file = params.get("scenario_file") break @@ -93,6 +116,48 @@ def load_json(path: Path) -> dict | None: return None +@contextlib.contextmanager +def pipeline_scenario_override(scenario_file: str | None): + """Temporarily point config/pipeline.json's sim_world node at ``scenario_file``. + + ``sim_world``'s ``params`` is a JSON-encoded string; we patch the embedded + ``scenario_file`` key, yield, then restore the original file byte-for-byte. + Passing ``None`` is a no-op so callers can use this unconditionally. + """ + if not scenario_file: + yield None + return + + original_text = PIPELINE_JSON.read_text(encoding="utf-8") + pipeline = json.loads(original_text) + patched = False + for node in _pipeline_nodes(pipeline): + if not isinstance(node, dict) or node.get("name") != "sim_world": + continue + params = node.get("params") + # params may be a JSON string (launcher format) or a plain dict. + if isinstance(params, str): + params_obj = json.loads(params) + params_obj["scenario_file"] = scenario_file + node["params"] = json.dumps(params_obj) + elif isinstance(params, dict): + params["scenario_file"] = scenario_file + else: + continue + patched = True + break + + if not patched: + raise RuntimeError("sim_world node with params not found in config/pipeline.json") + + try: + PIPELINE_JSON.write_text(json.dumps(pipeline, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8") + yield scenario_file + finally: + PIPELINE_JSON.write_text(original_text, encoding="utf-8") + + def topic_map(sample: dict) -> dict: topics = sample.get("metrics", {}).get("topics", []) return {t.get("topic"): t for t in topics if t.get("topic")} @@ -438,6 +503,10 @@ def main() -> int: parser.add_argument("--interval", type=float, default=0.25, help="JSON sample interval in seconds") parser.add_argument("--json-file", type=Path, default=DEFAULT_JSON) parser.add_argument("--no-run", action="store_true", help="evaluate current JSON/logs without starting demo.sh") + parser.add_argument("--scenario", type=str, default=None, + help="scenario JSON path; temporarily overrides sim_world.scenario_file for this run") + parser.add_argument("--json-out", type=Path, default=None, + help="write the machine-readable evaluation result to this JSON path") args = parser.parse_args() if args.no_run: @@ -445,7 +514,8 @@ def main() -> int: samples = [sample] if sample else [] returncode = 0 else: - samples, returncode = collect_samples(args.duration, args.json_file, args.interval) + with pipeline_scenario_override(args.scenario): + samples, returncode = collect_samples(args.duration, args.json_file, args.interval) if returncode != 0: print(f"demo.sh exited with code {returncode}") @@ -468,6 +538,20 @@ def main() -> int: for warning in warnings: print(f" - {warning}") + result = "FAIL" if failures else "PASS" + if args.json_out: + payload = { + "scenario": summary.get("scenario"), + "result": result, + "failures": failures, + "warnings": warnings, + "summary": summary, + } + args.json_out.parent.mkdir(parents=True, exist_ok=True) + args.json_out.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8") + print(f"\nwrote evaluation result to {args.json_out}") + if failures: print("\nFAIL:") for failure in failures: diff --git a/tools/scenario_regression.py b/tools/scenario_regression.py new file mode 100755 index 0000000..83d9c61 --- /dev/null +++ b/tools/scenario_regression.py @@ -0,0 +1,243 @@ +#!/usr/bin/env python3 +"""FlowEngine scenario-matrix regression harness. + +This is the "simulation-as-testing" driver (plan item B2). It runs the whole +scenario suite through ``demo_evaluator.py`` — one full simulation per scenario — +collects each PASS/FAIL result plus its numeric summary, and (optionally) diffs +those numbers against a saved *baseline* so that a code change that quietly +degrades behaviour shows up as a regression. + + # Run the full suite and print a matrix report + python3 tools/scenario_regression.py + + # Record the current results as the regression baseline + python3 tools/scenario_regression.py --update-baseline + + # Run and compare against the saved baseline (fail on regression) + python3 tools/scenario_regression.py --baseline + +Exit code is 0 only when every scenario PASSes and (when --baseline is given) +no numeric regression exceeds the suite tolerances. + +------------------------------------------------------------------------------ +Extending this harness (for follow-up implementers): + * Add scenarios by editing ``scenarios/suite.json`` — no code change needed. + * Tighten/loosen numeric regression gates via ``baseline_tolerances`` there. + * The only two functions worth touching are ``run_scenario`` (how one scenario + is executed/scored) and ``compare_summary`` (how a regression is decided). +------------------------------------------------------------------------------ +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +EVALUATOR = ROOT / "tools" / "demo_evaluator.py" +DEFAULT_SUITE = ROOT / "scenarios" / "suite.json" +DEFAULT_RESULTS_DIR = Path("/tmp/flow_regression") +DEFAULT_BASELINE_DIR = ROOT / "scenarios" / "baseline" + + +def load_json(path: Path) -> dict | None: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (FileNotFoundError, json.JSONDecodeError, OSError): + return None + + +def load_suite(path: Path) -> dict: + suite = load_json(path) + if not isinstance(suite, dict) or not isinstance(suite.get("scenarios"), list): + raise SystemExit(f"invalid suite manifest: {path}") + return suite + + +def enabled_scenarios(suite: dict) -> list[dict]: + return [s for s in suite["scenarios"] + if isinstance(s, dict) and s.get("file") and s.get("enabled", True)] + + +def scenario_key(entry: dict) -> str: + """Stable identifier for a scenario (its file stem).""" + return Path(entry["file"]).stem + + +def run_scenario(entry: dict, default_duration: int, interval: float, + results_dir: Path) -> dict: + """Run one scenario through demo_evaluator and return its result payload. + + Returns the dict written by ``demo_evaluator --json-out``: + {scenario, result, failures, warnings, summary} + On launch failure a synthetic FAIL payload is returned so the matrix stays + complete rather than aborting the whole suite. + """ + key = scenario_key(entry) + out_path = results_dir / f"{key}.json" + duration = int(entry.get("duration_s", default_duration)) + cmd = [ + sys.executable, str(EVALUATOR), + "--scenario", entry["file"], + "--duration", str(duration), + "--interval", str(interval), + "--json-out", str(out_path), + ] + print(f"\n─── running scenario '{key}' ({duration}s) ───") + proc = subprocess.run(cmd, cwd=ROOT) + payload = load_json(out_path) + if payload is None: + payload = { + "scenario": key, + "result": "FAIL", + "failures": [f"evaluator produced no result (exit={proc.returncode})"], + "warnings": [], + "summary": {}, + } + return payload + + +def compare_summary(baseline: dict, current: dict, tolerances: dict) -> list[str]: + """Return a list of regression messages (empty == no regression). + + Only keys present in ``tolerances`` are gated. Supported gate types: + * ``min_ratio`` — current >= baseline * ratio + * ``max_abs_increase`` — current <= baseline + delta + Missing/non-numeric values are skipped (they cannot regress meaningfully). + """ + regressions: list[str] = [] + for metric, rule in tolerances.items(): + if metric.startswith("_") or not isinstance(rule, dict): + continue + base = baseline.get(metric) + cur = current.get(metric) + if not isinstance(base, (int, float)) or not isinstance(cur, (int, float)): + continue + if "min_ratio" in rule: + threshold = base * rule["min_ratio"] + if cur < threshold: + regressions.append( + f"{metric}: {cur:.3f} < {threshold:.3f} " + f"(baseline {base:.3f} x {rule['min_ratio']})" + ) + if "max_abs_increase" in rule: + threshold = base + rule["max_abs_increase"] + if cur > threshold: + regressions.append( + f"{metric}: {cur:.3f} > {threshold:.3f} " + f"(baseline {base:.3f} + {rule['max_abs_increase']})" + ) + return regressions + + +def print_report(rows: list[dict]) -> None: + print("\n==================== Regression Matrix ====================") + header = f"{'scenario':<26} {'result':<6} {'regress':<8} notes" + print(header) + print("-" * len(header)) + for row in rows: + note = "" + if row["failures"]: + note = row["failures"][0] + elif row["regressions"]: + note = row["regressions"][0] + print(f"{row['scenario']:<26} {row['result']:<6} " + f"{('YES' if row['regressions'] else '-'):<8} {note}") + print("=" * len(header)) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--suite", type=Path, default=DEFAULT_SUITE, + help="scenario suite manifest (default: scenarios/suite.json)") + parser.add_argument("--interval", type=float, default=0.25, + help="sample interval passed to demo_evaluator") + parser.add_argument("--results-dir", type=Path, default=DEFAULT_RESULTS_DIR, + help="where per-scenario result JSON is written") + parser.add_argument("--baseline-dir", type=Path, default=DEFAULT_BASELINE_DIR, + help="directory holding baseline result JSON files") + parser.add_argument("--baseline", action="store_true", + help="compare results against the saved baseline and fail on regression") + parser.add_argument("--update-baseline", action="store_true", + help="write the current results into the baseline directory and exit 0") + parser.add_argument("--only", type=str, default=None, + help="run only the scenario whose file stem matches this value") + parser.add_argument("--dry-run", action="store_true", + help="list the scenarios that would run without executing the demo") + args = parser.parse_args() + + suite = load_suite(args.suite) + scenarios = enabled_scenarios(suite) + if args.only: + scenarios = [s for s in scenarios if scenario_key(s) == args.only] + if not scenarios: + raise SystemExit(f"no enabled scenario matches --only {args.only!r}") + default_duration = int(suite.get("default_duration_s", 30)) + raw_tolerances = suite.get("baseline_tolerances") + tolerances = raw_tolerances if isinstance(raw_tolerances, dict) else {} + + if args.dry_run: + print(f"suite: {suite.get('name')} ({len(scenarios)} scenarios)") + for entry in scenarios: + print(f" - {scenario_key(entry):<26} " + f"{entry.get('duration_s', default_duration)}s {entry['file']}") + return 0 + + args.results_dir.mkdir(parents=True, exist_ok=True) + + rows: list[dict] = [] + for entry in scenarios: + key = scenario_key(entry) + payload = run_scenario(entry, default_duration, args.interval, args.results_dir) + summary = payload.get("summary", {}) if isinstance(payload.get("summary"), dict) else {} + failures = payload.get("failures", []) or [] + regressions: list[str] = [] + + if args.baseline: + baseline_payload = load_json(args.baseline_dir / f"{key}.json") + if baseline_payload is None: + regressions.append("no baseline recorded for this scenario") + else: + base_summary = baseline_payload.get("summary", {}) + regressions = compare_summary(base_summary, summary, tolerances) + + rows.append({ + "scenario": key, + "result": payload.get("result", "FAIL"), + "failures": failures, + "warnings": payload.get("warnings", []) or [], + "regressions": regressions, + }) + + if args.update_baseline: + args.baseline_dir.mkdir(parents=True, exist_ok=True) + for entry in scenarios: + key = scenario_key(entry) + src = args.results_dir / f"{key}.json" + if src.exists(): + (args.baseline_dir / f"{key}.json").write_text( + src.read_text(encoding="utf-8"), encoding="utf-8") + print(f"\nupdated baseline in {args.baseline_dir} " + f"({len(scenarios)} scenarios)") + return 0 + + print_report(rows) + + failed = [r for r in rows if r["result"] != "PASS"] + regressed = [r for r in rows if r["regressions"]] + if failed: + print(f"\n{len(failed)} scenario(s) FAILED behavioral checks.") + if args.baseline and regressed: + print(f"{len(regressed)} scenario(s) REGRESSED vs baseline.") + if failed or (args.baseline and regressed): + return 2 + print("\nALL scenarios PASS within the regression envelope.") + return 0 + + +if __name__ == "__main__": + sys.exit(main())