lasagnastack.reel_pipeline

lasagnastack.reel_pipeline

ReelPipeline

ReelPipeline(
    ingest_max_workers: int = 2,
    analyse_max_workers: int = 4,
    analyse_client: LLMClient | None = None,
    direct_client: LLMClient | None = None,
    critique_client: LLMClient | None = None,
    enhance_client: LLMClient | None = None,
    post_caption_client: LLMClient | None = None,
)

Bases: Pipeline

The seven-stage raw video clips → CapCut draft + post caption pipeline.

Initialise the pipeline with optional per-stage LLM clients.

Each LLM-backed stage accepts its own client. When None, the stage constructs a fresh GeminiClient internally.

Parameters:

Name	Type	Description	Default
ingest_max_workers	int	Parallel worker processes for Stage 1.	2
analyse_max_workers	int	Concurrent LLM calls for Stage 2.	4
analyse_client	LLMClient | None	LLM client for Stage 2 (analyse).	None
direct_client	LLMClient | None	LLM client for Stage 3 (direct).	None
critique_client	LLMClient | None	LLM client for Stage 4 (critique).	None
enhance_client	LLMClient | None	LLM client for Stage 5 (enhance).	None
post_caption_client	LLMClient | None	LLM client for Stage 7 (post caption).	None

Source code in src/lasagnastack/reel_pipeline.py

def __init__(
    self,
    ingest_max_workers: int = 2,
    analyse_max_workers: int = 4,
    analyse_client: LLMClient | None = None,
    direct_client: LLMClient | None = None,
    critique_client: LLMClient | None = None,
    enhance_client: LLMClient | None = None,
    post_caption_client: LLMClient | None = None,
) -> None:
    """Initialise the pipeline with optional per-stage LLM clients.

    Each LLM-backed stage accepts its own client. When ``None``, the stage
    constructs a fresh ``GeminiClient`` internally.

    Args:
        ingest_max_workers: Parallel worker processes for Stage 1.
        analyse_max_workers: Concurrent LLM calls for Stage 2.
        analyse_client: LLM client for Stage 2 (analyse).
        direct_client: LLM client for Stage 3 (direct).
        critique_client: LLM client for Stage 4 (critique).
        enhance_client: LLM client for Stage 5 (enhance).
        post_caption_client: LLM client for Stage 7 (post caption).
    """
    self._ingest_max_workers = ingest_max_workers
    self._analyse_max_workers = analyse_max_workers
    self._analyse_client = analyse_client
    self._direct_client = direct_client
    self._critique_client = critique_client
    self._enhance_client = enhance_client
    self._post_caption_client = post_caption_client

stages property

stages: list[Stage]

Return the seven pipeline stages in execution order.

Returns:

Type	Description
list[Stage]	Ordered list of Stage instances.

run

run(
    state: ReelPipelineState, auto_confirm: bool = False
) -> ReelPipelineState

Run all seven stages, then log and print the completion summary.

Delegates the core orchestration to Pipeline.run() and appends a structured log entry plus human-readable output paths once all stages have completed.

Parameters:

Name	Type	Description	Default
state	ReelPipelineState	Initial pipeline state.	required
auto_confirm	bool	When True, skip the interactive confirmation prompt between stages.	False

Returns:

Type	Description
ReelPipelineState	Final pipeline state after all stages complete.

Source code in src/lasagnastack/reel_pipeline.py

def run(  # pyrefly: ignore[override, bad-override]
    self, state: ReelPipelineState, auto_confirm: bool = False
) -> ReelPipelineState:
    """Run all seven stages, then log and print the completion summary.

    Delegates the core orchestration to ``Pipeline.run()`` and appends a
    structured log entry plus human-readable output paths once all stages
    have completed.

    Args:
        state: Initial pipeline state.
        auto_confirm: When ``True``, skip the interactive confirmation
            prompt between stages.

    Returns:
        Final pipeline state after all stages complete.
    """
    state = super().run(state, auto_confirm=auto_confirm)
    assert isinstance(state, ReelPipelineState)
    log.info("pipeline_complete", draft=str(state.draft_path))
    print(f"\nDraft saved to: {state.draft_path}")
    print(f"Post caption saved to: {io.post_caption_path(state.output_dir)}")
    return state

find_brief

find_brief(input_dir: Path) -> Path

Find the single .txt brief file in input_dir.

Parameters:

Name	Type	Description	Default
input_dir	Path	Directory expected to contain exactly one .txt file.	required

Returns:

Type	Description
Path	Path to the brief file.

Raises:

Type	Description
ValueError	If there is not exactly one .txt file in the directory.

Source code in src/lasagnastack/reel_pipeline.py

def find_brief(input_dir: Path) -> Path:
    """Find the single .txt brief file in input_dir.

    Args:
        input_dir: Directory expected to contain exactly one ``.txt`` file.

    Returns:
        Path to the brief file.

    Raises:
        ValueError: If there is not exactly one ``.txt`` file in the directory.
    """
    txts = list(input_dir.glob("*.txt"))
    if len(txts) != 1:
        raise ValueError(
            f"Expected exactly 1 .txt brief in {input_dir}, found {len(txts)}."
        )
    return txts[0]