"""
Module for customizing af2 app features
Copyright Schrodinger, LLC. All rights reserved.
"""
import glob
import os
import schrodinger
from schrodinger import structure
from schrodinger.application.matsci import msprops
from schrodinger.application.matsci import jobutils
from schrodinger.project import utils as proj_utils
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtWidgets
from schrodinger.ui.qt import filedialog
from schrodinger.ui.qt import swidgets
from schrodinger.ui.qt import utils as qt_utils
from schrodinger.ui.qt.appframework2 import af2
from schrodinger.utils import fileutils
maestro = schrodinger.get_maestro()
[docs]class MatSciAppMixin:
"""
General mixin for MatSci panels
"""
WAM_LOAD_SELECTED = af2.input_selector.InputSelector.SELECTED_ENTRIES
WAM_LOAD_INCLUDED = af2.input_selector.InputSelector.INCLUDED_ENTRY
[docs] def initMixinOptions(self,
wam_input_state=None,
wam_load_function=None,
wam_run_singleton=True,
wam_set_panel_input_state=True):
"""
Initialize the options for the mixin
:param str wam_input_state: The input state to use to get the WAM entries
:param callable wam_load_function: Function that takes no arguments and
loads the entries into the panel.
:param bool wam_run_singleton: Whether the panel singleton should be
run (i.e. displayed) or not.
:param bool wam_set_panel_input_state: Whether to set panel's
input_selector state to the corresponding wam_input_state
"""
assert wam_input_state in (None, self.WAM_LOAD_SELECTED,
self.WAM_LOAD_INCLUDED)
self._wam_input_state = wam_input_state
self._wam_load_function = wam_load_function
self._wam_run_singleton = wam_run_singleton
self._wam_set_panel_input_state = wam_set_panel_input_state
[docs] @classmethod
def panel(cls, *entry_ids):
"""
Launch a singleton instance of the panel and load entries if applicable.
See `af2.App.panel` for more information.
:param tuple entry_ids: Entry ids to load into the panel
:raise RuntimeError: If the input state is invalid
"""
if not entry_ids:
return super().panel()
the_panel = super().panel(run=False)
if not hasattr(the_panel, '_wam_run_singleton'):
the_panel.error(
'"initMixinOptions" has not been called by the panel. Could'
' not open panel for entries.')
return the_panel
if the_panel._wam_run_singleton:
the_panel.run()
input_state = the_panel._wam_input_state
if input_state is None:
return the_panel
if input_state == cls.WAM_LOAD_INCLUDED:
command = 'entrywsincludeonly entry ' + str(entry_ids[0])
elif input_state == cls.WAM_LOAD_SELECTED:
command = 'entryselectonly entry ' + ' '.join(map(str, entry_ids))
if the_panel._wam_set_panel_input_state:
for selector in ('input_selector', '_if'):
input_selector = getattr(the_panel, selector, None)
if input_selector:
the_panel.input_selector.setInputState(input_state)
break
maestro.command(command)
if the_panel._wam_load_function:
the_panel._wam_load_function()
return the_panel
[docs]class ProcessPtStructuresApp(af2.App):
"""
Base class for panels that process pt structures and either replace them or
creates new entries
"""
REPLACE_ENTRIES = 'Replace current entries'
NEW_ENTRIES = 'Create new entries'
RUN_BUTTON_TEXT = 'Run' # Can be overwritten for custom button name
TEMP_DIR = fileutils.get_directory_path(fileutils.TEMP)
# Used for unittests. Should be overwritten in derived classes.
DEFAULT_GROUP_NAME = 'new_structures'
[docs] def setPanelOptions(self):
"""
Override the generic parent class to set panel options
"""
super().setPanelOptions()
self.input_selector_options = {
'file': False,
'selected_entries': True,
'included_entries': True,
'included_entry': False,
'workspace': False
}
[docs] def layOut(self):
"""
Lay out the widgets for the panel
"""
super().layOut()
layout = self.main_layout
# Any widgets for subclasses should be added to self.top_main_layout
self.top_main_layout = swidgets.SVBoxLayout(layout=layout)
output_gb = swidgets.SGroupBox("Output", parent_layout=layout)
self.output_rbg = swidgets.SRadioButtonGroup(
labels=[self.REPLACE_ENTRIES, self.NEW_ENTRIES],
layout=output_gb.layout,
command=self.outputTypeChanged,
nocall=True)
hlayout = swidgets.SHBoxLayout(layout=output_gb.layout, indent=True)
dator = swidgets.FileBaseNameValidator()
self.group_name_le = swidgets.SLabeledEdit(
"Group name: ",
edit_text=self.DEFAULT_GROUP_NAME,
validator=dator,
always_valid=True,
layout=hlayout)
self.outputTypeChanged()
layout.addStretch()
self.status_bar.showProgress()
self.app = QtWidgets.QApplication.instance()
# MATSCI-8244
size_hint = self.sizeHint()
size_hint.setWidth(410)
self.resize(size_hint)
# Set custom run button name if subclass defines it
self.bottom_bar.start_bn.setText(self.RUN_BUTTON_TEXT)
[docs] def outputTypeChanged(self):
"""
React to a change in output type
"""
self.group_name_le.setEnabled(
self.output_rbg.checkedText() == self.NEW_ENTRIES)
[docs] @af2.appmethods.start()
def myStartMethod(self):
"""
Process the selected or included rows' structures
"""
# Get rows
ptable = maestro.project_table_get()
input_state = self.input_selector.inputState()
if input_state == self.input_selector.INCLUDED_ENTRIES:
rows = [row for row in ptable.included_rows]
elif input_state == self.input_selector.SELECTED_ENTRIES:
rows = [row for row in ptable.selected_rows]
# Prepare progress bar
nouser = QtCore.QEventLoop.ExcludeUserInputEvents
num_structs = len(rows)
self.progress_bar.setValue(0)
self.progress_bar.setMaximum(num_structs)
self.app.processEvents(nouser)
structs_per_interval = max(1, num_structs // 20)
# Initialize output means
if self.output_rbg.checkedText() == self.REPLACE_ENTRIES:
modify = True
writer = None
else:
modify = False
file_path = os.path.join(self.TEMP_DIR,
self.group_name_le.text() + ".mae")
writer = structure.StructureWriter(file_path)
self.setUp()
# Process the structures
with qt_utils.wait_cursor:
for index, row in enumerate(rows, start=1):
with proj_utils.ProjectStructure(row=row, modify=modify) as \
struct:
try:
passed = self.processStructure(struct)
if passed and writer:
writer.append(struct)
except WorkflowError:
break
if not index % structs_per_interval:
self.progress_bar.setValue(index)
self.app.processEvents(nouser)
# Import file if applicable
if writer:
writer.close()
if os.path.exists(file_path): # No file will exist if all sts fail
ptable = maestro.project_table_get()
ptable.importStructureFile(file_path,
wsreplace=True,
creategroups='all')
fileutils.force_remove(file_path)
# Show 100%. Needed when num_structs is large and not a multiple of 20
self.progress_bar.setValue(num_structs)
self.app.processEvents(nouser)
# Panel-specific wrap up
self.wrapUp()
[docs] def setUp(self):
"""
Make any preparations required for processing structures
"""
pass
[docs] def processStructure(self, struct):
"""
Process each structure. Should be implemented in derived classes.
:param `structure.Structure` struct: The structure to process
"""
raise NotImplementedError
[docs] def wrapUp(self):
"""
Wrap up processing the structures
"""
pass
[docs] def reset(self):
"""
Reset the panel
"""
self.group_name_le.reset()
self.output_rbg.reset()
self.outputTypeChanged()
self.progress_bar._bar.reset(
) # af2.ProgressFrame doesn't have a reset method
[docs]class WorkflowError(ValueError):
"""
Custom exception for when the workflow should be stopped
"""
pass
[docs]class BaseAnalysisGui(object):
"""
Base class for other mixin gui class in the module
"""
[docs] def getIncludedEntry(self):
"""
Get included entry in maestro
:return `schrodinger.structure.Structure`: Structure in workspace
"""
# Tampering with licensing is a violation of the license agreement
if not jobutils.check_license(panel=self):
return
try:
struct = maestro.get_included_entry()
except RuntimeError as msg:
self.error(str(msg))
return
return struct
[docs] def resetPanel(self):
"""
Reset the panel variables and widgets set by this mixin
"""
self.struct = None
self.toggleStateMain(False)
self.load_btn.reset()
[docs] def toggleStateMain(self, state):
"""
Class to enable/disable widgets in panel when structure is added/removed.
:param bool state: If True it will enable the panel and will disable if
false
:raise NotImplementedError: Will raise error if it is not overwritten.
"""
raise NotImplementedError(
'toggleStateMain method should be implemented '
'to enable/disable panel widgets.')
[docs]class ViewerGuiMixin(MatSciAppMixin, BaseAnalysisGui):
"""
Class for extension of af2 to add widgets to gui for calculation viewer
"""
[docs] def setFilesUsingExt(self, path):
"""
Sets the data files using extension
:param path: The source/job path
:type path: str
"""
if not self.file_endings:
return
for file_ending in self.file_endings:
if not path or not os.path.exists(path):
filenames = []
else:
gen_filename = '*' + file_ending
filenames = glob.glob(os.path.join(path, gen_filename))
if len(filenames) == 1:
self.data_files[file_ending] = filenames[0]
continue
self.warning('The job directory for the workspace structure '
'could not be found. Please select the '
f'"*{file_ending}" data file.')
file_filter = f'Data file (*{file_ending})'
filename = filedialog.get_open_file_name(self,
filter=file_filter,
id=self.panel_id)
if not filename:
# User canceled hence clear all the data files collected
self.data_files = {}
return
self.data_files[file_ending] = filename
# Update dir for next file
path = os.path.dirname(filename)
[docs] def setFilesUsingStProp(self, path):
"""
Sets the data files using structure properties
:param path: The source/job path
:type path: str
"""
if not self.filename_props:
return None
for st_prop in self.filename_props:
# If the property is not loaded on the structure, then don't bother
# looking for the corresponding file
expected_file = self.struct.property.get(st_prop)
if not expected_file:
continue
# Get full path to the file
file_path = os.path.join(path, expected_file)
# Check for the file path, if the file is not found as user for
# for the file and update the source path location
if os.path.isfile(file_path):
filename = file_path
else:
self.warning(
f'The data file {expected_file} for property "{st_prop}" '
'could not be found')
file_ext = fileutils.splitext(expected_file)[-1]
file_filter = f'Data file (*{file_ext})'
filename = filedialog.get_open_file_name(self,
filter=file_filter,
id=self.panel_id)
if not filename:
# User canceled hence clear all the data files collected
self.data_files = {}
return
# Update dir for next file
path = os.path.dirname(filename)
self.data_files[st_prop] = filename
if not self.data_files:
self.warning(
f'No appropriate data files found for {self.struct.title}')
def _setDataFiles(self):
"""
Set the paths for the data files needed to setup the panel
"""
path = jobutils.get_source_path(self.struct)
self.setFilesUsingExt(path)
self.setFilesUsingStProp(path)
def _importStructData(self):
"""
Load the workspace file and data. Note create loadData method to load
the data
:return `schrodinger.structure.Structure`: Structure that was loaded
"""
self.resetPanel()
self.struct = self.getIncludedEntry()
if not self.struct:
return
# Get the data file
self._setDataFiles()
if not self.data_files:
return
if self.loadData() is False:
return
self.toggleStateMain(True)
return self.struct
[docs] def loadData(self):
"""
Class to load data to the panel
:raise NotImplementedError: Will raise error if it is not overwritten.
:rtype: bool or None
:return: False means the setup failed. True or None means it succeeded.
"""
raise NotImplementedError('loadData method should be implemented to '
'load results into the panel.')
[docs] def resetPanel(self):
"""
Reset the panel variables and widgets set by this mixin
"""
super().resetPanel()
self.data_files = {}