import contextlib
import os
from schrodinger import get_maestro
from schrodinger.application.job_monitor import file_browser_ui
from schrodinger.application.job_monitor import job_monitor_models
from schrodinger.job import jobcontrol
from schrodinger.models import mappers
from schrodinger.Qt import QtCore
from schrodinger.Qt import QtGui
from schrodinger.Qt.QtCore import Qt
from schrodinger.ui.qt.basewidgets import BaseWidget
from schrodinger.utils import fileutils
from schrodinger.utils import qt_utils
FILE_TRUNCATION_SIZE = -800000
maestro = get_maestro()
BROWSER_INFORMATIVE_TEXT_FORMAT = "<i style='color: #666666;'>{}</i>"
MISSING_FILE_MSG = "File no longer available - {} has been moved or deleted."
MISSING_DIR_MSG = "(directory not found)"
NO_FILES_TO_IMPORT_MSG = "No relevant files found - may have been moved or deleted"
[docs]def normalize_and_join_paths(*paths):
"""
Normalize and join paths.
:param paths: File paths to normalize and join.
:type paths: Iterable[str or os.PathLike]
"""
if len(paths) == 0:
return ''
# We skip adding '/./' in file path.
norm_paths = [
os.path.normpath(path) for path in paths if str(path) not in ('', '.')
]
return os.path.join(*norm_paths) if norm_paths else ''
[docs]class JobMonitorFileBrowser(mappers.MapperMixin, BaseWidget):
"""
The file browser is designed to have a combobox which contains a list of
files, and a text edit field to view them. Theoretically, the widget that creates
this should only be passing in valid files, but this dialog will have a
spot for error messages if the file 1) cant be found or 2) is not a text
file
:ivar updateStatusBar: a signal to update the status of job file being viewed currently.
:vartype updateStatusBar: QtCore.pyqtSignal
"""
ui_module = file_browser_ui
model_class = job_monitor_models.JobModel
updateStatusBar = QtCore.pyqtSignal(str)
[docs] def initSetUp(self):
super().initSetUp()
self.ui.file_combo.currentIndexChanged.connect(self.onFileIndexChanged)
self.ui.file_text_edit.setReadOnly(True)
self.setWindowFlags(Qt.FramelessWindowHint | Qt.Widget)
self._file_streamer = SubprocessStdOutStreamer()
self._file_streamer.textAppended.connect(self.appendFileContent)
# timer for updating the textedit of file browser
self.update_timer = QtCore.QTimer()
self.update_timer.setInterval(2000)
self.update_timer.timeout.connect(self._updateCurrentFile)
self.update_timer.start()
self._is_file_empty = True
self.ui.import_btn.clicked.connect(self._onImportBtnClicked)
self.ui.import_btn.setVisible(bool(maestro))
[docs] def defineMappings(self):
# @overrides: MapperMixin
M = self.model_class
return super().defineMappings() + [
(self._onJobUpdated, M.status),
(self._onJobUpdated, M.download_status),
]
[docs] def getSignalsAndSlots(self, model):
return [
(model.filesChanged, self.onFileListChanged),
]
[docs] def onFileListChanged(self):
if self.isVisible():
self.setFileList(self.model.files)
[docs] def showEvent(self, ev):
self.setFileList(self.model.files)
return super().showEvent(ev)
[docs] def setFileList(self, file_list):
"""
Set the list of files in the combo box. Attempt to retain current
selection, if possible.
:param file_list: A list of file paths
:type file_list: list[str]
"""
self.ui.file_combo.clear()
directory_exists = os.path.exists(self.model.directory)
self.ui.file_combo.setEnabled(directory_exists)
if directory_exists:
with self.maintainFileChoice():
for file_name in file_list:
file_path = os.path.join(self.model.directory, file_name)
if os.path.exists(file_path):
self.addFile(file_name)
self._updateImportBtnEnabled()
[docs] def addFile(self, file_name):
"""
Add a new file to the combo box. The visible text is the basename,
and the combo box data is the full path.
:param file_name: the full path to the file
:type file_name: str
"""
self.ui.file_combo.addItem(os.path.basename(file_name), file_name)
[docs] def getCurrentFile(self):
"""
Return the path of the current selected file relative to the job
directory.
:return: Current file path.
:rtype: str
"""
return self.ui.file_combo.currentData() or ''
[docs] def setCurrentFile(self, file_name):
"""
Set the current combo box selection to be the given file name.
:param file_name: The full path of the file
:type file_name: str
"""
idx = self.ui.file_combo.findData(file_name)
if idx == -1:
raise ValueError(f"{file_name} not in list")
self.ui.file_combo.setCurrentIndex(idx)
[docs] def onFileIndexChanged(self):
"""
When the chosen file changes in the combo box, set the contents of
the file to the text edit.
"""
self._file_streamer.stopStreaming()
self.updateStatusBar.emit("") # Clear the previous status bar message
self.loadCurrentFile()
self.moveBrowserToBottom()
self._updateImportBtnEnabled()
[docs] def loadCurrentFile(self):
"""
Load the currently selected file into the text edit
"""
file_name = self.getCurrentFile()
if file_name:
self._loadFile(file_name)
else:
self.clearBrowserText()
def _updateCurrentFile(self):
"""
Update the file browser's textedit with the content of the
log file being currently viewed from server.
"""
curr_file_name = self.getCurrentFile()
is_log_file = curr_file_name and curr_file_name in self.model.logfiles
if (not self.isVisible() # the file browser isn't visible
or not self.model.status is
jobcontrol.DisplayStatus.RUNNING # job isn't in RUNNING state
or not is_log_file # the file being viewed is not a log file
or self._file_streamer.isStreamingRunning()
): # there's a running file streamer already in place
return
jsc_exe = os.path.join(os.environ["SCHRODINGER"], "jsc")
tail_file_cmd = [
jsc_exe, "tail-file", "--follow", "--name", curr_file_name,
self.model.job_id
]
self.clearBrowserText()
self._is_file_empty = True
msg = "Retrieving file from server..."
self.setBrowserText(BROWSER_INFORMATIVE_TEXT_FORMAT.format(msg),
rich_text=True)
self._file_streamer.startStreaming(tail_file_cmd)
def _loadFile(self, file_name):
"""
Load the given file into the text edit
:param file_name: absolute file path
:type file_name: str
"""
file_path = normalize_and_join_paths(self.model.directory, file_name)
is_active = self.model.is_active
file_exists = os.path.isfile(file_path)
msg = ''
if file_name in self.model.logfiles:
if is_active:
# Skip loading a log file while job hasn't completed, file
# streamer is automatically updating the file in this case.
return
# TODO: MAE-45241 - Update API to check whether the current file
# is a Maestro importable structure file.
if self.model.is_downloaded:
if file_exists:
is_st_file = bool(
fileutils.get_structure_file_format(file_path))
if is_st_file:
msg = 'Structure file - cannot be displayed'
if maestro:
msg += ('<br>Use Import to add structures to current '
'Maestro project')
else:
# We will try to read the file contents later.
pass
else:
msg = MISSING_FILE_MSG.format(file_path)
elif not is_active:
# Currently there is no mechanism if a particular file was
# downloaded or not so we wait for all files to be downloaded.
msg = 'File not available - may be downloading'
else:
msg = 'File not available while job in progress'
if msg:
self.setBrowserText(BROWSER_INFORMATIVE_TEXT_FORMAT.format(msg),
rich_text=True)
return
file_size_in_mb = round(os.path.getsize(file_path) / (1000000), 2)
try:
with open(file_path, "rb") as fh:
if file_size_in_mb > 1:
fh.seek(FILE_TRUNCATION_SIZE,
os.SEEK_END) # Seek the last 800K data
msg = ('(File too large to load - last 800K of content '
'displayed)')
self.setBrowserText(
BROWSER_INFORMATIVE_TEXT_FORMAT.format(msg),
rich_text=True)
self._appendBrowserText(str(fh.read(), 'utf-8'))
self.updateStatusBar.emit(
f'End of file displayed; total file size: {str(file_size_in_mb)}M'
)
else:
self.setBrowserText(str(fh.read(), 'utf-8'))
except UnicodeDecodeError:
msg = "Binary file - cannot be displayed or imported"
self.setBrowserText(BROWSER_INFORMATIVE_TEXT_FORMAT.format(msg),
rich_text=True)
[docs] def appendFileContent(self, text):
"""
Append the latest content for the file being viewed currently to
the textedit.
"""
# text when the file is not present yet starts with the below message
if text.startswith("The requested file"):
return
# Clear the retrieving message when the first chunk of data is received
if self._is_file_empty:
self.clearBrowserText()
self._appendBrowserText(text)
self._is_file_empty = False
[docs] def setBrowserText(self, text, rich_text=False):
"""
Load the given text into the text edit while maintaining the vertical scroll
position.
:param text: Plain text to be loaded into the text edit
:type text: str
:param rich_text: Whether rich text.
:type rich_text: bool
"""
with self.maintainVerticalScroll():
if rich_text:
self.ui.file_text_edit.setHtml(text)
else:
# Reset the current character format, this is because in read
# only mode the text cursor can sometimes be at the top and
# thus setPlainText will skip resetting previous format.
self.ui.file_text_edit.setCurrentCharFormat(
QtGui.QTextCharFormat())
self.ui.file_text_edit.setPlainText(text)
def _appendBrowserText(self, text):
"""
Append the given text into the text edit while maintaining the vertical
scroll position.
:param text: Plain text to be appended into the text edit
:type text: str
"""
with self.maintainVerticalScroll():
# Always reset char format when appending.
self.ui.file_text_edit.setCurrentCharFormat(QtGui.QTextCharFormat())
self.ui.file_text_edit.append(text)
[docs] def clearBrowserText(self):
self.ui.file_text_edit.clear()
def _updateImportBtnEnabled(self):
"""
Update `Import` button enabled state.
"""
if not self.model.job_id or (not self.model.is_downloaded):
self.ui.import_btn.setEnabled(False)
return
curr_file = normalize_and_join_paths(self.model.directory,
self.getCurrentFile())
is_st_file = False
if curr_file:
# TODO: MAE-45241 - Update API to check whether the current file
# is a Maestro importable structure file.
is_st_file = bool(fileutils.get_structure_file_format(curr_file))
self.ui.import_btn.setEnabled(is_st_file and os.path.isfile(curr_file))
def _onImportBtnClicked(self):
"""
Import the current selected file in Maestro.
"""
curr_file_full_path = normalize_and_join_paths(self.model.directory,
self.getCurrentFile())
maestro.command(f'entryimport "{curr_file_full_path}"')
def _onJobUpdated(self):
"""
Update 'Import' button enabled state, list updated job files and update
the file text being viewed.
"""
self._updateImportBtnEnabled()
self.setFileList(self.model.files)
self.loadCurrentFile()
self._updateDisplayText()
def _updateDisplayText(self):
"""
Display proper message in the 'file_text_edit' when the job directory
doesn't exist, has no loadable files or is empty. The text edit is
cleared by 'loadCurrentFile', when no file is present in the file_combo.
"""
if self.getCurrentFile():
return
if not os.path.exists(self.model.directory):
msg = MISSING_DIR_MSG
else:
msg = NO_FILES_TO_IMPORT_MSG
self.setBrowserText(BROWSER_INFORMATIVE_TEXT_FORMAT.format(msg), True)
[docs] def moveBrowserToBottom(self):
"""
Move the text edit to the bottom of the file. This should be called
whenever viewing a new file, but not when the user is already viewing
the file.
TODO: Ideally, we want this method to move the scroll bar to the bottom
via the scroll bar object. This doesnt work when text is first loaded
into the text edit though, because the scrollbar hasnt been updated to
reflect the new text
"""
self.ui.file_text_edit.moveCursor(QtGui.QTextCursor.End)
self.ui.file_text_edit.moveCursor(QtGui.QTextCursor.StartOfLine)
[docs] @contextlib.contextmanager
def maintainFileChoice(self):
"""
Maintain file choice in the file combo box.
"""
cur_file = self.getCurrentFile()
yield
with qt_utils.suppress_signals(self.ui.file_combo):
try:
self.setCurrentFile(cur_file)
except ValueError:
pass
[docs] @contextlib.contextmanager
def maintainVerticalScroll(self):
"""
Maintain vertical scroll position in the text edit
"""
scroll_bar = self.ui.file_text_edit.verticalScrollBar()
old_value = scroll_bar.value()
was_max = scroll_bar.value() == scroll_bar.maximum()
yield
new_max = scroll_bar.maximum()
if was_max:
# keep scrollbar at end
new_value = new_max
else:
# maintain old value, can't be greater than maximum
new_value = min(new_max, old_value)
scroll_bar.setValue(new_value)
[docs]class SubprocessStdOutStreamer(QtCore.QObject):
"""
This class provides the support to stream files for data continuously
as and when new data is flushed into the file.
:ivar textAppended: a signal emitted when there's new data in the file
being streamed.
:vartype textAppended: QtCore.pyqtSignal
:ivar streamingDone: a signal emitted to indicate that the streaming
being run got finished.
:vartype streamingDone: QtCore.pyqtSignal
"""
textAppended = QtCore.pyqtSignal(str)
streamingDone = QtCore.pyqtSignal()
[docs] def __init__(self):
super().__init__()
self._tail_file_process = QtCore.QProcess()
self._tail_file_process.setProcessChannelMode(
QtCore.QProcess.MergedChannels)
self._tail_file_process.readyReadStandardOutput.connect(
self._onReadyReadyStandardOut)
self._tail_file_process.finished.connect(self.streamingDone)
def _onReadyReadyStandardOut(self):
stdout_txt = str(
self._tail_file_process.readAllStandardOutput().data().decode(
'utf-8'))
self.textAppended.emit(stdout_txt)
[docs] def startStreaming(self, cmd):
"""
:param cmd: command-line arguments for the streaming process
:type cmd: list[str]
"""
self._tail_file_process.start(cmd[0], cmd[1:])
[docs] def stopStreaming(self):
self._tail_file_process.kill()
[docs] def isStreamingRunning(self):
return bool(self._tail_file_process.processId())