Writer 把规范化记录追加到可 seek 的二进制端口。finalize 会写入规范化结构索引、 completion footer 和定长 trailer。Reader 只有在验证容器结构后才 暴露记录,并支持按 stream、generation、tick 或仿真时间离线定位。
发行版 0.3.3 默认写入 FSR binary 1.2,profile 为
structural-validation。写入热路径不计算 payload SHA。Reader 打开文件时仍会严格校验
frame magic 与长度、offset 和边界、规范化 metadata、index 投影、全局及流内序号、
generation 与时钟顺序、footer 计数和截断,但会直接 seek 跳过 payload。index/footer
继续使用小范围摘要,stream inventory 不会伪造或填充占位 payload 摘要。
只有 read_payload、read_record 或 typed decode 才读取 payload。公开属性
FSRReader.file_sha256 的返回类型仍为 str,但只有显式访问时才会扫描整文件并缓存结果。
需要带摘要的文件时可显式选择 integrity="fast" 写入 binary 1.1,或选择
integrity="strong" 写入 binary 1.0;Reader 继续兼容两种历史布局。
from io import BytesIO
from fastsim_recording_format import FSRReader, FSRWriter
port = BytesIO()
writer = FSRWriter(port)
writer.write_manifest({"run_id": "example"})
writer.define_schema("joint-state/1", {"type": "object"})
writer.define_stream("robot.joints", "joint-state/1", required=True)
writer.write_record(
"robot.joints",
record_sequence=1,
source_sequence=1,
generation=0,
tick=0,
sim_time_s=0.0,
attributes={"dtype": "float32"},
payload=b"binary sample",
)
writer.finalize(completion={"status": "succeeded"}, last_record_sequence=1)
reader = FSRReader(port)
record = reader.seek("robot.joints", generation=0, tick=0, mode="exact")
assert record is not None
assert reader.read_payload(record) == b"binary sample"
# 可选:显式访问该属性时才进行一次整文件 SHA-256。
print(reader.file_sha256)
不可变内嵌资产
0.3.3 新增可选 asset frame,用于场景几何目录、分块网格等非时间序列数据。
write_asset 只能在第一条时间记录之前调用;asset ID 唯一、有界、进入索引,并可通过
FSRReader.assets 和 FSRReader.read_asset 读取。没有写入 asset 的普通录制仍保持原有
帧布局。Asset 不占用 record sequence,因此不会改变控制、观测、事件或相机的全局顺序。
Typed 离线读取
FSRTypedReader 在不依赖仿真器的情况下解码官方 Recorder 写入的 portable JSON、
raw RGB24 和 little-endian float64。安装可选 images extra 后,它也会校验并把 JPEG/PNG
RGB payload 解码为相同的 RGBFrame 类型。它借用一个已经完成完整验证的 FSRReader:
构造过程不会再次读取文件、不会创建第二个 Reader,也不会接管调用方二进制端口的
所有权。
from fastsim_recording_format import FSRReader, FSRTypedReader
reader = FSRReader(port) # 结构校验不会扫描 record payload
typed = FSRTypedReader.from_validated_reader(reader)
record = typed.seek("robot.joints", generation=0, tick=0, mode="exact")
assert record is not None and record.decoded == b"binary sample"
# 也可以直接解码已验证的 RecordRef,适合构建有界缓存。
reference = reader.records("robot.joints")[0]
decoded = typed.read(reference)
公开 typed API 还包括 DecodedRecord、StreamKind、classify_stream、
decode_payload 和 FSRCodecError。JSON 解码会拒绝重复 key 和非有限数,并执行
节点数与深度限制。RGB24 和 float64 payload 必须与声明的 shape、dtype、layout、
color space 和有限值约束一致。未知 codec 会 fail-closed;raw 不透明 payload
保持为 bytes,但声称是 RGB 或 calibration 的流必须显式声明受支持的 codec。
该包不会解压文件、跟随容器内路径或遍历目录。Artifact 选择、路径策略和输出事务 属于 FastSim Core 与 Record/Replay 插件。
规范化 metadata profile
FSR v1 固定使用 fastsim-fsr-cjson/1。该 profile 与 FastSim Core 的
simulation-contract-digest 规范化器相互独立。它使用 UTF-8,不使用 BOM 或空白;
mapping key 按 Unicode code point 排序;数组保留原始顺序;非 ASCII 文本不转义;
字符串使用 JSON 强制转义;整数使用十进制 token;浮点数必须有限,并使用
Python 3.11/3.12 json 的最短 round-trip 形式。因此 1.0 和 -0.0 的书写
差异会保留,1e-07 这类指数补位也属于 FSR v1 规范。非 Python 实现必须逐字节
复现包内发布的 golden vector。如需更换数值 profile,必须发布新的 FSR 格式版本,
不能静默改变 v1 字节。
安装与验证
支持 Python 3.11 和 3.12。基础包不依赖 FastSim、UniRoboSim、仿真器 SDK、NumPy 或图像 库。需要解码压缩 RGB 时安装图像 extra:
python -m pip install "fastsim-recording-format[images]==0.3.3"
未安装该 extra 时,容器校验及 raw/JSON/数值读取仍可使用;只有 typed JPEG/PNG 解码会以
带明确安装提示的 FSRCodecError 失败。
在本仓库中验证:
python -m pytest packages/fastsim-recording-format/tests -q
python -m ruff check \
packages/fastsim-recording-format/src \
packages/fastsim-recording-format/tests
cd packages/fastsim-recording-format
python -m mypy --config-file pyproject.toml src/fastsim_recording_format
python -m build
许可证与作者
作者:Hofee lexhofee@gmail.com
仓库许可证确定前,本包暂用 LicenseRef-Pending。