GUI API¶

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        with as_file(files(__package__) / "resources" / "icon.ico") as icon_path:
            self.setWindowIcon(QIcon(str(icon_path)))

        self.experiment = _new_experiment()
        self._update_title()
        self._busy = False
        self._busy_source: BaseTab | None = None
        self._progress_label: str | None = None
        self._progress_start = 0.0
        self._dirty = False

        self._build_menu_bar()

        # make status label a QLabel so text can be selectable
        self.status_label = QLabel()
        self.status_label.setTextInteractionFlags(
            Qt.TextInteractionFlag.TextSelectableByMouse
        )
        self.status_label.setSizePolicy(
            QSizePolicy.Policy.Ignored, QSizePolicy.Policy.Preferred
        )
        self.statusBar().addWidget(self.status_label, 1)

        self.tabs = QTabWidget()
        self.tab_widgets: list[BaseTab] = []
        for tab_type in TAB_TYPES:
            tab = tab_type(self.experiment)
            tab.status_message.connect(self._show_status)
            tab.experiment_changed.connect(
                lambda source=tab: self._on_experiment_changed(source)
            )
            tab.busy_changed.connect(
                lambda busy, source=tab: self._set_busy(busy, source)
            )
            tab.finished.connect(lambda source=tab: self._on_tab_finished(source))
            tab.progress_changed.connect(
                lambda value, maximum, label, source=tab: self._set_progress(
                    source, value, maximum, label
                )
            )
            self.tabs.addTab(tab, tab_type.title)
            self.tab_widgets.append(tab)
        self.tabs.currentChanged.connect(self._on_current_tab_changed)

        self.progress_bar = QProgressBar()
        self.progress_bar.setTextVisible(True)
        self.progress_bar.setVisible(False)

        self.central = QWidget()
        central_layout = QVBoxLayout(self.central)
        central_layout.setContentsMargins(0, 0, 0, 0)
        central_layout.addWidget(self.tabs, stretch=1)
        central_layout.addWidget(self.progress_bar)
        self.setCentralWidget(self.central)

    def tab(self, tab_type: type[BaseTab]) -> BaseTab:
        """The window's instance of ``tab_type``."""
        return next(tab for tab in self.tab_widgets if isinstance(tab, tab_type))

    def load_experiment(self, folder: str | Path) -> None:
        self._load_experiment(Path(folder))

    def _build_menu_bar(self) -> None:
        file_menu = self.menuBar().addMenu("&File")

        self.new_action = QAction("&New", self)
        self.new_action.setShortcut(QKeySequence.StandardKey.New)
        self.new_action.triggered.connect(self._new_experiment)
        file_menu.addAction(self.new_action)

        self.open_action = QAction("&Open…", self)
        self.open_action.setShortcut(QKeySequence.StandardKey.Open)
        self.open_action.triggered.connect(self._choose_experiment)
        file_menu.addAction(self.open_action)

        self.save_action = QAction("&Save", self)
        self.save_action.setShortcut(QKeySequence.StandardKey.Save)
        self.save_action.triggered.connect(self._save_experiment)
        file_menu.addAction(self.save_action)

        file_menu.addSeparator()

        self.exit_action = QAction("E&xit", self)
        self.exit_action.setShortcut(QKeySequence.StandardKey.Quit)
        self.exit_action.triggered.connect(self.close)
        file_menu.addAction(self.exit_action)

    def _new_experiment(self) -> None:
        """Start again with an empty experiment, discarding the current one."""
        if self._busy or not self._confirm_discarding_changes():
            return
        self._set_experiment(_new_experiment())

    def _choose_experiment(self) -> None:
        if self._busy or not self._confirm_discarding_changes():
            return
        folder = QFileDialog.getExistingDirectory(self, "Open experiment folder")
        if folder:
            self._load_experiment(Path(folder))

    def _load_experiment(self, folder: Path) -> None:
        try:
            experiment = Experiment.load(folder)
        except (OSError, ValueError, ValidationError) as exc:
            QMessageBox.critical(self, "Could not open experiment", str(exc))
            return
        self._set_experiment(experiment)
        self._show_status(f"Opened experiment {folder}")

    def _set_experiment(self, experiment: Experiment) -> None:
        """Hand ``experiment`` to every tab, in place of the current one."""
        self.experiment = experiment
        self._dirty = False
        for tab in self.tab_widgets:
            tab.set_experiment(experiment)
        self._update_title()

    def _on_experiment_changed(self, _source: BaseTab) -> None:
        """Mark experiment as having unsaved changes."""
        self._dirty = True

    def _on_current_tab_changed(self, index: int) -> None:
        """Refresh a tab when it is opened."""
        if 0 <= index < len(self.tab_widgets):
            self.tab_widgets[index].refresh()

    def _on_tab_finished(self, source: BaseTab) -> None:
        if self._dirty and not self._save_experiment():
            return
        index = self.tabs.indexOf(source)
        if 0 <= index < self.tabs.count() - 1:
            self.tabs.setCurrentIndex(index + 1)

    def _save_experiment(self) -> bool:
        """Write the experiment (config and any computed results) to its folder.

        Returns True if the save was successful.
        """
        if self._busy:
            self._show_status("Cannot save while a step is running")
            return False
        folder = None
        if self.experiment.folder is None:
            folder = self._ask_save_folder()
            if folder is None:
                return False
        try:
            self.experiment.save(folder)
        except (OSError, ValueError, ValidationError) as exc:
            QMessageBox.critical(self, "Could not save experiment", str(exc))
            return False
        finally:
            # A failed save can still have created the folder, so the title is
            # brought up to date either way.
            self._update_title()
        self._dirty = False
        self._show_status(f"Saved experiment to {self.experiment.folder}")
        return True

    def _confirm_discarding_changes(self) -> bool:
        """Offer to save unsaved changes, and say whether to carry on.

        Returns True if the user discarded the changes, otherwise saves them and returns False
        """
        if not self._dirty:
            return True
        answer = QMessageBox.question(
            self,
            "Unsaved changes",
            "This experiment has changes that have not been saved.",
            QMessageBox.StandardButton.Save
            | QMessageBox.StandardButton.Discard
            | QMessageBox.StandardButton.Cancel,
            QMessageBox.StandardButton.Save,
        )
        if answer == QMessageBox.StandardButton.Save:
            return self._save_experiment()
        return answer == QMessageBox.StandardButton.Discard

    def _ask_save_folder(self) -> Path | None:
        """Ask where to save an experiment that has no folder yet."""
        location = QFileDialog.getExistingDirectory(self, "Save experiment in…")
        if not location:
            return None
        name, chosen = QInputDialog.getText(
            self,
            "Save experiment",
            f"Name of the folder to create in {location}\n(leave empty to use it):",
            text=_DEFAULT_FOLDER_NAME,
        )
        if not chosen:
            return None
        name = name.strip()
        return Path(location) / name if name else Path(location)

    def _update_title(self) -> None:
        folder = self.experiment.folder
        name = folder.name if folder is not None else _UNSAVED_TITLE
        self.setWindowTitle(f"{_BASE_TITLE} :: [{name}]")

    def _show_status(self, message: str) -> None:
        """Report ``message`` in the status bar until something replaces it."""
        self.status_label.setText(message)
        self.status_label.setToolTip(message)

    def _set_busy(self, busy: bool, source: BaseTab | None = None) -> None:
        """Lock other tabs and actions while the source tab is busy."""
        if busy:
            self._busy_source = source
            self._progress_label = None
            self.progress_bar.setRange(0, 0)
            self.progress_bar.setFormat("Working…")
            self.progress_bar.setVisible(True)
        elif self._busy_source is None or source is self._busy_source:
            self._busy_source = None
            self.progress_bar.setVisible(False)
        self._busy = busy
        self.new_action.setEnabled(not busy)
        self.open_action.setEnabled(not busy)
        self.save_action.setEnabled(not busy)
        for index, tab in enumerate(self.tab_widgets):
            self.tabs.setTabEnabled(index, not busy or tab is source)

    def _set_progress(
        self,
        source: BaseTab,
        value: int,
        maximum: int,
        label: str,
    ) -> None:
        """Display progress reported by the tab that owns the active task."""
        if source is not self._busy_source:
            return
        maximum = max(0, maximum)
        self.progress_bar.setRange(0, maximum)
        if maximum:
            value = max(0, min(value, maximum))
            self.progress_bar.setValue(value)
            self.progress_bar.setFormat(
                f"{label} — %p%{self._eta(value, maximum, label)}"
            )
        else:
            self.progress_bar.setFormat(label)

    def _eta(self, value: int, maximum: int, label: str) -> str:
        """How much longer this operation has left, from the progress so far.

        Timing restarts if the label changes.
        """
        now = time.monotonic()
        if label != self._progress_label:
            self._progress_label = label
            self._progress_start = now
        elapsed = now - self._progress_start
        remaining = tqdm.format_meter(
            n=value,
            total=maximum,
            elapsed=elapsed,
            bar_format="{remaining}",
        )
        return f" — {remaining} left"

    def closeEvent(self, event) -> None:
        if not self._confirm_discarding_changes():
            event.ignore()
            return
        for tab in self.tab_widgets:
            tab.shutdown()
        super().closeEvent(event)

class AlignmentTab(BaseTab):
    """Let the user align all kind of inputs in time with each other via setting their time_offset properties."""

    title = "Alignment"

    def __init__(self, experiment: Experiment) -> None:
        super().__init__(experiment)
        self.video_cards: list[_VideoAlignmentCard] = []
        self._play_all_primary: _VideoAlignmentCard | None = None
        self.align_button = QPushButton("Automatic alignment")
        self.align_button.setToolTip(
            "Estimate initial offsets for all recordings before fine-tuning them"
        )
        self.align_button.clicked.connect(self._align)
        self.reset_timeline_button = QToolButton()
        self.reset_timeline_button.setIcon(
            self.style().standardIcon(QStyle.StandardPixmap.SP_MediaSkipBackward)
        )
        self.reset_timeline_button.setToolTip("Go to timeline zero")
        self.reset_timeline_button.clicked.connect(self._go_to_timeline_zero)
        self.play_all_button = QToolButton()
        self.play_all_button.setCheckable(True)
        self.play_all_button.setIconSize(QSize(24, 24))
        self.play_all_button.setText("All")
        self.play_all_button.setToolButtonStyle(
            Qt.ToolButtonStyle.ToolButtonTextBesideIcon
        )
        self.play_all_button.setIcon(
            self.style().standardIcon(QStyle.StandardPixmap.SP_MediaPlay)
        )
        self.play_all_button.setToolTip("Play all videos")
        self.play_all_button.toggled.connect(self._on_play_all_toggled)
        self.done_button = QPushButton("Finish alignment")
        self.done_button.setDefault(True)
        self.done_button.clicked.connect(self._finish_alignment)

        layout = QVBoxLayout(self)
        layout.addWidget(self.align_button)
        self.scroll_area = QScrollArea()
        self.scroll_area.setWidgetResizable(True)
        self.video_grid_widget = QWidget()
        self.grid = QGridLayout(self.video_grid_widget)
        self.grid.setAlignment(Qt.AlignmentFlag.AlignTop)
        self.scroll_area.setWidget(self.video_grid_widget)
        layout.addWidget(self.scroll_area, stretch=1)
        buttons = QHBoxLayout()
        buttons.addWidget(self.reset_timeline_button)
        buttons.addWidget(self.play_all_button)
        buttons.addStretch(1)
        buttons.addWidget(self.done_button)
        layout.addLayout(buttons)
        self.refresh()

    def refresh(self) -> None:
        """Render every video input, with at most three videos per row."""
        self._stop_play_all()
        videos = [*self.experiment.glasses_videos, *self.experiment.fixed_videos]
        self.align_button.setEnabled(len(self._inputs()) >= 2)
        if (
            self.video_cards
            and len(videos) == len(self.video_cards)
            and all(
                card.video is video
                and card.loaded_path == video.video_path
                and card.loaded
                for card, video in zip(self.video_cards, videos)
            )
        ):
            for card in self.video_cards:
                card.input_label.setText(card.video.id)
                card.controls.spin.blockSignals(True)
                card.controls.spin.setValue(card.video.timeline.offset)
                card.controls.spin.blockSignals(False)
            self._show_shared_timeline_time(0.0)
            return

        for card in self.video_cards:
            self.grid.removeWidget(card)
            card.shutdown()
            card.deleteLater()
        self.video_cards = []

        column_count = min(_VIDEOS_PER_ROW, max(1, len(videos)))
        for column in range(_VIDEOS_PER_ROW):
            self.grid.setColumnStretch(column, int(column < column_count))
        for index, video in enumerate(videos):
            card = _VideoAlignmentCard(video)
            if card.load_error is not None:
                self.status_message.emit(f"Could not open video: {card.load_error}")
            card.changed.connect(self.experiment_changed)
            card.set_requested.connect(self._set_offset_from_current_frame)

            self.video_cards.append(card)
            self.grid.addWidget(card, index // _VIDEOS_PER_ROW, index % _VIDEOS_PER_ROW)
        self.play_all_button.setEnabled(
            # Only enable Play-all when every card can be played to avoid possible half-playing or mid-play loading states
            bool(self.video_cards) and all(card.loaded for card in self.video_cards)
        )
        self.reset_timeline_button.setEnabled(
            any(card.loaded for card in self.video_cards)
        )

    def _align(self) -> None:
        """Estimate initial offsets and show them in the manual controls."""
        if len(self._inputs()) < 2:
            return
        self._stop_play_all()
        self.busy_changed.emit(True)
        self.setEnabled(False)
        self.progress_changed.emit(0, 100, "Aligning recordings…")
        try:
            result = align_experiment(self.experiment, progress=self._progress)
            if result.offsets:
                self.experiment_changed.emit()
                self.status_message.emit("Automatic alignment finished")
        finally:
            self.setEnabled(True)
            self.busy_changed.emit(False)
            self.refresh()
        if result.offsets:
            self._show_shared_timeline_time(self._first_common_experiment_time())

    def _first_common_experiment_time(self) -> float:
        """First experiment time represented by every video timeline."""
        videos = [card.video for card in self.video_cards]
        return max(
            (video.timeline.to_experiment_time(0.0) for video in videos),
            default=0.0,
        )

    def _progress(self, value: float) -> bool:
        self.progress_changed.emit(round(100 * value), 100, "Aligning recordings…")
        QApplication.processEvents(QEventLoop.ProcessEventsFlag.ExcludeUserInputEvents)
        return True

    def _set_offset_from_current_frame(self, source: _VideoAlignmentCard) -> None:
        offset = round(
            -source.viewer.current_time_seconds * source.video.timeline.rate, 3
        )
        loaded_cards = [card for card in self.video_cards if card.loaded]
        message = QMessageBox(self)
        message.setWindowTitle("Zero current frame")
        message.setText(f"Apply offset {offset:.3f} s to this video, or to all videos?")
        message.setIcon(QMessageBox.Icon.Question)
        this_video_button = message.addButton(
            "This video", QMessageBox.ButtonRole.AcceptRole
        )
        all_videos_button = message.addButton(
            "All videos", QMessageBox.ButtonRole.DestructiveRole
        )
        cancel_button = message.addButton(QMessageBox.StandardButton.Cancel)
        this_video_button.setStyleSheet(_THIS_VIDEO_BUTTON_STYLE)
        all_videos_button.setStyleSheet(_ALL_VIDEOS_BUTTON_STYLE)
        message.setDefaultButton(this_video_button)
        message.setEscapeButton(cancel_button)
        message.exec()
        clicked_button = message.clickedButton()
        if clicked_button not in (this_video_button, all_videos_button):
            return
        self._stop_play_all()
        if clicked_button is this_video_button:
            source.set_offset(offset)
            self._show_shared_timeline_time(0.0)
            return
        for card in loaded_cards:
            card.set_offset(offset)

    def _go_to_timeline_zero(self) -> None:
        self._stop_play_all()
        self._show_shared_timeline_time(0.0)

    def _show_shared_timeline_time(self, seconds: float) -> None:
        for card in self.video_cards:
            if card.loaded:
                card.viewer.set_time_seconds(
                    card.video.timeline.to_local_time(seconds),
                    allow_negative=True,
                    show_requested_time=True,
                )
                card.controls._show_timeline_state(seconds)

    def _finish_alignment(self) -> None:
        self._stop_play_all()
        self.finished.emit()

    def _on_play_all_toggled(self, play: bool) -> None:
        """
        Start or stop shared playback; all videos must be loaded.
        """
        user_desires_pause = not play
        if user_desires_pause:
            self._stop_play_all()  # Cleanly stop all
            return
        if not self.video_cards or not all(card.loaded for card in self.video_cards):
            # To prevent unusual situations with some playing
            return
        primary = self.video_cards[0]
        # This is separate because we only stop videos other than the "primary one"
        for card in self.video_cards[1:]:
            card.viewer.stop()
        self._play_all_primary = primary
        primary.viewer.frame_changed.connect(self._sync_play_all_viewers)
        self.play_all_button.setIcon(
            self.style().standardIcon(QStyle.StandardPixmap.SP_MediaPause)
        )
        self.play_all_button.setToolTip("Pause all videos")
        self._sync_play_all_viewers()
        if self._play_all_primary is not None:
            primary.viewer._play_button.setChecked(True)

    def _sync_play_all_viewers(self, _frame: int = 0) -> None:
        """
        Use the primary videos current time seconds to icnrement the frames of the other videos.
        """
        primary = self._play_all_primary
        if primary is None:
            return
        shared_timeline_time = primary.video.timeline.to_experiment_time(
            primary.viewer.playback_time_seconds
        )
        for card in self.video_cards:
            if card is not primary:
                card.viewer.set_time_seconds(
                    card.video.timeline.to_local_time(shared_timeline_time),
                    allow_negative=True,
                    show_requested_time=True,
                    sync_audio=False,
                )
                card.controls._show_timeline_state(shared_timeline_time)
        if primary.viewer.current_frame + 1 >= primary.viewer.frame_count:
            self._stop_play_all()

    def _stop_play_all(self) -> None:
        if self._play_all_primary is not None:
            self._play_all_primary.viewer.frame_changed.disconnect(
                self._sync_play_all_viewers
            )
            self._play_all_primary = None
        for card in self.video_cards:
            if card.viewer._play_button.isChecked():
                card.viewer.stop()
        self.play_all_button.blockSignals(True)
        self.play_all_button.setChecked(False)
        self.play_all_button.blockSignals(False)
        self.play_all_button.setIcon(
            self.style().standardIcon(QStyle.StandardPixmap.SP_MediaPlay)
        )
        self.play_all_button.setToolTip("Play all videos")

class AudioProcessingTab(BaseTab):
    """Show one input's speech results and transcribe its audio."""

    title = "Audio processing"

    def __init__(self, experiment: Experiment) -> None:
        super().__init__(experiment)

        self._thread: threading.Thread | None = None
        self._worker: TranscriptionWorker | None = None
        self._pending_inputs: list[Video | Audio] = []
        self._recordings: list[Video | Audio] = []
        self._live_word_count = 0

        self.input_selector = QComboBox()
        self.input_selector.currentIndexChanged.connect(self._on_input_selected)

        self.summary_label = QLabel()
        self.summary_label.setWordWrap(True)

        self.audio_player = AudioPlaybackWidget()
        self.audio_player.position_changed.connect(self._highlight_transcript_at)
        self._highlighted_row = -1
        self._marker_row = -1

        self.transcript_table = QTableWidget(0, len(_COLUMNS))
        self.transcript_table.setHorizontalHeaderLabels(_COLUMNS)
        row_header = self.transcript_table.verticalHeader()
        row_header.setSectionResizeMode(QHeaderView.ResizeMode.Fixed)
        row_header.setFixedWidth(24)
        self.transcript_table.setSelectionMode(
            QAbstractItemView.SelectionMode.NoSelection
        )
        self.transcript_table.setEditTriggers(
            QAbstractItemView.EditTrigger.NoEditTriggers
        )
        self.transcript_table.cellDoubleClicked.connect(self._play_transcript_row)
        header = self.transcript_table.horizontalHeader()
        header.setSectionResizeMode(_TEXT, QHeaderView.ResizeMode.Stretch)

        self.cancel_button = QPushButton("Cancel")
        self.cancel_button.setVisible(False)
        self.cancel_button.clicked.connect(self._cancel_run)

        top_bar = QHBoxLayout()
        top_bar.addWidget(QLabel("Recording:"))
        top_bar.addWidget(self.input_selector, stretch=1)

        bottom_bar = QHBoxLayout()
        bottom_bar.addWidget(self.summary_label, stretch=1)
        bottom_bar.addWidget(self.cancel_button)

        results_layout = QVBoxLayout()
        results_layout.setContentsMargins(0, 0, 0, 0)
        results_layout.addLayout(top_bar)
        results_layout.addWidget(self.audio_player)
        results_layout.addWidget(self.transcript_table, stretch=1)
        results_layout.addLayout(bottom_bar)
        results_side = QWidget()
        results_side.setLayout(results_layout)

        self.pipeline_editor = PipelineEditor(SPEECH_STEPS)
        self.pipeline_editor.changed.connect(self._on_pipeline_edited)
        self.pipeline_editor.run_requested.connect(
            lambda _step_type: self._start_transcription()
        )
        self.pipeline_editor.run_all_requested.connect(self._start_run_all)
        self.transcription_checkbox = QCheckBox("Transcribe this experiment's speech")
        self.transcription_checkbox.toggled.connect(self._on_pipeline_toggled)
        self.pipeline_group = QGroupBox("Speech pipeline")
        pipeline_group_layout = QVBoxLayout(self.pipeline_group)
        pipeline_group_layout.addWidget(self.transcription_checkbox)
        pipeline_group_layout.addWidget(self.pipeline_editor)

        pipeline_layout = QVBoxLayout()
        pipeline_layout.setContentsMargins(0, 0, 0, 0)
        pipeline_layout.addWidget(self.pipeline_group)
        pipeline_layout.addStretch(1)
        pipeline_side = QWidget()
        pipeline_side.setLayout(pipeline_layout)

        self.splitter = QSplitter(Qt.Orientation.Horizontal)
        self.splitter.addWidget(results_side)
        self.splitter.addWidget(pipeline_side)
        self.splitter.setStretchFactor(0, 1)
        self.splitter.setStretchFactor(1, 0)

        layout = QVBoxLayout(self)
        layout.addWidget(self.splitter)

        self.refresh()

    def _inputs(self) -> dict[str, Video | Audio]:
        """The inputs that carry sound, the only ones with speech to transcribe."""
        return {
            input_id: data
            for input_id, data in super()._inputs().items()
            if data.has_audio_track()
        }

    def refresh(self) -> None:
        """Re-list the inputs, keeping the shown one if it is still there."""
        if self._thread is not None:
            return
        shown = self.input()
        self._recordings = list(self._inputs().values())
        self.input_selector.blockSignals(True)
        self.input_selector.clear()
        for data in self._recordings:
            self.input_selector.addItem(f"{data.id} ({_kind(data)})")
        index = next(
            (i for i, data in enumerate(self._recordings) if data is shown),
            0 if self._recordings else -1,
        )
        self.input_selector.setCurrentIndex(index)
        self.input_selector.blockSignals(False)
        self.input_selector.setEnabled(bool(self._recordings))
        self._show_selected_input()

    def input(self) -> Video | Audio | None:
        """The input being shown, or ``None`` if the experiment has none."""
        index = self.input_selector.currentIndex()
        if 0 <= index < len(self._recordings):
            return self._recordings[index]
        return None

    def speech(self) -> Speech | None:
        """The shown input's speech results, whichever kind of input it is."""
        data = self.input()
        return None if data is None else data.speech

    def is_busy(self) -> bool:
        """Whether transcription is currently running."""
        return self._thread is not None

    def _on_input_selected(self, _index: int) -> None:
        self._show_selected_input()

    def _show_selected_input(self) -> None:
        """List the chosen input's speech results and bind the editor to it."""
        data = self.input()
        if data is None:
            self.audio_player.clear()
        else:
            self.audio_player.load(data.path, data.loudness.levels)
        self._refresh_results()
        self._bind_editor_to_pipeline()
        self._update_run_availability()

    def _refresh_results(self) -> None:
        """Fill the transcript table and its summary from the shown input."""
        speech = self.speech()
        data = None if speech is None else speech.data
        self.transcript_table.clearContents()
        self.transcript_table.setRowCount(0 if data is None else len(data))
        self._highlighted_row = -1
        self._marker_row = -1
        if data is None:
            self.summary_label.setText(self._nothing_to_show())
            return
        for row, segment in enumerate(data.itertuples(index=False)):
            self._set_transcript_row(row, segment.start, segment.end, segment.text)
        self.summary_label.setText(self._summary(speech))

    def _set_transcript_row(
        self, row: int, start: float, end: float, text: str
    ) -> None:
        """Populate one row shared by loaded and live transcript segments."""
        self.transcript_table.setVerticalHeaderItem(row, QTableWidgetItem(""))
        alignment = Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter
        start_item = QTableWidgetItem(_time_text(start))
        start_item.setTextAlignment(alignment)
        # Playback highlighting needs the exact bounds, not the rounded text.
        start_item.setData(Qt.ItemDataRole.UserRole, (float(start), float(end)))
        end_item = QTableWidgetItem(_time_text(end))
        end_item.setTextAlignment(alignment)

        self.transcript_table.setItem(row, _START, start_item)
        self.transcript_table.setItem(row, _END, end_item)

        text_item = QTableWidgetItem(str(text))
        text_item.setToolTip(textwrap.fill(str(text), 80))
        self.transcript_table.setItem(row, _TEXT, text_item)
        self._tint_row(row, _REST_ALPHA)

    @Slot(float)
    def _highlight_transcript_at(self, seconds: float) -> None:
        """Select the transcript segment containing the playback position."""
        row_at_position = -1
        marker_row = -1
        for row in range(self.transcript_table.rowCount()):
            item = self.transcript_table.item(row, _START)
            bounds = None if item is None else item.data(Qt.ItemDataRole.UserRole)
            if bounds is None:
                continue
            if bounds[0] <= seconds:
                marker_row = row
            if bounds[0] <= seconds <= bounds[1] and row_at_position < 0:
                row_at_position = row
        self._mark_row(marker_row)
        if row_at_position == self._highlighted_row:
            return
        if self._highlighted_row >= 0:
            self._tint_row(self._highlighted_row, _REST_ALPHA)
        self._highlighted_row = row_at_position
        if row_at_position >= 0:
            self._tint_row(row_at_position, _HIGHLIGHT_ALPHA)
            self.transcript_table.scrollToItem(
                self.transcript_table.item(row_at_position, _TEXT),
                QAbstractItemView.ScrollHint.PositionAtCenter,
            )

    def _tint_row(self, row: int, alpha: int) -> None:
        """Wash one row in the highlight colour, at ``alpha`` out of 255."""
        color = QColor(self.palette().highlight().color())
        color.setAlpha(alpha)
        for column in range(self.transcript_table.columnCount()):
            item = self.transcript_table.item(row, column)
            if item is not None:
                item.setBackground(QBrush(color))

    def _mark_row(self, row: int) -> None:
        """Point the gutter at the last segment playback has reached.

        The selection only lasts as long as the segment is being spoken, so this
        is what holds the place in the table through the silence in between.
        """
        if row == self._marker_row:
            return
        for at, text in ((self._marker_row, ""), (row, _MARKER)):
            item = self.transcript_table.verticalHeaderItem(at) if at >= 0 else None
            if item is not None:
                item.setText(text)
        self._marker_row = row

    @Slot(int, int)
    def _play_transcript_row(self, row: int, _column: int) -> None:
        """Seek to a double-clicked segment and start or continue playback."""
        if self._thread is not None:
            return
        item = self.transcript_table.item(row, _START)
        bounds = None if item is None else item.data(Qt.ItemDataRole.UserRole)
        if bounds is None:
            return
        self.audio_player.seek(bounds[0])
        self.audio_player.play()

    def _nothing_to_show(self) -> str:
        """Why the transcript table is empty."""
        if self.input() is not None:
            return "No transcript yet; run transcription."
        if any(data.path is not None for data in self.experiment.inputs):
            return "None of this experiment's recordings carry audio."
        return "This experiment has no inputs."

    def _summary(self, speech: Speech) -> str:
        words = 0 if speech.words is None else len(speech.words)
        return f"{len(speech.data)} segment(s), {words} word(s)"

    def _bind_editor_to_pipeline(self) -> None:
        """Bind the checkbox and editor to the experiment's speech pipeline."""
        pipeline = self.experiment.pipeline.speech
        self.transcription_checkbox.blockSignals(True)
        self.transcription_checkbox.setChecked(pipeline is not None)
        self.transcription_checkbox.blockSignals(False)
        if pipeline is not None:
            self.pipeline_editor.set_from(pipeline)
        self.pipeline_editor.setEnabled(
            pipeline is not None and self.input() is not None
        )

    @Slot(bool)
    def _on_pipeline_toggled(self, enabled: bool) -> None:
        """Enable or disable transcription for the experiment."""
        if enabled:
            pipeline = SpeechPipeline()
            self.experiment.pipeline.speech = pipeline
            self.pipeline_editor.set_from(pipeline)
        else:
            self.experiment.pipeline.speech = None
            self._pending_inputs = []
        self.pipeline_editor.setEnabled(enabled and self.input() is not None)
        self._update_run_availability()
        self.experiment_changed.emit()

    def _on_pipeline_edited(self) -> None:
        """Adopt the editor's pipeline as the experiment's, when it is valid."""
        pipeline = self.experiment.pipeline.speech
        if pipeline is None:
            return
        try:
            self.pipeline_editor.apply_to(pipeline)
        except (ValidationError, ValueError):
            self.status_message.emit("Pipeline has invalid settings; not applied")
            return
        self.experiment_changed.emit()

    def _update_run_availability(self) -> None:
        """Enable the "Run" buttons when there are recordings to transcribe."""
        enabled = self.experiment.pipeline.speech is not None
        self.pipeline_editor.set_run_enabled(
            TranscriptionStep, enabled and self.input() is not None
        )
        self.pipeline_editor.set_run_all_enabled(enabled and bool(self._recordings))

    def _transcription_config(self) -> TranscriptionStep | None:
        """The editor's validated transcription settings, or ``None``."""
        if self.experiment.pipeline.speech is None:
            return None
        try:
            return self.pipeline_editor.config_for(TranscriptionStep)
        except (ValidationError, ValueError) as exc:
            QMessageBox.critical(self, "Invalid settings", str(exc))
            return None

    @Slot()
    def _start_transcription(self) -> None:
        """Transcribe the selected recording with the editor's settings."""
        if self._thread is not None:
            return
        data = self.input()
        if data is None:
            self._pending_inputs = []
            return
        settings = self._transcription_config()
        if settings is None:
            self._pending_inputs = []
            return

        speech = data.speech
        speech.begin_transcription()
        self._begin_run()

        self._worker = TranscriptionWorker(speech, data.loudness, data.path, settings)
        self._worker.progress.connect(self._on_progress)
        self._worker.new_frame.connect(self._on_new_segment)
        self._worker.finished.connect(self._on_transcription_finished)
        self._worker.failed.connect(self._on_failed)
        self._worker.cancelled.connect(self._on_cancelled)

        self._thread = threading.Thread(target=self._worker.run, daemon=True)
        self._thread.start()

    def _start_run_all(self) -> None:
        """Transcribe every recording in chooser order."""
        if self._thread is not None:
            return
        if self._transcription_config() is None:
            self._pending_inputs = []
            return
        self._pending_inputs = list(self._recordings)
        self._continue_run_all()

    def _continue_run_all(self) -> None:
        """Select and transcribe the next recording queued by "Run all"."""
        while self._pending_inputs and self._thread is None:
            data = self._pending_inputs.pop(0)
            index = self._recordings.index(data)
            self.input_selector.setCurrentIndex(index)
            self._start_transcription()

    def _begin_run(self) -> None:
        """Prepare the tab for a transcription run."""
        self.audio_player.pause()
        self._set_running(True)
        self.transcript_table.clearContents()
        self.transcript_table.setRowCount(0)
        self._live_word_count = 0
        self.summary_label.setText("Waiting for transcript segments…")
        self.progress_changed.emit(0, 0, "Downloading weights…")

    @Slot(object)
    def _on_new_segment(self, segment) -> None:
        """Append one provisional Whisper segment while transcription runs."""
        scrollbar = self.transcript_table.verticalScrollBar()
        following = scrollbar.value() == scrollbar.maximum()
        row = self.transcript_table.rowCount()
        self.transcript_table.insertRow(row)
        self._set_transcript_row(row, segment.start, segment.end, segment.text)
        self._live_word_count += len(segment.words)
        self.summary_label.setText(
            f"{row + 1} segment(s), {self._live_word_count} word(s) — transcribing…"
        )
        if following:
            self.transcript_table.scrollToBottom()

    @Slot(float)
    def _on_progress(self, fraction: float) -> None:
        self.progress_changed.emit(round(100 * fraction), 100, "Transcription…")

    def _cancel_run(self) -> None:
        if self._worker is not None:
            self._worker.cancel()
        self.cancel_button.setEnabled(False)
        self.cancel_button.setText("Cancelling…")

    @Slot()
    def _on_transcription_finished(self) -> None:
        speech = self.speech()
        n_words = 0 if speech.words is None else len(speech.words)
        self.status_message.emit(
            f"Transcription finished: {n_words} words over {len(speech.data)} segments"
        )
        self._set_running(False)
        self._continue_run_all()

    @Slot(str, str)
    def _on_failed(self, message: str, details: str) -> None:
        self._pending_inputs = []
        dialog = QMessageBox(self)
        dialog.setIcon(QMessageBox.Icon.Critical)
        dialog.setWindowTitle("Transcription failed")
        dialog.setText(message)
        dialog.setDetailedText(details)
        dialog.exec()
        self._set_running(False)

    @Slot()
    def _on_cancelled(self) -> None:
        self._pending_inputs = []
        self.status_message.emit("Transcription cancelled")
        self._set_running(False)

    def _set_running(self, running: bool) -> None:
        if not running:
            self._thread = None
            self._worker = None
            self.refresh()
            self.experiment_changed.emit()
        self.input_selector.setEnabled(not running and bool(self._recordings))
        self.audio_player.setEnabled(not running)
        self.transcription_checkbox.setEnabled(not running)
        self.pipeline_editor.setEnabled(
            not running
            and self.input() is not None
            and self.experiment.pipeline.speech is not None
        )
        self.cancel_button.setVisible(running)
        self.cancel_button.setEnabled(True)
        self.cancel_button.setText("Cancel")
        self.busy_changed.emit(running)