crosstyan/cvmmap-streamer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(zed): recover corrupted frame gaps in MCAP export

Browse Source 
Make ZED MCAP export skip corrupted frame runs until recovery and

treat unreadable tail frames as end-of-stream instead of hard

failing conversion.

Update bundled nearest-mode export to emit partial bundles during

corruption gaps, extend BundleManifest with explicit member status

and skipped-frame counts, and only write payload messages for

present cameras.

Tighten batch probing so bundled MCAP validation checks /bundle

coverage and per-camera message counts, and improve failure

excerpts to include stderr tail output.

Also add a local cppzmq CMake fallback, refresh the multi-record

tester for the new bundle schema, and document the mixed NVENC

limitations in the README.
This commit is contained in:
crosstyan
2026-03-24 03:49:35 +00:00
parent e3a423433e
commit 807a73b480
8 changed files with 1006 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/zed_batch_svo_to_mcap.py
+171 -11
View File
@@ -10,6 +10,7 @@ import re
import subprocess
import sys
import time
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
@@ -34,6 +35,8 @@ class BatchConfig:
mcap_compression: str
depth_mode: str
depth_size: str
bundle_policy: str
bundle_topic: str | None
with_pose: bool
pose_config: Path | None
world_frame_id: str | None
@@ -102,6 +105,7 @@ class ActiveJobState:
_MCAP_READER_MODULE = None
_BUNDLE_MANIFEST_CLASS_CACHE: dict[bytes, tuple[object, int | None]] = {}
TABLE_REFRESH_SECONDS = 1.0
TEXT_HEARTBEAT_SECONDS = 30.0
@@ -500,7 +504,11 @@ def command_for_job(job: ConversionJob, config: BatchConfig, encoder_device: str
config.depth_mode,
"--depth-size",
config.depth_size,
"--bundle-policy",
config.bundle_policy,
]
if config.bundle_topic:
command.extend(["--bundle-topic", config.bundle_topic])
if config.with_pose:
command.append("--with-pose")
if config.pose_config is not None:
@@ -569,22 +577,131 @@ def required_topics_for(camera_labels: tuple[str, ...]) -> set[str]:
return topics
def probe_output(output_path: Path, camera_labels: tuple[str, ...]) -> OutputProbeResult:
def load_bundle_manifest_type(schema_data: bytes) -> tuple[object, int | None]:
cached = _BUNDLE_MANIFEST_CLASS_CACHE.get(schema_data)
if cached is not None:
return cached
from google.protobuf import descriptor_pb2, descriptor_pool, message_factory, timestamp_pb2
descriptor_set = descriptor_pb2.FileDescriptorSet()
descriptor_set.ParseFromString(schema_data)
pool = descriptor_pool.DescriptorPool()
has_embedded_timestamp = any(
file_descriptor.name == "google/protobuf/timestamp.proto"
for file_descriptor in descriptor_set.file
)
if has_embedded_timestamp:
for file_descriptor in descriptor_set.file:
if file_descriptor.name == "google/protobuf/timestamp.proto":
pool.Add(file_descriptor)
break
else:
pool.AddSerializedFile(timestamp_pb2.DESCRIPTOR.serialized_pb)
for file_descriptor in descriptor_set.file:
if file_descriptor.name == "google/protobuf/timestamp.proto":
continue
pool.Add(file_descriptor)
message_descriptor = pool.FindMessageTypeByName("cvmmap_streamer.BundleManifest")
message_class = message_factory.GetMessageClass(message_descriptor)
present_value = None
if "BundleMemberStatus" in message_descriptor.enum_types_by_name:
status_enum = message_descriptor.enum_types_by_name["BundleMemberStatus"]
present_value = status_enum.values_by_name["BUNDLE_MEMBER_STATUS_PRESENT"].number
_BUNDLE_MANIFEST_CLASS_CACHE[schema_data] = (message_class, present_value)
return message_class, present_value
def probe_output(
output_path: Path,
camera_labels: tuple[str, ...],
*,
bundle_topic: str | None,
) -> OutputProbeResult:
if not output_path.is_file():
return OutputProbeResult(output_path=output_path, status="missing")
reader_module = load_mcap_reader()
expected_topics = required_topics_for(camera_labels)
require_bundle = len(camera_labels) > 1 and bool(bundle_topic)
if require_bundle:
expected_topics.add(bundle_topic or "/bundle")
found_topics: set[str] = set()
video_counts: Counter[str] = Counter()
depth_counts: Counter[str] = Counter()
bundle_present_counts: Counter[str] = Counter()
expected_camera_labels = set(camera_labels)
try:
with output_path.open("rb") as stream:
reader = reader_module.make_reader(stream)
for _schema, channel, _message in reader.iter_messages():
for schema, channel, message in reader.iter_messages():
if channel.topic in expected_topics:
found_topics.add(channel.topic)
if found_topics == expected_topics:
return OutputProbeResult(output_path=output_path, status="valid")
if channel.topic.endswith("/video"):
video_counts[channel.topic.removeprefix("/").removesuffix("/video")] += 1
continue
if channel.topic.endswith("/depth"):
depth_counts[channel.topic.removeprefix("/").removesuffix("/depth")] += 1
continue
if require_bundle and channel.topic == bundle_topic:
if schema is None or schema.name != "cvmmap_streamer.BundleManifest":
return OutputProbeResult(
output_path=output_path,
status="invalid",
reason=f"bundle topic '{bundle_topic}' is missing the BundleManifest schema",
)
try:
bundle_class, present_value = load_bundle_manifest_type(schema.data)
bundle = bundle_class()
bundle.ParseFromString(message.data)
except Exception as error: # noqa: BLE001
return OutputProbeResult(
output_path=output_path,
status="invalid",
reason=f"failed to parse bundle manifest: {error}",
)
bundle_labels: set[str] = set()
for member in bundle.members:
label = str(member.camera_label)
if label not in expected_camera_labels:
return OutputProbeResult(
output_path=output_path,
status="invalid",
reason=f"bundle manifest referenced unknown camera label '{label}'",
)
if label in bundle_labels:
return OutputProbeResult(
output_path=output_path,
status="invalid",
reason=f"bundle manifest duplicated camera label '{label}'",
)
bundle_labels.add(label)
is_present = member.HasField("timestamp")
if present_value is not None:
is_present = member.status == present_value
if is_present and not member.HasField("timestamp"):
return OutputProbeResult(
output_path=output_path,
status="invalid",
reason=f"bundle member '{label}' is present but missing a timestamp",
)
if is_present:
bundle_present_counts[label] += 1
if bundle_labels != expected_camera_labels:
missing_labels = sorted(expected_camera_labels - bundle_labels)
extra_labels = sorted(bundle_labels - expected_camera_labels)
details = []
if missing_labels:
details.append("missing=" + ",".join(missing_labels))
if extra_labels:
details.append("extra=" + ",".join(extra_labels))
return OutputProbeResult(
output_path=output_path,
status="invalid",
reason="bundle manifest camera coverage mismatch: " + " ".join(details),
)
except Exception as error: # noqa: BLE001
return OutputProbeResult(output_path=output_path, status="invalid", reason=str(error))
@@ -595,6 +712,27 @@ def probe_output(output_path: Path, camera_labels: tuple[str, ...]) -> OutputPro
status="invalid",
reason="missing expected topics: " + ", ".join(missing_topics),
)
if require_bundle:
for label in camera_labels:
present_count = bundle_present_counts[label]
if video_counts[label] != present_count:
return OutputProbeResult(
output_path=output_path,
status="invalid",
reason=(
f"video count mismatch for {label}: "
f"bundle_present={present_count} video_messages={video_counts[label]}"
),
)
if depth_counts[label] != present_count:
return OutputProbeResult(
output_path=output_path,
status="invalid",
reason=(
f"depth count mismatch for {label}: "
f"bundle_present={present_count} depth_messages={depth_counts[label]}"
),
)
return OutputProbeResult(output_path=output_path, status="valid")
@@ -635,8 +773,13 @@ def split_lines_for_excerpt(text: str, max_lines: int = 8) -> list[str]:
lines = [line.rstrip() for line in text.splitlines() if line.strip()]
if len(lines) <= max_lines:
return lines
excerpt = lines[:max_lines]
excerpt.append(f"... ({len(lines) - max_lines} more lines)")
head_count = max(1, max_lines // 2)
tail_count = max_lines - head_count
excerpt = lines[:head_count]
omitted = len(lines) - head_count - tail_count
if omitted > 0:
excerpt.append(f"... ({omitted} omitted line(s))")
excerpt.extend(lines[-tail_count:])
return excerpt
@@ -1011,6 +1154,19 @@ def build_worker_slots(
default="optimal",
show_default=True,
)
@click.option(
"--bundle-policy",
type=click.Choice(("nearest", "strict")),
default="nearest",
show_default=True,
help="Bundling policy for multi-camera MCAP export.",
)
@click.option(
"--bundle-topic",
default="/bundle",
show_default=True,
help="Topic used for bundled multi-camera manifest messages.",
)
@click.option("--with-pose", is_flag=True, help="Enable per-camera positional tracking export when available.")
@click.option(
"--pose-config",
@@ -1026,19 +1182,19 @@ def build_worker_slots(
"--start-frame",
type=click.IntRange(min=0),
default=None,
help="First synced frame group to export (inclusive) in bundled multi-camera mode.",
help="First bundle index to export (inclusive) in bundled multi-camera mode.",
)
@click.option(
"--end-frame",
type=click.IntRange(min=0),
default=None,
help="Last synced frame group to export (inclusive) in bundled multi-camera mode.",
help="Last bundle index to export (inclusive) in bundled multi-camera mode.",
)
@click.option(
"--sync-tolerance-ms",
type=click.FloatRange(min=0.0, min_open=True),
default=None,
help="Override the maximum timestamp delta used for bundled multi-camera sync.",
help="Override the maximum timestamp delta used by strict bundled sync.",
)
@click.option(
"--progress-ui",
@@ -1070,6 +1226,8 @@ def main(
mcap_compression: str,
depth_mode: str,
depth_size: str,
bundle_policy: str,
bundle_topic: str,
with_pose: bool,
pose_config: Path | None,
world_frame_id: str | None,
@@ -1102,6 +1260,8 @@ def main(
mcap_compression=mcap_compression,
depth_mode=depth_mode,
depth_size=depth_size,
bundle_policy=bundle_policy,
bundle_topic=bundle_topic,
with_pose=with_pose,
pose_config=pose_config.expanduser().resolve() if pose_config is not None else None,
world_frame_id=world_frame_id,
@@ -1156,7 +1316,7 @@ def main(
continue
if report_existing:
probe_result = probe_output(output_path, job.camera_labels)
probe_result = probe_output(output_path, job.camera_labels, bundle_topic=config.bundle_topic)
if probe_result.status == "valid":
valid_existing.append(probe_result)
elif probe_result.status == "invalid":
@@ -1171,7 +1331,7 @@ def main(
continue
if config.probe_existing:
probe_result = probe_output(output_path, job.camera_labels)
probe_result = probe_output(output_path, job.camera_labels, bundle_topic=config.bundle_topic)
if probe_result.status == "valid":
valid_existing.append(probe_result)
skipped_results.append(