Export API¶

def construct_video_grid(
    experiment: Experiment,
    output_path: str | Path,
    *,
    layout: LayoutKind | str = LayoutKind.grid,
    video_ids: Sequence[str | None] | None = None,
    cell_size: tuple[int, int] = (640, 360),
    show_labels: bool = True,
    input_ids: Iterable[str] | None = None,
    include_merged_audio: bool = False,
    overwrite: bool = False,
    progress: Callable[[float], bool] | None = None,
) -> VideoGridResult:
    """Write selected experiment videos as one synchronized 25 fps video.

    The output interval is the union of all selected media on the experiment
    clock. ``input_ids`` defaults to every experiment input. Every selected
    input that carries audio contributes a separately selectable, full-duration
    audio track named after its input id; selected audio-only inputs contribute
    a track but no picture. ``include_merged_audio`` appends a track mixing all
    of those synchronized source tracks. Before and after a recording, and
    wherever it lost content, its slot shows whatever is behind it -- black, or
    the video it overlaps -- and its audio track is silent.

    ``layout`` arranges the pictures: the grid holds every selected video, while
    ``"2+1"`` shows two videos over a third and ``"4+1"`` shows one central video
    with a video lapping over each of its corners. Those two leave any slot they
    are not given a video for empty.
    ``video_ids`` says which video fills each slot of that layout, in
    drawing order, with ``None`` for a slot left empty; it defaults to the
    selected videos in experiment order.

    ``progress`` receives fractions between zero and one. Returning ``False``
    cancels construction and leaves no partial output behind.
    """
    kind = LayoutKind(layout)
    all_inputs = {data.id: data for data in experiment.inputs}
    if input_ids is None:
        selected_ids = set(all_inputs)
    else:
        if isinstance(input_ids, (str, bytes)):
            raise TypeError("input_ids must be an iterable of input ids, not a string")
        selected_ids = set(input_ids)
        unknown_ids = selected_ids - set(all_inputs)
        if unknown_ids:
            raise ValueError(f"unknown input ids: {sorted(unknown_ids)}")

    selected_inputs = [data for data in experiment.inputs if data.id in selected_ids]
    videos = [
        video
        for video in [*experiment.glasses_videos, *experiment.fixed_videos]
        if video.id in selected_ids
    ]
    if not videos:
        raise ValueError("no video inputs selected")
    if len(cell_size) != 2 or any(value <= 0 or value % 2 for value in cell_size):
        raise ValueError("cell dimensions must be positive even integers")

    slot_ids = _slot_ids(kind, videos, video_ids)
    frame_layout = layout_frame(kind, len(slot_ids), cell_size)

    output_path = Path(output_path)
    if output_path.suffix.lower() != ".mp4":
        raise ValueError("video-grid output path must have an .mp4 extension")
    if output_path.exists() and not overwrite:
        raise FileExistsError(f"output already exists: {output_path}")
    output_path.parent.mkdir(parents=True, exist_ok=True)

    sources = [_probe_source(data) for data in selected_inputs]
    sources_by_id = {source.data.id: source for source in sources}

    slot_sources = [
        None if slot_id is None else sources_by_id[slot_id] for slot_id in slot_ids
    ]
    video_sources = [source for source in slot_sources if source is not None]
    for source in video_sources:
        if source.video is None:
            raise VideoGridError(f"input {source.data.id!r} has no usable video stream")
    audio_sources = [source for source in sources if source.audio is not None]

    streams = [(source.data.timeline, source.video) for source in video_sources] + [
        (source.data.timeline, source.audio) for source in audio_sources
    ]
    output_start = min(
        timeline.to_experiment_time(stream.start) for timeline, stream in streams
    )
    output_end = max(
        timeline.to_experiment_time(stream.end) for timeline, stream in streams
    )

    temporary_output = output_path.with_name(
        f".{output_path.stem}.{uuid.uuid4().hex}{output_path.suffix}"
    )
    try:
        try:
            _render(
                temporary_output,
                slot_sources,
                audio_sources,
                output_start,
                output_end,
                frame_layout,
                show_labels,
                include_merged_audio,
                progress,
            )
        except (OSError, ValueError, av.FFmpegError) as exc:
            raise VideoGridError(f"could not construct video grid: {exc}") from exc
        if output_path.exists() and not overwrite:
            raise FileExistsError(f"output already exists: {output_path}")
        temporary_output.replace(output_path)
    finally:
        temporary_output.unlink(missing_ok=True)

    logger.info("wrote synchronized combined video to %s", output_path)
    return VideoGridResult(
        path=output_path,
        experiment_start=output_start,
        experiment_end=output_end,
    )

def layout_frame(
    kind: LayoutKind | str,
    slots: int,
    cell_size: tuple[int, int] = DEFAULT_CELL_SIZE,
) -> LayoutFrame:
    """The output frame and slot rectangles of one layout.

    ``cell_size`` is the size of one grid cell, and sets the scale of the whole
    frame.
    """
    if slots <= 0:
        raise ValueError("a layout needs at least one slot")
    if len(cell_size) != 2 or any(value <= 0 for value in cell_size):
        raise ValueError("cell dimensions must be positive")
    match LayoutKind(kind):
        case LayoutKind.two_plus_one:
            return _two_plus_one_frame(cell_size)
        case LayoutKind.four_plus_one:
            return _four_plus_one_frame(cell_size)
        case _:
            return _grid_frame(slots, cell_size)

def slot_count(kind: LayoutKind | str, videos: int) -> int:
    """How many videos this layout shows, given how many are available.

    The grid grows to hold every video; the other layouts have a fixed number
    of slots, which may be left empty.
    """
    match LayoutKind(kind):
        case LayoutKind.grid:
            return max(1, videos)
        case LayoutKind.two_plus_one:
            return 3
        case LayoutKind.four_plus_one:
            return 5

def export_elan(
    experiment: Experiment,
    video: VideoGridResult,
    *,
    overwrite: bool = False,
    author: str = "body-eye-sync",
) -> Path:
    """Write the experiment's speech turns and words beside the synchronized video."""
    turns = experiment.speech_turns.data
    if turns is None:
        raise ValueError(
            "this experiment has no speech turns; run speech post processing first"
        )
    if video.experiment_end <= video.experiment_start:
        raise ValueError("video result has an invalid experiment interval")

    output = video.path.with_suffix(".eaf")
    if output.exists() and not overwrite:
        raise FileExistsError(f"output already exists: {output}")

    speech_tiers = _speech_tiers(turns, video)
    word_tiers = _word_tiers(experiment, turns, video)
    document = _document(
        video,
        output,
        speech_tiers,
        word_tiers,
        author,
    )

    output.parent.mkdir(parents=True, exist_ok=True)
    if overwrite:
        output.unlink(missing_ok=True)
    document.to_file(str(output))

    return output