"""Native clip review, Wan training and MP4 generation workspace.""" from __future__ import annotations import json import shutil from datetime import datetime from pathlib import Path from adam.ui.asset_refresh import asset_refresh from uuid import uuid4 from PySide6.QtCore import Qt, QUrl, Signal, QTimer, QItemSelectionModel, QObject, QRunnable, QThreadPool from PySide6.QtGui import QDesktopServices, QImage, QPixmap from PySide6.QtMultimedia import QMediaPlayer from PySide6.QtMultimediaWidgets import QVideoWidget from PySide6.QtWidgets import ( QAbstractItemView, QComboBox, QDialog, QFileDialog, QFormLayout, QHBoxLayout, QLabel, QLineEdit, QListWidget, QListWidgetItem, QMessageBox, QPlainTextEdit, QPushButton, QDoubleSpinBox, QSpinBox, QStackedLayout, QTabWidget, QVBoxLayout, QWidget, QInputDialog, QTreeWidget, QTreeWidgetItem, ) from adam.models import ExecutionPlan, PlanStep, JobStatus from adam.ui.settings_ui import SettingsForm from adam.video_lora import ( VIDEO_EXTENSIONS, clips_in, frames_for_seconds, generation_plan, inspect_clip, safe_name, setup_errors, trainer_root, training_plan, ) from adam.model_plugins_builtin.wan_video.manifest import TRAINING_SETTINGS, GENERATION_SETTINGS from adam.video_characters import ( Character, add_character_triggers, load_characters, remove_character, suggest_characters, upsert_character, ) from adam.ui.background import BackgroundRead from adam.ollama import OllamaClient from adam.video_captioning import draft_video_caption class _CaptionSignals(QObject): progress = Signal(int, int, str) caption_ready = Signal(str, str, str) finished = Signal(int, int) class _CaptionBatch(QRunnable): def __init__(self, client: OllamaClient, clips: list[Path], trigger_word: str, signals: _CaptionSignals): super().__init__() self.client, self.clips, self.trigger_word, self.signals = client, clips, trigger_word, signals def run(self): successes = 0 failures = 0 for index, clip in enumerate(self.clips, 1): self.signals.progress.emit(index, len(self.clips), clip.name) try: caption = draft_video_caption(self.client, clip, self.trigger_word) self.signals.caption_ready.emit(str(clip.resolve()), caption, "") successes += 1 except Exception as exc: self.signals.caption_ready.emit(str(clip.resolve()), "", str(exc)) failures += 1 self.signals.finished.emit(successes, failures) class VideoCaptionDraftDialog(QDialog): """Generate draft captions in the background and save only checked reviews.""" def __init__(self, parent, client: OllamaClient, clips: list[Path], trigger_word: str): super().__init__(parent) self.setWindowTitle("AI video caption drafts") self.resize(900, 620) self.drafts: dict[str, str] = {} self.loading_caption = False self.successes = 0 self.failures = 0 self.completed = 0 self.total = len(clips) layout = QVBoxLayout(self) intro = QLabel( "ADAM samples six frames across each selected clip. Review and edit each draft, " "then check the captions you want written beside the videos. The active training trigger is added automatically." ) intro.setWordWrap(True) layout.addWidget(intro) content = QHBoxLayout() self.items = QListWidget() self.items.setMinimumWidth(300) self.items.currentItemChanged.connect(self._select_item) content.addWidget(self.items, 1) self.caption = QPlainTextEdit() self.caption.setPlaceholderText("A generated caption will appear here for review.") self.caption.setEnabled(False) self.caption.textChanged.connect(self._caption_changed) content.addWidget(self.caption, 2) layout.addLayout(content, 1) self.status = QLabel("Starting caption requests…") self.status.setWordWrap(True) layout.addWidget(self.status) buttons = QHBoxLayout() check_all = QPushButton("Check all drafts") check_all.clicked.connect(lambda: self._set_all_checked(True)) clear_checks = QPushButton("Clear checks") clear_checks.clicked.connect(lambda: self._set_all_checked(False)) buttons.addStretch(1) self.save_button = QPushButton("Save checked captions") self.save_button.clicked.connect(self._save_checked) self.save_button.setEnabled(False) close = QPushButton("Close") close.clicked.connect(self.reject) buttons.addWidget(check_all) buttons.addWidget(clear_checks) buttons.addStretch(1) buttons.addWidget(self.save_button) buttons.addWidget(close) layout.addLayout(buttons) # Keep worker signals alive if the user closes the review dialog mid-run. self.signals = _CaptionSignals() self.signals.progress.connect(self._progress) self.signals.caption_ready.connect(self._caption_ready) self.signals.finished.connect(self._finished) QThreadPool.globalInstance().start(_CaptionBatch(client, clips, trigger_word, self.signals)) def _progress(self, index: int, total: int, name: str): self.status.setText(f"Drafting caption {index} of {total}: {name}") def _caption_ready(self, path: str, caption: str, error: str): clip = Path(path) item = QListWidgetItem(f"{clip.name} · {'Needs review' if not error else 'Could not caption'}") item.setData(Qt.UserRole, path) if error: item.setToolTip(error) self.failures += 1 else: item.setFlags(item.flags() | Qt.ItemIsUserCheckable) item.setCheckState(Qt.Unchecked) self.drafts[path] = caption self.successes += 1 self.save_button.setEnabled(True) self.completed += 1 self.items.addItem(item) if not self.caption.hasFocus(): self.items.setCurrentItem(item) self.status.setText( f"Completed {self.completed} of {self.total} · drafts ready: {self.successes} · errors: {self.failures}. " "Review completed clips now; the next clip may already be processing." ) def _finished(self, successes: int, failures: int): self.status.setText( f"Caption batch finished · drafts ready: {successes} · errors: {failures}. " "Review and edit drafts, check the captions to keep, then save." ) def _select_item(self, item, _previous): self.loading_caption = True path = str(item.data(Qt.UserRole)) if item else "" draft = self.drafts.get(path, "") self.caption.setPlainText(draft) self.caption.setEnabled(bool(draft)) self.loading_caption = False def _caption_changed(self): if self.loading_caption: return item = self.items.currentItem() if item: path = str(item.data(Qt.UserRole)) if path in self.drafts: self.drafts[path] = self.caption.toPlainText().strip() def _set_all_checked(self, checked: bool): state = Qt.Checked if checked else Qt.Unchecked for index in range(self.items.count()): item = self.items.item(index) if str(item.data(Qt.UserRole)) in self.drafts: item.setCheckState(state) def _save_checked(self): saved = 0 try: for index in range(self.items.count()): item = self.items.item(index) path = str(item.data(Qt.UserRole)) caption = self.drafts.get(path, "").strip() if item.checkState() != Qt.Checked or not caption: continue target = Path(path).with_suffix(".txt") temporary = target.with_suffix(".txt.adam-tmp") temporary.write_text(caption + "\n", encoding="utf-8") temporary.replace(target) saved += 1 self.status.setText(f"Saved {saved} reviewed caption(s). Other drafts remain unsaved.") except OSError as exc: QMessageBox.warning(self, "Captions could not be saved", f"Saved {saved} caption(s) before the error.\n\n{exc}") class CharacterEditor(QDialog): def __init__(self, parent, character: Character | None = None): super().__init__(parent) self.setWindowTitle("Video character") self.resize(520, 360) layout = QVBoxLayout(self) form = QFormLayout() self.name = QLineEdit(character.name if character else "") self.trigger = QLineEdit(character.trigger_word if character else "") form.addRow("Character name", self.name) form.addRow("Caption trigger word", self.trigger) layout.addLayout(form) layout.addWidget(QLabel("Reference images")) self.references = QListWidget() self.references.setSelectionMode(QAbstractItemView.ExtendedSelection) self.references.addItems(character.references if character else []) layout.addWidget(self.references, 1) row = QHBoxLayout() add = QPushButton("Add reference images…") add.clicked.connect(self._add_images) remove = QPushButton("Remove selected") remove.clicked.connect(lambda: [self.references.takeItem(self.references.row(item)) for item in self.references.selectedItems()]) row.addWidget(add) row.addWidget(remove) layout.addLayout(row) buttons = QHBoxLayout() buttons.addStretch(1) save = QPushButton("Save character") save.setProperty("primary", True) save.clicked.connect(self.accept) cancel = QPushButton("Cancel") cancel.clicked.connect(self.reject) buttons.addWidget(save) buttons.addWidget(cancel) layout.addLayout(buttons) def _add_images(self): paths, _ = QFileDialog.getOpenFileNames(self, "Choose character references", "", "Images (*.png *.jpg *.jpeg *.webp *.bmp)") existing = {self.references.item(i).text().casefold() for i in range(self.references.count())} for path in paths: if path.casefold() not in existing: self.references.addItem(path) def values(self): return self.name.text(), self.trigger.text(), [self.references.item(i).text() for i in range(self.references.count())] class CharacterReviewDialog(QDialog): """Edit a local character library and review multi-character clip matches.""" def __init__(self, parent, root: Path, dataset: Path): super().__init__(parent) self.root, self.dataset = root, dataset self.setWindowTitle("Video characters and recognition review") self.resize(920, 640) layout = QVBoxLayout(self) layout.addWidget(QLabel("Reference matching is a suggestion. Check names or select their rows to accept them; review each clip and multiple names can be selected.")) library_row = QHBoxLayout() self.library = QListWidget() self.library.setMaximumWidth(300) library_row.addWidget(self.library) controls = QVBoxLayout() self.add_button = QPushButton("Add character…") self.add_button.clicked.connect(self._add) self.edit_button = QPushButton("Edit selected…") self.edit_button.clicked.connect(self._edit) self.remove_button = QPushButton("Remove selected") self.remove_button.clicked.connect(self._remove) controls.addWidget(self.add_button) controls.addWidget(self.edit_button) controls.addWidget(self.remove_button) controls.addStretch(1) library_row.addLayout(controls) layout.addLayout(library_row) self.analyze_button = QPushButton("Analyze clips in current dataset") self.analyze_button.setProperty("primary", True) self.analyze_button.clicked.connect(self._analyze) analyze_row = QHBoxLayout() analyze_row.addWidget(self.analyze_button, 1) analyze_row.addWidget(QLabel("Similarity threshold")) self.threshold = QDoubleSpinBox() self.threshold.setRange(0.10, 0.95) self.threshold.setSingleStep(0.02) self.threshold.setDecimals(2) self.threshold.setValue(0.62) self.threshold.setToolTip("Lower it to see more possible matches, or raise it to reduce suggestions. Review every result.") analyze_row.addWidget(self.threshold) layout.addLayout(analyze_row) self.results = QTreeWidget() self.results.setHeaderLabels(["Clip and suggested character", "Similarity", "Review"]) self.results.setColumnWidth(0, 540) self.results.itemDoubleClicked.connect(self._open_clip) layout.addWidget(self.results, 1) open_row = QHBoxLayout() check_all = QPushButton("Check all suggestions") check_all.clicked.connect(lambda: self._set_suggestions_checked(True)) clear_checks = QPushButton("Clear checks") clear_checks.clicked.connect(lambda: self._set_suggestions_checked(False)) open_clip = QPushButton("Open selected clip") open_clip.clicked.connect(self._open_clip) open_row.addWidget(check_all) open_row.addWidget(clear_checks) open_row.addWidget(open_clip) open_row.addStretch(1) layout.addLayout(open_row) self.status = QLabel("Add characters and reference images, then analyze this dataset.") self.status.setWordWrap(True) layout.addWidget(self.status) buttons = QHBoxLayout() buttons.addStretch(1) self.apply_button = QPushButton("Apply checked names to captions") self.apply_button.clicked.connect(self._apply) close = QPushButton("Close") close.clicked.connect(self.reject) buttons.addWidget(self.apply_button) buttons.addWidget(close) layout.addLayout(buttons) self.characters: list[Character] = [] self._refresh_library() self.background = BackgroundRead(self, max_age=0) self.background.ready.connect(self._show_results) self.background.failed.connect(self._analysis_failed) def _refresh_library(self): self.characters = load_characters(self.root) self.library.clear() for character in self.characters: item = QListWidgetItem(f"{character.name} · {character.trigger_word} · {len(character.references)} refs") item.setData(Qt.UserRole, character.id) self.library.addItem(item) self.edit_button.setEnabled(bool(self.characters)) self.remove_button.setEnabled(bool(self.characters)) def _save_editor(self, character: Character | None = None): editor = CharacterEditor(self, character) if editor.exec() != QDialog.Accepted: return name, trigger, refs = editor.values() try: upsert_character(self.root, name, trigger, refs, character_id=character.id if character else "") self._refresh_library() self.status.setText("Character saved. Analyze clips to get reviewable suggestions.") except (OSError, ValueError) as exc: QMessageBox.warning(self, "Character could not be saved", str(exc)) def _selected_character(self): item = self.library.currentItem() return next((character for character in self.characters if item and character.id == item.data(Qt.UserRole)), None) def _add(self): self._save_editor() def _edit(self): character = self._selected_character() if character: self._save_editor(character) def _remove(self): character = self._selected_character() if character: remove_character(self.root, character.id) self._refresh_library() def _analyze(self): if not self.characters: QMessageBox.information(self, "Add characters first", "Add a character and one or more reference images before analyzing clips.") return self.analyze_button.setEnabled(False) self.apply_button.setEnabled(False) self.status.setText("Sampling clips and comparing reference images locally. The first run may download DINOv2.") threshold = self.threshold.value() self.background.request(lambda: suggest_characters(self.root, self.dataset, threshold=threshold), force=True) def _analysis_failed(self, error: str): self.analyze_button.setEnabled(True) self.apply_button.setEnabled(False) self.status.setText("Recognition could not finish.") QMessageBox.warning(self, "Character analysis failed", error) def _show_results(self, results): self.results.clear() matches = 0 for result in results: clip = Path(result["clip"]) parent = QTreeWidgetItem([clip.name, "", "No match suggested" if not result["suggestions"] else "Review suggestions"]) parent.setData(0, Qt.UserRole, str(clip)) self.results.addTopLevelItem(parent) for suggestion in result["suggestions"]: child = QTreeWidgetItem([suggestion["name"], f"{suggestion['score']:.3f}", "Include in caption"]) child.setData(0, Qt.UserRole, suggestion["trigger_word"]) child.setFlags(child.flags() | Qt.ItemIsUserCheckable) child.setCheckState(0, Qt.Unchecked) parent.addChild(child) matches += 1 parent.setExpanded(True) self.analyze_button.setEnabled(True) self.apply_button.setEnabled(matches > 0) self.status.setText(f"Reviewed {len(results)} clips · {matches} character suggestion(s). Similarity is a ranking aid, not a certainty score.") def _open_clip(self, item=None, _column=0): selected = item or self.results.currentItem() while selected and selected.parent(): selected = selected.parent() if selected: QDesktopServices.openUrl(QUrl.fromLocalFile(str(selected.data(0, Qt.UserRole)))) def _set_suggestions_checked(self, checked: bool): state = Qt.Checked if checked else Qt.Unchecked for row in range(self.results.topLevelItemCount()): parent = self.results.topLevelItem(row) for child in range(parent.childCount()): parent.child(child).setCheckState(0, state) def _apply(self): changed = 0 already_present = 0 accepted = 0 try: for index in range(self.results.topLevelItemCount()): parent = self.results.topLevelItem(index) clip = Path(parent.data(0, Qt.UserRole)) triggers = [parent.child(child).data(0, Qt.UserRole) for child in range(parent.childCount()) if parent.child(child).checkState(0) == Qt.Checked or parent.child(child).isSelected()] if not triggers: continue accepted += len(triggers) known = {item.trigger_word.casefold(): item for item in self.characters} selected = [known[trigger.casefold()] for trigger in triggers if trigger.casefold() in known] caption_path = clip.with_suffix(".txt") caption = caption_path.read_text(encoding="utf-8-sig") if caption_path.is_file() else "" updated = add_character_triggers(caption, selected) if updated != caption.strip(): temporary = caption_path.with_suffix(".txt.adam-tmp") temporary.write_text(updated + "\n", encoding="utf-8") temporary.replace(caption_path) changed += 1 else: already_present += 1 if not accepted: self.status.setText("Nothing selected. Check suggestions or click a character suggestion row, then apply.") else: self.status.setText( f"Accepted {accepted} character suggestion(s): updated {changed} caption(s); " f"{already_present} already contained those names. Existing descriptions were kept." ) except (OSError, UnicodeError) as exc: QMessageBox.warning(self, "Captions could not be updated", str(exc)) class CaptionEditor(QPlainTextEdit): """A multiline editor with an explicit save-and-advance shortcut.""" save_next_requested = Signal() def keyPressEvent(self, event): if event.key() in {Qt.Key_Return, Qt.Key_Enter} and event.modifiers() & Qt.ShiftModifier: event.accept() self.save_next_requested.emit() return super().keyPressEvent(event) class VideoLoraPage(QWidget): open_jobs_requested = Signal() def __init__(self, root, registry, jobs, assets, config): super().__init__() self.root, self.registry, self.jobs, self.assets, self.config = root, registry, jobs, assets, config self.current_clip: Path | None = None self.active_id: str | None = None self.refreshing = False self.preview_original: QPixmap | None = None self.preview_autoplay = False layout = QVBoxLayout(self) layout.setContentsMargins(22, 18, 22, 18) title = QLabel("Video LoRA") title.setObjectName("PageTitle") layout.addWidget(title) self.setup_status = QLabel() self.setup_status.setWordWrap(True) layout.addWidget(self.setup_status) self.tabs = QTabWidget() layout.addWidget(self.tabs, 1) self._dataset_tab() self._training_tab() self._generation_tab() self._history_tab() footer = QHBoxLayout() self.status = QLabel("Ready") self.status.setWordWrap(True) footer.addWidget(self.status, 1) self.pause_button = QPushButton("Pause / resume") self.pause_button.clicked.connect(self._pause) self.stop_button = QPushButton("Stop") self.stop_button.clicked.connect(self._stop) history = QPushButton("Jobs / approval / logs") history.clicked.connect(self.open_jobs_requested) footer.addWidget(self.pause_button) footer.addWidget(self.stop_button) footer.addWidget(history) layout.addLayout(footer) self.save_timer = QTimer(self) self.save_timer.setSingleShot(True) self.save_timer.setInterval(500) self.save_timer.timeout.connect(self._save) self._restore() self.training.changed.connect(self._schedule_save) self.training.changed.connect(self._update_trigger_labels) self.generation.changed.connect(self._generation_changed) for edit in (self.dataset_folder, self.model_name, self.resume_path, self.output_folder): edit.editingFinished.connect(self._schedule_save) self.epochs.valueChanged.connect(self._schedule_save) self.tabs.currentChanged.connect(self._tab_changed) self.jobs.job_updated.connect(self._job_updated) if hasattr(self.jobs, "active_changed"): self.jobs.active_changed.connect(lambda _job: self._job_controls()) self.refresh() self._update_trigger_labels() self._generation_changed() self._job_controls() def _path_row(self, layout, edit, label, *, file=False, callback=None): row = QHBoxLayout() row.addWidget(QLabel(label)) row.addWidget(edit, 1) button = QPushButton("Browse") def browse(): selected = (QFileDialog.getOpenFileName(self, label, edit.text(), "LoRA (*.safetensors)")[0] if file else QFileDialog.getExistingDirectory(self, label, edit.text())) if selected: edit.setText(selected) if callback: callback() self._schedule_save() button.clicked.connect(browse) row.addWidget(button) layout.addLayout(row) def _dataset_tab(self): tab = QWidget() layout = QVBoxLayout(tab) self.dataset_folder = QLineEdit() self._path_row(layout, self.dataset_folder, "Clip folder", callback=self._refresh_clips) self.dataset_folder.editingFinished.connect(self._refresh_clips) saved_row = QHBoxLayout() saved_row.addWidget(QLabel("Saved datasets")) self.saved_datasets = QComboBox() self.saved_datasets.setToolTip("Recently used and ADAM video datasets. Choose one to make it the active clip folder.") self.saved_datasets.currentIndexChanged.connect(self._select_saved_dataset) saved_row.addWidget(self.saved_datasets, 1) reload_saved = QPushButton("Reload") reload_saved.setToolTip("Find video datasets created in ADAM's video dataset folders.") reload_saved.clicked.connect(self._refresh_dataset_choices) saved_row.addWidget(reload_saved) layout.addLayout(saved_row) buttons = QHBoxLayout() self.import_button = None for label, action in (("New dataset…", self._new_dataset), ("Archive current & new…", self._archive_and_new_dataset), ("Import clips…", self._import_clips), ("Split long video…", self._split), ("Validate dataset", self._validate), ("Characters & recognition…", self._characters), ("AI draft captions…", self._ai_draft_captions), ("Refresh", self._refresh_clips)): button = QPushButton(label) if label == "Import clips…": self.import_button = button elif label == "Archive current & new…": self.archive_dataset_button = button button.clicked.connect(action) buttons.addWidget(button) layout.addLayout(buttons) content = QHBoxLayout() self.clips = QListWidget() self.clips.setSelectionMode(QAbstractItemView.ExtendedSelection) self.clips.setToolTip("Select several clips with Ctrl-click or Shift-click, then apply one caption to all of them.") self.clips.currentItemChanged.connect(self._clip_changed) content.addWidget(self.clips, 1) detail = QVBoxLayout() preview = QWidget() preview.setMinimumSize(280, 160) self.preview_stack = QStackedLayout(preview) self.preview_frame = QLabel("Select a clip to preview it.") self.preview_frame.setAlignment(Qt.AlignCenter) self.preview_frame.setWordWrap(True) self.video = QVideoWidget() self.preview_stack.addWidget(self.preview_frame) self.preview_stack.addWidget(self.video) self.player = QMediaPlayer(self) self.player.setVideoOutput(self.video) self.player.setLoops(QMediaPlayer.Infinite) self.player.mediaStatusChanged.connect(self._preview_media_status) self.player.errorOccurred.connect(lambda _error, text: self.status.setText(f"Preview: {text}. You can open the clip in your default video player.")) detail.addWidget(preview, 1) playback = QHBoxLayout() play = QPushButton("Play / pause preview") play.clicked.connect(self._play_pause) external = QPushButton("Open clip externally") external.clicked.connect(lambda: self._open(self.current_clip)) playback.addWidget(play) playback.addWidget(external) detail.addLayout(playback) self.clip_info = QLabel("Select a clip to inspect its duration and caption.") self.clip_info.setWordWrap(True) detail.addWidget(self.clip_info) self.trigger_hint = QLabel() self.trigger_hint.setWordWrap(True) self.trigger_hint.setProperty("muted", True) detail.addWidget(self.trigger_hint) self.caption_label = QLabel() detail.addWidget(self.caption_label) self.caption = CaptionEditor() self.caption.setMaximumHeight(110) self.caption.setPlaceholderText("Describe what is visible and what moves in this clip. Shift+Enter saves and opens the next clip.") detail.addWidget(self.caption) self.caption.save_next_requested.connect(self._save_caption_and_next) save = QPushButton("Save caption") save.clicked.connect(self._save_caption) detail.addWidget(save) self.apply_caption_button = QPushButton("Apply caption to selected clips") self.apply_caption_button.setToolTip("Writes the caption above to every selected clip, replacing any existing captions.") self.apply_caption_button.clicked.connect(self._apply_caption_to_selected) detail.addWidget(self.apply_caption_button) content.addLayout(detail, 2) layout.addLayout(content, 1) note = QLabel("Start with short continuous shots. Training uses paired video/.txt files in this folder. Press Shift+Enter in a caption to save it and move to the next clip. Ctrl-click or Shift-click clips, then use Apply caption to selected clips for batch captions. New dataset creates a separate folder. Archive current & new keeps the old dataset available from Saved datasets.") note.setWordWrap(True) layout.addWidget(note) self.tabs.addTab(tab, "Dataset") def _training_tab(self): tab = QWidget() layout = QVBoxLayout(tab) form = QFormLayout() self.model_name = QLineEdit("my_video_lora") self.epochs = QSpinBox() self.epochs.setRange(1, 100000) self.epochs.setValue(12) form.addRow("New run name", self.model_name) form.addRow("Epochs", self.epochs) layout.addLayout(form) self.output_folder = QLineEdit() self.output_folder.setPlaceholderText("Automatic: ADAM/data/video_models/") self._path_row(layout, self.output_folder, "New output folder") self.resume_path = QLineEdit() self.resume_path.setPlaceholderText("Optional: continue an existing Wan LoRA into a new run") self._path_row(layout, self.resume_path, "Continue weights", file=True) self.training = SettingsForm(TRAINING_SETTINGS) self.training.scroll.setMaximumHeight(16777215) self.training.layout().setStretch(1, 1) layout.addWidget(self.training, 1) note = QLabel("Preset engine: Wan 2.1 T2V 1.3B, BF16, FP8 base, AdamW8bit and gradient checkpointing. Continuation starts a fresh optimizer; checkpoint rank and alpha take precedence. Each run gets a separate cache.") note.setWordWrap(True) layout.addWidget(note) train = QPushButton("Review full training pipeline") train.setProperty("primary", True) train.clicked.connect(self._train) layout.addWidget(train) self.tabs.addTab(tab, "Training") def _generation_tab(self): tab = QWidget() layout = QVBoxLayout(tab) row = QHBoxLayout() row.addWidget(QLabel("Wan checkpoint")) self.models = QComboBox() row.addWidget(self.models, 1) browse = QPushButton("Browse…") browse.clicked.connect(self._browse_model) row.addWidget(browse) refresh = QPushButton("Refresh models") refresh.clicked.connect(lambda: self.refresh(force=True)) row.addWidget(refresh) layout.addLayout(row) self.generation = SettingsForm(GENERATION_SETTINGS) self.generation.scroll.setMaximumHeight(16777215) self.generation.layout().setStretch(1, 1) layout.addWidget(self.generation, 1) self.duration_hint = QLabel() self.duration_hint.setWordWrap(True) layout.addWidget(self.duration_hint) buttons = QHBoxLayout() fast = QPushButton("Fast preview preset") fast.clicked.connect(lambda: self.generation.set_values({"duration": 2.0, "fps": 12, "steps": 12, "blocks_to_swap": 24})) buttons.addWidget(fast) generate = QPushButton("Generate video") generate.setProperty("primary", True) generate.clicked.connect(self._generate) buttons.addWidget(generate) layout.addLayout(buttons) self.tabs.addTab(tab, "Generate") def _history_tab(self): tab = QWidget() layout = QVBoxLayout(tab) self.videos = QListWidget() self.videos.itemDoubleClicked.connect(lambda item: self._open(Path(item.data(Qt.UserRole)))) self.videos.currentItemChanged.connect(self._show_video_details) layout.addWidget(self.videos, 1) self.video_details = QLabel("Generated videos retain their prompt, seed, checkpoint and settings. Double-click a take to play it.") self.video_details.setWordWrap(True) layout.addWidget(self.video_details) open_video = QPushButton("Play selected video") open_video.clicked.connect(lambda: self._open(Path(self.videos.currentItem().data(Qt.UserRole))) if self.videos.currentItem() else None) layout.addWidget(open_video) self.tabs.addTab(tab, "Videos / takes") def _schedule_save(self, *_args): if hasattr(self, "save_timer"): self.save_timer.start() def _update_trigger_labels(self, *_args): if not hasattr(self, "training") or not hasattr(self, "trigger_hint"): return trigger = str(self.training.value("trigger_word") or "").strip() shown = trigger or "not set" self.trigger_hint.setText(f"Active training trigger: {shown} · You can change it in the Training tab or when splitting a video.") self.caption_label.setText(f"Visual caption (must describe the clip; ADAM expects the exact trigger “{shown}”)") def _save(self): self._remember_dataset(self.dataset_folder.text()) self.config.update({"video_lora_settings": {"dataset": self.dataset_folder.text(), "dataset_history": self.dataset_history, "model_name": self.model_name.text(), "epochs": self.epochs.value(), "output": self.output_folder.text(), "resume": self.resume_path.text(), "training": self.training.values(), "generation": self.generation.values(), "model": self.models.currentData() or ""}}) def _restore(self): saved = self.config.get("video_lora_settings", {}) external = trainer_root(self.root, self.config) if not saved and external: try: old = json.loads((external / "app_settings.json").read_text(encoding="utf-8")) except (OSError, ValueError): old = {} saved = {"dataset": str(external / "dataset" / "videos"), "model_name": old.get("run_name", "my_video_lora"), "epochs": old.get("epochs", 12), "model": old.get("generation_checkpoint", ""), "training": {}, "generation": {}} for old_key, key in (("trigger", "trigger_word"), ("rank", "rank"), ("alpha", "alpha"), ("blocks", "blocks_to_swap"), ("learning_rate", "learning_rate")): if old_key in old: saved["training"][key] = old[old_key] for old_key, key in (("generation_prompt", "prompt"), ("generation_duration", "duration"), ("generation_fps", "fps"), ("generation_steps", "steps"), ("generation_strength", "lora_strength"), ("generation_seed", "seed")): if old_key in old: saved["generation"][key] = old[old_key] if str(old.get("generation_resolution", "")).startswith("Portrait"): saved["generation"]["format"] = "Portrait 480x832" if not isinstance(saved, dict): saved = {} history = saved.get("dataset_history", []) self.dataset_history = [str(path) for path in history if isinstance(path, str) and path.strip()] if isinstance(history, list) else [] self.dataset_folder.setText(str(saved.get("dataset", ""))) self._remember_dataset(self.dataset_folder.text()) self.model_name.setText(str(saved.get("model_name", "my_video_lora"))) try: self.epochs.setValue(int(saved.get("epochs", 12))) except (ValueError, TypeError): pass self.output_folder.setText(str(saved.get("output", ""))) self.resume_path.setText(str(saved.get("resume", ""))) self.training.set_values(saved.get("training", {})) self.generation.set_values(saved.get("generation", {})) self.saved_model = str(saved.get("model", "")) def refresh(self, *, force=False): if not hasattr(self, "_asset_refresh"): self._asset_refresh = asset_refresh(self.assets, self.config) self._asset_refresh.changed.connect(self._refresh_models) self._asset_refresh.request(force=force) self._refresh_models() self._refresh_clips() self._refresh_dataset_choices() self._refresh_videos() def _refresh_models(self): errors = setup_errors(trainer_root(self.root, self.config)) self.setup_status.setText("Ready: connected LoRAVideoTrainer · Wan 2.1 T2V 1.3B" if not errors else "Setup needed: " + " · ".join(errors[:3])) selected = self.models.currentData() or self.saved_model self.models.clear() for asset in self.assets.assets: if asset.kind == "model" and asset.trainer == "wan_video" and Path(asset.path).is_file(): self.models.addItem(asset.name, asset.path) if selected and self.models.findData(selected) < 0 and Path(selected).is_file(): self.models.addItem(Path(selected).stem, selected) index = self.models.findData(selected) if index >= 0: self.models.setCurrentIndex(index) def _video_dataset_root(self) -> Path: return self.root / "ADAM_Datasets" / "Video" def _remember_dataset(self, raw_path: str) -> None: if not raw_path.strip(): return try: path = str(Path(raw_path).expanduser().resolve()) except OSError: return prior = getattr(self, "dataset_history", []) self.dataset_history = [path] + [item for item in prior if item.casefold() != path.casefold()][:19] def _dataset_display_name(self, path: Path) -> str: try: relative = path.relative_to(self._video_dataset_root()) return f"ADAM Video / {relative}" if relative.parts else "ADAM Video" except ValueError: return f"{path.name or path.drive} — {path.parent}" def _refresh_dataset_choices(self) -> None: """Show recent folders and ADAM-owned datasets without changing the active one.""" current = self.dataset_folder.text().strip() candidates = list(getattr(self, "dataset_history", [])) if current: candidates.insert(0, current) base = self._video_dataset_root() if base.is_dir(): candidates.append(str(base)) candidates.extend(str(folder) for folder in base.iterdir() if folder.is_dir()) archive = base / "Archived" if archive.is_dir(): candidates.extend(str(folder) for folder in archive.iterdir() if folder.is_dir()) seen: set[str] = set() paths: list[Path] = [] for raw_path in candidates: try: path = Path(raw_path).expanduser().resolve() except OSError: continue key = str(path).casefold() if key not in seen and path.is_dir(): seen.add(key) paths.append(path) self.saved_datasets.blockSignals(True) self.saved_datasets.clear() self.saved_datasets.addItem("Choose a saved dataset…", "") for path in paths: self.saved_datasets.addItem(self._dataset_display_name(path), str(path)) index = self.saved_datasets.findData(str(Path(current).expanduser().resolve())) if current else 0 self.saved_datasets.setCurrentIndex(index if index >= 0 else 0) self.saved_datasets.blockSignals(False) def _select_saved_dataset(self, index: int) -> None: selected = self.saved_datasets.itemData(index) if not selected or str(selected) == self.dataset_folder.text().strip(): return if not self._save_caption(): return self.dataset_folder.setText(str(selected)) self._remember_dataset(str(selected)) self._refresh_clips() self._schedule_save() def _activate_dataset(self, folder: Path) -> None: self.dataset_folder.setText(str(folder.resolve())) self._remember_dataset(self.dataset_folder.text()) self._refresh_clips() self._refresh_dataset_choices() self._save() def _ask_dataset_name(self, title: str, prompt: str, default: str) -> str | None: name, accepted = QInputDialog.getText(self, title, prompt, text=default) if not accepted: return None try: return safe_name(name) except ValueError as exc: QMessageBox.warning(self, "Dataset name needs attention", str(exc).replace("model name", "dataset name")) return None def _new_dataset(self) -> None: name = self._ask_dataset_name("New video dataset", "Dataset folder name:", "new_video_dataset") if not name: return folder = self._video_dataset_root() / name if folder.exists(): QMessageBox.warning(self, "Dataset already exists", "Choose a different dataset name or select the saved dataset.") return try: folder.mkdir(parents=True) self._activate_dataset(folder) except OSError as exc: QMessageBox.warning(self, "Dataset could not be created", str(exc)) def _archive_and_new_dataset(self) -> None: if not self._save_caption(): return raw_source = self.dataset_folder.text().strip() if not raw_source: self._new_dataset() return source = Path(raw_source).expanduser().resolve() if not source.is_dir() or source == source.parent or source == self.root.resolve(): QMessageBox.warning(self, "Dataset cannot be archived", "Choose a specific video dataset folder first.") return active = self.jobs.active_job if active and any( step.tool_id == "wan_video_trainer" and Path(str(step.arguments.get("dataset_dir", ""))).expanduser().resolve() == source for step in active.plan.steps ): QMessageBox.warning(self, "Dataset is in use", "Stop or wait for the active training job before archiving this dataset.") return stamp = datetime.now().strftime("%Y%m%d-%H%M%S") archive_name = self._ask_dataset_name("Archive current dataset", "Name for the saved dataset:", f"{source.name}_{stamp}") if not archive_name: return new_name = self._ask_dataset_name("New video dataset", "New dataset folder name:", "new_video_dataset") if not new_name: return archive = self._video_dataset_root() / "Archived" / archive_name new_folder = self._video_dataset_root() / new_name if archive.exists() or new_folder.exists(): QMessageBox.warning(self, "Folder already exists", "Choose unique names for both the archived and new datasets.") return confirm = QMessageBox.question( self, "Archive current dataset?", f"Move the contents of\n{source}\ninto\n{archive}\n\nand start with an empty dataset? Nothing will be deleted.", QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if confirm != QMessageBox.Yes: return try: archive.mkdir(parents=True) archive_root = archive.parent.resolve() for item in source.iterdir(): # When the active folder is ADAM_Datasets/Video itself, do not move # its archive container into the archive being created inside it. if item.resolve() == archive_root: continue shutil.move(str(item), str(archive / item.name)) new_folder.mkdir(parents=True) self._remember_dataset(str(archive)) self._activate_dataset(new_folder) self.status.setText(f"Archived {source.name} as {archive_name}. Ready for a new dataset.") except OSError as exc: QMessageBox.warning(self, "Dataset could not be archived", f"Some files may already be in {archive}.\n\n{exc}") def _refresh_clips(self): if not self._save_caption(): return selected = str(self.current_clip) if self.current_clip else "" self.refreshing = True self.clips.clear() for path in clips_in(Path(self.dataset_folder.text())) if self.dataset_folder.text().strip() else []: item = QListWidgetItem(path.name + (" • needs caption" if not path.with_suffix(".txt").is_file() else "")) item.setData(Qt.UserRole, str(path)) self.clips.addItem(item) self.refreshing = False self.current_clip = None self.caption.clear() self.preview_autoplay = False self.player.stop() self.preview_original = None self.preview_frame.setPixmap(QPixmap()) self.preview_frame.setText("Select a clip to preview it.") self.preview_stack.setCurrentWidget(self.preview_frame) for i in range(self.clips.count()): if self.clips.item(i).data(Qt.UserRole) == selected: self.clips.setCurrentRow(i) break def _clip_changed(self, item, _previous): if self.refreshing: return if not self._save_caption(): self.clips.blockSignals(True) self.clips.setCurrentItem(_previous) self.clips.blockSignals(False) return self.preview_autoplay = False self.player.stop() self.current_clip = Path(item.data(Qt.UserRole)) if item else None self.caption.clear() if self.current_clip: try: path = self.current_clip.with_suffix(".txt") self.caption.setPlainText(path.read_text(encoding="utf-8-sig") if path.is_file() else "") self.caption.document().setModified(False) info = inspect_clip(self.current_clip) self.clip_info.setText(f"{info['width']}×{info['height']} · {info['duration']:.2f}s · {info['fps']:.2f} FPS · {info['frames']} frames") self._show_clip_still(self.current_clip) self.preview_autoplay = True self.player.setSource(QUrl.fromLocalFile(str(self.current_clip))) except (OSError, ValueError) as exc: self.clip_info.setText(str(exc)) self.preview_frame.setPixmap(QPixmap()) self.preview_frame.setText("Preview unavailable for this clip.") self.preview_stack.setCurrentWidget(self.preview_frame) self._job_controls() def _show_clip_still(self, clip: Path): """Show an immediate first-frame preview while the media player prepares.""" try: import cv2 capture = cv2.VideoCapture(str(clip)) try: available, frame = capture.read() finally: capture.release() if not available: raise ValueError("Could not read a preview frame") height, width = frame.shape[:2] image = QImage(frame.data, width, height, frame.strides[0], QImage.Format_BGR888).copy() self.preview_original = QPixmap.fromImage(image) self._scale_preview_still() self.preview_frame.setText("") except (ImportError, OSError, ValueError): self.preview_original = None self.preview_frame.setPixmap(QPixmap()) self.preview_frame.setText("Preparing video preview…") self.preview_stack.setCurrentWidget(self.preview_frame) def _scale_preview_still(self): if self.preview_original: self.preview_frame.setPixmap(self.preview_original.scaled( self.preview_frame.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation)) def _preview_media_status(self, status): if status == QMediaPlayer.LoadedMedia and self.preview_autoplay and self.current_clip: self.preview_stack.setCurrentWidget(self.video) self.player.play() def _save_caption(self): if self.current_clip and self.caption.document().isModified(): try: path = self.current_clip.with_suffix(".txt") temp = path.with_suffix(".txt.adam-tmp") temp.write_text(self.caption.toPlainText().strip() + "\n", encoding="utf-8") temp.replace(path) self.caption.document().setModified(False) self._mark_caption_saved(self.current_clip) except OSError as exc: QMessageBox.warning(self, "Caption could not be saved", str(exc)) return False return True def _mark_caption_saved(self, clip: Path) -> None: for index in range(self.clips.count()): item = self.clips.item(index) if item.data(Qt.UserRole) == str(clip): item.setText(clip.name) return def _save_caption_and_next(self) -> None: """Support uninterrupted captioning from the keyboard.""" if not self._save_caption(): return row = self.clips.currentRow() if row < 0: return next_row = row + 1 if next_row >= self.clips.count(): self.status.setText("Caption saved. This is the last clip in the dataset.") return self.clips.setCurrentRow(next_row, QItemSelectionModel.ClearAndSelect) self.caption.setFocus() self.status.setText(f"Caption saved. Ready for clip {next_row + 1} of {self.clips.count()}.") def _apply_caption_to_selected(self) -> None: selected = self.clips.selectedItems() caption = self.caption.toPlainText().strip() if not selected: QMessageBox.information(self, "Select clips", "Select one or more clips, then enter the caption to apply.") return if not caption: QMessageBox.warning(self, "Caption needed", "Enter a caption before applying it to the selected clips.") return saved = 0 try: for item in selected: clip = Path(item.data(Qt.UserRole)) path = clip.with_suffix(".txt") temp = path.with_suffix(".txt.adam-tmp") temp.write_text(caption + "\n", encoding="utf-8") temp.replace(path) self._mark_caption_saved(clip) saved += 1 if self.current_clip: self.caption.document().setModified(False) self.status.setText(f"Applied the caption to {saved} clip{'s' if saved != 1 else ''}.") except OSError as exc: QMessageBox.warning(self, "Captions could not be saved", f"Saved {saved} clip{'s' if saved != 1 else ''} before the error.\n\n{exc}") def _play_pause(self): if self.player.playbackState() == QMediaPlayer.PlayingState: self.player.pause() else: self.preview_autoplay = True self.preview_stack.setCurrentWidget(self.video) self.player.play() def resizeEvent(self, event): self._scale_preview_still() super().resizeEvent(event) def _tab_changed(self, *_args): self._save_caption() self.player.pause() self._save() def hideEvent(self, event): self._save_caption() self.player.pause() self._save() super().hideEvent(event) def _import_clips(self): destination = self.dataset_folder.text().strip() if not destination: destination = QFileDialog.getExistingDirectory(self, "Choose a dataset folder") if not destination: return self.dataset_folder.setText(destination) paths, _ = QFileDialog.getOpenFileNames(self, "Import short clips", "", "Videos (*.mp4 *.mov *.mkv *.webm *.m4v *.avi)") try: for raw in paths: source = Path(raw) target = Path(destination) / source.name if source.resolve() == target.resolve(): continue if target.exists() or target.with_suffix(".txt").exists() or any(target.with_suffix(ext).exists() for ext in VIDEO_EXTENSIONS): target = target.with_name(f"{target.stem}_{uuid4().hex[:8]}{target.suffix}") shutil.copy2(source, target) if source.with_suffix(".txt").is_file(): shutil.copy2(source.with_suffix(".txt"), target.with_suffix(".txt")) self._refresh_clips() self._save() except OSError as exc: QMessageBox.warning(self, "Import failed", str(exc)) def _submit(self, plan): self._save() job = self.jobs.submit(plan) self.active_id = job.id self._job_updated(job) if job.status == JobStatus.AWAITING_CONFIRMATION: self.open_jobs_requested.emit() def _train(self): if not self._save_caption(): return try: plan = training_plan(self.root, self.dataset_folder.text(), self.model_name.text(), self.epochs.value(), self.training.values(), self.resume_path.text(), self.output_folder.text()) self._submit(plan) except (OSError, ValueError) as exc: QMessageBox.warning(self, "Training needs attention", str(exc)) def _validate(self): if not self._save_caption(): return self._submit(ExecutionPlan("Validate video clips", "Validate every clip and caption.", [PlanStep("wan_video_dataset", "Validate video clips", "Check paired clips and captions.", {"operation": "validate", "dataset_dir": self.dataset_folder.text(), "trigger_word": self.training.value("trigger_word")})], project_name="Video LoRA")) def _characters(self): raw = self.dataset_folder.text().strip() if not raw or not Path(raw).is_dir(): QMessageBox.information(self, "Choose a video dataset", "Select an existing clip folder before managing characters or running recognition.") return dialog = CharacterReviewDialog(self, self.root, Path(raw).expanduser().resolve()) dialog.exec() def _ai_draft_captions(self): if not self._save_caption(): return clips = [Path(item.data(Qt.UserRole)) for item in self.clips.selectedItems()] if not clips: QMessageBox.information( self, "Select clips first", "Select one or more clips in the Dataset list, then choose AI draft captions. Use Ctrl+A to select every listed clip.", ) return if len(clips) > 12 and QMessageBox.question( self, "Draft many captions?", f"ADAM will send six sampled frames for each of {len(clips)} clips to the configured Ollama model, one clip at a time. This can take a while. Continue?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) != QMessageBox.Yes: return model = str(self.config.get("ollama_model", "")).strip() if not model: QMessageBox.warning(self, "Ollama model not configured", "Choose an Ollama model in Settings → Planning model first.") return client = OllamaClient( str(self.config.get("ollama_url", "http://localhost:11434")), model, timeout=600, chat_max_tokens=min(512, int(self.config.get("ollama_chat_max_tokens", 512))), chat_response_length="balanced", ) dialog = VideoCaptionDraftDialog( self, client, clips, str(self.training.value("trigger_word") or "").strip(), ) dialog.exec() self._refresh_clips() def _split(self): from PySide6.QtWidgets import QInputDialog source, _ = QFileDialog.getOpenFileName(self, "Choose a long source video", "", "Videos (*.mp4 *.mov *.mkv *.webm *.m4v *.avi)") if not source: return count, ok = QInputDialog.getInt(self, "Extract clips", "How many evenly spaced clips? (49 frames at 12 FPS)", 32, 1, 100) if not ok: return trigger, ok = QInputDialog.getText( self, "Training trigger word", "Unique name used to identify this person or subject:", text=str(self.training.value("trigger_word") or ""), ) trigger = trigger.strip() if not ok: return if not trigger: QMessageBox.warning(self, "Trigger word needed", "Enter a unique subject name, such as Joe_Bartolozzi.") return description, ok = QInputDialog.getMultiLineText( self, "Clip captions", f"Describe what {trigger} is doing in the source video. ADAM will add the trigger word automatically. Review individual captions after splitting.", "speaking directly to the camera", ) if not ok: return description = description.strip().lstrip(",").strip() if not description: QMessageBox.warning(self, "Description needed", "Describe the visible subject and action in the video.") return caption = f"{trigger}, {description}" self.training.set_values({"trigger_word": trigger}) self._update_trigger_labels() self._schedule_save() output = str(self.root / "ADAM_Datasets" / "Video") self._submit(ExecutionPlan("Split source video", "Extract short clips into a new video dataset folder.", [PlanStep("wan_video_dataset", "Split source video", "Keep the source and create captioned clips.", {"operation": "split", "dataset_dir": output, "trigger_word": trigger, "source": source, "count": count, "caption": caption})], project_name="Video LoRA")) def _browse_model(self): selected, _ = QFileDialog.getOpenFileName(self, "Choose Wan checkpoint", "", "LoRA (*.safetensors)") if selected: from adam.video_lora import checkpoint_info try: checkpoint_info(Path(selected)) self.assets.register(kind="model", name=Path(selected).stem, path=selected, trainer="wan_video", checkpoint=selected) self.saved_model = selected self.models.setCurrentIndex(-1) self.refresh() self._save() except ValueError as exc: QMessageBox.warning(self, "Incompatible checkpoint", str(exc)) def _generation_changed(self): values = self.generation.values() frames = frames_for_seconds(values["duration"], values["fps"]) self.duration_hint.setText(f"Will generate {frames} frames: {frames / values['fps']:.2f} seconds at {values['fps']} FPS (Wan requires 4N+1 frames). Longer clips need more memory and time. Viewable pixels appear after denoising and decoding.") self._schedule_save() def _generate(self): try: if not self.models.currentData(): raise ValueError("Choose or train a Wan checkpoint first.") self._submit(generation_plan(self.root, self.models.currentData(), self.models.currentText(), self.generation.values())) except (OSError, ValueError) as exc: QMessageBox.warning(self, "Generation needs attention", str(exc)) def _refresh_videos(self): self.videos.clear() entries = [] for metadata in (self.root / "data" / "video_generations").glob("*/generation.json"): try: info = json.loads(metadata.read_text(encoding="utf-8")) path = Path(info.get("video", "")) if info.get("status") == "finished" and path.is_file(): entries.append((path, info)) except (OSError, ValueError): continue external = trainer_root(self.root, self.config) if external: entries.extend((path, {}) for path in (external / "samples").glob("*.mp4")) for path, info in sorted(entries, key=lambda entry: entry[0].stat().st_mtime, reverse=True): item = QListWidgetItem(f"{info.get('model_name', 'Existing sample')} · {path.name}") item.setData(Qt.UserRole, str(path)) item.setData(Qt.UserRole + 1, info) self.videos.addItem(item) def _show_video_details(self, item, _previous): if item: info = item.data(Qt.UserRole + 1) or {} settings = info.get("settings", {}) self.video_details.setText(f"{settings.get('prompt', 'Imported sample — generation settings unavailable')}\nSeed: {settings.get('seed', 'unknown')} · {info.get('width', '?')}×{info.get('height', '?')} · {info.get('frames', '?')} frames\n{item.data(Qt.UserRole)}") def _job_updated(self, job): if not any(step.tool_id.startswith("wan_video_") for step in job.plan.steps): return if not self.active_id or job.id == self.active_id: self.active_id = job.id self.status.setText(f"{job.id}: {job.status.value} · {job.progress}%" + (f" · {job.error}" if job.error else "")) if job.status == JobStatus.FINISHED: if any(step.tool_id == "wan_video_dataset" and step.arguments.get("operation") == "split" for step in job.plan.steps) and job.output_folder: self.dataset_folder.setText(job.output_folder) self.refresh() self._job_controls() def _job_controls(self): active = self.jobs.active_job relevant = bool(active and any(step.tool_id.startswith("wan_video_") for step in active.plan.steps)) self.pause_button.setEnabled(relevant) self.stop_button.setEnabled(relevant) dataset_busy = bool(active and self.dataset_folder.text().strip() and any( step.tool_id == "wan_video_trainer" and Path(str(step.arguments.get("dataset_dir", ""))).resolve() == Path(self.dataset_folder.text()).resolve() for step in active.plan.steps )) self.caption.setReadOnly(dataset_busy) self.import_button.setEnabled(not dataset_busy) self.archive_dataset_button.setEnabled(not dataset_busy) self.apply_caption_button.setEnabled(not dataset_busy) def _pause(self): job = self.jobs.active_job if job and any(step.tool_id.startswith("wan_video_") for step in job.plan.steps): self.jobs.resume(job.id) if job.status == JobStatus.PAUSED else self.jobs.pause(job.id) def _stop(self): job = self.jobs.active_job if job and any(step.tool_id.startswith("wan_video_") for step in job.plan.steps): self.jobs.cancel(job.id) @staticmethod def _open(path): if path and path.is_file(): QDesktopServices.openUrl(QUrl.fromLocalFile(str(path.resolve())))