Mercurial Repositories > Flask Project / file revision

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

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

"""
Module implementing a dialog to run the Flask server.
"""

import re

from PyQt6.QtCore import pyqtSlot, Qt, QProcess, QTimer
from PyQt6.QtGui import QTextCharFormat
from PyQt6.QtWidgets import QDialog, QDialogButtonBox, QMenu

from EricWidgets import EricMessageBox
from EricWidgets.EricApplication import ericApp

from .Ui_RunServerDialog import Ui_RunServerDialog

from . import AnsiTools
from .ServerStartOptionsDialog import ServerStartOptionsDialog

import UI.PixmapCache


# TODO: should this be placed into the sidebar as a sidebar widget?
class RunServerDialog(QDialog, Ui_RunServerDialog):
    """
    Class implementing a dialog to run the Flask server.
    """
    def __init__(self, plugin, project, parent=None):
        """
        Constructor
        
        @param plugin reference to the plug-in object
        @type PluginProjectFlask
        @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.__plugin = plugin
        self.__project = project
        
        self.__serverOptions = {
            "development": False
        }
        
        self.__process = None
        self.__serverUrl = ""
        
        self.__ansiRe = re.compile(r"(\x1b\[\d+m)")
        
        self.__urlRe = re.compile(r" \* Running on ([^(]+) \(.*")
        
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close).setEnabled(True)
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close).setDefault(True)
        
        self.__defaultTextFormat = self.outputEdit.currentCharFormat()
        
        self.__initActionsMenu()
    
    def __initActionsMenu(self):
        """
        Private method to populate the actions button menu.
        """
        self.__actionsMenu = QMenu()
        self.__actionsMenu.setTearOffEnabled(True)
        self.__actionsMenu.setToolTipsVisible(True)
        self.__actionsMenu.aboutToShow.connect(self.__showActionsMenu)
        
        # re-start server
        self.__actionsMenu.addAction(
            self.tr("Re-start Server"),
            self.__restartServer)
        self.__restartModeAct = self.__actionsMenu.addAction(
            self.tr("Re-start Server"),
            self.__restartServerDifferentMode)
        self.__actionsMenu.addSeparator()
        # re-start server with options
        self.__actionsMenu.addAction(
            self.tr("Re-start Server with Options"),
            self.__restartServerWithOptions)
        
        self.menuButton.setIcon(UI.PixmapCache.getIcon("actionsToolButton"))
        self.menuButton.setMenu(self.__actionsMenu)
    
    @pyqtSlot()
    def __showActionsMenu(self):
        """
        Private slot handling the actions menu about to be shown.
        """
        if self.__serverOptions["development"]:
            self.__restartModeAct.setText(
                self.tr("Re-start Server (Production Mode)"))
        else:
            self.__restartModeAct.setText(
                self.tr("Re-start Server (Development Mode)"))
    
    def startServer(self, development=False, askForOptions=False):
        """
        Public method to start the Flask server process.
        
        @param development flag indicating development mode
        @type bool
        @param askForOptions flag indicating to ask for server start options
            first
        @type bool
        @return flag indicating success
        @rtype bool
        """
        self.__serverOptions["development"] = development
        
        if askForOptions:
            dlg = ServerStartOptionsDialog(self.__serverOptions)
            if dlg.exec() != QDialog.DialogCode.Accepted:
                return False
            
            self.__serverOptions.update(dlg.getDataDict())
        
        workdir, env = self.__project.prepareRuntimeEnvironment(
            development=self.__serverOptions["development"])
        if env is not None:
            command = self.__project.getFlaskCommand()
            
            self.__process = QProcess()
            self.__process.setProcessEnvironment(env)
            self.__process.setWorkingDirectory(workdir)
            self.__process.setProcessChannelMode(
                QProcess.ProcessChannelMode.MergedChannels)
            
            self.__process.readyReadStandardOutput.connect(self.__readStdOut)
            self.__process.finished.connect(self.__processFinished)
            
            self.outputEdit.clear()
            
            args = ["run"]
            if "host" in self.__serverOptions and self.__serverOptions["host"]:
                args += ["--host", self.__serverOptions["host"]]
            if "port" in self.__serverOptions and self.__serverOptions["port"]:
                args += ["--port", self.__serverOptions["port"]]
            if "cert" in self.__serverOptions and self.__serverOptions["cert"]:
                args += ["--cert", self.__serverOptions["cert"]]
            if "key" in self.__serverOptions and self.__serverOptions["key"]:
                args += ["--key", self.__serverOptions["key"]]
            
            self.__process.start(command, args)
            ok = self.__process.waitForStarted(10000)
            if not ok:
                EricMessageBox.critical(
                    None,
                    self.tr("Run Flask Server"),
                    self.tr("""The Flask server process could not be"""
                            """ started."""))
            else:
                self.buttonBox.button(
                    QDialogButtonBox.StandardButton.Close).setEnabled(False)
                self.stopServerButton.setEnabled(True)
                self.stopServerButton.setDefault(True)
                self.startBrowserButton.setEnabled(True)
        else:
            ok = False
        
        return ok
    
    def closeEvent(self, evt):
        """
        Protected method handling a close event.
        
        @param evt reference to the close event
        @type QCloseEvent
        """
        self.on_stopServerButton_clicked()
        evt.accept()
    
    @pyqtSlot()
    def __readStdOut(self):
        """
        Private slot to add the server process output to the output pane.
        """
        if self.__process is not None:
            out = str(self.__process.readAllStandardOutput(), "utf-8")
            if not self.__serverUrl:
                urlMatch = self.__urlRe.search(out)
                if urlMatch:
                    self.__serverUrl = urlMatch.group(1)
            
            for txt in self.__ansiRe.split(out):
                if txt.startswith("\x1b["):
                    color = int(txt[2:-1])      # strip off ANSI command parts
                    if color == 0:
                        self.outputEdit.setCurrentCharFormat(
                            self.__defaultTextFormat)
                    elif 30 <= color <= 37:
                        brush = AnsiTools.getColor(
                            self.__plugin.getPreferences("AnsiColorScheme"),
                            color - 30)
                        if brush is not None:
                            charFormat = QTextCharFormat(
                                self.__defaultTextFormat)
                            charFormat.setForeground(brush)
                            self.outputEdit.setCurrentCharFormat(charFormat)
                else:
                    self.outputEdit.insertPlainText(txt)
    
    @pyqtSlot()
    def __processFinished(self):
        """
        Private slot handling the finishing of the server process.
        """
        if (
            self.__process is not None and
            self.__process.state() != QProcess.ProcessState.NotRunning
        ):
            self.__process.terminate()
            QTimer.singleShot(2000, self.__process.kill)
            self.__process.waitForFinished(3000)
        
        self.__process = None
        
        self.startBrowserButton.setEnabled(False)
        self.stopServerButton.setEnabled(False)
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close).setEnabled(True)
        self.buttonBox.button(
            QDialogButtonBox.StandardButton.Close).setDefault(True)
        self.buttonBox.button(QDialogButtonBox.StandardButton.Close).setFocus(
            Qt.FocusReason.OtherFocusReason)
    
    @pyqtSlot()
    def on_stopServerButton_clicked(self):
        """
        Private slot to stop the running server.
        """
        self.__processFinished()
    
    @pyqtSlot()
    def on_startBrowserButton_clicked(self):
        """
        Private slot to start a web browser with the server URL.
        """
        if self.__plugin.getPreferences("UseExternalBrowser"):
            import webbrowser
            webbrowser.open(self.__serverUrl)
        else:
            ericApp().getObject("UserInterface").launchHelpViewer(
                self.__serverUrl)
    
    @pyqtSlot()
    def __restartServer(self):
        """
        Private slot to restart the server process.
        """
        # step 1: stop the current server
        self.on_stopServerButton_clicked()
        
        # step 2: start a new server
        self.startServer(development=self.__serverOptions["development"])
    
    @pyqtSlot()
    def __restartServerDifferentMode(self):
        """
        Private slot to restart the server process with the opposite mode.
        """
        # step 1: stop the current server
        self.on_stopServerButton_clicked()
        
        # step 2: start a new server
        self.startServer(development=not self.__serverOptions["development"])
    
    @pyqtSlot()
    def __restartServerWithOptions(self):
        """
        Private slot to restart the server asking for start options.
        """
        # step 1: stop the current server
        self.on_stopServerButton_clicked()
        
        # step 2: start a new server
        self.startServer(development=self.__serverOptions["development"],
                         askForOptions=True)