Mercurial Repositories > Pyramid Project / file revision

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

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

"""
Module implementing a dialog starting a process and showing its output.
"""

import os

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

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

try:
    from eric7.SystemUtilities.OSUtilities import isMacPlatform, isWindowsPlatform
except ImportError:
    # imports for eric < 23.1
    from eric7.Globals import isWindowsPlatform, isMacPlatform

from .Ui_PyramidDialog import Ui_PyramidDialog


class PyramidDialog(QDialog, Ui_PyramidDialog):
    """
    Class implementing a dialog starting a process and showing its output.

    It starts a QProcess and displays a dialog that
    shows the output of the process. The dialog is modal,
    which causes a synchronized execution of the process.
    """

    def __init__(
        self,
        text,
        fixed=False,
        linewrap=True,
        msgSuccess=None,
        msgError=None,
        combinedOutput=False,
        parent=None,
    ):
        """
        Constructor

        @param text text to be shown by the label
        @type str
        @param fixed flag indicating a fixed font should be used
        @type bool
        @param linewrap flag indicating to wrap long lines
        @type bool
        @param msgSuccess optional string to show upon successful execution
        @type str
        @param msgError optional string to show upon unsuccessful execution
        @type str
        @param combinedOutput flag indicating to combine the output into the
            output pane
        @type bool
        @param parent 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.__proc = None
        self.__argsLists = []
        self.__workingDir = None
        self.__msgSuccess = msgSuccess
        self.__msgError = msgError
        self.__combinedOutput = combinedOutput
        self.__batchMode = False

        self.outputGroup.setTitle(text)

        if fixed:
            if isWindowsPlatform():
                self.resultbox.setFontFamily("Lucida Console")
            elif isMacPlatform():
                self.resultbox.setFontFamily("Menlo")
            else:
                self.resultbox.setFontFamily("Monospace")

        if not linewrap:
            self.resultbox.setLineWrapMode(QTextEdit.LineWrapMode.NoWrap)

        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.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
        )

        if self.__argsLists:
            args = self.__argsLists.pop(0)[:]
            self.startProcess(args[0], args[1:], self.__workingDir)

    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()

        if self.normal and self.__msgSuccess:
            self.resultbox.insertPlainText(self.__msgSuccess)
        elif not self.normal and self.__msgError:
            self.resultbox.insertPlainText(self.__msgError)
        self.resultbox.ensureCursorVisible()

    def startProcess(self, command, args, workingDir=None, showArgs=True):
        """
        Public slot used to start the process.

        @param command command to start
        @type str
        @param args list of arguments for the process
        @type list of str
        @param workingDir working directory for the process
        @type str
        @param showArgs flag indicating to show the arguments
        @type bool
        @return flag indicating a successful start of the process
        @rtype bool
        """
        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
        )

        if self.__batchMode:
            self.resultbox.append(80 * "#")

        if showArgs:
            self.resultbox.append(command + " " + " ".join(args))
            self.resultbox.append("")

        self.__proc = QProcess()
        if self.__combinedOutput:
            self.__proc.setProcessChannelMode(
                QProcess.ProcessChannelMode.MergedChannels
            )

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

        if workingDir:
            self.__proc.setWorkingDirectory(workingDir)
        self.__proc.start(command, 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(command),
            )
        else:
            self.inputGroup.setEnabled(True)
            self.inputGroup.show()
        return procStarted

    def startBatchProcesses(self, argsLists, workingDir=None):
        """
        Public slot used to start a batch of processes.

        @param argsLists list of lists of arguments for the processes
        @type list of list of str
        @param workingDir working directory for the process
        @type str
        @return flag indicating a successful start of the first process
        @rtype bool
        """
        self.__argsLists = argsLists[:]
        self.__workingDir = workingDir
        self.__batchMode = True

        # start the first process
        args = self.__argsLists.pop(0)[:]
        res = self.startProcess(args[0], args[1:], self.__workingDir)
        if not res:
            self.__argsLists = []

        return res

    def normalExit(self):
        """
        Public method to check for a normal process termination.

        @return flag indicating normal process termination
        @rtype bool
        """
        return self.normal

    def normalExitWithoutErrors(self):
        """
        Public method to check for a normal process termination without
        error messages.

        @return flag indicating normal process termination
        @rtype bool
        """
        return self.normal and self.errors.toPlainText() == ""

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

        It reads the output of the process, formats it and inserts it into
        the contents pane.
        """
        if self.__proc is not None:
            out = str(
                self.__proc.readAllStandardOutput(),
                Preferences.getSystem("IOEncoding"),
                "replace",
            )
            self.resultbox.insertPlainText(out)
            self.resultbox.ensureCursorVisible()

            QCoreApplication.processEvents()

    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()
        else:
            self.resultbox.insertPlainText(inputTxt)
            self.resultbox.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)