twotalesanimation
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""
|
|
scanner.py — Recursive EXR sequence discovery.
|
|
|
|
Uses a background QThread so the GUI stays responsive during scanning.
|
|
The scan logic itself (_scan_folder) is kept free of Qt types so it can
|
|
be tested independently.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
from PySide6.QtCore import QThread, Signal
|
|
|
|
from parser import SEQ_FILE_RE, Shot, build_shot, parse_basename
|
|
|
|
|
|
class ScanWorker(QThread):
|
|
"""Scans one or more folders for EXR sequences in a background thread.
|
|
|
|
Signals
|
|
-------
|
|
log_message(str)
|
|
Informational text suitable for display in a log window.
|
|
scan_complete(list[Shot])
|
|
Emitted when scanning finishes (always, even on error).
|
|
"""
|
|
|
|
log_message: Signal = Signal(str)
|
|
scan_complete: Signal = Signal(list)
|
|
|
|
def __init__(
|
|
self,
|
|
folders: list[str],
|
|
output_root: str,
|
|
parent=None,
|
|
) -> None:
|
|
super().__init__(parent)
|
|
self.folders = folders
|
|
self.output_root = output_root
|
|
|
|
def run(self) -> None:
|
|
shots: list[Shot] = []
|
|
try:
|
|
for folder in self.folders:
|
|
self.log_message.emit(f"Scanning: {folder}")
|
|
found = scan_folder(folder, self.output_root, self.log_message.emit)
|
|
shots.extend(found)
|
|
except Exception as exc:
|
|
self.log_message.emit(f"ERROR during scan: {exc}")
|
|
finally:
|
|
self.log_message.emit(f"Scan complete — {len(shots)} sequence(s) found.")
|
|
self.scan_complete.emit(shots)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Core scan logic (Qt-free, testable standalone)
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
def scan_folder(
|
|
root: str,
|
|
output_root: str,
|
|
log: Optional[Callable[[str], None]] = None,
|
|
) -> list[Shot]:
|
|
"""Recursively scan *root* for EXR sequences.
|
|
|
|
Parameters
|
|
----------
|
|
root:
|
|
Top-level directory to walk.
|
|
output_root:
|
|
Base path used to derive output folder/movie/exr paths on each Shot.
|
|
log:
|
|
Optional callable that receives progress strings (e.g. ``print``).
|
|
|
|
Returns
|
|
-------
|
|
list[Shot]
|
|
One Shot per discovered sequence, sorted by basename.
|
|
"""
|
|
|
|
def _log(msg: str) -> None:
|
|
if log:
|
|
log(msg)
|
|
|
|
# Collect files: dir → basename → [(frame_int, frame_str)]
|
|
dir_seqs: dict[str, dict[str, list[tuple[int, str]]]] = defaultdict(
|
|
lambda: defaultdict(list)
|
|
)
|
|
|
|
for dirpath, _dirs, files in os.walk(root):
|
|
for fname in files:
|
|
m = SEQ_FILE_RE.match(fname)
|
|
if not m:
|
|
continue
|
|
basename, frame_str = m.group(1), m.group(2)
|
|
dir_seqs[dirpath][basename].append((int(frame_str), frame_str))
|
|
|
|
shots: list[Shot] = []
|
|
|
|
for dirpath in sorted(dir_seqs):
|
|
for basename in sorted(dir_seqs[dirpath]):
|
|
parsed = parse_basename(basename)
|
|
if parsed is None:
|
|
_log(f" Skipping (unrecognised name): {basename}")
|
|
continue
|
|
|
|
frame_list = dir_seqs[dirpath][basename]
|
|
frames = [f for f, _ in frame_list]
|
|
# Detect padding length from the first filename encountered
|
|
frame_padding = len(frame_list[0][1])
|
|
|
|
_log(
|
|
f" Found: {basename} "
|
|
f"({len(frames)} frames, padding={frame_padding})"
|
|
)
|
|
|
|
shot = build_shot(
|
|
basename=basename,
|
|
source_dir=dirpath,
|
|
frames=frames,
|
|
output_root=output_root,
|
|
parsed=parsed,
|
|
frame_padding=frame_padding,
|
|
)
|
|
shots.append(shot)
|
|
|
|
return shots
|