Sequence Replacement

Utilities for replacing detected source frame sequences with reviewed MP4 files.

The public CLI commands are documented in the Usage Guide:

• renderkit replace-sequence-with-mp4
• renderkit batch-replace

Use this API reference when integrating the same audit and verification behavior into pipeline tools.

renderkit.core.sequence_replacement

Safe replacement of frame sequences with verified MP4 files.

SequenceReplacementResult dataclass

Result of a sequence replacement operation.

Source code in src/renderkit/core/sequence_replacement.py

@dataclass(frozen=True)
class SequenceReplacementResult:
    """Result of a sequence replacement operation."""

    source_pattern: str
    source_frames: list[Path]
    replacement_mp4: Path
    copied_mp4: Path
    audit_report: Path
    deleted_frames: list[Path]
    deleted_count: int
    reclaimed_bytes: int
    dry_run: bool
    verified: bool
    copied: bool

    def to_audit_record(self) -> dict[str, object]:
        """Return a JSON-serializable audit record."""
        return {
            "timestamp": datetime.now(UTC).isoformat(),
            "source_pattern": self.source_pattern,
            "source_frames": [str(path) for path in self.source_frames],
            "source_count": len(self.source_frames),
            "replacement_mp4": str(self.replacement_mp4),
            "copied_mp4": str(self.copied_mp4),
            "deleted_frames": [str(path) for path in self.deleted_frames],
            "deleted_count": self.deleted_count,
            "reclaimed_bytes": self.reclaimed_bytes,
            "dry_run": self.dry_run,
            "verified": self.verified,
            "copied": self.copied,
        }

to_audit_record()

Return a JSON-serializable audit record.

Source code in src/renderkit/core/sequence_replacement.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-serializable audit record."""
    return {
        "timestamp": datetime.now(UTC).isoformat(),
        "source_pattern": self.source_pattern,
        "source_frames": [str(path) for path in self.source_frames],
        "source_count": len(self.source_frames),
        "replacement_mp4": str(self.replacement_mp4),
        "copied_mp4": str(self.copied_mp4),
        "deleted_frames": [str(path) for path in self.deleted_frames],
        "deleted_count": self.deleted_count,
        "reclaimed_bytes": self.reclaimed_bytes,
        "dry_run": self.dry_run,
        "verified": self.verified,
        "copied": self.copied,
    }

find_exr_sequences(root)

Find EXR sequence patterns below a directory.

Parameters:

Name	Type	Description	Default
root	Path	Directory to scan recursively.	required

Returns:

Type	Description
list[str]	Detected sequence patterns sorted by path.

Source code in src/renderkit/core/sequence_replacement.py

def find_exr_sequences(root: Path) -> list[str]:
    """Find EXR sequence patterns below a directory.

    Args:
        root: Directory to scan recursively.

    Returns:
        Detected sequence patterns sorted by path.
    """
    groups: dict[tuple[Path, str, str, int], list[int]] = {}
    for file_path in root.rglob("*.exr"):
        if not file_path.is_file():
            continue
        numbered_name = split_numbered_frame_name(file_path.name, required_extension=".exr")
        if numbered_name is None:
            continue
        prefix, frame, suffix = numbered_name
        key = (file_path.parent, prefix, suffix, len(frame))
        groups.setdefault(key, []).append(int(frame))

    patterns = []
    for directory, prefix, suffix, padding in groups:
        if not groups[(directory, prefix, suffix, padding)]:
            continue
        patterns.append(str(directory / f"{prefix}%0{padding}d{suffix}"))
    return sorted(patterns)

find_replacement_mp4(sequence_pattern, mp4_dir, root=None)

Return the expected MP4 replacement path for a sequence pattern.

Parameters:

Name	Type	Description	Default
sequence_pattern	str	Detected sequence pattern.	required
mp4_dir	Path	Directory containing replacement MP4s.	required
root	Optional[Path]	Optional scan root used to match batch-convert's nested output names.	None

Returns:

Type	Description
Path	Expected MP4 path.

Source code in src/renderkit/core/sequence_replacement.py

def find_replacement_mp4(sequence_pattern: str, mp4_dir: Path, root: Optional[Path] = None) -> Path:
    """Return the expected MP4 replacement path for a sequence pattern.

    Args:
        sequence_pattern: Detected sequence pattern.
        mp4_dir: Directory containing replacement MP4s.
        root: Optional scan root used to match batch-convert's nested output names.

    Returns:
        Expected MP4 path.
    """
    if root is not None:
        sequence_path = Path(sequence_pattern)
        sequence_parts = _split_sequence_pattern_name(sequence_path.name)
        if sequence_parts is not None:
            prefix, suffix, padding = sequence_parts
            sequence = BatchSequence(
                directory=sequence_path.parent,
                prefix=prefix,
                suffix=suffix,
                padding=padding,
                frame_numbers=[0],
            )
            return build_safe_output_path(root, mp4_dir, sequence)

    pattern_name = Path(sequence_pattern).name
    stem = Path(_remove_frame_token(pattern_name)).stem.strip(" ._-")
    return mp4_dir / f"{stem}.mp4"

replace_sequence_with_mp4(input_pattern, output_mp4, *, delete_source=False, verify=False, dry_run=False, audit_report=None, verifier=verify_mp4_readable)

Replace one detected frame sequence with an MP4 and write an audit record.

Parameters:

Name	Type	Description	Default
input_pattern	str	Sequence pattern such as render.%04d.exr.	required
output_mp4	Path	MP4 file to copy into the source sequence directory.	required
delete_source	bool	Whether to delete the source frame files after replacement.	False
verify	bool	Whether to verify the MP4 with FFprobe before proceeding.	False
dry_run	bool	Whether to only report planned actions.	False
audit_report	Optional[Path]	Optional JSONL audit path.	None
verifier	Mp4Verifier	Verification callback used by tests.	verify_mp4_readable

Returns:

Type	Description
SequenceReplacementResult	Result describing the replacement.

Raises:

Type	Description
SequenceReplacementError	If verification, copy, or deletion cannot proceed safely.

Source code in src/renderkit/core/sequence_replacement.py

def replace_sequence_with_mp4(
    input_pattern: str,
    output_mp4: Path,
    *,
    delete_source: bool = False,
    verify: bool = False,
    dry_run: bool = False,
    audit_report: Optional[Path] = None,
    verifier: Mp4Verifier = verify_mp4_readable,
) -> SequenceReplacementResult:
    """Replace one detected frame sequence with an MP4 and write an audit record.

    Args:
        input_pattern: Sequence pattern such as ``render.%04d.exr``.
        output_mp4: MP4 file to copy into the source sequence directory.
        delete_source: Whether to delete the source frame files after replacement.
        verify: Whether to verify the MP4 with FFprobe before proceeding.
        dry_run: Whether to only report planned actions.
        audit_report: Optional JSONL audit path.
        verifier: Verification callback used by tests.

    Returns:
        Result describing the replacement.

    Raises:
        SequenceReplacementError: If verification, copy, or deletion cannot proceed safely.
    """
    sequence = SequenceDetector.detect_sequence(input_pattern)
    source_frames = _existing_sequence_frames(sequence)
    if not source_frames:
        raise SequenceReplacementError(f"No source frames found for pattern: {input_pattern}")

    source_mp4 = output_mp4.resolve()
    destination_mp4 = sequence.base_path / output_mp4.name
    report_path = audit_report or (sequence.base_path / _DEFAULT_AUDIT_FILENAME)
    _prepare_audit_report(report_path)
    verified = _verify_source_mp4_if_needed(
        source_mp4,
        delete_source=delete_source,
        verify=verify,
        dry_run=dry_run,
        verifier=verifier,
    )
    copied, destination_verified = _copy_replacement_mp4(
        source_mp4,
        destination_mp4,
        delete_source=delete_source,
        dry_run=dry_run,
        verifier=verifier,
    )
    verified = verified or destination_verified
    deleted_frames, reclaimed_bytes = _delete_or_plan_frames(
        source_frames,
        delete_source=delete_source,
        dry_run=dry_run,
    )

    result = SequenceReplacementResult(
        source_pattern=input_pattern,
        source_frames=source_frames,
        replacement_mp4=source_mp4,
        copied_mp4=destination_mp4,
        audit_report=report_path,
        deleted_frames=deleted_frames,
        deleted_count=len(deleted_frames) if delete_source else 0,
        reclaimed_bytes=reclaimed_bytes if delete_source else 0,
        dry_run=dry_run,
        verified=verified,
        copied=copied,
    )
    _append_audit_record(report_path, result.to_audit_record())
    return result

verify_mp4_readable(mp4_path)

Verify that an MP4 exists and can be read by FFprobe.

Parameters:

Name	Type	Description	Default
mp4_path	Path	MP4 file to verify.	required

Raises:

Type	Description
SequenceReplacementError	If the MP4 does not exist or FFprobe cannot read it.

Source code in src/renderkit/core/sequence_replacement.py

def verify_mp4_readable(mp4_path: Path) -> None:
    """Verify that an MP4 exists and can be read by FFprobe.

    Args:
        mp4_path: MP4 file to verify.

    Raises:
        SequenceReplacementError: If the MP4 does not exist or FFprobe cannot read it.
    """
    if not mp4_path.is_file():
        raise SequenceReplacementError(f"Replacement MP4 does not exist: {mp4_path}")

    ffprobe_exe = get_ffprobe_exe()
    command = [
        ffprobe_exe,
        "-v",
        "error",
        "-show_entries",
        "format=duration",
        "-of",
        "default=noprint_wrappers=1:nokey=1",
        str(mp4_path),
    ]
    try:
        # FFprobe receives fixed arguments and the MP4 path as a single argv item.
        result = subprocess.run(  # NOSONAR
            command,
            capture_output=True,
            text=True,
            check=False,
            shell=False,
            **popen_kwargs(),
        )
    except OSError as exc:
        raise SequenceReplacementError(f"Could not run ffprobe: {exc}") from exc

    if result.returncode != 0:
        details = result.stderr.strip() or result.stdout.strip() or "ffprobe returned an error"
        raise SequenceReplacementError(f"Replacement MP4 is not readable: {mp4_path} ({details})")