Source code for SDF.GUI.converter.sdf_converter

import logging
import os
import sys
import threading
from contextlib import redirect_stdout, redirect_stderr
from traceback import print_exc
from typing import List, Optional, Union

from PyQt5.QtCore import QDir, QThread, Qt, QObject, pyqtSignal
from PyQt5.QtGui import QResizeEvent, QKeyEvent, QDropEvent, QDragEnterEvent, QDragMoveEvent
from PyQt5.QtWidgets import QApplication, QWidget, QVBoxLayout, QHBoxLayout, QCheckBox, QPushButton, QProgressBar, \
    QLabel, QRadioButton, QSpacerItem, QSizePolicy, QTabWidget, QListWidget, QMainWindow, QFileDialog, QTextEdit, \
    QAbstractItemView, QListWidgetItem, QFrame

from SDF.file_io import SDFConverter


[docs]class HLine(QFrame):
    def __init__(self):
        super().__init__()
        self.setFrameShape(QFrame.HLine)
        self.setFrameShadow(QFrame.Sunken)


[docs]class FileListWidget(QListWidget):
    def __init__(self):
        super().__init__()
        self.setAcceptDrops(True)
        self.setSelectionMode(QAbstractItemView.ExtendedSelection)

[docs]    def addItem(self, item: Union[QListWidgetItem, str]) -> None:
        if isinstance(item, QListWidgetItem):
            item = item.text()
        if not os.path.isfile(item):
            raise FileNotFoundError("Can only add existing files")
        item = os.path.realpath(item)  # realpath resolves links
        if item in self.filenames:  # ignore duplicates
            return
        super().addItem(item)

[docs]    def dropEvent(self, event: QDropEvent) -> None:
        for url in event.mimeData().urls():
            if url.isLocalFile() and os.path.isfile(url.toLocalFile()):
                # isLocalFile() is True for files and directories
                self.addItem(url.toLocalFile())

[docs]    def dragEnterEvent(self, event: QDragEnterEvent) -> None:
        # we must allow drags to enter the widget (-> dragMoveEvent on move)
        event.accept()

[docs]    def dragMoveEvent(self, event: QDragMoveEvent) -> None:
        # we must allow mouse movement while dragging (-> dropEvent on drop)
        event.accept()

    @property
    def filenames(self) -> List[str]:
        return [self.item(i).text() for i in range(self.count())]

[docs]    def keyPressEvent(self, keypress: QKeyEvent):
        # enable item deletion via Delete Key
        if keypress.key() == Qt.Key_Delete:
            for item in self.selectedItems():
                self.takeItem(self.row(item))
        super().keyPressEvent(keypress)


[docs]class SDFConverterGUI(QMainWindow):
    def __init__(self):
        super().__init__()

        # sizes
        self.margin = 5
        self.bottom_panel_height = 100
        self.sidebar_width = 150

        # main widgets, must be members, because resizeEvent needs to access them
        self.tabs_panel = QWidget(parent=self)
        self.bottom_panel = QWidget(parent=self)
        self.right_panel = QWidget(parent=self)

        # tabs
        self.file_list = FileListWidget()
        self.output_textbox = QTextEdit()

        # output directory
        self.output_directory_label = QTextEdit()
        self.output_directory_selection_button = QPushButton(text="Select")
        self.output_directory_reset_button = QPushButton(text="Reset")

        # run panel components
        self.progress_bar = QProgressBar()
        self.run_button = QPushButton(text="Run")

        # ForceSDF options
        self.checkbox_mat = QCheckBox(text="Generate matlab\nfile")
        self.checkbox_no_force = QCheckBox(text="Don't generate\nForceSDF file")

        # verbosity options
        self.radiobutton_verbosity_normal = QRadioButton(text="Normal")
        self.radiobutton_verbosity_verbose = QRadioButton(text="Verbose")
        self.radiobutton_verbosity_debug = QRadioButton(text="Debug")

        # file selection
        self.file_selection_button = QPushButton(text="Add files")
        self.clear_file_list_button = QPushButton(text="Clear file list")

        # organize everything
        self._init_layout()
        self._init_components()
        self._init_logging()

        # worker thread
        self.worker: Optional[WorkerThread] = None

    def _init_logging(self):
        handler = LoggingHandler()
        handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
        logging.getLogger().addHandler(handler)
        handler.output_stream.output_ready.connect(self.write_output)

    def _init_components(self):
        # output tab
        self.output_textbox.setReadOnly(True)
        self.output_textbox.setPlaceholderText("Output")

        # output directory label
        self.output_directory_label.setReadOnly(True)
        self.output_directory_label.setLineWrapMode(QTextEdit.LineWrapMode.NoWrap)
        self.output_directory_label.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
        self.output_directory_label.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
        self.output_directory_label.setPlaceholderText("[Same as input files]")

        self.radiobutton_verbosity_normal.setChecked(True)

        # connect buttons
        self.output_directory_reset_button.clicked.connect(self.output_directory_reset)
        self.output_directory_selection_button.clicked.connect(self.output_directory_selection)

        self.file_selection_button.clicked.connect(self.select_files)
        self.clear_file_list_button.clicked.connect(self.file_list.clear)

        self.run_button.clicked.connect(self.run)

    def _init_layout(self):
        # main window
        self.setWindowTitle("SDF Converter")
        self.resize(700, 500)

        # left panel
        left_panel_layout = QVBoxLayout(self.tabs_panel)

        tab_container = QTabWidget()

        file_tab = QWidget()
        file_tab_layout = QVBoxLayout()
        file_tab.setLayout(file_tab_layout)
        file_tab_layout.addWidget(self.file_list)

        output_tab = QWidget()
        output_tab_layout = QVBoxLayout()
        output_tab.setLayout(output_tab_layout)
        output_tab_layout.addWidget(self.output_textbox)

        tab_container.addTab(file_tab, "Files")
        tab_container.addTab(output_tab, "Output")
        left_panel_layout.addWidget(tab_container)

        # output directory panel
        output_directory_layout = QHBoxLayout()
        output_directory_label = QLabel(text="<b>Output directory:</b>")
        output_directory_layout.addWidget(output_directory_label)
        output_directory_layout.addWidget(self.output_directory_label)
        output_directory_layout.addWidget(self.output_directory_selection_button)
        output_directory_layout.addWidget(self.output_directory_reset_button)

        output_directory_widget = QWidget()
        output_directory_widget.setLayout(output_directory_layout)

        # run panel
        run_layout = QHBoxLayout()
        run_layout.addWidget(self.progress_bar)
        run_layout.addWidget(self.run_button)

        run_widget = QWidget()
        run_widget.setLayout(run_layout)

        # bottom panel
        bottom_panel_layout = QVBoxLayout(self.bottom_panel)
        bottom_panel_layout.setSpacing(0)

        bottom_panel_layout.addWidget(HLine())
        bottom_panel_layout.addWidget(output_directory_widget)
        bottom_panel_layout.addWidget(run_widget)

        # right panel
        right_panel_layout = QVBoxLayout(self.right_panel)

        force_sdf_container = QVBoxLayout()
        force_sdf_container.addWidget(QLabel(text="<b>ForceSDF options</b>"))
        force_sdf_container.addWidget(self.checkbox_mat)
        force_sdf_container.addWidget(self.checkbox_no_force)

        verbosity_container = QVBoxLayout()
        verbosity_container.addWidget(QLabel(text="<b>Verbosity level</b>"))
        verbosity_container.addWidget(self.radiobutton_verbosity_normal)
        verbosity_container.addWidget(self.radiobutton_verbosity_verbose)
        verbosity_container.addWidget(self.radiobutton_verbosity_debug)

        file_selection_container = QVBoxLayout()
        file_selection_container.addWidget(self.file_selection_button)
        file_selection_container.addWidget(self.clear_file_list_button)

        right_panel_layout.addLayout(force_sdf_container)
        right_panel_layout.addLayout(verbosity_container)
        right_panel_layout.addItem(QSpacerItem(2 * self.margin, self.bottom_panel_height, QSizePolicy.Minimum,
                                               QSizePolicy.Expanding))
        right_panel_layout.addLayout(file_selection_container)

        # set sizes
        self.output_directory_selection_button.setFixedWidth(self.sidebar_width - 2*self.margin)
        self.output_directory_reset_button.setFixedWidth(self.sidebar_width - 2*self.margin)
        self.file_selection_button.setFixedWidth(self.sidebar_width-2*self.margin)
        self.clear_file_list_button.setFixedWidth(self.sidebar_width-2*self.margin)
        self.output_directory_label.setFixedHeight(25)
        output_directory_label.setFixedWidth(self.sidebar_width - 2*self.margin)

        # set margins
        left_panel_layout.setContentsMargins(self.margin, self.margin, self.margin, self.margin)
        output_directory_layout.setContentsMargins(self.margin, self.margin, self.margin, self.margin)
        run_layout.setContentsMargins(self.margin, self.margin, self.margin, self.margin)
        right_panel_layout.setContentsMargins(self.margin, self.margin, self.margin, self.margin)

[docs]    def resizeEvent(self, resize_event: QResizeEvent):
        super().resizeEvent(resize_event)
        size = resize_event.size()
        width, height = size.width(), size.height()

        self.tabs_panel.setGeometry(0, 0, width - self.sidebar_width, height - self.bottom_panel_height)
        self.bottom_panel.setGeometry(0, height - self.bottom_panel_height, width, self.bottom_panel_height)
        self.right_panel.setGeometry(width - self.sidebar_width, 0, self.sidebar_width,
                                     height - self.bottom_panel_height)

[docs]    def output_directory_reset(self):
        self.output_directory_label.setText("")

[docs]    def output_directory_selection(self):
        self.output_directory = QFileDialog.getExistingDirectory(self, "Select output directory", QDir.currentPath())
        self.output_directory_label.setText(self.output_directory)

[docs]    def select_files(self):
        filenames, _ = QFileDialog.getOpenFileNames(self, "Select input files", QDir.currentPath())
        for filename in filenames:
            self.file_list.addItem(filename)

[docs]    def run(self):
        if not self.file_list.filenames:
            return

        converter = SDFConverter(
            generate_mat=self.checkbox_mat.isChecked(),
            generate_force=not self.checkbox_no_force.isChecked(),
            output_directory=self.output_directory,
        )
        filenames = self.file_list.filenames

        if self.radiobutton_verbosity_verbose.isChecked():
            log_level = logging.INFO
        elif self.radiobutton_verbosity_debug.isChecked():
            log_level = logging.DEBUG
        else:
            log_level = logging.WARNING
        logging.getLogger().setLevel(log_level)

        self.worker = WorkerThread(converter, filenames)
        self.worker.finished.connect(self.job_done)
        self.worker.progress.connect(self.update_progressbar)
        self.worker.output.connect(self.write_output)
        # delete worker after job is done
        self.worker.finished.connect(self.worker.deleteLater)
        # qt deleted the c++ object, but we still have a reference to it (usage leads to RuntimeError)
        self.worker.destroyed.connect(self.remove_worker_reference)

        self.progress_bar.setMaximum(len(filenames))
        self.update_progressbar(value=0)
        self.run_button.setText("Abort")
        self.run_button.clicked.disconnect()
        self.run_button.clicked.connect(self.abort)

        self.worker.start()

[docs]    def remove_worker_reference(self):
        self.worker = None

[docs]    def write_output(self, text: str):
        scrollbar = self.output_textbox.verticalScrollBar()
        autoscroll = scrollbar.value() == scrollbar.maximum()

        self.output_textbox.insertHtml(text.replace("\n", "<br>"))

        if autoscroll:
            scrollbar.setValue(scrollbar.maximum())

[docs]    def update_progressbar(self, value: int):
        self.progress_bar.setValue(value)
        percentage = value / self.progress_bar.maximum() * 100
        self.progress_bar.setFormat(f"Converted {value}/{self.progress_bar.maximum()} files ({int(percentage)} %)")

[docs]    def job_done(self):
        self.run_button.setText("Run")
        self.run_button.clicked.disconnect()
        self.run_button.clicked.connect(self.run)

[docs]    def abort(self):
        # request interruption, which currently means finishing the current file and then exiting the loop
        if self.worker is not None:
            self.run_button.setText("Aborting...")
            self.worker.requestInterruption()

    # TODO: decide if we want a clean shutdown and how it should look like
    #  (not handling it means just destroying the qt objects, leading to Python errors)
    # def closeEvent(self, close_event: QCloseEvent) -> None:
    #     # freezes window until abort action succeeded
    #     self.abort()
    #     if self.worker is not None:
    #         self.worker.wait()  # blocks and waits for job abortion (else it would crash immediately)
    #     close_event.accept()

    @property
    def output_directory(self) -> Optional[str]:
        if self.output_directory_label.toPlainText() == "":
            return None
        return self.output_directory_label.toPlainText()

    @output_directory.setter
    def output_directory(self, directory: str) -> None:
        if directory is None:
            self.output_directory_label.setPlainText("")
        else:
            if os.path.isdir(directory):
                self.output_directory_label.setPlainText(directory)


[docs]class WorkerThread(QThread):
    # TODO: Python GIL -> shares CPU time with GUI thread. Use QProcess?
    # TODO: Find a way to safely close the program without producing errors while this Thread is running
    progress = pyqtSignal(int)
    output = pyqtSignal(str)

    def __init__(self, converter: SDFConverter, files: List[str]):
        super().__init__()
        self.converter = converter
        self.files = files

[docs]    def run(self):
        stdout_stream = OutputStream(thread_id=threading.get_ident(), replaces=sys.__stdout__)
        stderr_stream = OutputStream(thread_id=threading.get_ident(), textcolor_html="#FF0000", replaces=sys.__stderr__)
        stdout_stream.output_ready.connect(self.output.emit)
        stderr_stream.output_ready.connect(self.output.emit)

        with redirect_stdout(stdout_stream), redirect_stderr(stderr_stream):
            for i, filename in enumerate(self.files, start=1):
                if self.isInterruptionRequested():
                    print("Interrupted")
                    break

                print(f"<b>Converting file {i}/{len(self.files)}:'{filename}'</b>")
                try:
                    self.converter.convert_file(filename)
                except:
                    print_exc(file=sys.stderr)
                self.progress.emit(i)


[docs]class OutputStream(QObject):
    output_ready = pyqtSignal(str)

    def __init__(self, thread_id: Optional[int] = None, textcolor_html: str = None,
                 replaces=sys.__stdout__):
        """
        To be used as output stream, emits output string as Qt Signal. Uses a buffer to emit signals only

        :param thread_id: As returned by threading.get_ident(). If set, only output of this thread is emitted
        :param textcolor_html: e.g. '#ff0000', ignored when writing to `replaces`
        :param replaces: The stream it replaces. If thread_id is not None and output from another thread is caught,
                         the output is forwarded to this stream
        """
        super().__init__()
        self.thread_id = thread_id
        self.textcolor_html = textcolor_html  # e.g. '#ff0000' for red
        self.replaces = replaces
        self.buffer = ""

[docs]    def write_to_buffer(self, text: str):
        self.buffer += text

[docs]    def write_from_buffer(self):
        text = self.buffer
        self.buffer = ""

        if self.thread_id is None or threading.get_ident() == self.thread_id:
            if self.textcolor_html is not None:
                text = f"<font color='{self.textcolor_html}'>{text}</font>"
            self.output_ready.emit(text)
        else:
            self.replaces.write(text)

[docs]    def write(self, text: str):
        self.write_to_buffer(text)

        if self.buffer.endswith("\n"):
            self.write_from_buffer()

    def __del__(self):
        if self.buffer:
            self.output_ready.emit(self.buffer)


[docs]class LoggingHandler(logging.Handler):
    def __init__(self):
        super().__init__()
        self.output_stream = OutputStream(replaces=sys.__stderr__)

[docs]    def emit(self, record):
        record = self.format(record)
        self.output_stream.write(f"{record}\n")


[docs]def main():
    app = QApplication([])
    gui = SDFConverterGUI()
    gui.show()
    exit(app.exec_())