import inspect
import sys
from pathlib import Path
from typing import Any, Union, Optional
from qtpy.QtCore import Qt, QModelIndex
from qtpy import QtWidgets, QtCore, QtGui
from qtpy.QtGui import QKeySequence
from pymodaq_gui.utils.styling import create_icon
from pymodaq_utils.logger import set_logger, get_module_name
from pymodaq_gui.utils.custom_app import CustomApp
from pymodaq_utils.enums import StrEnum
from pymodaq_gui.messenger import dialog
from pymodaq_gui.qt_utils import center_widget_on_screen_and_show, mkQApp
from pymodaq_gui.utils.splash import get_pymodaq_pixmap
from pymodaq_gui.utils.widgets.combo import ComboBox
from pymodaq_gui.utils.widget_sync import WidgetSync, SyncMode
logger = set_logger(get_module_name(__file__))
[docs]
class ManagerActions(StrEnum):
OPEN = 'open_manager'
COPY = 'copy_entry'
NEW = 'create_new_entry'
DELETE = 'delete_entry'
SAVE = 'save_entry'
RELOAD = 'reload_entry'
EXECUTE = 'execute_entry'
LIST = 'list_entries'
LABEL_EXTERNAL = 'label_external'
LIST_EXTERNAL = 'list_entries_external'
[docs]
class ManagerSubEntriesModel(QtCore.QAbstractListModel):
def __init__(self, entries: list[str]):
super().__init__()
self._data = entries
self.colors: list[QtCore.Qt.GlobalColor] = [QtCore.Qt.GlobalColor.darkGray for _ in range(len(self._data))]
self.status: list[Union[bool, None]] = [None for _ in range(len(self._data))]
[docs]
def set_status(self, index: int, status: bool):
self.status[index] = status
if status:
self.colors[index] = QtCore.Qt.GlobalColor.darkGreen
else:
self.colors[index] = QtCore.Qt.GlobalColor.darkRed
model_index = self.index(index, 0)
self.dataChanged.emit(model_index, model_index,
[Qt.ItemDataRole.DecorationRole,
Qt.ItemDataRole.BackgroundColorRole])
[docs]
def rowCount(self, *args, **kwargs) -> int:
return len(self._data)
[docs]
def data(self, index: QModelIndex, role: Qt.ItemDataRole):
if index.isValid():
if role == Qt.ItemDataRole.DisplayRole:
return str(self._data[index.row()])
elif role == Qt.ItemDataRole.DecorationRole:
if self.status[index.row()] is None:
return create_icon('yellow_light')
elif self.status[index.row()]:
return create_icon('greenLight2')
else:
return create_icon('red_light')
elif role == Qt.ItemDataRole.FontRole:
return QtGui.QFont('Arial', 10)
elif role == Qt.ItemDataRole.ForegroundRole:
if self.status[index.row()] is None:
return QtGui.QBrush(QtCore.Qt.GlobalColor.darkGray)
elif self.status[index.row()]:
return QtGui.QBrush(QtCore.Qt.GlobalColor.darkGreen)
else:
return QtGui.QBrush(QtCore.Qt.GlobalColor.darkRed)
elif role == Qt.ItemDataRole.BackgroundColorRole:
return Qt.BrushStyle.NoBrush
[docs]
class ManagerBase(CustomApp):
new_entry = QtCore.Signal(str)
applied_entry = QtCore.Signal(str)
deleted_entry = QtCore.Signal(str)
updated_entry = QtCore.Signal(str)
entry_type: str
entry_extension: str
execute_action_checkable = False
icon_name = 'build_circle' # the default icon to represent the manager in Toolbar and Menus
default_entry_name = 'default'
def __init__(self, **kwargs):
# what CustomApp does not know about goes to init_extra, once the Qt side is initialized
app_params = inspect.signature(CustomApp.__init__).parameters
app_kwargs = {key: kwargs.pop(key) for key in list(kwargs) if key in app_params}
super().__init__(parent=QtWidgets.QMainWindow(), **app_kwargs)
self.init_extra(**kwargs)
self._entry_applied = False
self._applied_entry_name: str = None
self.main_widget = QtWidgets.QWidget()
self.mainwindow.setCentralWidget(self.main_widget)
self._entry_slots = {}
self.splash_subentries: Optional[SubEntriesSplash] = None
self.subentries_model: Optional[ManagerSubEntriesModel] = None
self._updating_entry = False
#first create the object
self.entries_sync = WidgetSync(
initial_value={
'items': [],
'current': None,
'enabled': False,
})
self.setup_ui()
self.enable_actions(False)
# then update it using a call to self.entries (itself needing entries_sync)
self.entries_sync.update_key('items', self.entries)
self.entries_sync.update_key('current', self.default_entry_name)
self.update_action_list()
self.update_execute_action_tooltip(self.entry)
self.finish_init()
[docs]
def finish_init(self):
""" Called at the end of the initialization of the manager, to be reimplemented if needed"""
pass
[docs]
def enable_actions(self, enable=True):
self.entries_sync.update_key('enabled', enable)
for action in self.actions_names:
self.set_action_enabled(action, enable)
[docs]
def show_subentries(self, subentries: list[Any], title=''):
self.splash_subentries = SubEntriesSplash(title=title)
self.subentries_model = ManagerSubEntriesModel(self.format_subentries(subentries))
self.splash_subentries.setModel(self.subentries_model)
self.splash_subentries.show_splash()
[docs]
def close_subentries_display(self, after_ms=1000):
if self.splash_subentries is not None:
QtCore.QTimer.singleShot(after_ms, self.splash_subentries.close)
@property
def manager_name(self) -> str:
return self.__class__.__name__
[docs]
def get_entry_folder(self, **kwargs_to_entry_folder) -> Path:
"""Get the folder path where the managed entries are stored."""
raise NotImplementedError
[docs]
def setup_ui(self):
self.setup_docks_and_widgets()
self.setup_menus_and_toolbars()
self.setup_actions_base()
self.setup_actions()
self.connect_things_base()
self.connect_things()
self.do_things_after_ui_setup()
@property
def entry(self) -> str:
""" Get/Set the name of the current entry """
return self.entries_sync.value['current']
@entry.setter
def entry(self, entry_name: str):
self.entries_sync.set_value({**self.entries_sync.value, 'current': entry_name})
@property
def entry_filepath(self) -> Path:
""" Get the full path of the current entry file """
return self.entry_path_from_name(self.entry)
[docs]
def entry_path_from_name(self, entry_name: str) -> Path:
kwargs_to_entry_folder = {} # reimplement if needed
return self.get_entry_folder(**kwargs_to_entry_folder).joinpath(
entry_name + self.entry_extension)
@property
def entries(self) -> list[str]:
""" Get the name of all existing entries """
return self.list_managed_entries()
@property
def entries_filepath(self) -> list[Path]:
""" Get the full path of all entries file """
return self.list_managed_entries_path()
[docs]
def list_managed_entries(self, **kwargs_to_entry_folder) -> list[str]:
"""Returns a list of names of managed entries with the default one as first
See default_entry_name class attribute
"""
entries = [path.stem for path in self.list_managed_entries_path(**kwargs_to_entry_folder)]
entries.remove(self.default_entry_name)
entries = [self.default_entry_name] + entries # always shows default as first
return entries
[docs]
def list_managed_entries_path(self, **kwargs_to_entry_folder) -> list[Path]:
"""Should return a list of Path objects representing managed entries.
Example:
--------
[path for path in get_set_experiment_path().iterdir() if path.suffix == self.entry_extension]
"""
entry_path = self.get_entry_folder(**kwargs_to_entry_folder)
if not entry_path.exists():
entry_path.mkdir(parents=True)
if not entry_path.joinpath(f'{self.default_entry_name}{self.entry_extension}').exists():
self.create_entry(self.default_entry_name, bypass_dialog=True)
return [path for path in entry_path.iterdir() if path.suffix == self.entry_extension]
[docs]
def get_action_list(self) -> ComboBox:
""" Convenience method to get right return type """
return self.get_action(ManagerActions.LIST).widget
[docs]
def get_action_from_file(self, file: Path) -> str:
""" Get an action name given a file and the manager name"""
return f"{file.stem}_{self.manager_name}"
[docs]
def setup_actions_base(self):
# ACTIONS in Manager
# self.add_widget('entry_label', QtWidgets.QLabel(
# f'{self.entry_type.capitalize()}:'))
self.add_widget(ManagerActions.LIST, ComboBox(),
tip=f'Name of the current {self.entry_type}',
kwargs={'setReadOnly': True})
self.get_action_list().addItems(self.entries)
self.add_action(ManagerActions.COPY, f'Copy {self.entry_type.capitalize()}',
'file_copy')
self.add_action(ManagerActions.NEW,
f'New {self.entry_type.capitalize()}', 'add_circle',
tip=f'Create a new {self.entry_type} file ("Ctrl+N")',
shortcut=QKeySequence(Qt.Modifier.CTRL | Qt.Key.Key_N),
icon_color=self.get_theme().green)
self.add_action(ManagerActions.DELETE,
f'Delete {self.entry_type.capitalize()}', 'do_not_disturb_on',
icon_color=self.get_theme().red,
tip=f'Delete the current {self.entry_type.capitalize()} ("Ctrl+Delete")',
shortcut=QKeySequence(Qt.Modifier.CTRL | Qt.Key.Key_Delete))
self.add_action(ManagerActions.SAVE,
f'Save {self.entry_type.capitalize()}', 'save_as',
icon_color=self.get_theme().blue,
tip=f'Save/Update the current {self.entry_type.capitalize()} ("Ctrl+S")',
shortcut=QKeySequence(Qt.Modifier.CTRL | Qt.Key.Key_S))
self.add_action(ManagerActions.RELOAD,
f'Reload {self.entry_type.capitalize()}', 'refresh',
tip=f'Reload the current {self.entry_type} file ("Ctrl+Shift+R")',
icon_color=self.get_theme().orange,
shortcut=QKeySequence(Qt.Modifier.CTRL | Qt.Modifier.SHIFT | Qt.Key.Key_R))
self.add_action(ManagerActions.EXECUTE,
f'Execute {self.entry_type.capitalize()}', 'start',
icon_color=self.get_theme().magenta,
checkable=self.execute_action_checkable,
icon_checked_color=QtGui.QColor(255, 0, 201),
tip=f'Execute the current {self.entry_type} file ("Ctrl+Shift+E")',
shortcut=QKeySequence(Qt.Modifier.CTRL | Qt.Modifier.SHIFT | Qt.Key.Key_E))
self.add_action(ManagerActions.OPEN, f"{self.entry_type.capitalize()} Manager",
self.icon_name,
icon_color=self.get_theme().blue,
tip=f'Open the {self.entry_type.capitalize()} Manager',
checkable=True,
icon_checked_color=self.get_theme().cyan,
auto_toolbar=False, auto_menu=False)
[docs]
def get_sync_combo_entry(self) -> ComboBox:
combo = ComboBox()
self.sync_entries_with(combo)
return combo
[docs]
def connect_things_base(self):
self.connect_action(ManagerActions.COPY, lambda: self.copy_entry())
self.connect_action(ManagerActions.NEW, lambda: self.create_entry())
self.connect_action(ManagerActions.DELETE, lambda: self.delete_entry())
self.connect_action(ManagerActions.SAVE, lambda: self.save_check())
self.connect_action(ManagerActions.RELOAD, lambda: self.update_entry())
self.connect_action(ManagerActions.EXECUTE, lambda: self.execute_current_entry())
self.connect_action(ManagerActions.OPEN, lambda: self.show())
self.entries_sync.value_changed.connect(lambda value: self.update_entry(value['current']))
self.sync_entries_with(self.get_action_list())
[docs]
def execute_current_entry(self):
self.execute_entry(self.entry)
[docs]
def sync_entries_with(self, combo: ComboBox):
self.entries_sync.bind_properties(
combo,
property_map={
'items': {
'signal': combo.items_changed,
'getter': combo.get_items,
'setter': combo.set_items,
'mode': SyncMode.BIDIRECTIONAL,
},
'current': {
'signal': combo.currentTextChanged,
'getter': combo.currentText,
'setter': combo.setCurrentText,
'mode': SyncMode.BIDIRECTIONAL,
},
'enabled': {
'signal': combo.enabled_changed,
'getter': combo.isEnabled,
'setter': combo.setEnabled,
'mode': SyncMode.BIDIRECTIONAL,
},
},
)
[docs]
def quit_fun(self):
self.entries_sync.unbind_all()
return super().quit_fun()
[docs]
def create_entry(self, entry: str = None, bypass_dialog=False):
if entry is not None:
ok = True
else:
entry, ok = QtWidgets.QInputDialog.getText(
None,
f'Enter a NEW {self.entry_type.capitalize()} name',
f'{self.entry_type.capitalize()} name:', QtWidgets.QLineEdit.Normal)
self.do_things_for_new_creation()
if ok and entry != '' and self.save_check(entry, bypass_dialog=bypass_dialog):
self.entries_sync.append_to_list('items', entry)
self.entries_sync.update_key('current', entry)
self.update_action_list()
self.new_entry.emit(entry)
[docs]
def do_things_for_new_creation(self):
""" To be reimplemented if needed """
pass
[docs]
def save_check(self, entry: str = None, bypass_dialog=False) -> bool:
if entry is not None:
entry_path = self.get_entry_folder().joinpath(entry+self.entry_extension)
else:
entry_path = self.entry_filepath
if entry_path.exists() and not bypass_dialog:
user_agreed = dialog(
title='Overwrite confirmation',
message='File exist do you want to overwrite it ?',
)
if not user_agreed:
return False
self.save_entries(entry_path)
return True
[docs]
def save_entries(self, entry_path: Path = None):
""" Particular implementation to save entries for this inherited Manager """
raise NotImplementedError
[docs]
def copy_entry(self, entry: str = None, bypass_dialog=False):
if entry is None:
entry, ok = QtWidgets.QInputDialog.getText(
None, f'Enter a NEW {self.entry_type.capitalize()} name',
f'{self.entry_type.capitalize()} name:', QtWidgets.QLineEdit.Normal)
if not ok or entry == '':
return
if self.save_check(entry, bypass_dialog=bypass_dialog):
self.entries_sync.append_to_list('items', entry)
self.entries_sync.update_key('current', entry)
self.update_action_list()
self.new_entry.emit(entry)
[docs]
def delete_entry(self, entry: str = None, bypass_dialog=False):
if entry is None:
entry = self.entry
else:
self.entry = entry
if bypass_dialog:
user_agreed = True
else:
user_agreed = dialog(
title='Delete confirmation',
message=f'Are you sure you want to delete the {self.entry_type.capitalize()}'
f' {entry} ?',
)
if user_agreed:
entries = self.entries[:] # to get before unlinking
self.entry_filepath.unlink(missing_ok=True)
logger.info(f'{self.entry_type.capitalize()} file {self.entry} deleted')
index = entries.index(entry)
entries.pop(index)
if len(entries) != 0:
current = entries[max(0, index - 1)]
else: # should trigger the default entry creation!
entries = self.entries # this recreate default
current = entries[0]
self.entries_sync.set_value({'items': entries,
'current': current}) # deleting will update current and fire update_entry_base
self.deleted_entry.emit(entry) # notify that an entry has been deleted
@property
def entry_applied(self) -> bool:
""" Get the status of the execution of the last applied entry
If True, the :attr:`applied_entry_name` property will reflect the last successfully applied/executed entry
"""
return self._entry_applied
@entry_applied.setter
def entry_applied(self, applied: bool):
self._entry_applied = applied
if applied:
self._applied_entry_name = self.entry
self.applied_entry.emit(self._applied_entry_name)
[docs]
@QtCore.Slot(bool)
def set_entry_applied(self, value: bool):
self.entry_applied = value
@property
def applied_entry_name(self) -> str | None:
""" Get the name of the last entry that has been successfully applied/executed """
return self._applied_entry_name
[docs]
def execute_entry(self, entry_path: str | Path = None, **kwargs):
""" To be called to execute the selected entry """
if entry_path is None:
self.save_check(self.entry, bypass_dialog=True)
entry_path = self.entry_filepath
if isinstance(entry_path, str):
entry_path = self.entry_path_from_name(entry_path)
self.update_entry(entry_path)
if not self.can_execute_entry():
logger.info(f"Cannot Load {self.entry_type.capitalize()} file: {entry_path.stem} "
f"as the manager cannot execute entries (no Dashboard initialized?)")
return
res = self._execute_entry(entry_path, **kwargs)
if isinstance(res, bool):
# covers the cases where _execute_entry is async and return None, while the async
# execution will trigger set_entry_applied method (qt slot or manual call)
self.entry_applied = res
[docs]
def can_execute_entry(self) -> bool:
""" Whether entries can be executed, to be reimplemented
The base manager does not know about the dashboard: the managers working on it reimplement this method,
usually returning ``DashboardMixin.is_dashboard_available``
"""
return False
def _execute_entry(self, entry_path: Path = None, **kwargs) -> bool:
"""Particular implementation of the entry execution for this manager
Applies the entry from the given file in the manager.
Should not be called directly, use :attr:`execute_entry` instead.
To be reimplemented
Parameters:
-----------
file : Path
The path to the configuration file to be applied.
"""
checked = self.is_action_checked(ManagerActions.EXECUTE)
# you may want to use the checked state of this action in your implementation
# Checkable EXECUTE action is optional and dependant if the class attribute
# execute_action_checkable is set to True
return False
[docs]
def update_entry(self, entry: Union[str, Path] = None, **kwargs):
""" Load and display the given entry """
if self._updating_entry:
return
self._updating_entry = True
try:
if entry is None:
if self.entry is None:
return
entry = self.entry_filepath
elif isinstance(entry, str):
entry = self.entry_path_from_name(entry)
self.entry = entry.stem # syncs the combo; if called from an action slot this fires
# value_changed → lambda → update_entry (re-entrant, blocked above)
# but value_changed sync-to-widget still runs → external combos updated
self._update_entry(entry)
self.update_execute_action_tooltip(entry.stem)
self.updated_entry.emit(entry.stem)
finally:
self._updating_entry = False
def _update_entry(self, entry_path: Path):
""" Particular implementation to update entries for this inherited Manager """
raise NotImplementedError
[docs]
def show(self):
if self.is_action_checked(ManagerActions.OPEN):
self.mainwindow.show()
self.mainwindow.raise_()
else:
self.mainwindow.hide()
[docs]
def force_show(self, show: bool = True):
self.set_action_checked(ManagerActions.OPEN, show)
self.show()
[docs]
def update_action_list(self):
with QtCore.QSignalBlocker(self.get_action_list()) as blocker:
try:
for ind_file, file in enumerate(self.list_managed_entries_path()):
if not self.has_action(self.get_action_from_file(file)):
self.add_action(
self.get_action_from_file(file),
file.stem,
"",
f"Load the {file.stem} entry",
auto_toolbar=False,
)
self.update_actions_connection()
self.update_menu()
except KeyError as e:
pass
[docs]
def update_actions_connection(self):
for ind_file, file in enumerate(self.list_managed_entries_path()):
self.connect_action(self.get_action_from_file(file), connect=False)
self._entry_slots[file.stem] = self.create_slot_from_file(
self.get_entry_folder().joinpath(file.stem + self.entry_extension))
self.connect_action(self.get_action_from_file(file),
self._entry_slots[file.stem])
[docs]
def create_slot_from_file(self, filename: Path):
return lambda: self.execute_entry(filename)
[docs]
def save_new_history_entry(self):
"""Save a new history entry in the history file"""
raise NotImplementedError
[docs]
class ListView(QtWidgets.QListView):
def __init__(self, parent=None):
super().__init__(parent)
self.setSizeAdjustPolicy(QtWidgets.QAbstractScrollArea.AdjustToContents)
self.setSelectionMode(QtWidgets.QAbstractItemView.NoSelection)
[docs]
class SubEntriesSplash(QtWidgets.QLabel):
def __init__(self, parent=None, title=''):
super().__init__(parent)
self.list_view = ListView()
self.title_label = QtWidgets.QLabel(title)
[docs]
def mousePressEvent(self, event):
self.close()
[docs]
def setModel(self, model):
self.list_view.setModel(model)
[docs]
def show_splash(self):
self.setPixmap(get_pymodaq_pixmap())
self.list_view.viewport().setAutoFillBackground(False)
self.list_view.setStyleSheet("QListView { outline: none; }")
self.list_view.clearFocus()
self.title_label.setFont(QtGui.QFont("Arial", 12))
self.setWindowFlags(Qt.FramelessWindowHint | Qt.WindowSystemMenuHint)
vlayout = QtWidgets.QVBoxLayout()
hlayout = QtWidgets.QHBoxLayout()
self.setLayout(hlayout)
hlayout.addStretch(10)
hlayout.addLayout(vlayout)
vlayout.addWidget(self.title_label)
vlayout.addWidget(self.list_view)
vlayout.addStretch(10)
center_widget_on_screen_and_show(self)
self.raise_()
if __name__ == '__main__':
import numpy as np
# Create application and main window
app = mkQApp('Dashboard')
entries = [f'Module {ind:03.0f} si about to do something incredible' for ind in range(10)]
widget = SubEntriesSplash()
entry_sublist_model = ManagerSubEntriesModel(entries)
widget.setModel(entry_sublist_model)
def execute_entries(sublist: list[str]):
for ind in range(len(sublist)):
entry_sublist_model.set_status(ind, bool(np.random.randint(2)))
QtWidgets.QApplication.processEvents()
QtCore.QThread.msleep(200)
QtCore.QThread.msleep(2000)
#widget.close()
widget.show_splash()
execute_entries(entries)
sys.exit(app.exec())