JingsongLi commented on code in PR #9494:
URL: https://github.com/apache/paimon/pull/9494#discussion_r3975958288
##########
paimon-python/pypaimon/write/table_write.py:
##########
@@ -85,6 +85,13 @@ def write_arrow_batch(self, data: pa.RecordBatch):
sub_table = pa.compute.take(data, row_indices)
self._write_partition_bucket_batch(partition, bucket, sub_table)
+ def begin_video_episode(self, row_count: int):
Review Comment:
Could we use more general names for these interfaces in the shared write
layer? For example, `begin_video_episode(row_count)` could become
`roll_before_group_if_needed(row_count)`, and
`should_roll_before_video_episode(row_count)` could become
`should_roll_before_group(row_count)`, consistently across the writer layers.
The loader can treat each LeRobot episode as one logical write group. The
current method only checks whether to roll before the next group; it does not
establish a group lifecycle or track its remaining rows, so naming it after
that operation would describe its behavior more precisely. The documentation
should also state that the current implementation applies to writers with the
video grouping policy, rather than implying that arbitrary groups are
guaranteed to remain unsplit.
##########
paimon-python/pypaimon/multimodal/lerobot/loader.py:
##########
@@ -291,32 +321,316 @@ def _read_batch(dataset, info, begin, end, schema):
elif not isinstance(raw, pa.Table):
raw = pa.Table.from_pydict(raw)
features = info["features"]
+ video_rows = None
+ if any(feature.get("dtype") == "video"
+ for feature in features.values()):
+ video_rows = _validate_video_rows(
+ raw, info, episode, begin, end)
arrays = []
fields = []
for name, feature in features.items():
field = schema.field(name)
dtype = feature["dtype"]
- if name not in raw.column_names:
+ if dtype == "video":
+ values = _video_frame_descriptors(
+ dataset,
+ info,
+ episode,
+ video_rows,
+ name,
+ feature,
+ begin,
+ end,
+ video_sources if video_sources is not None else {},
+ )
+ elif name not in raw.column_names:
raise ValueError(
"LeRobot data is missing metadata feature %s." % name)
- values = raw.column(name).to_pylist()
- if dtype == "image":
- image_reader = getattr(dataset, "image_bytes", None)
- if callable(image_reader):
- values = [image_reader(value) for value in values]
+ else:
+ values = raw.column(name).to_pylist()
+ if dtype == "image":
+ image_reader = getattr(dataset, "image_bytes", None)
+ if callable(image_reader):
+ values = [image_reader(value) for value in values]
+ else:
+ values = [_image_bytes(value, dataset.root)
+ for value in values]
else:
- values = [_image_bytes(value, dataset.root)
+ values = [_normalize_value(value, feature, name)
for value in values]
- else:
- values = [_normalize_value(value, feature, name)
- for value in values]
arrays.append(_safe_array(values, field, name, dtype))
fields.append(field)
return pa.Table.from_arrays(arrays, schema=pa.schema(fields))
+def _video_frame_descriptors(
+ dataset, info, episode, video_rows, name, feature, begin, end, cache):
+ if episode is None:
+ raise ValueError("LeRobot video import requires Episode metadata.")
+ episode_begin = _nonnegative_integer(
+ episode["dataset_from_index"], "dataset_from_index")
+ episode_end = _nonnegative_integer(
+ episode["dataset_to_index"], "dataset_to_index")
+ if begin < episode_begin or end > episode_end:
+ raise ValueError(
+ "LeRobot video batch [%d, %d) crosses Episode range [%d, %d)."
+ % (begin, end, episode_begin, episode_end)
+ )
+
+ fps = _video_fps(info, feature, name)
+ prefix = "videos/%s/" % name
+ try:
+ chunk_index = _nonnegative_integer(
+ episode[prefix + "chunk_index"], prefix + "chunk_index")
+ file_index = _nonnegative_integer(
+ episode[prefix + "file_index"], prefix + "file_index")
+ from_timestamp = float(_python_scalar(
+ episode[prefix + "from_timestamp"]))
+ to_timestamp = float(_python_scalar(
+ episode[prefix + "to_timestamp"]))
+ except (KeyError, TypeError, ValueError) as error:
+ raise ValueError(
+ "LeRobot Episode metadata is missing video mapping for %s."
+ % name
+ ) from error
+
+ episode_length = episode_end - episode_begin
+ _validate_video_timestamp_range(
+ from_timestamp, to_timestamp, fps, episode_length, name)
+
+ source_key = (name, chunk_index, file_index)
+ source = cache.get(source_key)
+ if source is None:
+ uri, length = _video_source(
+ dataset, info, episode, name, chunk_index, file_index)
+ source = (uri, length, _video_sample_timestamps(dataset, uri))
+ cache[source_key] = source
+ uri, length, sample_timestamps = source
+ return [
+ VideoFrameDescriptor(
+ uri,
+ 0,
+ length,
+ _video_frame_ordinal(
+ sample_timestamps, from_timestamp + timestamp, name),
+ ).serialize()
+ for unused_frame_index, timestamp in video_rows
+ ]
+
+
+def _video_sample_timestamps(dataset, uri):
+ resolver = getattr(dataset, "video_sample_timestamps", None)
+ if callable(resolver):
+ values = resolver(uri)
+ else:
+ try:
+ import av
+ except ImportError as error:
+ raise ImportError(
+ "LeRobot video import requires PyAV. Install it with "
+ "`pip install 'pypaimon[lerobot]'`.") from error
+
+ input_stream = None
+ parsed = urlparse(uri)
+ if parsed.scheme in ("", "file"):
+ source = unquote(parsed.path) if parsed.scheme else uri
+ else:
+ factory = getattr(dataset, "video_uri_reader_factory", None)
+ if factory is None:
+ raise ValueError(
+ "LeRobot video source %s cannot be inspected." % uri)
+ input_stream = factory.create(uri).new_input_stream(uri)
+ source = input_stream
+ try:
+ with av.open(source) as container:
+ stream = container.streams.video[0]
+ values = [
+ float(packet.pts * (packet.time_base or stream.time_base))
+ for packet in container.demux(stream)
+ if packet.pts is not None and not packet.is_discard
+ ]
+ finally:
+ if input_stream is not None:
+ input_stream.close()
+
+ timestamps = array("d", sorted(float(value) for value in values))
+ if not timestamps or any(not math.isfinite(value) for value in timestamps):
+ raise ValueError(
+ "LeRobot video source %s has no valid frame timestamps." % uri)
+ return timestamps
+
+
+def _video_frame_ordinal(timestamps, timestamp, name):
+ position = bisect_left(timestamps, timestamp)
+ candidates = []
+ if position:
+ candidates.append(position - 1)
+ if position < len(timestamps):
+ candidates.append(position)
+ ordinal = min(
+ candidates,
+ key=lambda index: (abs(timestamps[index] - timestamp), index),
+ )
+ distance = abs(timestamps[ordinal] - timestamp)
+ if distance >= _VIDEO_TIMESTAMP_TOLERANCE:
+ raise ValueError(
Review Comment:
[P2] Please account for float32 timestamp rounding when matching video
frames.
A valid long episode can fail this check even when the MP4 and frame rows
are synchronized. At 30 FPS, frame index `61441` (about 34 minutes into the
episode) has an MP4 PTS of `2048.0333333333333`, while the standard float32
frame timestamp is `2048.033447265625`. The difference is approximately
`1.1393e-4`, exceeding the fixed `1e-4` tolerance.
I reproduced this with a real 61,442-frame MP4: PyAV decodes all frames, the
existing frame-control validation passes because it casts the expected
timestamp to the source dtype, but `_read_batch` raises `ValueError` here and
the import is aborted.
Could we account for the source timestamp dtype's quantization error when
validating the nearest PTS, while still rejecting mismatched frames? A
regression test crossing this float32 precision boundary would cover the case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
