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_payloadread_record 或 typed decode 才读取 payload。公开属性 FSRReader.file_sha256 的返回类型仍为 str,但只有显式访问时才会扫描整文件并缓存结果。 需要带摘要的文件时可显式选择 integrity="fast" 写入 binary 1.1,或选择 integrity="strong" 写入 binary 1.0;Reader 继续兼容两种历史布局。

python
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.assetsFSRReader.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,也不会接管调用方二进制端口的 所有权。

python
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 还包括 DecodedRecordStreamKindclassify_streamdecode_payloadFSRCodecError。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:

bash
python -m pip install "fastsim-recording-format[images]==0.3.3"

未安装该 extra 时,容器校验及 raw/JSON/数值读取仍可使用;只有 typed JPEG/PNG 解码会以 带明确安装提示的 FSRCodecError 失败。

在本仓库中验证:

bash
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