Mercurial Repositories > Pyramid Project / file revision

# -*- coding: utf-8 -*-

# Copyright (c) 2012 - 2024 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing a dialog showing the available routes.
"""

import os

from PyQt6.QtCore import QCoreApplication, QProcess, Qt, QTimer, pyqtSlot
from PyQt6.QtWidgets import QDialog, QDialogButtonBox, QLineEdit, QTreeWidgetItem

from eric7 import Preferences
from eric7.EricWidgets import EricMessageBox
from eric7.Globals import strToQByteArray

from .Ui_PyramidRoutesDialog import Ui_PyramidRoutesDialog


class PyramidRoutesDialog(QDialog, Ui_PyramidRoutesDialog):
    """
    Class implementing a dialog showing the available routes.
    """

    def __init__(self, project, parent=None):
        """
        Constructor

        @param project reference to the project object
        @type Project
        @param parent reference to the parent widget
        @type QWidget
        """
        super().__init__(parent)
        self.setupUi(self)

        self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setEnabled(False)
        self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setDefault(True)

        self.__project = project
        self.proc = None
        self.buffer = ""

        self.show()
        QCoreApplication.processEvents()

    def finish(self):
        """
        Public slot called when the process finished or the user pressed the
        button.
        """
        if (
            self.proc is not None
            and self.proc.state() != QProcess.ProcessState.NotRunning
        ):
            self.proc.terminate()
            QTimer.singleShot(2000, self.proc.kill)
            self.proc.waitForFinished(3000)

        self.inputGroup.setEnabled(False)
        self.inputGroup.hide()

        self.proc = None

        self.__processBuffer()

        self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setEnabled(True)
        self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setEnabled(False)
        self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setDefault(True)
        self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setFocus(
            Qt.FocusReason.OtherFocusReason
        )

    def on_buttonBox_clicked(self, button):
        """
        Private slot called by a button of the button box clicked.

        @param button button that was clicked
        @type QAbstractButton
        """
        if button == self.buttonBox.button(QDialogButtonBox.StandardButton.Close):
            self.close()
        elif button == self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel):
            self.finish()

    def __procFinished(self, exitCode, exitStatus):
        """
        Private slot connected to the finished signal.

        @param exitCode exit code of the process
        @type int
        @param exitStatus exit status of the process
        @type QProcess.ExitStatus
        """
        self.normal = exitStatus == QProcess.ExitStatus.NormalExit and exitCode == 0
        self.finish()

    def __processBuffer(self):
        """
        Private slot to process the output buffer of the proutes command.
        """
        self.routes.clear()

        if not self.buffer:
            QTreeWidgetItem(self.routes, [self.tr("No routes found.")])
            self.routes.setHeaderHidden(True)
        else:
            lines = self.buffer.splitlines()
            row = 0
            headers = []
            while row < len(lines):
                if lines[row].strip().startswith("---"):
                    headerLine = lines[row - 1]
                    headers = headerLine.split()
                    break
                row += 1
            if headers:
                self.routes.setHeaderLabels(headers)
                self.routes.setHeaderHidden(False)
                row += 1
                splitCount = len(headers) - 1
                while row < len(lines):
                    line = lines[row].strip()
                    parts = line.split(None, splitCount)
                    QTreeWidgetItem(self.routes, parts)
                    row += 1
                for column in range(len(headers)):
                    self.routes.resizeColumnToContents(column)

    def start(self, projectPath):
        """
        Public slot used to start the process.

        @param projectPath path to the Pyramid project
        @type str
        @return flag indicating a successful start of the process
        @rtype bool
        """
        QTreeWidgetItem(self.routes, [self.tr("Getting routes...")])
        self.routes.setHeaderHidden(True)

        self.errorGroup.hide()
        self.normal = False
        self.intercept = False

        self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setEnabled(False)
        self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setEnabled(True)
        self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setDefault(True)
        self.buttonBox.button(QDialogButtonBox.StandardButton.Cancel).setFocus(
            Qt.FocusReason.OtherFocusReason
        )

        self.proc = QProcess()

        self.proc.finished.connect(self.__procFinished)
        self.proc.readyReadStandardOutput.connect(self.__readStdout)
        self.proc.readyReadStandardError.connect(self.__readStderr)

        cmd = self.__project.getPyramidCommand(
            "proutes", virtualEnv=self.__project.getProjectVirtualEnvironment()
        )
        args = []
        args.append("development.ini")

        if projectPath:
            self.proc.setWorkingDirectory(projectPath)
        self.proc.start(cmd, args)
        procStarted = self.proc.waitForStarted()
        if not procStarted:
            self.buttonBox.setFocus()
            self.inputGroup.setEnabled(False)
            EricMessageBox.critical(
                self,
                self.tr("Process Generation Error"),
                self.tr(
                    "The process {0} could not be started. "
                    "Ensure, that it is in the search path."
                ).format(cmd),
            )
        else:
            self.inputGroup.setEnabled(True)
            self.inputGroup.show()
        return procStarted

    def __readStdout(self):
        """
        Private slot to handle the readyReadStandardOutput signal.

        It reads the output of the process and appends it to a buffer for
        delayed processing.
        """
        if self.proc is not None:
            out = str(
                self.proc.readAllStandardOutput(),
                Preferences.getSystem("IOEncoding"),
                "replace",
            )
            self.buffer += out

    def __readStderr(self):
        """
        Private slot to handle the readyReadStandardError signal.

        It reads the error output of the process and inserts it into the
        error pane.
        """
        if self.proc is not None:
            err = str(
                self.proc.readAllStandardError(),
                Preferences.getSystem("IOEncoding"),
                "replace",
            )
            self.errorGroup.show()
            self.errors.insertPlainText(err)
            self.errors.ensureCursorVisible()

            QCoreApplication.processEvents()

    def on_passwordCheckBox_toggled(self, isOn):
        """
        Private slot to handle the password checkbox toggled.

        @param isOn flag indicating the status of the check box
        @type bool
        """
        if isOn:
            self.input.setEchoMode(QLineEdit.EchoMode.Password)
        else:
            self.input.setEchoMode(QLineEdit.EchoMode.Normal)

    @pyqtSlot()
    def on_sendButton_clicked(self):
        """
        Private slot to send the input to the subversion process.
        """
        inputTxt = self.input.text()
        inputTxt += os.linesep

        if self.passwordCheckBox.isChecked():
            self.errors.insertPlainText(os.linesep)
            self.errors.ensureCursorVisible()

        self.proc.write(strToQByteArray(inputTxt))

        self.passwordCheckBox.setChecked(False)
        self.input.clear()

    def on_input_returnPressed(self):
        """
        Private slot to handle the press of the return key in the input field.
        """
        self.intercept = True
        self.on_sendButton_clicked()

    def keyPressEvent(self, evt):
        """
        Protected slot to handle a key press event.

        @param evt the key press event
        @type QKeyEvent
        """
        if self.intercept:
            self.intercept = False
            evt.accept()
            return
        super().keyPressEvent(evt)