#!/usr/bin/env python #============================================================== # # Abs: Module that implements a wizard in QT # # Name: wizard.py # # Facility: SLAC Controls (Development) # # Auth: 18-May-2016, M. Shankar (mshankar) #-------------------------------------------------------------- # Mods: (Latest to oldest) # 19-Mar-2024, A. Le (anle) # Use software rendering if running over SSH # 24-Jul-2019, A. Le (anle) # Use logging module instead of custom logger # 23-Jul-2019, A. Le (anle) # Use custom logger and attach log to email # #============================================================== ''' Python based wizard implemented on top of the webkit in QT. We start a HTTP listener on a random port; and then launch a WebKit panel against this. Content is fed from the staticcontent folder. Responses to GET's come from the staticcontent folder ''' from __future__ import print_function import os import sys import shutil import string import random import json import threading import types import mimetypes import logging from six.moves.urllib.parse import parse_qs, urlparse from six import iteritems try: import http.server as SimpleHTTPServer import socketserver as SocketServer from PyQt5.QtWebEngineWidgets import QWebEngineView as QWebView, QWebEnginePage as QWebPage, QWebEngineSettings as QWebSettings from PyQt5.QtWidgets import QApplication from PyQt5.QtCore import QUrl except ImportError: import SimpleHTTPServer import SocketServer from PyQt4.QtWebKit import QWebView, QWebSettings, QWebPage from PyQt4.QtGui import QApplication from PyQt4.QtCore import QUrl logger = logging.getLogger() global enableLogs enableLogs = False # Fix for rendering over SSH, taken from https://github.com/slaclab/pydm/pull/661/files def is_ssh_session(): """ Whether or not this is a SSH session. Returns ------- bool True if it is a ssh session, False otherwise. """ return os.getenv("SSH_CONNECTION") is not None def setup_renderer(): """ This utility function reverts the renderer to Software rendering if it is running in a SSH session. """ if is_ssh_session(): logger.info("Using cram via SSH. Reverting to Software Rendering.") from qtpy.QtCore import QCoreApplication, Qt from qtpy.QtQuick import QQuickWindow, QSGRendererInterface QCoreApplication.setAttribute(Qt.AA_UseSoftwareOpenGL) QQuickWindow.setSceneGraphBackend(QSGRendererInterface.Software) class WizardContext: ''' Context for the wizard. We use the same object for the entire wizard This also implements a bunch pattern; all keyword arguments are accessible as attributes of the context. ''' def __init__(self, wizard, steps, contentFolder, **kwargs): self.currentStep = 0 self.data = {} self.wizard = wizard self.steps = steps self.contentFolder = contentFolder self.UIsteps = [] # If this is true, we are running a silent install and no UI's should be shown. # All the needed variables are expected to come from the file used to drive the silent install self.silentInstall = False # Bunch pattern; all keywords are added as attributes of this wizard context instance. self.__dict__.update(kwargs) def nextStep(self): self.currentStep = self.currentStep + 1 if enableLogs: print("Next step in WizardContext {0}".format(self.currentStep)) def previousStep(self): self.currentStep = self.currentStep - 1 if enableLogs: print("Previous step in WizardContext {0}".format(self.currentStep)) def skipNextStep(self): self.currentStep = self.currentStep + 2 if enableLogs: print("Skipping next step in WizardContext {0}".format(self.currentStep)) def gotToStep(self, stepId): self.currentStep = stepId if enableLogs: print("Going to step in WizardContext {0}".format(self.currentStep)) def pushUIStep(self): '''The current step generated UI; so push it to the UI stack''' if self.currentStep not in self.UIsteps: self.UIsteps.append(self.currentStep) def previousUIStep(self): self.UIsteps.pop() self.currentStep = self.UIsteps[-1] def resetPreviousSteps(self): self.UIsteps = [] def resetForSilentInstall(self, initialSteps): '''Reset the wizardContext for the next IOC/Item in a silent install. Be very careful to pass in copies of the steps list; otherwise we'll keep adding steps to the same list. This will make the list grow continuously and you'll get bizarre results. ''' self.steps = initialSteps self.currentStep = 0 self.UIsteps = [] class TemplateWithValues: '''Every action returns one of these. The templateName is relative to the rootFolder for the wizard; contains a python String template. values is a name/value dict used in template replacement''' def __init__(self, templateName, values, wizardContext): self.templateName = templateName self.values = values self.context = wizardContext def fetchTemplateAndGenerateStaticContent(self, rootFolder): # We first offer the client the chance to override templates. contentFileName = os.path.join(self.context.contentFolder, self.templateName) if os.path.exists(contentFileName): if enableLogs: print("Serving template from client {0}".format(contentFileName)) return self.templateReplace(contentFileName) contentFileName = os.path.join(os.path.dirname(__file__), "wizardcontent", self.templateName) if os.path.exists(contentFileName): if enableLogs: print("Serving template from myself {0}".format(contentFileName)) return self.templateReplace(contentFileName) raise Exception("Cannot find template " + self.templateName) def templateReplace(self, contentFileName): with open(contentFileName, 'r') as f: data = f.read() if enableLogs: print("List of previous UI steps {0}".format(self.context.UIsteps)) valscopy = self.values.copy() valscopy.update(self.context.data) valscopy['currentStep'] = self.context.currentStep valscopy['previousStepExists'] = True if len(self.context.UIsteps) > 1 else False return data.replace("//{{jsvars}}", "var serverState = " + json.dumps(valscopy) + ";" ) class WizardStep: ''' Wizards are a list of steps; this represents such a step. There are various type of steps; to create a wizard, you create a sequence of wizardsteps. ''' def enter(self, wizardContext): '''Execute this step; when executing the step; the context's current step will be set to this step.''' pass class CallableWizardStep(WizardStep): '''This lets you implement business logic using a step. Construct using a function and this will be called as entry into this step''' def __init__(self, callable): self.callable = callable def enter(self, wizardContext): return self.callable(wizardContext) class MyHTTPHandler(SimpleHTTPServer.SimpleHTTPRequestHandler): wizard = None def do_GET(self): if self.path.startswith("/wizard.html"): content = next(self.wizard.execute()) if content: self.send_response(200) self.send_header("Content-type", "text/html") self.send_header("Content-length", len(content)) self.end_headers() self.wfile.write(content.encode("utf-8")) else: if enableLogs: print("All done. Shutting down the wizard") print(self.wizard.context.data) self.browser.close() elif self.path.startswith("/updateState"): query_components = parse_qs(urlparse(self.path).query) for (k,v) in iteritems(query_components): if enableLogs: print("Setting server state {0} to {1} in context".format(k, v[0])) self.wizard.context.data[k] = v[0] self.sendOk() elif self.path.startswith("/addToState"): query_components = parse_qs(urlparse(self.path).query) for (k,v) in iteritems(query_components): if enableLogs: print("Appending to server state {0} the value {1} in context".format(k, v[0])) if k not in self.wizard.context.data: self.wizard.context.data[k] = [] self.wizard.context.data[k].append(v[0]) self.sendOk() elif self.path.startswith("/nextStep"): if enableLogs: print("Client made call to go to next step") self.wizard.context.nextStep() self.sendOk() elif self.path.startswith("/previousStep"): if enableLogs: print("Client made call to go to previous step") self.wizard.context.previousUIStep() self.sendOk() else: contentFileName = os.path.join(os.path.dirname(__file__), "staticcontent" + urlparse(self.path).path) if(os.path.exists(contentFileName)): if enableLogs: print("Serving {0}".format(contentFileName)) self.sendFile(contentFileName) return contentFileName = os.path.join(self.wizard.context.contentFolder, "staticcontent" + urlparse(self.path).path) if(os.path.exists(contentFileName)): if enableLogs: print("Serving {0}".format(contentFileName)) self.sendFile(contentFileName) return def sendOk(self): content = "Ok" self.send_response(200) self.send_header("Content-type", "text/html") self.send_header("Content-length", len(content)) self.end_headers() self.wfile.write(content.encode("utf-8")) def sendFile(self, contentFileName): with open(contentFileName, 'rb') as f: self.send_response(200) self.send_header("Content-type", mimetypes.guess_type(contentFileName)[0]) self.send_header("Content-length", os.path.getsize(contentFileName)) self.end_headers() shutil.copyfileobj(f, self.wfile) def log_message(self, format, *args): if enableLogs: sys.stderr.write("%s - - [%s] %s\n" % (self.address_string(), self.log_date_time_string(), format%args)) class SeparateThreadHTTPServer: def startServer(self, PORT, wizard, browser): def launchHTTPServer(httpd): httpd.serve_forever() self.handler = MyHTTPHandler self.handler.wizard = wizard self.handler.browser = browser self.httpd = SocketServer.TCPServer(("", PORT), self.handler) print("HTTP server serving at port {0}".format(PORT)) self.t1 = threading.Thread(target=launchHTTPServer, args=(self.httpd,)) self.t1.start() def shutdown(self): self.httpd.shutdown() self.t1.join(10) def showConsoleLogInTerminal(browser): ''' Display the browser log on the terminal. However, this seems to cause a core dump on exit''' class MyWebPage(QWebPage): def javaScriptConsoleMessage(self, msg, line, source): print('%s line %d: %s' % (source, line, msg)) page = MyWebPage() browser.setPage(page) class TemplateWithValuesWizardStep(WizardStep): def __init__(self, pageName, values): self.pageName = pageName self.values = values def enter(self, wizardContext): return TemplateWithValues(self.pageName, self.values, wizardContext) class Wizard: ''' Implements a web based wizard using a QTWebView. Pass in a list of steps and a folder with static content In addition, the keyword arguments into the wizard are passed to the wizardContext which implements a bunch pattern. ''' def __init__(self, title, contentFolder, steps, verbose=False, **kwargs): global enableLogs self.context = WizardContext(self, steps, contentFolder, **kwargs) self.context.data['wizardTitle'] = title self.showConsoleOnTerminal = False enableLogs = verbose def execute(self): while self.context.currentStep < len(self.context.steps): curStepId = self.context.currentStep if enableLogs: print("Executing stepid {0}".format(curStepId) ) curStep = self.context.steps[curStepId] logger.debug("Executing step #{0} - {1}".format(curStepId, curStep.__class__.__name__)) curTemplate = curStep.enter(self.context) if curTemplate == True: if enableLogs: print("Step {0} returned True; moving on to the next step".format(curStepId)) self.context.nextStep() continue else: if self.context.silentInstall: if enableLogs: print("Skipping a UI step for silent install") self.context.nextStep() continue self.context.pushUIStep() yield curTemplate.fetchTemplateAndGenerateStaticContent(self.context.contentFolder) yield None def launchWizard(self): if self.context.silentInstall: next(self.execute()) return setup_renderer() PORT = random.randrange(15000, 16000) app = QApplication(sys.argv) # PyQt5 does not support DeveloperExtrasEnabled try: QWebSettings.globalSettings().setAttribute(QWebSettings.DeveloperExtrasEnabled, True) except AttributeError: pass browser = QWebView() browser.setFixedSize(800, 640); # Uncomment the following line if you want to see the console logs in the terminal. # Note that this causes QT to dump core. if self.showConsoleOnTerminal: showConsoleLogInTerminal(browser) server = SeparateThreadHTTPServer() server.startServer(PORT, self, browser) browser.load(QUrl("http://localhost:" + str(PORT) + "/wizard.html")) browser.show() app.exec_() server.shutdown() if enableLogs: print("Shutdown the server final context data follows") print(json.dumps(self.context.data, sort_keys=True, indent=4, separators=(',', ': '))) if __name__ == '__main__': Wizard.launchWizard()