From 0bae12f78795113dbc492b188d0c010434ea6c83 Mon Sep 17 00:00:00 2001 From: chaoliu Date: Wed, 14 Oct 2026 16:14:58 +0800 Subject: [PATCH 1/2] feat(stereo-split): validate archebase topic prefix --- ...split-prefixed-topic-output-design.zh.html | 566 ++++++++++++++++++ internal/services/stereosplit/manager_test.go | 94 ++- internal/services/stereosplit/reconciler.go | 22 +- 3 files changed, 672 insertions(+), 10 deletions(-) create mode 100644 docs/designs/v2/stereo-split-prefixed-topic-output-design.zh.html diff --git a/docs/designs/v2/stereo-split-prefixed-topic-output-design.zh.html b/docs/designs/v2/stereo-split-prefixed-topic-output-design.zh.html new file mode 100644 index 0000000..e6aab80 --- /dev/null +++ b/docs/designs/v2/stereo-split-prefixed-topic-output-design.zh.html @@ -0,0 +1,566 @@ + + + + + + + Stereo Split / archebase Topic 前缀兼容方案 + + + +
+
+ STEREO SPLIT / MINIMAL COMPATIBILITY DESIGN +

两种 Topic
两种输出

+

+ 支持 /decxin/... 与 /archebase/decxin/... 两种固定布局；输入是哪一种，输出就保持哪一种，Keystone QA 同时接受两种。 +

+
只做
两种！
+
+
+ +
+
+ 不做通用 namespace 系统。 + 只新增对 /archebase 前缀的明确兼容；不带前缀的现有 MCAP 行为完全保持不变；不考虑 v1。 +
+ +
+
01

需求定义

+
+
+

现有模式

+ 
/decxin/rgb/compressed
+/decxin/imu
+/decxin/serial_number
+
+→ 输出
+/decxin/left_rgb/h264
+/decxin/right_rgb/h264
+/decxin/imu
+/decxin/serial_number
+
+
+

新增模式

+ 
/archebase/decxin/rgb/compressed
+/archebase/decxin/imu
+/archebase/decxin/serial_number
+
+→ 输出
+/archebase/decxin/left_rgb/h264
+/archebase/decxin/right_rgb/h264
+/archebase/decxin/imu
+/archebase/decxin/serial_number
+
+
+
+ 只有两种合法前缀：空字符串与 /archebase。不支持任意 /foo、多级 namespace 或不同 topic 使用不同前缀。 +
+
+ +
+
02

最小方案

+
+
① 镜像探测先找无前缀 input；找不到再找 /archebase input。
+
② 同前缀处理input、IMU、serial、left、right 使用同一个 prefix。
+
③ Manifest 标记输出 topic_prefix，值只能为空或 /archebase。
+
④ Keystone QA按 manifest prefix 拼出待验证的 left/right/imu topic。
+
+

+ 不需要数据库迁移，不需要新增管理配置，不需要 Keystone 预读源 MCAP，也不需要把完整 topic 列表设计成通用动态契约。 +

+
+ +
+
03

镜像改造

+

改动集中在当前构建使用的 orbit/jobs/stereo-split/ 中。

+ +

3.1 固定前缀探测

+ 
SUPPORTED_TOPIC_PREFIXES = ("", "/archebase")
+
+def detect_topic_prefix(topics):
+    canonical = "/decxin/rgb/compressed"
+    prefixed = "/archebase/decxin/rgb/compressed"
+
+    if canonical in topics and prefixed in topics:
+        raise RuntimeError("ambiguous stereo input topics")
+    if canonical in topics:
+        return ""
+    if prefixed in topics:
+        return "/archebase"
+    raise RuntimeError("no supported CompressedImage topic found")
+ +

3.2 根据 prefix 构造有效配置

+ 
def topic(prefix, canonical):
+    return prefix + canonical
+
+config = ConverterConfig(
+    input_topic=topic(prefix, "/decxin/rgb/compressed"),
+    left_topic=topic(prefix, "/decxin/left_rgb/h264"),
+    right_topic=topic(prefix, "/decxin/right_rgb/h264"),
+    imu_topic=topic(prefix, "/decxin/imu"),
+    serial_topic=topic(prefix, "/decxin/serial_number"),
+)
+

+ 这样现有转换主流程基本不需要理解 namespace：它仍然使用 config.input_topic、 + config.imu_topic、config.left_topic 等字段，只是 config 由输入自动决定。 +

+ +

3.3 Manifest 增加一个简单字段

+ 
{
+  "schema_version": 2,
+  "output_format": "stereo_h264",
+  "topic_prefix": "/archebase",
+  "stats": {
+    "input_messages": 1238,
+    "left_videos": 1214,
+    "right_videos": 1214,
+    "imu_messages": 27236
+  }
+}
+
    +
  • 无前缀任务写 "topic_prefix": ""，或者省略该字段。
    • +
  • 旧镜像生成的 manifest 没有此字段，Keystone 读取零值 ""，继续按无前缀处理。
    • +
  • 只接受 "" 和 "/archebase"。
    • +
+ +

3.4 已拆分输入

+

+ 为保持行为一致，split H.264 timestamp-repair 分支也使用同一套 prefix 探测。带前缀的 left/right/imu 输入仍带前缀写出；无前缀行为不变。 +

+
+ +
+
04

Keystone 改造

+

Keystone 只需要知道本次结果属于哪一种固定布局，不需要解析源文件，也不需要支持任意 namespace。

+ +

4.1 Manifest 结构增加字段

+ 
type processingManifest struct {
+    // existing fields...
+    TopicPrefix string `json:"topic_prefix,omitempty"`
+}
+ +

4.2 QA contract 增加 IMU topic

+ 
type mcapOutputContract struct {
+    // existing fields...
+    LeftTopic  string
+    RightTopic string
+    IMUTopic   string
+}
+ +

4.3 v2/v3 contract 构造

+ 
prefix := strings.TrimSpace(manifest.TopicPrefix)
+if prefix != "" && prefix != "/archebase" {
+    return mcapOutputContract{}, fmt.Errorf(
+        "unsupported stereo split topic prefix %q", prefix,
+    )
+}
+
+contract.LeftTopic = prefix + leftVideoTopic
+contract.RightTopic = prefix + rightVideoTopic
+contract.IMUTopic = prefix + imuTopic
+

+ 其中现有常量保持为 canonical suffix：/decxin/left_rgb/h264、 + /decxin/right_rgb/h264、/decxin/imu。 +

+ +

4.4 inspectOutputMCAP 使用 contract.IMUTopic

+ 
states := map[string]*topicQAState{
+    contract.LeftTopic:  {},
+    contract.RightTopic: {},
+    contract.IMUTopic:   {},
+}
+
+expectedSchemas := map[string]string{
+    contract.LeftTopic:  compressedVideoSchema,
+    contract.RightTopic: compressedVideoSchema,
+    contract.IMUTopic:   imuSchema,
+}
+
+imuCount := states[contract.IMUTopic].Count
+

+ 左右视频 schema、protobuf encoding、IMU schema、cdr encoding、消息数量和时间戳检查逻辑都保持不变，只是 topic key 根据固定 prefix 切换。 +

+ +
+

不需要做的 Keystone 改动

+
    +
  • 不需要新增数据库字段或 migration。
    • +
  • 不需要修改 Episode、Derivative 或 Bulk API。
    • +
  • 不需要修改 Synapse 页面。
    • +
  • 不需要 Keystone 读取源 MCAP summary。
    • +
  • 不需要新增系统设置或管理员选择项。
    • +
+
+
+ +
+
05

兼容性行为

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
输入镜像输出ManifestKeystone QA
/decxin/.../decxin/...topic_prefix="" 或缺省使用现有 topic；行为不变
/archebase/decxin/.../archebase/decxin/...topic_prefix="/archebase"使用带前缀 topic；验证通过
两种 input 同时存在不生成无作业失败，避免猜测
其它前缀不生成无明确报 unsupported
旧镜像的无 prefix manifest历史产物没有 topic_prefixGo 零值为空，继续通过
+
+ +
+
06

测试清单

+
+
+

镜像测试

+
    +
  • 无前缀 joined H.264 回归测试。
    • +
  • /archebase joined H.264 输出带前缀。
    • +
  • /archebase IMU 与 serial 被正确识别和保留。
    • +
  • /archebase split H.264 timestamp-repair。
    • +
  • 两种 input 同时存在时失败。
    • +
  • manifest 的 topic_prefix 与输出一致。
    • +
+
+
+

Keystone 测试

+
    +
  • 缺省 prefix 构造现有 QA contract。
    • +
  • /archebase 构造带前缀 QA contract。
    • +
  • 其它 prefix 被拒绝。
    • +
  • 带前缀 MCAP 的 left/right/imu 计数通过。
    • +
  • 带前缀 MCAP 缺失 IMU 时失败。
    • +
  • 现有无前缀测试全部保持通过。
    • +
+
+
+ +

真实文件验收

+

+ 使用 ~/Downloads/session_004(1).mcap 作为集成样本。作业应识别 + topic_prefix=/archebase，输出 left/right/imu/serial 全部保留前缀，并由 Keystone 完成产物验证与自动 QA。 +

+
+ +
+
07

实施顺序

+
    +
  1. 先在镜像 converter 中增加两种固定 prefix 的探测与有效 config 构造。
    2. +
  2. 让 run_processing.py 把 topic_prefix 写入 metadata 与 manifest。
    3. +
  3. 修改 Keystone manifest struct 和动态 QA contract。
    4. +
  4. 补镜像与 Keystone 单测。
    5. +
  5. 用 session_004(1).mcap 跑一次真实 Orbit Job。
    6. +
  6. Keystone 升级后再切换新的 immutable image digest。
    7. +
+
+ 发布顺序必须是 Keystone 先支持 /archebase QA，再启用会输出该前缀的新镜像；否则新镜像的带前缀产物会被旧 Keystone 判定失败。 +
+
+ +
+
08

结论

+
+

这是一个小范围兼容改造

+

+ 镜像增加两种固定 topic 布局的选择；manifest 增加一个 topic_prefix 字段；Keystone QA 根据该字段在两套固定 topic 之间切换。 +

+

+ 不引入通用 namespace、不增加数据库、不改变 API、不影响现有无前缀 MCAP，也不考虑 v1。 +

+
+
+
+ + + + diff --git a/internal/services/stereosplit/manager_test.go b/internal/services/stereosplit/manager_test.go index 2e7025b..79fc5ac 100644 --- a/internal/services/stereosplit/manager_test.go +++ b/internal/services/stereosplit/manager_test.go @@ -1820,6 +1820,52 @@ func TestReconcileQAApprovesStereoH264ManifestV2(t *testing.T) { } } +func TestReconcileQAApprovesArchebasePrefixedStereoH264ManifestV2(t *testing.T) { + db := newTestDB(t) + insertTestEpisode(t, db, 38, "keystone_tos", `{"bucket":"source-bucket","object_key":"raw/source.mcap"}`, "") + outputMCAP, outputChecksum := makeStereoSplitH264QAOutputWithPrefix(t, "/archebase", 10, 10, 100) + outputKey := "derived/qa-h264-archebase/output_bag.mcap" + objects := &fakeObjectStore{objects: map[string]fakeStoredObject{ + outputKey: {size: int64(len(outputMCAP)), etag: "mcap-etag", body: string(outputMCAP)}, + }} + manager := NewManager(db, nil, objects, testManagerConfig()) + derivative, _, err := manager.Start(context.Background(), 38, "admin") + if err != nil { + t.Fatalf("Start() error = %v", err) + } + processingResult := `{ + "schema_version":2, + "output_format":"stereo_h264", + "topic_prefix":"/archebase", + "stats":{ + "input_messages":10, + "decoded_images":10, + "left_videos":10, + "right_videos":10, + "imu_messages":100, + "skipped_messages":0 + } + }` + if _, err := db.Exec(` + UPDATE episode_derivatives SET processing_status = ?, qa_status = ?, mcap_path = ?, + checksum = ?, processing_result = ?, orbit_delete_status = ? + WHERE id = ? + `, ProcessingSucceeded, QAPending, outputKey, outputChecksum, processingResult, DeletePending, derivative.ID); err != nil { + t.Fatalf("prepare prefixed H.264 QA derivative: %v", err) + } + + if _, err := manager.ReconcileOnce(context.Background()); err != nil { + t.Fatalf("ReconcileOnce() error = %v", err) + } + derivative, err = manager.Get(context.Background(), 38) + if err != nil { + t.Fatalf("Get() error = %v", err) + } + if derivative.QAStatus != QAApproved || derivative.QAError != "" { + t.Fatalf("prefixed H.264 QA derivative = %+v", derivative) + } +} + func TestReconcileQAApprovesMatchingCalibrationAttachment(t *testing.T) { db := newTestDB(t) insertTestEpisode(t, db, 35, "keystone_tos", `{"bucket":"source-bucket","object_key":"raw/source.mcap"}`, "") @@ -2138,6 +2184,36 @@ func TestValidateManifestSnapshotSupportsStereoSplitV1AndV2(t *testing.T) { } } +func TestValidateManifestStatsUsesSupportedTopicPrefix(t *testing.T) { + manifest := processingManifest{ + SchemaVersion: manifestSchemaV2, + OutputFormat: stereoH264OutputFormat, + TopicPrefix: "/archebase", + Stats: manifestStats{ + InputMessages: 10, + DecodedImages: 10, + LeftVideos: 10, + RightVideos: 10, + IMUMessages: 100, + }, + } + + contract, err := validateManifestStats(manifest) + if err != nil { + t.Fatalf("validateManifestStats() error = %v", err) + } + if contract.LeftTopic != "/archebase/decxin/left_rgb/h264" || + contract.RightTopic != "/archebase/decxin/right_rgb/h264" || + contract.IMUTopic != "/archebase/decxin/imu" { + t.Fatalf("prefixed contract = %+v", contract) + } + + manifest.TopicPrefix = "/unsupported" + if _, err := validateManifestStats(manifest); err == nil || !strings.Contains(err.Error(), "unsupported topic prefix") { + t.Fatalf("validateManifestStats() unsupported prefix error = %v", err) + } +} + func TestValidateManifestStatsAcceptsTimestampRepairMode(t *testing.T) { manifest := processingManifest{ SchemaVersion: manifestSchemaV2, @@ -2223,6 +2299,7 @@ func makeStereoSplitQAOutput(t *testing.T, leftCount, rightCount, imuCount int) "ros2msg", leftImageTopic, rightImageTopic, + imuTopic, "cdr", leftCount, rightCount, @@ -2231,12 +2308,21 @@ func makeStereoSplitQAOutput(t *testing.T, leftCount, rightCount, imuCount int) } func makeStereoSplitH264QAOutput(t *testing.T, leftCount, rightCount, imuCount int) ([]byte, string) { + return makeStereoSplitH264QAOutputWithPrefix(t, "", leftCount, rightCount, imuCount) +} + +func makeStereoSplitH264QAOutputWithPrefix( + t *testing.T, + prefix string, + leftCount, rightCount, imuCount int, +) ([]byte, string) { return makeStereoSplitQAOutputForContract( t, compressedVideoSchema, "protobuf", - leftVideoTopic, - rightVideoTopic, + prefix+leftVideoTopic, + prefix+rightVideoTopic, + prefix+imuTopic, "protobuf", leftCount, rightCount, @@ -2255,6 +2341,7 @@ func makeStereoSplitH264QAOutputWithAttachments( "protobuf", leftVideoTopic, rightVideoTopic, + imuTopic, "protobuf", leftCount, rightCount, @@ -2269,6 +2356,7 @@ func makeStereoSplitQAOutputForContract( stereoSchemaEncoding string, leftTopic string, rightTopic string, + imuTopicName string, stereoMessageEncoding string, leftCount int, rightCount int, @@ -2295,7 +2383,7 @@ func makeStereoSplitQAOutputForContract( for _, channel := range []*mcap.Channel{ {ID: 1, SchemaID: 1, Topic: leftTopic, MessageEncoding: stereoMessageEncoding}, {ID: 2, SchemaID: 1, Topic: rightTopic, MessageEncoding: stereoMessageEncoding}, - {ID: 3, SchemaID: 2, Topic: imuTopic, MessageEncoding: "cdr"}, + {ID: 3, SchemaID: 2, Topic: imuTopicName, MessageEncoding: "cdr"}, } { if err := writer.WriteChannel(channel); err != nil { t.Fatalf("write MCAP channel: %v", err) diff --git a/internal/services/stereosplit/reconciler.go b/internal/services/stereosplit/reconciler.go index 59e900e..21cdd60 100644 --- a/internal/services/stereosplit/reconciler.go +++ b/internal/services/stereosplit/reconciler.go @@ -1084,6 +1084,7 @@ type processingManifest struct { Generation int `json:"generation"` ProcessorImage string `json:"processor_image"` CameraSerial string `json:"camera_serial,omitempty"` + TopicPrefix string `json:"topic_prefix,omitempty"` Source struct { URI string `json:"uri"` SizeBytes int64 `json:"size_bytes"` @@ -1524,6 +1525,7 @@ type mcapOutputContract struct { SchemaVersion int LeftTopic string RightTopic string + IMUTopic string LeftSchema string RightSchema string LeftSchemaEncoding string @@ -1548,6 +1550,7 @@ func validateManifestStats(manifest processingManifest) (mcapOutputContract, err contract := mcapOutputContract{ SchemaVersion: manifest.SchemaVersion, ExpectedIMU: stats.IMUMessages, + IMUTopic: imuTopic, } switch manifest.SchemaVersion { case manifestSchemaV1: @@ -1574,8 +1577,13 @@ func validateManifestStats(manifest processingManifest) (mcapOutputContract, err if stats.CopiedMessages < 0 || stats.CopiedTopics < 0 { return mcapOutputContract{}, fmt.Errorf("processing manifest contains negative copied-topic statistics") } - contract.LeftTopic = leftVideoTopic - contract.RightTopic = rightVideoTopic + topicPrefix := strings.TrimSpace(manifest.TopicPrefix) + if topicPrefix != "" && topicPrefix != "/archebase" { + return mcapOutputContract{}, fmt.Errorf("processing manifest has unsupported topic prefix %q", topicPrefix) + } + contract.LeftTopic = topicPrefix + leftVideoTopic + contract.RightTopic = topicPrefix + rightVideoTopic + contract.IMUTopic = topicPrefix + imuTopic contract.LeftSchema = compressedVideoSchema contract.RightSchema = compressedVideoSchema contract.LeftSchemaEncoding = "protobuf" @@ -1657,22 +1665,22 @@ func (m *Manager) inspectOutputMCAP( states := map[string]*topicQAState{ contract.LeftTopic: {}, contract.RightTopic: {}, - imuTopic: {}, + contract.IMUTopic: {}, } expectedSchemas := map[string]string{ contract.LeftTopic: contract.LeftSchema, contract.RightTopic: contract.RightSchema, - imuTopic: imuSchema, + contract.IMUTopic: imuSchema, } expectedSchemaEncodings := map[string]string{ contract.LeftTopic: contract.LeftSchemaEncoding, contract.RightTopic: contract.RightSchemaEncoding, - imuTopic: "ros2msg", + contract.IMUTopic: "ros2msg", } expectedEncodings := map[string]string{ contract.LeftTopic: contract.LeftMessageEncoding, contract.RightTopic: contract.RightMessageEncoding, - imuTopic: "cdr", + contract.IMUTopic: "cdr", } var firstLogTime uint64 var lastLogTime uint64 @@ -1767,7 +1775,7 @@ func (m *Manager) inspectOutputMCAP( } leftCount := states[contract.LeftTopic].Count rightCount := states[contract.RightTopic].Count - imuCount := states[imuTopic].Count + imuCount := states[contract.IMUTopic].Count if leftCount <= 0 || leftCount != rightCount || imuCount <= 0 { return mcapQAObservation{}, fmt.Errorf("output MCAP required topic counts are invalid") } From 74b7c15581f93ca4644aee0a3d8f9ef70a79cfe1 Mon Sep 17 00:00:00 2001 From: chaoliu Date: Wed, 14 Oct 2026 16:24:02 +0800 Subject: [PATCH 2/2] feat(stereo-split): preserve topic prefix in job --- jobs/stereo-split/run_processing.py | 46 +++++++++- .../stereo-split/tests/test_run_processing.py | 89 ++++++++++++++++++- 2 files changed, 133 insertions(+), 2 deletions(-) diff --git a/jobs/stereo-split/run_processing.py b/jobs/stereo-split/run_processing.py index 7b0d415..2775089 100644 --- a/jobs/stereo-split/run_processing.py +++ b/jobs/stereo-split/run_processing.py @@ -32,6 +32,7 @@ MCAP_MAGIC = b"\x89MCAP0\r\n" CALIBRATION_ATTACHMENT_NAME = "calibration.json" CALIBRATION_MEDIA_TYPE = "application/json" +ARCHEBASE_TOPIC_PREFIX = "/archebase" @dataclass(frozen=True) @@ -329,6 +330,44 @@ def load_embedded_calibration(path: Path) -> dict[str, object] | None: } +def detect_topic_prefix(path: Path) -> str: + with path.open("rb") as stream: + reader = make_reader(stream) + summary = reader.get_summary() + if summary is not None: + topics = {channel.topic for channel in summary.channels.values()} + else: + stream.seek(0) + topics = { + channel.topic + for _, channel, _ in make_reader(stream).iter_messages(log_time_order=False) + } + + def matches(prefix: str) -> bool: + joined = prefix + "/decxin/rgb/compressed" + left = prefix + "/decxin/left_rgb/h264" + right = prefix + "/decxin/right_rgb/h264" + return joined in topics or {left, right}.issubset(topics) + + canonical = matches("") + archebase = matches(ARCHEBASE_TOPIC_PREFIX) + if canonical and archebase: + raise RuntimeError("input contains ambiguous stereo topic prefixes") + if archebase: + return ARCHEBASE_TOPIC_PREFIX + return "" + + +def converter_config_for_prefix(prefix: str) -> ConverterConfig: + return ConverterConfig( + input_topic=prefix + "/decxin/rgb/compressed", + left_topic=prefix + "/decxin/left_rgb/h264", + right_topic=prefix + "/decxin/right_rgb/h264", + imu_topic=prefix + "/decxin/imu", + serial_topic=prefix + "/decxin/serial_number", + ) + + def run(args: argparse.Namespace) -> dict[str, object]: source = args.input.resolve() output_binding = args.output_binding.resolve() @@ -357,7 +396,8 @@ def run(args: argparse.Namespace) -> dict[str, object]: if source_identity.sha256 != expected: raise RuntimeError(f"source checksum mismatch: expected {expected}, got {source_identity.sha256}") - config = ConverterConfig() + topic_prefix = detect_topic_prefix(local_input) + config = converter_config_for_prefix(topic_prefix) stats = StereoSplitH264Converter(config).convert( local_input, local_mcap, @@ -393,6 +433,8 @@ def run(args: argparse.Namespace) -> dict[str, object]: "imu_topic": config.imu_topic, "stats": stats_payload, } + if topic_prefix: + metadata["topic_prefix"] = topic_prefix if camera_serial: metadata["camera_serial"] = camera_serial if color_consistency is not None: @@ -427,6 +469,8 @@ def run(args: argparse.Namespace) -> dict[str, object]: "started_at": started_at, "finished_at": utc_now(), } + if topic_prefix: + manifest["topic_prefix"] = topic_prefix if camera_serial: manifest["camera_serial"] = camera_serial if color_consistency is not None: diff --git a/jobs/stereo-split/tests/test_run_processing.py b/jobs/stereo-split/tests/test_run_processing.py index 735c1ec..55d73fa 100644 --- a/jobs/stereo-split/tests/test_run_processing.py +++ b/jobs/stereo-split/tests/test_run_processing.py @@ -125,8 +125,13 @@ def make_source_mcap( ) if serial is not None: + serial_topic = ( + "/archebase/decxin/serial_number" + if topic.startswith("/archebase/") + else "/decxin/serial_number" + ) serial_connection = writer.add_connection( - "/decxin/serial_number", + serial_topic, "std_msgs/msg/String", typestore=typestore, ) @@ -339,6 +344,52 @@ def test_publishes_outputs_and_manifest_for_valid_source(self) -> None: self.assertAlmostEqual(first_imu.angular_velocity.y, -5000 * gyro_scale) self.assertAlmostEqual(first_imu.angular_velocity.z, 6000 * gyro_scale) + def test_preserves_archebase_topic_prefix_in_outputs_and_manifest(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source = make_source_mcap( + root, + topic="/archebase/decxin/rgb/compressed", + serial="CMD-000148", + ) + output_binding = root / "published" + output_binding.mkdir() + + result = run_job(source, output_binding, root / "scratch") + + self.assertEqual(result.returncode, 0, result.stderr) + manifest = json.loads( + (output_binding / "processing_manifest.json").read_text(encoding="utf-8") + ) + metadata = json.loads( + (output_binding / "metadata.yaml").read_text(encoding="utf-8") + ) + self.assertEqual(manifest["topic_prefix"], "/archebase") + self.assertEqual(metadata["topic_prefix"], "/archebase") + self.assertEqual( + metadata["video"]["topics"], + [ + "/archebase/decxin/left_rgb/h264", + "/archebase/decxin/right_rgb/h264", + ], + ) + self.assertEqual(metadata["imu_topic"], "/archebase/decxin/imu") + + output_topics = set() + with (output_binding / "output_bag.mcap").open("rb") as stream: + summary = make_reader(stream).get_summary() + self.assertIsNotNone(summary) + output_topics = {channel.topic for channel in summary.channels.values()} + self.assertEqual( + output_topics, + { + "/archebase/decxin/left_rgb/h264", + "/archebase/decxin/right_rgb/h264", + "/archebase/decxin/imu", + "/archebase/decxin/serial_number", + }, + ) + def test_processes_already_split_input_and_records_repair_mode(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) @@ -360,6 +411,42 @@ def test_processes_already_split_input_and_records_repair_mode(self) -> None: self.assertEqual(manifest["stats"]["left_videos"], 2) self.assertEqual(manifest["stats"]["right_videos"], 2) + def test_preserves_archebase_prefix_for_already_split_input(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + prefix = "/archebase" + joined_source = make_source_mcap( + root, + topic=prefix + "/decxin/rgb/compressed", + ) + split_source = root / "split-source.mcap" + StereoSplitH264Converter( + processing_runner.converter_config_for_prefix(prefix) + ).convert(joined_source, split_source) + output_binding = root / "published" + output_binding.mkdir() + + result = run_job(split_source, output_binding, root / "scratch") + + self.assertEqual(result.returncode, 0, result.stderr) + manifest = json.loads( + (output_binding / "processing_manifest.json").read_text(encoding="utf-8") + ) + self.assertEqual(manifest["topic_prefix"], prefix) + self.assertEqual(manifest["processing_mode"], "timestamp_repair") + with (output_binding / "output_bag.mcap").open("rb") as stream: + summary = make_reader(stream).get_summary() + self.assertIsNotNone(summary) + topics = {channel.topic for channel in summary.channels.values()} + self.assertEqual( + topics, + { + prefix + "/decxin/left_rgb/h264", + prefix + "/decxin/right_rgb/h264", + prefix + "/decxin/imu", + }, + ) + def test_does_not_preserve_embedded_calibration_in_split_manifest(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir)