# -*- coding: utf-8 -*-
"""
Base classes used for LiveDesign import/export GUIs.
Copyright Schrödinger, LLC. All rights reserved.
"""
from contextlib import contextmanager
import requests
from schrodinger.Qt import QtCore
from schrodinger.Qt.QtGui import QColor
from schrodinger.ui.qt.appframework2 import tasks
from schrodinger.ui.qt.appframework2.af2 import StatusBar
from schrodinger.ui.qt.appframework2.wizards import BaseWizardPanel
from . import login
from . import panel_components
from .live_report_widget import LiveReportWidget
LR_ID = 'live_report_id'
LR_ROWS = 'lr_rows'
LR_COLUMNS = 'live_report_columns'
LR_STRUCTS = 'live_report_structs'
# User message colors
WARNING_COLOR = 'firebrick'
SUCCESS_COLOR = 'green'
INFO_COLOR = 'teal'
[docs]class HandledError(Exception):
pass
[docs]class ImportExportBasePanel(BaseWizardPanel):
"""
Main panel for specifying how to import or export the data to/from LD.
The following signals may be emitted for updating status text and color
during export process.
"""
allow_add_live_reports = False
statusChanged = QtCore.pyqtSignal(str, int, QColor)
warningRaised = QtCore.pyqtSignal(str, str)
errorRaised = QtCore.pyqtSignal(str, str)
questionRaised = QtCore.pyqtSignal(str, str)
infoUpdated = QtCore.pyqtSignal(str, str)
TITLE_BASE = ''
[docs] def initSetOptions(self):
super().initSetOptions()
self.ld_client = None
self.ld_models = None
[docs] def initSetUp(self):
super().initSetUp()
self.lr_widget = LiveReportWidget(
parent=self, allow_add_live_reports=self.allow_add_live_reports)
# Add task bar and status bar widgets
self.setupTaskWidget()
self.setupStatusBar()
self.setupTaskMessagingSignals()
self._setUpProjectsAndLiveReports()
[docs] def initSetDefaults(self):
super().initSetDefaults()
self.lr_widget.initSetDefaults()
[docs] def setupStatusBar(self):
"""
Set up the status bar with the help topic and product for the help
icon. A separate parent for the status bar is created so the help
dialog is opened correctly.
Note: The help topic is set here so an auto help dialog button isn't
added next to the 'Back' button.
"""
status_bar_parent = panel_components.StatusBarDialog(self)
# FIXME Set correct help topic
status_bar_parent.help_topic = self.HELP_TOPIC
self.status_bar = StatusBar(status_bar_parent)
self.status_bar.setMaximumHeight(25)
def _setUpProjectsAndLiveReports(self):
"""
Sets up LD projects and live reports combo boxes. Should be implemented
in subclasses.
"""
pass
[docs] def setupTaskMessagingSignals(self):
"""
Connect the signals for showing QMessageBoxes to the user or updating
the status bar message from processTaskMessage().
"""
self.statusChanged.connect(self.status_bar.showMessage)
self.warningRaised.connect(lambda title, msg: self.warning(msg, title))
self.errorRaised.connect(lambda title, msg: self.error(msg, title))
self.questionRaised.connect(
lambda title, msg: self.question(msg, title))
self.infoUpdated.connect(lambda title, msg: self.info(msg, title))
[docs] def initLayOut(self):
"""
See parent class for method documentation.
"""
super().initLayOut()
self.ui.lr_widget_layout.addWidget(self.lr_widget)
self.ui.bottom_layout.insertWidget(0, self.task_bar)
self.ui.status_layout.addWidget(self.status_bar)
# Remove 'Finish' button for run button from task bar is used instead.
self.next_btn.hide()
# Also remove 'Cancel' button.
self.cancel_btn.hide()
[docs] def processPrevPanel(self, state):
"""
This is called in BaseWizardPanel.run method.
:param state: State of the previous panel
:type state: `af2.settings.PanelState`
"""
self.ld_client = state[login.CLIENT]
self.ld_models = state[login.MODELS]
host = login.get_host()
window_title = f'{self.TITLE_BASE} ({host})'
self.setWindowTitle(window_title)
[docs] def processTaskMessage(self, message_type, text, options=None, runner=None):
"""
This method is used as a callback to the task runner, and emits
signals to interact with the main GUI. Signals are used here,
instead of directly interacting with the GUI, which causes threading
and parenting issues.
:param message_type: the type of message being sent
:type message_type: int
:param text: the main text to show the user
:type text: str
:param options: extra options
:type options: dict
:param runner: Task runner.
:type runner: `tasks.ThreadRunner`
"""
if options is None:
options = {}
caption = options.get('caption', '')
if message_type == tasks.WARNING:
self.warningRaised.emit(caption, text)
elif message_type == tasks.ERROR:
self.errorRaised.emit(caption, text)
elif message_type == tasks.QUESTION:
self.questionRaised.emit(caption, text)
elif message_type == tasks.INFO:
self.infoUpdated.emit(caption, text)
elif message_type == tasks.STATUS:
if not hasattr(self, 'status_bar'):
return
timeout = options.get('timeout', 3000)
color = options.get('color')
if not color:
# Set a default black color. Note: this cannot be set through
# the dict get method as options might hold None as the value.
color = QColor(INFO_COLOR)
if (text.startswith('Task completed') or
text.startswith('Task status:')):
# Do not show the default completion messages from task runner.
return
self.statusChanged.emit(text, timeout, color)
else:
raise ValueError('Unexpected message_type %d for message:\n%s' %
(message_type, text))
[docs] def processSettings(self, settings=None, runner=None):
"""
See parent class for method documentation.
"""
return self.getAliasedSettings()
[docs] def reject(self):
"""
See parent class for more info. Resets the custom LiveReport combo box
as the SettingsPanelMixin will not know how to set the default option.
"""
self.lr_widget.initSetDefaults()
super().reject()
[docs] def run(self, prev_panel_state=None):
super().run(prev_panel_state)
self.lr_widget.refresh()
[docs] def closeEvent(self, event):
"""
After the panel is closed, forget that the current live report URL has
been used before. This allows the URL to be re-evaluated when the panel
is opened again.
"""
self.initSetDefaults()
super().closeEvent(event)
@property
def project_name(self):
return self.lr_widget.getProjectName()
@property
def project_id(self):
return self.lr_widget.getProjectID()
@property
def live_report_id(self):
return self.lr_widget.getLiveReportID()
[docs]class ImportExportThreadRunner(tasks.ThreadRunner):
process_type = 'import/export'
use_event_loop = True
[docs] @contextmanager
def errorHandler(self):
"""
A context manager for handling errors - catches a few specific error
types, creating an error message as a function of the error type and
calling. Reraises HandledError(error_msg) if any errors are caught.
:raise: HandledError if any errors are caught.
"""
error_msg = ''
try:
yield
except requests.exceptions.ConnectionError as e:
error_msg = 'Maestro was unable to connect to the LiveDesign ' \
'server due to a Connection Error.'
except requests.exceptions.Timeout as e:
error_msg = 'Attempt to connect to the LiveDesign server timed out.'
except Exception as e: # client.py raises a bare exception
error_msg = str(e)
if error_msg:
raise HandledError(error_msg)
[docs] def setErrorStatus(self, curr_name, error_msg):
"""
Update self.status and self.error with the proper error messages.
:param curr_name: the name of the task that is reporting the error
:param error_msg: the error message
"""
self.status('{0}: {1}'.format(curr_name, error_msg),
timeout=0,
color=QColor(WARNING_COLOR))
error_str = ('An error has occurred during the {}'
' process: {}'.format(self.process_type, error_msg))
self.error(error_str)