toolbar ¶

def abort(self) -> None:
    if self.running:
        self.update_info()
        self._save_benchmark_results()

    self.running = False
    QMetaObject.invokeMethod(self.update_info_timer, 'stop', Qt.ConnectionType.QueuedConnection)

    if self.run_abort_button.isChecked():
        self.run_abort_button.click()

def get_notches(self) -> Notches:
    from .custom import Notches
    return Notches()

def get_separator(self, horizontal: bool = False) -> QFrame:
    separator = QFrame(self)
    separator.setFrameShape(QFrame.Shape.HLine if horizontal else QFrame.Shape.VLine)
    separator.setFrameShadow(QFrame.Shadow.Sunken)
    return separator

def init_notches(self, main: MainWindow = ...) -> None:
    self.notches_changed.connect(main.timeline.update_notches)

def on_current_frame_changed(self, frame: Frame) -> None:
    pass

def on_current_output_changed(self, index: int, prev_index: int) -> None:
    max_frames = 1000 if self.main.current_output is None else self.main.current_output.total_frames
    self.start_frame_control.setMaximum(max_frames - 1)
    self.end_frame_control.setMaximum(max_frames - 1)
    self.total_frames_control.setMaximum(max_frames)
    self.total_frames_control.setValue(min(self.total_frames_control.value() or 1000, max_frames))

def on_prefetch_changed(self, new_state: Qt.CheckState) -> None:
    if new_state == Qt.CheckState.Checked:
        self.unsequenced_checkbox.setEnabled(True)
        self.usable_cpus_spinbox.setEnabled(True)
    elif new_state == Qt.CheckState.Unchecked:
        self.unsequenced_checkbox.setChecked(False)
        self.unsequenced_checkbox.setEnabled(False)
        self.usable_cpus_spinbox.setEnabled(False)

def on_run_abort_pressed(self, checked: bool) -> None:
    self.set_ui_editable(not checked)
    if checked:
        self.run()
    else:
        self.abort()

def on_toggle(self, new_state: bool) -> None:
    if new_state == self.visibility:
        return

    # invoking order matters
    self.setVisible(new_state)
    self.visibility = new_state
    self.toggle_button.setChecked(new_state)
    self.resize_main_window(new_state)

def resize_main_window(self, expanding: bool) -> None:
    if self.main.windowState() in {Qt.WindowState.WindowMaximized, Qt.WindowState.WindowFullScreen}:
        return

    if expanding:
        self.main.resize(self.main.width(), self.main.height() + self.height() + round(6 * self.main.display_scale))
    if not expanding:
        self.main.resize(self.main.width(), self.main.height() - self.height() - round(6 * self.main.display_scale))
        self.main.timeline.update()

def run(self) -> None:
    if self.settings.clear_cache_enabled:
        from vstools.utils.vs_proxy import clear_cache
        clear_cache()

    if self.settings.frame_data_sharing_fix_enabled:
        self.main.current_output.update_graphic_item(
            self.main.current_scene.pixmap().copy(),
            graphics_scene_item=self.main.current_output.graphics_scene_item
        )

    self.frames_done = 0

    self.start_frame = self.start_frame_control.value()
    self.end_frame = self.end_frame_control.value()
    self.total_frames = self.total_frames_control.value()

    if self.prefetch_checkbox.isChecked():
        concurrent_requests_count = self.usable_cpus_spinbox.value()
    else:
        concurrent_requests_count = 1

    self.unsequenced = self.unsequenced_checkbox.isChecked()
    if not self.unsequenced:
        self.buffer = deque([], concurrent_requests_count)
        self.sequenced_timer.start()

    self.running = True
    self.run_start_time = perf_counter()

    # Initialize benchmark data if logging is enabled
    if self.settings.log_results_enabled:
        self.benchmark_data = {
            'Date & Time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'Node': self.main.current_output.name,
            'Index': self.main.current_output.index,
            'Start': int(self.start_frame),
            'End': int(self.end_frame),
            'Frames Processed': 0,
            'Average FPS': 0.0,
            'Total Time (s)': 0.0,
            'Thread Count': concurrent_requests_count,
            'Prefetch': self.prefetch_checkbox.isChecked(),
            'Unsequenced': self.unsequenced_checkbox.isChecked(),
        }

        logging.debug(f"Initial benchmark data: {self.benchmark_data}")

    self.update_info()

    for offset in range(min(int(self.end_frame - self.frames_done), concurrent_requests_count)):
        if self.unsequenced:
            self._request_next_frame_unsequenced()
        else:
            self.buffer.appendleft(
                self.main.current_output.source.original_clip.get_frame_async(self.start_frame + offset)
            )

    self.update_info_timer.setInterval(round(float(self.settings.refresh_interval) * 1000))
    self.update_info_timer.start()

def set_qobject_names(self) -> None:
    if not hasattr(self, '__slots__'):
        return

    slots = list(self.__slots__)

    if isinstance(self, AbstractToolbar) and 'main' in slots:
        slots.remove('main')

    for attr_name in slots:
        attr = getattr(self, attr_name)
        if not isinstance(attr, QObject):
            continue
        attr.setObjectName(type(self).__name__ + '.' + attr_name)

def set_ui_editable(self, new_state: bool) -> None:
    self. start_frame_control.setEnabled(new_state)
    self.end_frame_control.setEnabled(new_state)
    self.total_frames_control.setEnabled(new_state)
    self.prefetch_checkbox.setEnabled(new_state)
    self. unsequenced_checkbox.setEnabled(new_state)

def setup_ui(self) -> None:
    super().setup_ui()

    self.start_frame_control = FrameEdit(
        self, 0, maximum=10000, valueChanged=lambda value: self.update_controls(start=value)
    )
    self.end_frame_control = FrameEdit(
        self, maximum=10000, valueChanged=lambda value: self.update_controls(end=value)
    )
    self.total_frames_control = FrameEdit(
        self, 1, maximum=10000, valueChanged=lambda value: self.update_controls(total=value)
    )
    self.total_frames_control.setValue(1000)

    self.unsequenced_checkbox = CheckBox(
        'Unsequenced', self, checked=True, tooltip=(
            "If enabled, next frame will be requested each time frameserver returns completed frame.\n"
            "If disabled, first frame that's currently processing will be waited before requesting the next one."
        )
    )

    self.prefetch_checkbox = CheckBox(
        'Prefetch', self, checked=True, tooltip='Request multiple frames in advance.',
        stateChanged=self.on_prefetch_changed
    )

    self.run_abort_button = PushButton('Run', self, checkable=True, clicked=self.on_run_abort_pressed)

    self.info_label = QLabel(self)

    self.usable_cpus_spinbox = SpinBox(self, 1, self.settings.default_usable_cpus_spinbox.maximum())
    self.usable_cpus_spinbox.setValue(self.settings.default_usable_cpus_count)

    self.hlayout.addWidgets([
        QLabel('Start:'), self.start_frame_control,
        QLabel('End:'), self.end_frame_control,
        QLabel('Total:'), self.total_frames_control,
        QLabel('Usable CPUs Count:'), self.usable_cpus_spinbox,
        self.prefetch_checkbox,
        self.unsequenced_checkbox,
        self.settings.log_results_checkbox,
        self.run_abort_button,
        self.info_label
    ])
    self.hlayout.addStretch()

def update_controls(
    self, start: Frame | None = None, end: Frame | None = None, total: Frame | None = None
) -> None:
    if not hasattr(self.main, 'current_output'):
        return

    if self.main.current_output is None:
        max_frames = 1000
    else:
        max_frames = self.main.current_output.total_frames

    if start is not None:
        end = self.end_frame_control.value()
        total = self.total_frames_control.value()

        if start > end:
            end = start
        total = end - start + Frame(1)
    elif end is not None:
        start = self.start_frame_control.value()
        total = self.total_frames_control.value()

        if end < start:
            start = end
        total = end - start + Frame(1)
    elif total is not None:
        start = self.start_frame_control.value()
        end = self.end_frame_control.value()
        old_total = end - start + Frame(1)
        delta = total - old_total

        end += delta
        if end > (e := max_frames - 1):
            start -= end - e
            end = e
    else:
        return

    qt_silent_call(self.start_frame_control.setValue, start)
    qt_silent_call(self.end_frame_control.setValue, end)
    qt_silent_call(self.total_frames_control.setValue, total)

def update_info(self) -> None:
    run_time = Time(seconds=(perf_counter() - self.run_start_time))
    fps = int(self.frames_done) / float(run_time)

    self.info_label.setText(
        f"{self.frames_done}/{self.total_frames} frames in {strfdelta(run_time, '%M:%S.%Z')}, {fps:.4f} fps"
    )