Fix export range, transition frame counting, session restore, and video encoding

- Fix export range not covering TRANSITION folder middle frames: range max
  was based on MAIN-only file count, causing blends at sequence end to be
  silently skipped. Now uses full sequence frame count from preview table.
- Fix preview table not counting TRANSITION middle frames: these frames are
  output as symlinks in export but were shown without sequence numbers in
  preview. Now displayed as [T] entries with proper output_seq numbering.
- Fix session restore path resolution: all folder paths now .resolve()'d on
  save and restored with _resolve_lookup() fallback for both raw and resolved
  forms. Fixes folder order corruption on restore.
- Fix legacy session restore: detect pre-migration sessions (all folder_order=0)
  and fall back to symlink-derived ordering with get_all_folder_settings().
- Fix ffmpeg concat demuxer duration format: use decimal instead of fraction.
- Fix QProgressDialog false cancellation from autoReset at max value.
- Fix Export with Transitions skipping TRANSITION folders entirely while
  preview processed them, causing cutoff at blend boundaries.
- Fix Encode Video Only not finding transition-exported files in trans_dest.
- Add video encoding module (core/video.py) with concat demuxer support.
- Add direct_transition_settings DB table and persistence.
- Add sticky folder types on reorder and placeholder transition slots.
- Add blend-skipped-range counter to export completion dialog.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-11 01:35:59 +01:00
parent 78a1c8b795
commit 82a1c2ff9f
5 changed files with 1504 additions and 191 deletions

View File

@@ -8,6 +8,8 @@ from .models import (
TransitionSettings,
PerTransitionSettings,
DirectTransitionSettings,
VideoPreset,
VIDEO_PRESETS,
BlendResult,
TransitionSpec,
LinkResult,
@@ -23,6 +25,7 @@ from .models import (
from .database import DatabaseManager
from .blender import ImageBlender, TransitionGenerator, RifeDownloader, PracticalRifeEnv, FilmEnv, OPTICAL_FLOW_PRESETS
from .manager import SymlinkManager
from .video import encode_image_sequence, encode_from_file_list, find_ffmpeg
__all__ = [
'BlendCurve',
@@ -32,6 +35,8 @@ __all__ = [
'TransitionSettings',
'PerTransitionSettings',
'DirectTransitionSettings',
'VideoPreset',
'VIDEO_PRESETS',
'BlendResult',
'TransitionSpec',
'LinkResult',
@@ -51,4 +56,7 @@ __all__ = [
'FilmEnv',
'SymlinkManager',
'OPTICAL_FLOW_PRESETS',
'encode_image_sequence',
'encode_from_file_list',
'find_ffmpeg',
]

View File

@@ -90,6 +90,16 @@ class DatabaseManager:
filename TEXT NOT NULL,
UNIQUE(session_id, source_folder, filename)
);
CREATE TABLE IF NOT EXISTS direct_transition_settings (
id INTEGER PRIMARY KEY,
session_id INTEGER REFERENCES symlink_sessions(id) ON DELETE CASCADE,
after_folder TEXT NOT NULL,
frame_count INTEGER DEFAULT 16,
method TEXT DEFAULT 'film',
enabled INTEGER DEFAULT 1,
UNIQUE(session_id, after_folder)
);
""")
# Migration: add folder_type column if it doesn't exist
@@ -122,9 +132,27 @@ class DatabaseManager:
except sqlite3.OperationalError:
conn.execute("ALTER TABLE transition_settings ADD COLUMN rife_binary_path TEXT")
# Migration: add folder_order column if it doesn't exist
try:
conn.execute("SELECT folder_order FROM sequence_trim_settings LIMIT 1")
except sqlite3.OperationalError:
conn.execute("ALTER TABLE sequence_trim_settings ADD COLUMN folder_order INTEGER DEFAULT 0")
# Migration: remove overlap_frames from transition_settings (now per-transition)
# We'll keep it for backward compatibility but won't use it
def clear_session_data(self, session_id: int) -> None:
"""Delete all data for a session (symlinks, settings, etc.) but keep the session row."""
try:
with self._connect() as conn:
for table in (
'symlinks', 'sequence_trim_settings', 'transition_settings',
'per_transition_settings', 'removed_files', 'direct_transition_settings',
):
conn.execute(f"DELETE FROM {table} WHERE session_id = ?", (session_id,))
except sqlite3.Error as e:
raise DatabaseError(f"Failed to clear session data: {e}") from e
def _connect(self) -> sqlite3.Connection:
"""Create a database connection with foreign keys enabled."""
conn = sqlite3.connect(self.db_path)
@@ -328,7 +356,8 @@ class DatabaseManager:
source_folder: str,
trim_start: int,
trim_end: int,
folder_type: FolderType = FolderType.AUTO
folder_type: FolderType = FolderType.AUTO,
folder_order: int = 0,
) -> None:
"""Save trim settings for a folder in a session.
@@ -338,6 +367,7 @@ class DatabaseManager:
trim_start: Number of images to trim from start.
trim_end: Number of images to trim from end.
folder_type: The folder type (auto, main, or transition).
folder_order: Position of this folder in source_folders list.
Raises:
DatabaseError: If saving fails.
@@ -346,13 +376,14 @@ class DatabaseManager:
with self._connect() as conn:
conn.execute(
"""INSERT INTO sequence_trim_settings
(session_id, source_folder, trim_start, trim_end, folder_type)
VALUES (?, ?, ?, ?, ?)
(session_id, source_folder, trim_start, trim_end, folder_type, folder_order)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(session_id, source_folder)
DO UPDATE SET trim_start=excluded.trim_start,
trim_end=excluded.trim_end,
folder_type=excluded.folder_type""",
(session_id, source_folder, trim_start, trim_end, folder_type.value)
folder_type=excluded.folder_type,
folder_order=excluded.folder_order""",
(session_id, source_folder, trim_start, trim_end, folder_type.value, folder_order)
)
except sqlite3.Error as e:
raise DatabaseError(f"Failed to save trim settings: {e}") from e
@@ -404,6 +435,62 @@ class DatabaseManager:
return {row[0]: (row[1], row[2]) for row in rows}
def get_all_folder_settings(self, session_id: int) -> dict[str, tuple[int, int, FolderType]]:
"""Get all folder settings (trim + type) for a session, unordered.
Returns:
Dict mapping source_folder to (trim_start, trim_end, folder_type).
"""
with self._connect() as conn:
rows = conn.execute(
"""SELECT source_folder, trim_start, trim_end, folder_type
FROM sequence_trim_settings WHERE session_id = ?""",
(session_id,)
).fetchall()
result = {}
for row in rows:
try:
ft = FolderType(row[3]) if row[3] else FolderType.AUTO
except ValueError:
ft = FolderType.AUTO
result[row[0]] = (row[1], row[2], ft)
return result
def get_ordered_folders(self, session_id: int) -> list[tuple[str, FolderType, int, int]]:
"""Get all folders for a session in saved order.
Returns:
List of (source_folder, folder_type, trim_start, trim_end) sorted by folder_order.
Returns empty list if folder_order is not meaningful (all zeros from
pre-migration sessions), so the caller falls back to symlink-derived order.
"""
with self._connect() as conn:
rows = conn.execute(
"""SELECT source_folder, folder_type, trim_start, trim_end, folder_order
FROM sequence_trim_settings WHERE session_id = ?
ORDER BY folder_order""",
(session_id,)
).fetchall()
if not rows:
return []
# If all folder_order values are 0, this is a pre-migration session
# where the ordering is not meaningful — return empty to trigger
# the legacy symlink-derived ordering path.
if len(rows) > 1 and all(row[4] == 0 for row in rows):
return []
result = []
for row in rows:
try:
ft = FolderType(row[1]) if row[1] else FolderType.AUTO
except ValueError:
ft = FolderType.AUTO
result.append((row[0], ft, row[2], row[3]))
return result
def save_transition_settings(
self,
session_id: int,
@@ -669,3 +756,41 @@ class DatabaseManager:
result[folder] = set()
result[folder].add(filename)
return result
def save_direct_transition(
self,
session_id: int,
after_folder: str,
frame_count: int,
method: str,
enabled: bool,
) -> None:
"""Save direct interpolation settings for a folder transition."""
try:
with self._connect() as conn:
conn.execute(
"""INSERT INTO direct_transition_settings
(session_id, after_folder, frame_count, method, enabled)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(session_id, after_folder)
DO UPDATE SET frame_count=excluded.frame_count,
method=excluded.method,
enabled=excluded.enabled""",
(session_id, after_folder, frame_count, method, 1 if enabled else 0)
)
except sqlite3.Error as e:
raise DatabaseError(f"Failed to save direct transition: {e}") from e
def get_direct_transitions(self, session_id: int) -> list[tuple[str, int, str, bool]]:
"""Get direct interpolation settings for a session.
Returns:
List of (after_folder, frame_count, method, enabled) tuples.
"""
with self._connect() as conn:
rows = conn.execute(
"SELECT after_folder, frame_count, method, enabled "
"FROM direct_transition_settings WHERE session_id = ?",
(session_id,)
).fetchall()
return [(r[0], r[1], r[2], bool(r[3])) for r in rows]

View File

@@ -83,6 +83,29 @@ class DirectTransitionSettings:
enabled: bool = True
@dataclass
class VideoPreset:
"""Preset for video encoding via ffmpeg."""
label: str # Display name
container: str # 'mp4' or 'webm'
codec: str # ffmpeg codec: libx264, libx265, libvpx-vp9, libaom-av1
crf: int
pixel_format: str = 'yuv420p'
preset: str = 'medium' # x264/x265 speed preset
max_height: Optional[int] = None # Downscale filter
extra_args: list[str] = field(default_factory=list)
VIDEO_PRESETS: dict[str, VideoPreset] = {
'web_streaming': VideoPreset('Web Streaming', 'mp4', 'libx264', 23, preset='medium'),
'high_quality': VideoPreset('High Quality', 'mp4', 'libx264', 18, preset='slow'),
'archive': VideoPreset('Archive (H.265)', 'mp4', 'libx265', 18, preset='slow', extra_args=['-tag:v', 'hvc1']),
'social_media': VideoPreset('Social Media', 'mp4', 'libx264', 23, preset='fast', max_height=1080),
'fast_preview': VideoPreset('Fast Preview', 'mp4', 'libx264', 28, preset='ultrafast'),
'webm_vp9': VideoPreset('WebM VP9', 'webm', 'libvpx-vp9', 30, extra_args=['-b:v', '0']),
'webm_av1': VideoPreset('WebM AV1', 'webm', 'libaom-av1', 30, extra_args=['-b:v', '0', '-strict', 'experimental']),
}
@dataclass
class BlendResult:
"""Result of an image blend operation."""

259
core/video.py Normal file
View File

@@ -0,0 +1,259 @@
"""Video encoding utilities wrapping ffmpeg."""
import re
import shutil
import subprocess
import tempfile
from pathlib import Path
from typing import Callable, Optional
from .models import VideoPreset
def find_ffmpeg() -> Optional[Path]:
"""Find the ffmpeg binary on the system PATH."""
result = shutil.which('ffmpeg')
return Path(result) if result else None
def encode_image_sequence(
input_dir: Path,
output_path: Path,
fps: int,
preset: VideoPreset,
input_pattern: Optional[str] = None,
progress_callback: Optional[Callable[[int, int], bool]] = None,
total_frames: Optional[int] = None,
) -> tuple[bool, str]:
"""Encode an image sequence directory to a video file using ffmpeg.
Args:
input_dir: Directory containing sequentially named image files.
output_path: Output video file path.
fps: Frames per second.
preset: VideoPreset with codec settings.
input_pattern: ffmpeg input pattern (e.g. 'seq_%06d.png').
Auto-detected from first seq_* file if not provided.
progress_callback: Called with (current_frame, total_frames).
Return False to cancel encoding.
total_frames: Total number of frames for progress reporting.
Auto-counted from input_dir if not provided.
Returns:
(success, message) — message is output_path on success or error text on failure.
"""
ffmpeg = find_ffmpeg()
if not ffmpeg:
return False, "ffmpeg not found. Install ffmpeg to encode video."
# Auto-detect input pattern from first seq_* file
if input_pattern is None:
input_pattern = _detect_input_pattern(input_dir)
if input_pattern is None:
return False, f"No seq_* image files found in {input_dir}"
# Auto-count frames
if total_frames is None:
ext = Path(input_pattern).suffix
total_frames = len(list(input_dir.glob(f"seq_*{ext}")))
if total_frames == 0:
return False, f"No matching frames found in {input_dir}"
# Build ffmpeg command
cmd = [
str(ffmpeg), '-y',
'-framerate', str(fps),
'-i', str(input_dir / input_pattern),
'-c:v', preset.codec,
'-crf', str(preset.crf),
'-pix_fmt', preset.pixel_format,
]
# Add speed preset for x264/x265
if preset.codec in ('libx264', 'libx265'):
cmd += ['-preset', preset.preset]
# Add downscale filter if max_height is set
if preset.max_height is not None:
cmd += ['-vf', f'scale=-2:{preset.max_height}']
# Add any extra codec-specific args
if preset.extra_args:
cmd += preset.extra_args
# Progress parsing via -progress pipe:1
cmd += ['-progress', 'pipe:1']
cmd.append(str(output_path))
try:
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
cancelled = False
if proc.stdout:
for line in proc.stdout:
line = line.strip()
m = re.match(r'^frame=(\d+)', line)
if m and progress_callback is not None:
current = int(m.group(1))
if not progress_callback(current, total_frames):
cancelled = True
proc.terminate()
proc.wait()
break
proc.wait()
if cancelled:
# Clean up partial file
if output_path.exists():
output_path.unlink()
return False, "Encoding cancelled by user."
if proc.returncode != 0:
stderr = proc.stderr.read() if proc.stderr else ""
return False, f"ffmpeg exited with code {proc.returncode}:\n{stderr}"
return True, str(output_path)
except FileNotFoundError:
return False, "ffmpeg binary not found."
except Exception as e:
return False, f"Encoding error: {e}"
def _detect_input_pattern(input_dir: Path) -> Optional[str]:
"""Detect the ffmpeg input pattern from seq_* files in a directory.
Looks for files like seq_000000.png and returns a pattern like seq_%06d.png.
"""
for f in sorted(input_dir.iterdir()):
m = re.match(r'^(seq_)(\d+)(\.\w+)$', f.name)
if m:
prefix = m.group(1)
digits = m.group(2)
ext = m.group(3)
width = len(digits)
return f"{prefix}%0{width}d{ext}"
return None
def encode_from_file_list(
file_paths: list[Path],
output_path: Path,
fps: int,
preset: VideoPreset,
progress_callback: Optional[Callable[[int, int], bool]] = None,
) -> tuple[bool, str]:
"""Encode a video from an explicit list of image file paths.
Uses ffmpeg's concat demuxer so files can be scattered across directories.
Args:
file_paths: Ordered list of image file paths.
output_path: Output video file path.
fps: Frames per second.
preset: VideoPreset with codec settings.
progress_callback: Called with (current_frame, total_frames).
Return False to cancel encoding.
Returns:
(success, message) — message is output_path on success or error text on failure.
"""
ffmpeg = find_ffmpeg()
if not ffmpeg:
return False, "ffmpeg not found. Install ffmpeg to encode video."
if not file_paths:
return False, "No files provided."
total_frames = len(file_paths)
frame_duration = f"{1.0 / fps:.10f}"
# Write a concat-demuxer file listing each image with its duration
try:
concat_file = tempfile.NamedTemporaryFile(
mode='w', suffix='.txt', delete=False, prefix='vml_concat_'
)
concat_path = Path(concat_file.name)
for p in file_paths:
# Escape single quotes for ffmpeg concat format
escaped = str(p.resolve()).replace("'", "'\\''")
concat_file.write(f"file '{escaped}'\n")
concat_file.write(f"duration {frame_duration}\n")
# Repeat last file so the last frame displays for its full duration
escaped = str(file_paths[-1].resolve()).replace("'", "'\\''")
concat_file.write(f"file '{escaped}'\n")
concat_file.close()
except OSError as e:
return False, f"Failed to create concat file: {e}"
cmd = [
str(ffmpeg), '-y',
'-f', 'concat', '-safe', '0',
'-i', str(concat_path),
'-c:v', preset.codec,
'-crf', str(preset.crf),
'-pix_fmt', preset.pixel_format,
]
if preset.codec in ('libx264', 'libx265'):
cmd += ['-preset', preset.preset]
if preset.max_height is not None:
cmd += ['-vf', f'scale=-2:{preset.max_height}']
if preset.extra_args:
cmd += preset.extra_args
cmd += ['-progress', 'pipe:1']
cmd.append(str(output_path))
try:
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
cancelled = False
if proc.stdout:
for line in proc.stdout:
line = line.strip()
m = re.match(r'^frame=(\d+)', line)
if m and progress_callback is not None:
current = int(m.group(1))
if not progress_callback(current, total_frames):
cancelled = True
proc.terminate()
proc.wait()
break
proc.wait()
if cancelled:
if output_path.exists():
output_path.unlink()
return False, "Encoding cancelled by user."
if proc.returncode != 0:
stderr = proc.stderr.read() if proc.stderr else ""
return False, f"ffmpeg exited with code {proc.returncode}:\n{stderr}"
return True, str(output_path)
except FileNotFoundError:
return False, "ffmpeg binary not found."
except Exception as e:
return False, f"Encoding error: {e}"
finally:
try:
concat_path.unlink(missing_ok=True)
except OSError:
pass

File diff suppressed because it is too large Load Diff