Description #

Timestamp merging operator – normalization, merging, and splitting of second-level intervals

Core capabilities #

• Normalization and sorting: unify the input format as (start, end) floating-point seconds and validate legality.
• Pre-merge small gaps: merge overlapping segments, and merge adjacent segments if the gap is less than or equal to the threshold (pre_merge_gap_seconds).
• Maximum silence priority splitting: prioritize splitting at the longest silence to ensure that each segment span does not exceed max_span_seconds.
• Forced chunking (optional): split overly long segments into fixed windows to ensure that each segment length does not exceed the upper limit (enforce_chunking).

Recommended practices #

• When VAD output contains short silences and semantic continuity needs to be preserved, prioritize using "maximum silence priority splitting" instead of simple sequential chunking.
• To strictly limit segment length as required by business, enable enforce_chunking to achieve continuous window splitting.
• Set pre_merge_gap_seconds slightly greater than the typical duration of noise silence (for example, 0.2–1.0 seconds).

Output format #

• Each row returns List[List[float]], with elements as [start, end] (unit: seconds), sorted in ascending order by start; output column type is List[List[float32]].

Processing flow #

1. Normalization: add the unified offset start_time, sort, and validate legality.
2. Pre-merge: merge overlapping segments and those with small gaps (≤ pre_merge_gap_seconds).
3. Maximum silence priority splitting: split at the longest silence to ensure each segment span ≤ max_span_seconds; for a single segment, there is no ≤ max_span_seconds requirement for the time range.
4. (Optional) Forced chunking: when enforce_chunking=True, split continuously into fixed windows to ensure time length ≤ max_span_seconds.

Operator parameters #

Input #

Output #

Each element is a two-dimensional list of floating-point numbers (unit: seconds), of type List[List[float32]]; invalid rows return None.

Parameters #

If a parameter does not have a default value, it is required.

Examples #

The following code demonstrates how to use daft to run the operator for normalization operations such as merging and splitting timestamp lists.

from __future__ import annotations

import os

import daft
from daft import col
from daft.las.functions.other.timestamps_merge import TimestampsMerge
from daft.las.functions.udf import las_udf

if __name__ == "__main__":

    if os.getenv("DAFT_RUNNER", "native") == "ray":
        import logging

        import ray

        def configure_logging():
            logging.basicConfig(
                level=logging.INFO,
                format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S.%s".format(),
            )
            logging.getLogger("tracing.span").setLevel(logging.WARNING)
            logging.getLogger("daft_io.stats").setLevel(logging.WARNING)
            logging.getLogger("DaftStatisticsManager").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaScheduler").setLevel(logging.WARNING)
            logging.getLogger("DaftFlotillaDispatcher").setLevel(logging.WARNING)

        ray.init(dashboard_host="0.0.0.0", runtime_env={"worker_process_setup_hook": configure_logging})
        daft.set_runner_ray()
    daft.set_execution_config(actor_udf_ready_timeout=600)
    daft.set_execution_config(min_cpu_per_task=0)

    samples = {"timestamps": [[[0.0, 4.34], [5.50, 7.12], [8.10, 8.34], [8.50, 10.12]]]}
    ds = daft.from_pydict(samples)

    start_time = 0.0
    pre_merge_gap_seconds = 0.5
    max_span_seconds = 10.0
    enforce_chunking = True
    ds = ds.with_column(
        "timestamps_merged",
        las_udf(
            TimestampsMerge,
            construct_args={
                "start_time": start_time,
                "pre_merge_gap_seconds": pre_merge_gap_seconds,
                "max_span_seconds": max_span_seconds,
                "enforce_chunking": enforce_chunking,
            },
            num_cpus=1,
            batch_size=1,
            concurrency=1,
        )(col("timestamps")),
    )
    ds.show()

    # ╭────────────────────────────────┬───────────────────────────╮
    # │ timestamps                     ┆ timestamps_merged         │
    # │ ---                            ┆ ---                       │
    # │ List[List[Float64]]            ┆ List[List[Float32]]       │
    # ╞════════════════════════════════╪═══════════════════════════╡
    # │ [[0, 4.34], [5.5, 7.12], [8.1… ┆ [[0, 4.34], [5.5, 10.12]] │
    # ╰────────────────────────────────┴───────────────────────────╯