Skip to content

lasagnastack.stages.reel_pipeline.ingest

lasagnastack.stages.reel_pipeline.ingest

IngestStage 

IngestStage(max_workers: int = 1)

Bases: Stage

Initialise IngestStage.

Parameters:

Name Type Description Default
max_workers int

Number of parallel worker processes for clip normalisation and scene detection. 1 runs serially.

 1
Source code in src/lasagnastack/stages/reel_pipeline/ingest.py 
170
171
172
173
174
175
176
177
def __init__(self, max_workers: int = 1) -> None:
    """Initialise IngestStage.

    Args:
        max_workers: Number of parallel worker processes for clip
            normalisation and scene detection. ``1`` runs serially.
    """
    self._max_workers = max_workers

run 

run(
    input_dir: Path, output_dir: Path, max_workers: int = 1
) -> list[NormalisedClip]

Normalise all clips to 480×854 H.264 and detect scene cuts.

Parameters:

Name Type Description Default
input_dir Path

Folder containing raw MP4/MOV clips and the brief .txt.

 required
output_dir Path

Destination root; normalised clips written to output_dir/normalised/.

 required
max_workers int

Number of parallel worker processes. 1 runs serially in the current process; >1 uses a ProcessPoolExecutor.

 1

Returns:

Type Description
list[NormalisedClip]

One NormalisedClip per source file, in discovery order.
Source code in src/lasagnastack/stages/reel_pipeline/ingest.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def run(
    input_dir: Path,
    output_dir: Path,
    max_workers: int = 1,
) -> list[NormalisedClip]:
    """Normalise all clips to 480×854 H.264 and detect scene cuts.

    Args:
        input_dir: Folder containing raw MP4/MOV clips and the brief .txt.
        output_dir: Destination root; normalised clips written to output_dir/normalised/.
        max_workers: Number of parallel worker processes. ``1`` runs serially
            in the current process; ``>1`` uses a ``ProcessPoolExecutor``.

    Returns:
        One NormalisedClip per source file, in discovery order.
    """
    clips = _find_clips(input_dir)
    if not clips:
        raise ValueError(f"No MP4/MOV files found in {input_dir}")

    normalised_dir = output_dir / "normalised"
    normalised_dir.mkdir(parents=True, exist_ok=True)
    dests = [normalised_dir / f"{src.stem}_norm.mp4" for src in clips]
    cache_dir = io.cache_dir(output_dir)

    if max_workers > 1:
        with ProcessPoolExecutor(
            max_workers=max_workers, initializer=configure_logging
        ) as executor:
            process_results = list(
                executor.map(_process_clip, clips, dests, itertools.repeat(cache_dir))
            )
    else:
        process_results = [
            _process_clip(src, dest, cache_dir) for src, dest in zip(clips, dests)
        ]

    return [
        NormalisedClip(
            source_path=src,
            normalised_path=dest,
            duration_sec=duration,
            scene_cut_times=cuts,
        )
        for src, dest, (duration, cuts) in zip(clips, dests, process_results)
    ]