持久层

数据库选型

系统支持双数据库后端，通过环境变量切换：

环境	数据库	连接串示例
开发/本地	SQLite	sqlite+pysqlite:///./unet_dt.db
生产/部署	MySQL 8.x	mysql+pymysql://user:pass@host:3306/db?charset=utf8mb4

MySQL 连接池调优

MySQL 模式下额外配置：

• pool_size=10：基础连接池大小
• max_overflow=20：溢出连接上限
• pool_recycle=3600：连接回收时间（防止 MySQL wait_timeout 断连）
• pool_pre_ping=True：每次取连接前验活
• connect_timeout=10：连接超时

数据表结构（6 表）

runs

主表，存储运行记录。

列名	类型	约束	说明
id	VARCHAR(128)	PK	运行 ID，格式 run-YYYYMMDD-HHMMSS-{hex8}
scenario_id	VARCHAR(128)	NOT NULL	关联场景 ID
strategy_id	VARCHAR(64)	NOT NULL	调度策略 ID
mode	VARCHAR(32)	NOT NULL, DEFAULT 'mock'	运行模式（mock/gateway/logonly）
status	VARCHAR(32)	NOT NULL, DEFAULT 'created'	状态机状态
steps	INTEGER	NULLABLE	总步数
current_step	INTEGER	NULLABLE, DEFAULT 0	当前执行步
dt_seconds	FLOAT	NULLABLE	步间隔（秒）
seed	INTEGER	NULLABLE	随机种子
created_at	DATETIME	NOT NULL	创建时间
started_at	DATETIME	NULLABLE	启动时间
finished_at	DATETIME	NULLABLE	结束时间
output_dir	VARCHAR(512)	NULLABLE	产物目录路径
error_message	TEXT	NULLABLE	错误信息

run_summary

运行摘要，一对一关联 runs。

列名	类型	约束	说明
run_id	VARCHAR(128)	PK, FK→runs.id (CASCADE)	运行 ID
avg_throughput_bps	FLOAT	NULLABLE	平均吞吐量
avg_loss_rate	FLOAT	NULLABLE	平均丢包率
p95_delay_ms	FLOAT	NULLABLE	P95 延迟
trace_count	INTEGER	NULLABLE	总 trace 数
received_count	INTEGER	NULLABLE	成功接收数
updated_at	DATETIME	NOT NULL	最后更新时间

run_events

运行事件日志。

列名	类型	约束	说明
id	INTEGER	PK, AUTO_INCREMENT	自增主键
run_id	VARCHAR(128)	FK→runs.id (CASCADE), INDEX	运行 ID
event_type	VARCHAR(64)	NOT NULL	事件类型（state/location_init_warning 等）
step_id	INTEGER	NULLABLE	关联步序号
payload_json	TEXT	NULLABLE	JSON 载荷
created_at	DATETIME	NOT NULL	事件时间

scenarios

场景定义，支持 DB + YAML 双源。

列名	类型	约束	说明
id	VARCHAR(128)	PK	场景 ID
name	VARCHAR(256)	NOT NULL	场景名称
nodes_json	TEXT	NOT NULL, DEFAULT '[]'	节点定义 JSON
traffic_json	TEXT	NOT NULL, DEFAULT '[]'	流量定义 JSON
run_config_json	TEXT	NOT NULL, DEFAULT '{}'	运行配置 JSON
unet_config_json	TEXT	NULLABLE	UNET 配置 JSON
created_at	DATETIME	NOT NULL	创建时间
updated_at	DATETIME	NOT NULL	更新时间

run_traces

原始 trace 记录，运行结束后从 CSV 批量导入。

列名	类型	约束	说明
id	INTEGER	PK, AUTO_INCREMENT	自增主键
run_id	VARCHAR(128)	FK→runs.id (CASCADE), INDEX	运行 ID
step_id	INTEGER	NOT NULL	步序号
seq	INTEGER	NOT NULL	帧序号
tx_time_ms	INTEGER	NOT NULL	发送时间（ms）
rx_time_ms	INTEGER	NULLABLE	接收时间（ms）
src	INTEGER	NOT NULL	源节点地址
dst	INTEGER	NOT NULL	目标节点地址
payload_bytes	INTEGER	NOT NULL, DEFAULT 0	载荷字节数
received	BOOLEAN	NOT NULL, DEFAULT FALSE	是否接收成功
source	VARCHAR(32)	NOT NULL, DEFAULT 'unknown'	数据来源

索引：ix_run_traces_run_step_seq (run_id, step_id, seq)

run_metrics

聚合指标，每步一行，由 step_callback 实时写入。

列名	类型	约束	说明
id	INTEGER	PK, AUTO_INCREMENT	自增主键
run_id	VARCHAR(128)	FK→runs.id (CASCADE), INDEX	运行 ID
step_id	INTEGER	NOT NULL	步序号
t_window_start_ms	INTEGER	NOT NULL, DEFAULT 0	窗口起始时间
t_window_end_ms	INTEGER	NOT NULL, DEFAULT 0	窗口结束时间
throughput_bps	FLOAT	NOT NULL, DEFAULT 0.0	吞吐量
loss_rate	FLOAT	NOT NULL, DEFAULT 0.0	丢包率
delay_p50_ms	FLOAT	NOT NULL, DEFAULT 0.0	P50 延迟
delay_p95_ms	FLOAT	NOT NULL, DEFAULT 0.0	P95 延迟
sent_count	INTEGER	NOT NULL, DEFAULT 0	发送帧数
recv_count	INTEGER	NOT NULL, DEFAULT 0	接收帧数
sync_error_ms	FLOAT	NOT NULL, DEFAULT 0.0	同步误差
overhead_ms	FLOAT	NOT NULL, DEFAULT 0.0	调度开销

索引：ix_run_metrics_run_step (run_id, step_id)

ER 关系图

┌──────────┐     1:1     ┌─────────────┐
│  runs    │────────────▶│ run_summary  │
│          │             └─────────────┘
│          │     1:N     ┌─────────────┐
│          │────────────▶│ run_events   │
│          │             └─────────────┘
│          │     1:N     ┌─────────────┐
│          │────────────▶│ run_traces   │
│          │             └─────────────┘
│          │     1:N     ┌─────────────┐
│          │────────────▶│ run_metrics  │
└──────────┘             └─────────────┘

┌──────────┐
│scenarios │  （独立表，通过 scenario_id 逻辑关联 runs）
└──────────┘

Repository 层

RunRepository (repo.py)

职责：runs / run_summary / run_events 三表操作

• prepare_schema()：自动建表 + 补列（via ensure_tables_with_columns）
• create_run() / get_run() / list_runs()：基础 CRUD
• transition_run_status()：带状态机校验的状态转换
• set_run_fields()：通用字段更新
• upsert_summary()：INSERT or UPDATE 摘要
• append_event() / list_events()：事件日志
• ping() / table_status()：健康检查支持
• 使用 Table 反射（autoload_with）适配可能存在的表结构差异

TraceRepository (trace_repo.py)

职责：run_traces / run_metrics 两表操作

• bulk_insert_traces()：批量写入 trace
• bulk_insert_metrics()：批量写入指标
• has_traces() / has_metrics()：判断是否有 DB 数据
• get_traces() / get_metrics()：分页查询
• get_latest_metric()：获取最新一条指标（供轮询端点使用）

ScenarioRepository (scenario_repo.py)

职责：scenarios 表 + YAML 文件同步

• create_scenario() / get_scenario() / update_scenario() / delete_scenario()
• list_scenarios_merged()：合并 DB 和 YAML 两种来源
• import_from_yaml() / export_to_yaml() / sync_to_yaml()：双向同步

自动 Schema 迁移

db.py 中的 ensure_table_with_columns() 实现轻量级自动迁移：

1. 表不存在 → CREATE TABLE
2. 表存在但缺列 → ALTER TABLE ADD COLUMN
3. MySQL 的 TEXT/BLOB 类型列添加为 NULL（不设 DEFAULT）
4. 其他类型根据 SQLAlchemy model 的 default 值自动设置

DbTableMapping

RuntimeSettings 中的 DbTableMapping 支持自定义表名：

class DbTableMapping(BaseModel):
    runs: str = "runs"
    run_summary: str = "run_summary"
    run_events: str = "run_events"
    scenarios: str = "scenarios"
    run_traces: str = "run_traces"
    run_metrics: str = "run_metrics"

通过环境变量 UNET_DT_DB_TABLE_{RUNS|RUN_SUMMARY|...} 可分别覆盖每张表的物理表名。