import json
import os
from schrodinger.models import mappers
from schrodinger.models import parameters
from schrodinger.Qt import QtCore
from schrodinger.ui.qt import basewidgets
from schrodinger.utils import fileutils
from . import constants
from . import data_classes
from . import panel_components
from . import save_mapping_dialog_ui
# LiveDesign export JSON properties
ASSAY = 'assay'
DATA_NAME = 'data_name'
FAMILY_NAME = 'family_name'
USER_NAME = 'user_name'
ROWS = 'rows'
DECIMAL = 'decimal'
OPTION = 'option'
ASSAY_FOLDER_PATH = 'assay_folder_path'
HOST = 'host'
LD_PROJECT = 'ld_project'
[docs]class ExportMapTypeError(TypeError):
"""
Exception to raise when the panel attempts to load a map that has host and
project values that do not match those of the panel.
"""
[docs]class ExportMap:
"""
A wrapper for holding fields required for repopulating the export table.
:cvar EXPORT_ROW_CLASS: class to use for the export table row objects
:vartype EXPORT_ROW_CLASS: panel_components.ExportRow
"""
EXPORT_ROW_CLASS = panel_components.ExportRow
[docs] def __init__(self,
map_name,
map_file_path,
host=None,
proj_id=None,
export_rows=None):
"""
:param map_name: name of map
:type map_name: str
:param map_file_path: file path to map
:type map_file_path: str
:param host: LD server - used to confirm the same conditions exist for
a successful application of the map.
:type host: str or None
:param proj_id: project ID used to confirm the same conditions exist
for a successful application of the map.
:type proj_id: str or None
:param export_rows: rows of the export table
:type export_rows: [panel_components.ExportRow] or None
"""
self.map_name = map_name
self.map_file_path = map_file_path
self.host = host
self.proj_id = proj_id
self.export_rows = export_rows
[docs] def writeMapToFile(self):
"""
Write the mapping as json to file with name 'self.map_name'.
"""
row_maps = self._getRowMaps()
data = {HOST: self.host, LD_PROJECT: self.proj_id, ROWS: row_maps}
with open(self.map_file_path, 'w') as json_file:
json.dump(data, json_file, indent=4)
def _getRowMaps(self):
"""
:return: a list of dictionaries containing data for each export row in
the table
:rtype: List[Dict]
"""
row_maps = []
for exp_row in self.export_rows:
row_map = {
DATA_NAME: exp_row.ld_data.data_name,
FAMILY_NAME: exp_row.ld_data.family_name,
USER_NAME: exp_row.ld_data.user_name or '',
ASSAY: exp_row.assay,
constants.LD_PROP_ENDPOINT: exp_row.endpoint,
constants.LD_PROP_UNITS: exp_row.units,
DECIMAL: exp_row.decimal,
OPTION: exp_row.option,
ASSAY_FOLDER_PATH: exp_row.assay_folder_path
}
row_maps.append(row_map)
return row_maps
[docs] def readMapFile(self):
"""
Read mappings file and set up
"""
with open(self.map_file_path, "r") as json_file:
data = json.load(json_file)
self.host = data[HOST]
self.proj_id = data[LD_PROJECT]
self.export_rows = self._getExportRows(data[ROWS])
def _is3DRow(self, row_map):
'''
:param row_map: Dict of data row read from map file
:type row_map: `dict`
:return: True if row is 3D row else False
:rtype: `bool`
'''
return row_map[FAMILY_NAME] == constants.FAMNAME_3D_DATA
def _getExportRows(self, row_maps):
"""
Given a list of row data dictionaries, return a list of properly-
formatted row objects.
:param row_maps: the mapping file data dictionary
:type row_maps: List[Dict]
:return: a list of export rows in the table
:rtype: List[panel_components.ExportRow]
"""
export_rows = []
for row_map in row_maps:
data_name = row_map[DATA_NAME]
user_name = row_map[USER_NAME]
family_name = row_map[FAMILY_NAME]
ld_data = data_classes.LDData(data_name=data_name,
family_name=family_name,
user_name=user_name,
requires_3d=self._is3DRow(row_map))
exp_row = self.EXPORT_ROW_CLASS(
ld_data=ld_data,
assay=row_map[ASSAY],
endpoint=row_map[constants.LD_PROP_ENDPOINT],
units=row_map[constants.LD_PROP_UNITS],
decimal=row_map[DECIMAL],
# PANEL-16830 row_map will not have OPTION, if mappings saved
# before 2020-2
option=row_map.get(OPTION),
assay_folder_path=row_map[ASSAY_FOLDER_PATH])
export_rows.append(exp_row)
return export_rows
[docs]class ExportMapManager:
"""
Interface to manage Export Map files.
:cvar EXPORT_MAP_CLASS: the class to use for mapping export rows
:vartype EXPORT_MAP_CLASS: ExportMap
"""
EXPORT_MAP_CLASS = ExportMap
MAESTRO_EXPORT_MAP_DIR_PATH = os.path.join(
fileutils.get_directory_path(fileutils.USERDATA),
'maestro_export_mappings')
[docs] def __init__(self, host, proj_id):
"""
Creates the custom dir to store mappings, if it doesn't already exist.
:param host: LD server being used to export to.
:type host: str
:param proj_id: LD project to export to.
:type proj_id: str
"""
fileutils.mkdir_p(self.MAESTRO_EXPORT_MAP_DIR_PATH)
self.host = host
self.proj_id = proj_id
self.recently_used_maps = sorted(self.getAvailableMappings())
[docs] def saveNewMapping(self, map_name, export_rows):
"""
Save the given mapping to disk under given map_name.
:param map_name: name to save the new map under
:type map_name: str
:param export_rows: of the export table used to generate the map
:type export_rows: [panel_components.ExportRow]
"""
map_fp = self.getMapFilePath(map_name)
new_map = self.EXPORT_MAP_CLASS(map_name, map_fp, self.host,
self.proj_id, export_rows)
new_map.writeMapToFile()
self.recently_used_maps.insert(0, map_name)
[docs] def openMapping(self, map_file):
"""
Open and read the mapping from the map_file
:param map_file: Path to a map file
:type map_file: str
:return: mapping of rows of the export table
:rtype: [panel_components.ExportRow]
"""
map_name = os.path.splitext(os.path.basename(map_file))[0]
op_map = self.EXPORT_MAP_CLASS(map_name, map_file)
op_map.readMapFile()
# PANEL-16830
# map_file may contain an integer value for the project id if saved by
# verions before 2020-2 build-049. So explicitly comparing as string
if op_map.host != self.host or str(op_map.proj_id) != self.proj_id:
raise ExportMapTypeError(
'Export Mapping Error: The following mapping, {0}, is '
'configured for the host: {1}, and project: {2}.'.format(
map_name, op_map.host, op_map.proj_id))
if map_name in self.recently_used_maps:
self.recently_used_maps.remove(map_name)
self.recently_used_maps.insert(0, map_name)
return op_map.export_rows
[docs] def deleteMapping(self, map_name):
"""
Delete the mapping under giving map_name from disk.
:param map_name: name of map to delete
:type map_name: str
"""
del_map_file_path = self.getMapFilePath(map_name)
fileutils.force_remove(del_map_file_path)
if map_name in self.recently_used_maps:
self.recently_used_maps.remove(map_name)
[docs] def getAvailableMappings(self):
"""
Return a list of mappings available for use.
:return: list of available mappings
:rtype: list of str
"""
files = os.listdir(self.MAESTRO_EXPORT_MAP_DIR_PATH)
mapping_names = []
for fn in files:
mapping_names.append(os.path.splitext(fn)[0])
return mapping_names
[docs] def getMostRecentMappings(self, cutoff=10):
"""
Returns the most recent map files opened/created
:param cutoff: How many mappings to return
:type cutoff: int
:return: list of available mappings
:rtype: list of str
"""
available_mappings = self.getAvailableMappings()
for map_name in reversed(self.recently_used_maps):
if map_name in available_mappings:
map_idx = available_mappings.index(map_name)
available_mappings.pop(map_idx)
available_mappings.insert(0, map_name)
return available_mappings[:cutoff]
[docs] def getMapFilePath(self, map_name):
"""
Given the map name, generate the file path.
:param map_name: name of map to generate file path for
:type map_name: str
:return: file path to map on disk
:rtype: str
"""
return os.path.join(self.MAESTRO_EXPORT_MAP_DIR_PATH,
map_name + '.json')
[docs]class SaveMappingDialogModel(parameters.CompoundParam):
mapping_name: str = ''
[docs]class SaveMappingDialog(mappers.MapperMixin, basewidgets.BaseDialog):
"""
Dialog for saving a mapping name
"""
ui_module = save_mapping_dialog_ui
model_class = SaveMappingDialogModel
saveMappingRequested = QtCore.pyqtSignal(str)
[docs] def initSetOptions(self):
super().initSetOptions()
self.std_btn_specs = {
self.StdBtn.Ok: (self.saveMapping, 'Save'),
self.StdBtn.Cancel: self.clearMappingName,
}
[docs] def initSetUp(self):
super().initSetUp()
self.setWindowTitle('Save New Mapping')
[docs] def defineMappings(self):
M = self.model_class
ui = self.ui
return [
(ui.mapping_name_le, M.mapping_name),
]
[docs] def saveMapping(self):
self.saveMappingRequested.emit(self.model.mapping_name)
self.clearMappingName()
[docs] def clearMappingName(self):
self.model.reset()