# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.
"""
Main window
===========
The :mod:`datalab.gui.main` module provides the main window of the
DataLab project.
.. autoclass:: DLMainWindow
"""
# pylint: disable=invalid-name # Allows short reference names like x, y, ...
# This module intentionally concentrates the application shell, menus, panels,
# status widgets and shutdown logic. A file-length disable is more honest than
# splitting high-coupling UI code only to satisfy a metric.
# pylint: disable=too-many-lines
from __future__ import annotations
import functools
import logging
import os
import os.path as osp
import sys
import time
import traceback
from datetime import datetime
from typing import TYPE_CHECKING
import guidata.dataset as gds
import numpy as np
import scipy.ndimage as spi
import scipy.signal as sps
from guidata.configtools import get_icon
from guidata.qthelpers import add_actions, create_action
from plotpy.builder import make
from plotpy.constants import PlotType
from qtpy import QtCore as QC
from qtpy import QtGui as QG
from qtpy import QtWidgets as QW
from sigima.config import options as sigima_options
from sigima.objects import ImageObj, SignalObj, create_image, create_signal
from sigimax.mainwindow import SGMXMainWindow
from sigimax.utils import qthelpers as sgmx_qth
import datalab
from datalab.adapters_metadata.common import have_geometry_results
from datalab.adapters_plotpy import create_adapter_from_object
from datalab.config import (
APP_DESC,
APP_NAME,
Conf,
_,
save_runtime_option,
)
from datalab.control.baseproxy import AbstractDLControl
from datalab.control.remote import RemoteServer
from datalab.env import execenv
from datalab.gui.actionhandler import ActionCategory
from datalab.gui.commandpalette import (
CommandPaletteDialog,
CommandSearchField,
collect_commands,
)
from datalab.gui.docks import DockablePlotWidget
from datalab.gui.h5io import H5InputOutput
from datalab.gui.panel import base, history, image, macro, signal
from datalab.gui.pluginconfig import PluginConfigDialog
from datalab.gui.settings import AI_OPTION_NAMES, edit_settings
from datalab.objectmodel import ObjectGroup, get_uuid
from datalab.plugins import PluginRegistry, discover_plugins, discover_v020_plugins
from datalab.utils import qthelpers as qth
from datalab.utils.qthelpers import configure_menu_about_to_show
from datalab.webapi import WEBAPI_AVAILABLE, get_webapi_controller
from datalab.webapi.actions import WebApiActions
from datalab.widgets import instconfviewer
from datalab.widgets import status as dl_status
if TYPE_CHECKING:
from typing import Literal
from datalab.gui.historysession_ops import SessionBehavior
from datalab.gui.panel.base import AbstractPanel, BaseDataPanel
from datalab.gui.panel.image import ImagePanel
from datalab.gui.panel.macro import MacroPanel
from datalab.gui.panel.signal import SignalPanel
from datalab.plugins import PluginBase
def remote_controlled(func):
"""Decorator for remote-controlled methods"""
@functools.wraps(func)
def method_wrapper(*args, **kwargs):
"""Decorator wrapper function"""
win = args[0] # extracting 'self' from method arguments
already_busy = not win.ready_flag
win.ready_flag = False
try:
output = func(*args, **kwargs)
finally:
if not already_busy:
win.SIG_READY.emit()
win.ready_flag = True
QW.QApplication.processEvents()
return output
return method_wrapper
# DLMainWindow is the top-level UI shell, so it legitimately owns many widget
# references and public control methods used by the rest of the application.
[docs]
class DLMainWindow( # pylint: disable=too-many-instance-attributes,too-many-public-methods
SGMXMainWindow, AbstractDLControl
):
"""DataLab main window
Args:
console: enable internal console
hide_on_close: True to hide window on close
"""
def __init__(self, console=None, hide_on_close=False):
"""Initialize main window"""
self.started_at = datetime.now().astimezone()
self.plugins_last_load_at = self.started_at
self.webapistatus: dl_status.WebAPIStatus | None = None
self.pluginstatus: dl_status.PluginStatus | None = None
self._startup_errors: list[str] = []
self.macropanel: MacroPanel | None = None
self.historypanel: history.HistoryPanel | None = None
self.aiassistantpanel = None # type: ignore[assignment]
self.signalpanel_toolbar: QW.QToolBar | None = None
self.imagepanel_toolbar: QW.QToolBar | None = None
self.signalpanel: SignalPanel | None = None
self.imagepanel: ImagePanel | None = None
self.signalview: DockablePlotWidget | None = None
self.imageview: DockablePlotWidget | None = None
self.h5inputoutput: H5InputOutput | None = None
self.webapi_actions: WebApiActions | None = None
self.openh5_action: QW.QAction | None = None
self.saveh5_action: QW.QAction | None = None
self.browseh5_action: QW.QAction | None = None
self.settings_action: QW.QAction | None = None
self.command_palette_action: QW.QAction | None = None
self.quit_action: QW.QAction | None = None
self.autorefresh_action: QW.QAction | None = None
self.reload_plugins_action: QW.QAction | None = None
self.configure_plugins_action: QW.QAction | None = None
self.create_menu: QW.QMenu | None = None
self.edit_menu: QW.QMenu | None = None
self.roi_menu: QW.QMenu | None = None
self.operation_menu: QW.QMenu | None = None
self.processing_menu: QW.QMenu | None = None
self.analysis_menu: QW.QMenu | None = None
self.plugins_menu: QW.QMenu | None = None
self.remote_server: RemoteServer | None = None
super().__init__(console=console, hide_on_close=hide_on_close)
def _before_setup(self, console: bool) -> None:
"""Initialize DataLab-specific services before setting up the UI."""
super()._before_setup(console)
self.h5inputoutput = H5InputOutput(self)
# Starting XML-RPC server thread
self.remote_server = RemoteServer(self)
if Conf.rpc_server_enabled.get():
self.remote_server.SIG_SERVER_PORT.connect(self.xmlrpc_server_started)
self.remote_server.start()
self.__register_plugins()
self.webapi_actions = WebApiActions(self)
# ------API related to XML-RPC remote control
[docs]
@staticmethod
def xmlrpc_server_started(port):
"""XML-RPC server has started, writing comm port in configuration file"""
Conf.rpc_server_port.set(port)
# Persist the port with a direct single-key write: it is a shared IPC
# value read by remote clients across processes, and must not be
# clobbered by unrelated bulk saves (e.g. window geometry persisted on
# close by another DataLab instance sharing the same INI file).
save_runtime_option(Conf, "rpc_server_port")
def __get_current_basedatapanel(self) -> BaseDataPanel:
"""Return the current BaseDataPanel,
or the signal panel if macro panel is active
Returns:
BaseDataPanel: current panel
"""
panel = self.tabwidget.currentWidget()
if not isinstance(panel, base.BaseDataPanel):
panel = self.signalpanel
return panel
def __get_datapanel(
self, panel: Literal["signal", "image"] | None
) -> BaseDataPanel:
"""Return a specific BaseDataPanel.
Args:
panel: panel name. If None, current panel is used.
Returns:
Panel widget
Raises:
ValueError: if panel is unknown
"""
if not panel:
return self.__get_current_basedatapanel()
if panel == "signal":
return self.signalpanel
if panel == "image":
return self.imagepanel
raise ValueError(f"Unknown panel: {panel}")
[docs]
@remote_controlled
def get_group_titles_with_object_info(
self,
) -> tuple[list[str], list[list[str]], list[list[str]]]:
"""Return groups titles and lists of inner objects uuids and titles.
Returns:
Groups titles, lists of inner objects uuids and titles
"""
panel = self.__get_current_basedatapanel()
return panel.objmodel.get_group_titles_with_object_info()
[docs]
@remote_controlled
def get_object_titles(
self, panel: Literal["signal", "image", "macro"] | None = None
) -> list[str]:
"""Get object (signal/image) list for current panel.
Objects are sorted by group number and object index in group.
Args:
panel: panel name. If None, current data panel is used (i.e. signal or
image panel).
Returns:
List of object titles
Raises:
ValueError: if panel is unknown
"""
if not panel or panel in ("signal", "image"):
return self.__get_datapanel(panel).objmodel.get_object_titles()
if panel == "macro":
return self.macropanel.get_macro_titles()
raise ValueError(f"Unknown panel: {panel}")
[docs]
@remote_controlled
def get_object(
self,
nb_id_title: int | str | None = None,
panel: Literal["signal", "image"] | None = None,
) -> SignalObj | ImageObj:
"""Get object (signal/image) from index.
Args:
nb_id_title: Object number, or object id, or object title.
Defaults to None (current object).
panel: Panel name. Defaults to None (current panel).
Returns:
Object
Raises:
KeyError: if object not found
TypeError: if index_id_title type is invalid
"""
panelw = self.__get_datapanel(panel)
if nb_id_title is None:
return panelw.objview.get_current_object()
if isinstance(nb_id_title, int):
return panelw.objmodel.get_object_from_number(nb_id_title)
if isinstance(nb_id_title, str):
try:
return panelw.objmodel[nb_id_title]
except KeyError:
try:
return panelw.objmodel.get_object_from_title(nb_id_title)
except KeyError as exc:
raise KeyError(
f"Invalid object index, id or title: {nb_id_title}"
) from exc
raise TypeError(f"Invalid index_id_title type: {type(nb_id_title)}")
[docs]
def find_object_by_uuid(
self, uuid: str
) -> SignalObj | ImageObj | ObjectGroup | None:
"""Find an object by UUID, searching across all panels.
This method searches for an object in both signal and image panels,
making it suitable for cross-panel operations (e.g., radial profile that
takes an ImageObj and produces a SignalObj).
Difference from get_object():
- get_object() requires specifying a panel and accepts number/id/title
- find_object_by_uuid() searches all panels automatically using only UUID
Args:
uuid: UUID of the object to find
Returns:
The object if found in any panel, None otherwise
"""
for panel in (self.signalpanel, self.imagepanel):
if panel is not None:
try:
return panel.objmodel[uuid]
except KeyError:
continue
return None
[docs]
@remote_controlled
def get_object_uuids(
self,
panel: Literal["signal", "image"] | None = None,
group: int | str | None = None,
) -> list[str]:
"""Get object (signal/image) uuid list for current panel.
Objects are sorted by group number and object index in group.
Args:
panel: panel name. If None, current panel is used.
group: Group number, or group id, or group title.
Defaults to None (all groups).
Returns:
List of object uuids
Raises:
ValueError: if panel is unknown
"""
objmodel = self.__get_datapanel(panel).objmodel
if group is None:
return objmodel.get_object_ids()
if isinstance(group, int):
grp = objmodel.get_group_from_number(group)
else:
try:
grp = objmodel.get_group(group)
except KeyError:
grp = objmodel.get_group_from_title(group)
if grp is None:
raise KeyError(f"Invalid group index, id or title: {group}")
return grp.get_object_ids()
[docs]
@remote_controlled
def get_sel_object_uuids(self, include_groups: bool = False) -> list[str]:
"""Return selected objects uuids.
Args:
include_groups: If True, also return objects from selected groups.
Returns:
List of selected objects uuids.
"""
panel = self.__get_current_basedatapanel()
return panel.objview.get_sel_object_uuids(include_groups)
[docs]
@remote_controlled
def get_current_object_uuid(self) -> str | None:
"""Return current object uuid in current panel.
Returns:
UUID of the current object, or None if no object is current.
"""
panel = self.__get_current_basedatapanel()
return panel.objview.get_current_object_uuid()
[docs]
@remote_controlled
def add_group(
self,
title: str,
panel: Literal["signal", "image"] | None = None,
select: bool = False,
) -> None:
"""Add group to DataLab.
Args:
title: Group title
panel: Panel name. Defaults to None.
select: Select the group after creation. Defaults to False.
"""
self.__get_datapanel(panel).add_group(title, select)
[docs]
@remote_controlled
def select_objects(
self,
selection: list[int | str],
panel: Literal["signal", "image"] | None = None,
) -> None:
"""Select objects in current panel.
Args:
selection: List of object numbers (1 to N) or uuids to select
panel: panel name. If None, current panel is used. Defaults to None.
"""
panel = self.__get_datapanel(panel)
panel.objview.select_objects(selection)
[docs]
@remote_controlled
def select_groups(
self,
selection: list[int | str] | None = None,
panel: Literal["signal", "image"] | None = None,
) -> None:
"""Select groups in current panel.
Args:
selection: List of group numbers (1 to N), or list of group uuids,
or None to select all groups. Defaults to None.
panel: panel name. If None, current panel is used. Defaults to None.
"""
panel = self.__get_datapanel(panel)
panel.objview.select_groups(selection)
[docs]
@remote_controlled
def delete_metadata(
self, refresh_plot: bool = True, keep_roi: bool = False
) -> None:
"""Delete metadata of selected objects
Args:
refresh_plot: Refresh plot. Defaults to True.
keep_roi: Keep ROI. Defaults to False.
"""
panel = self.__get_current_basedatapanel()
panel.delete_metadata(refresh_plot, keep_roi)
[docs]
@remote_controlled
def call_method(
self,
method_name: str,
*args,
panel: Literal["signal", "image"] | None = None,
**kwargs,
):
"""Call a public method on a panel or main window.
This generic method allows calling any public method that is not explicitly
exposed in the proxy API. The method resolution follows this order:
1. If panel is specified: call method on that specific panel
2. If panel is None:
a. Try to call method on main window (DLMainWindow)
b. If not found, try to call method on current panel (BaseDataPanel)
This makes it convenient to call panel methods without specifying the panel
parameter when working on the current panel.
Args:
method_name: Name of the method to call
*args: Positional arguments to pass to the method
panel: Panel name ("signal", "image", or None for auto-detection).
Defaults to None.
**kwargs: Keyword arguments to pass to the method
Returns:
The return value of the called method
Raises:
AttributeError: If the method does not exist or is not public
ValueError: If the panel name is invalid
Examples:
>>> # Call remove_object on current panel (auto-detected)
>>> win.call_method("remove_object", force=True)
>>> # Call a signal panel method specifically
>>> win.call_method("delete_all_objects", panel="signal")
>>> # Call main window method
>>> win.call_method("get_current_panel")
"""
# Security check: only allow public methods (not starting with _)
if method_name.startswith("_"):
raise AttributeError(
f"Cannot call private method '{method_name}' through proxy"
)
# If panel is specified, use that panel directly
if panel is not None:
target = self.__get_datapanel(panel)
if not hasattr(target, method_name):
raise AttributeError(
f"Method '{method_name}' does not exist on {panel} panel"
)
method = getattr(target, method_name)
if not callable(method):
raise AttributeError(f"'{method_name}' is not a callable method")
return method(*args, **kwargs)
# Panel is None: try main window first, then current panel
# Try main window first
if hasattr(self, method_name):
method = getattr(self, method_name)
if callable(method):
return method(*args, **kwargs)
# Method not found on main window, try current panel
current_panel = self.__get_current_basedatapanel()
if hasattr(current_panel, method_name):
method = getattr(current_panel, method_name)
if callable(method):
return method(*args, **kwargs)
# Method not found anywhere
raise AttributeError(
f"Method '{method_name}' does not exist on main window or current panel"
)
[docs]
@remote_controlled
def call_method_slot(
self,
method_name: str,
args: list,
panel: Literal["signal", "image"] | None,
kwargs: dict,
) -> None:
"""Slot to call a method from RemoteServer thread in GUI thread.
This slot receives signals from RemoteServer and executes the method in
the GUI thread, avoiding thread-safety issues with Qt widgets and dialogs.
Args:
method_name: Name of the method to call
args: Positional arguments as a list
panel: Panel name or None for auto-detection
kwargs: Keyword arguments as a dict
"""
# Call the method and store result in RemoteServer
try:
result = self.call_method(method_name, *args, panel=panel, **kwargs)
# Store result in RemoteServer for retrieval by XML-RPC thread
self.remote_server.result = result
self.remote_server.exception = None # Clear any previous exception
except Exception as exc: # pylint: disable=broad-except
# Store exception for re-raising in XML-RPC thread
self.remote_server.result = None
self.remote_server.exception = exc
[docs]
@remote_controlled
def get_object_shapes(
self,
nb_id_title: int | str | None = None,
panel: Literal["signal", "image"] | None = None,
) -> list:
"""Get plot item shapes associated to object (signal/image).
Args:
nb_id_title: Object number, or object id, or object title.
Defaults to None (current object).
panel: Panel name. Defaults to None (current panel).
Returns:
List of plot item shapes
"""
obj = self.get_object(nb_id_title, panel)
return list(create_adapter_from_object(obj).iterate_shape_items(editable=False))
[docs]
@remote_controlled
def add_annotations_from_items(
self,
items: list,
refresh_plot: bool = True,
panel: Literal["signal", "image"] | None = None,
) -> None:
"""Add object annotations (annotation plot items).
Args:
items: annotation plot items
refresh_plot: refresh plot. Defaults to True.
panel: panel name. If None, current panel is used.
"""
panel = self.__get_datapanel(panel)
panel.add_annotations_from_items(items, refresh_plot)
[docs]
@remote_controlled
def add_label_with_title(
self, title: str | None = None, panel: Literal["signal", "image"] | None = None
) -> None:
"""Add a label with object title on the associated plot
Args:
title: Label title. Defaults to None.
If None, the title is the object title.
panel: panel name. If None, current panel is used.
"""
self.__get_datapanel(panel).add_label_with_title(title)
[docs]
@remote_controlled
def run_macro(self, number_or_title: int | str | None = None) -> None:
"""Run macro.
Args:
number: Number of the macro (starting at 1). Defaults to None (run
current macro, or does nothing if there is no macro).
"""
self.macropanel.run_macro(number_or_title)
[docs]
@remote_controlled
def stop_macro(self, number_or_title: int | str | None = None) -> None:
"""Stop macro.
Args:
number: Number of the macro (starting at 1). Defaults to None (stop
current macro, or does nothing if there is no macro).
"""
self.macropanel.stop_macro(number_or_title)
[docs]
@remote_controlled
def import_macro_from_file(self, filename: str) -> None:
"""Import macro from file
Args:
filename: Filename.
"""
self.macropanel.import_macro_from_file(filename)
# ------WebAPI control
[docs]
@remote_controlled
def start_webapi_server(
self,
host: str | None = None,
port: int | None = None,
) -> dict:
"""Start the Web API server.
Args:
host: Host address to bind to. Defaults to "127.0.0.1".
port: Port number. Defaults to auto-detect available port.
Returns:
Dictionary with "url" and "token" keys.
Raises:
RuntimeError: If Web API deps not installed or server already running.
"""
if not WEBAPI_AVAILABLE:
raise RuntimeError(
"Web API dependencies not installed. "
"Install with: pip install datalab-platform[webapi]"
)
controller = get_webapi_controller()
controller.set_main_window(self)
url, token = controller.start(host=host, port=port)
return {"url": url, "token": token}
[docs]
@remote_controlled
def stop_webapi_server(self) -> None:
"""Stop the Web API server."""
if not WEBAPI_AVAILABLE:
return
controller = get_webapi_controller()
controller.stop()
[docs]
@remote_controlled
def get_webapi_status(self) -> dict:
"""Get Web API server status.
Returns:
Dictionary with "running", "url", and "token" keys.
"""
if not WEBAPI_AVAILABLE:
return {"running": False, "url": None, "token": None, "available": False}
controller = get_webapi_controller()
info = controller.get_connection_info()
info["available"] = True
return info
# ------Misc.
@property
def panels(self) -> tuple[AbstractPanel, ...]:
"""Return the tuple of implemented panels (signal, image, macro, history)
Returns:
Tuple of panels
"""
return (
self.signalpanel,
self.imagepanel,
self.macropanel,
self.historypanel,
)
def __show_webapi_info(self) -> None:
"""Show Web API connection info when status widget is clicked."""
if self.webapi_actions is not None:
self.webapi_actions.show_connection_info()
def __start_webapi_server(self) -> None:
"""Start Web API server when status widget is clicked."""
if self.webapi_actions is not None:
self.webapi_actions.start_server_from_status_widget()
[docs]
def check_for_v020_plugins(self) -> None: # pragma: no cover
"""Check for v0.20 plugins and warn user if any are found"""
if Conf.v020_plugins_warning_ignore.get(False):
return
v020_plugins = discover_v020_plugins()
if execenv.unattended or not v020_plugins:
return
# Build plugin list with clickable directory paths
plugin_items = []
for name, directory_path in v020_plugins:
if directory_path:
# Create clickable file:// link to directory
dir_url = QC.QUrl.fromLocalFile(directory_path).toString()
plugin_items.append(
f'<li>{name} (<a href="{dir_url}">{directory_path}</a>)</li>'
)
else:
plugin_items.append(f"<li>{name}</li>")
plugin_list = "<ul>" + "".join(plugin_items) + "</ul>"
txtlist = [
"<b>" + _("DataLab v0.20 plugins detected") + "</b>",
"",
_("The following plugins are using the old DataLab v0.20 format:"),
plugin_list,
_(
"These plugins will <b>not be loaded</b> in DataLab v1.0 because "
"they are not compatible with the new architecture."
),
"",
_(
"To use these plugins with DataLab v1.0, you need to update them. "
"Please refer to the migration guide on the DataLab website "
)
+ '(<a href="https://datalab-platform.com/en/features/advanced/'
'migration_v020_to_v100.html">Migration guide</a>)'
+ _(" or in the PDF documentation."),
"",
_("Choosing to ignore this message will prevent it from appearing again."),
]
answer = QW.QMessageBox.question(
self,
APP_NAME,
"<br>".join(txtlist),
QW.QMessageBox.Ok | QW.QMessageBox.Ignore,
)
if answer == QW.QMessageBox.Ignore:
Conf.v020_plugins_warning_ignore.set(True)
[docs]
def execute_post_show_actions(self) -> None:
"""Execute post-show actions"""
super().execute_post_show_actions()
self.check_for_v020_plugins()
if not execenv.unattended and Conf.tour_enabled.get():
Conf.tour_enabled.set(False)
self.show_tour()
# Auto-start WebAPI server if environment variable is set
if os.environ.get("DATALAB_WEBAPI_ENABLED") == "1":
try:
self.start_webapi_server()
except Exception as e: # pylint: disable=broad-exception-caught
print(f"Warning: Failed to auto-start WebAPI server: {e}")
[docs]
def take_menu_screenshots(self) -> None: # pragma: no cover
"""Take menu screenshots"""
for panel in self.panels:
if isinstance(panel, base.BaseDataPanel):
self.tabwidget.setCurrentWidget(panel)
for name in (
"file",
"create",
"edit",
"roi",
"view",
"operation",
"processing",
"analysis",
"help",
):
menu = getattr(self, f"{name}_menu")
menu.popup(self.pos())
qth.grab_save_window(menu, f"{panel.objectName()}_{name}")
menu.close()
if panel in (self.signalpanel, self.imagepanel):
panel: BaseDataPanel
# Take screenshots of Edit menu submenus (Metadata and Annotations)
for submenu, suffix in (
(panel.acthandler.metadata_submenu, "_edit_metadata"),
(panel.acthandler.annotations_submenu, "_edit_annotations"),
):
submenu.popup(self.pos())
qth.grab_save_window(submenu, f"{panel.objectName()}{suffix}")
submenu.close()
# ------GUI setup
def _setup_docks(self) -> None:
"""Add the macro, history and AI assistant docks"""
self.__add_macro_panel()
self.__add_history_panel()
self.__add_aiassistant_panel()
def _post_setup(self, console: bool) -> None:
"""Create plugin actions and wire panels, once the whole UI exists"""
self.__create_plugins_actions()
self.__update_actions(update_other_data_panel=True)
self.__configure_panels()
def _restore_state(self) -> None:
"""Restore the persisted layout, then realign the view with the current tab"""
super()._restore_state()
# A persisted layout may leave the macro or AI assistant dock raised, which
# would hide the view of the current panel tab:
self.__raise_view_dock(self.tabwidget.currentWidget())
def __register_plugins(self) -> None:
"""Discover and register third-party plugins at startup
The discovery phase imports all modules following the plugin
naming convention. Plugin classes are then provided by
:class:`PluginRegistry` and instantiated/registered here.
Errors are captured per-plugin so that one failing plugin does not
prevent the others from loading. Because this method runs before
the internal console is available, error tracebacks are buffered
in ``_startup_errors`` and replayed to the console later (see
:meth:`setup`).
"""
# Clear plugin class registry to avoid duplicate registration
# when reloading modules or running tests
PluginRegistry.clear_plugin_classes()
with sgmx_qth.try_or_log_error("Discovering plugins"):
# Discovering plugins
plugin_nb = len(discover_plugins())
execenv.log(self, f"{plugin_nb} plugin(s) found")
# Buffer any import errors that occurred during discovery
self._startup_errors.extend(PluginRegistry.get_discovery_errors())
# Get enabled plugins list from configuration
# None = all plugins enabled (default), [] = no plugins, list = specific plugins
enabled_list = Conf.plugins_enabled_list.get(None)
if not Conf.plugins_enabled.get():
self.plugins_last_load_at = datetime.now().astimezone()
return
for plugin_class in PluginRegistry.get_plugin_classes():
try:
# Check if plugin is enabled before instantiation
# None means all plugins are enabled
if enabled_list is not None:
plugin_name = plugin_class.PLUGIN_INFO.name
if plugin_name not in enabled_list:
execenv.log(
self,
f"Plugin {plugin_name} is disabled, skipping registration",
)
continue
# Instantiate and register plugin
plugin: PluginBase = plugin_class()
plugin.register(self)
# Plugin registration executes third-party code. We intentionally
# isolate any exception here so the failure is still reported in the
# internal console, log files, and plugin configuration dialog.
except Exception: # pylint: disable=broad-except
if sgmx_qth.is_running_tests():
raise
# Log to file (same mechanism as try_or_log_error)
tb_text = traceback.format_exc()
traceback.print_exc()
logger = logging.getLogger(__name__)
logger.error(
"Error in Instantiating and registering plugin %s",
plugin_class.__name__,
exc_info=True,
)
Conf.traceback_log_available.set(True)
# Buffer for replay in console once it is ready
self._startup_errors.append(tb_text)
# Record structured info about the failed plugin
mod = sys.modules.get(plugin_class.__module__)
filepath = getattr(mod, "__file__", "") if mod else ""
PluginRegistry.add_failed_plugin(
plugin_class.__name__, filepath or "", tb_text
)
self.plugins_last_load_at = datetime.now().astimezone()
def __flush_startup_errors(self) -> None:
"""Write any buffered startup errors to the internal console.
Called right after :meth:`_setup_console` so that plugin-import
tracebacks captured during :meth:`__register_plugins` become
visible to the user in the console widget.
"""
if self.console is None or not self._startup_errors:
return
for tb_text in self._startup_errors:
self.console.write_error(tb_text)
self._startup_errors.clear()
def __create_plugins_actions(self) -> None:
"""Ask each registered plugin to create its UI actions
Actions created while the PLUGINS category is active are stored
in the panels' action handlers and later exposed through the
*Plugins* menu.
"""
with self.signalpanel.acthandler.new_category(ActionCategory.PLUGINS):
with self.imagepanel.acthandler.new_category(ActionCategory.PLUGINS):
for plugin in PluginRegistry.get_plugins():
with sgmx_qth.try_or_log_error(
f"Create actions for {plugin.info.name}"
):
plugin.create_actions()
@staticmethod
def __unregister_plugins() -> None:
"""Unregister all plugins and let them cleanup their hooks"""
with sgmx_qth.try_or_log_error("Unregistering plugins"):
PluginRegistry.unregister_all_plugins()
def __restart_processor_pool(self) -> None:
"""Restart the shared pool after plugin paths change at runtime."""
for processor in (self.imagepanel.processor, self.signalpanel.processor):
if processor.worker is not None:
processor.worker.restart_pool()
return
def __configure_plugins(self) -> None:
"""Open plugin configuration dialog"""
dialog = PluginConfigDialog(self)
dialog.exec()
[docs]
def set_plugins_enabled(self, enabled: bool) -> None:
"""Apply the global third-party plugin enabled state."""
Conf.plugins_enabled.set(enabled)
self.__apply_plugins_enabled_setting()
[docs]
def reload_plugins(self) -> None:
"""Reload third-party plugins at runtime.
This unregisters active plugins, clears plugin actions from both panels,
re-discovers plugin modules (reloading code changes from disk),
re-registers enabled plugins, then recreates plugin actions and refreshes
menus.
"""
with sgmx_qth.try_or_log_error("Reloading plugins"):
if not Conf.plugins_enabled.get():
QW.QMessageBox.information(
self,
_("Plugins"),
_(
"Third-party plugins are disabled. Enable them again "
"from the plugin configuration dialog to use this "
"feature."
),
)
return
# Unregister existing plugin instances
self.__unregister_plugins()
# Clear existing plugin actions on both panels so that
# removed plugins no longer appear in menus.
for panel in (self.signalpanel, self.imagepanel):
panel.acthandler.clear_plugin_actions()
# Reset plugin class registry and rediscover plugins. The
# discovery step will reload already-imported modules so
# that code changes are picked up.
PluginRegistry.clear_plugin_classes()
with sgmx_qth.try_or_log_error("Discovering plugins (reload)"):
plugin_nb = len(discover_plugins())
execenv.log(self, f"{plugin_nb} plugin(s) found (reloaded)")
self.__restart_processor_pool()
# Get enabled plugins list from configuration
# None = all enabled (default), [] = none, list = specific plugins
enabled_list = Conf.plugins_enabled_list.get(None)
# Instantiate and register plugins again
for plugin_class in PluginRegistry.get_plugin_classes():
try:
# Check if plugin is enabled before instantiation
# None means all plugins are enabled
if enabled_list is not None:
plugin_name = plugin_class.PLUGIN_INFO.name
if plugin_name not in enabled_list:
execenv.log(
self,
f"Plugin {plugin_name} is disabled, "
"skipping registration (reload)",
)
continue
plugin: PluginBase = plugin_class()
plugin.register(self)
# Plugin registration executes third-party code. We intentionally
# isolate any exception here so the failure is still reported in the
# internal console, log files, and plugin configuration dialog.
except Exception: # pylint: disable=broad-except
if sgmx_qth.is_running_tests():
raise
context = (
f"Instantiating and registering plugin "
f"{plugin_class.__name__} (reload)"
)
tb_text = traceback.format_exc()
traceback.print_exc()
logger = logging.getLogger(__name__)
logger.error("Error in %s", context, exc_info=True)
Conf.traceback_log_available.set(True)
# Write error to console (available during reload)
if self.console is not None:
self.console.write_error(tb_text)
# Record structured info about the failed plugin
mod = sys.modules.get(plugin_class.__module__)
filepath = getattr(mod, "__file__", "") if mod else ""
PluginRegistry.add_failed_plugin(
plugin_class.__name__, filepath or "", tb_text
)
# Recreate plugin actions for the new plugin set
self.__create_plugins_actions()
# Update actions and menus to reflect new plugin set
self.__update_actions(update_other_data_panel=True)
# Update plugin status in the status bar
self.pluginstatus.update_status()
self.__update_plugins_availability()
self.plugins_last_load_at = datetime.now().astimezone()
def _configure_statusbar(self, console: bool) -> None:
"""Configure status bar
Args:
console: True if console is enabled
"""
super()._configure_statusbar(console)
self.__update_plugins_availability()
def _get_extra_status_widgets(self) -> list[QW.QWidget]:
"""Return the plugin, XML-RPC server and Web API server status widgets"""
self.pluginstatus = dl_status.PluginStatus()
xmlrpcstatus = dl_status.XMLRPCStatus()
xmlrpcstatus.set_port(self.remote_server.port)
self.webapistatus = dl_status.WebAPIStatus()
self.webapistatus.SIG_SHOW_INFO.connect(self.__show_webapi_info)
self.webapistatus.SIG_START_SERVER.connect(self.__start_webapi_server)
return [self.pluginstatus, xmlrpcstatus, self.webapistatus]
def __update_plugins_availability(self) -> None:
"""Update plugin-related UI according to third-party plugin setting."""
plugins_enabled = Conf.plugins_enabled.get()
if self.reload_plugins_action is not None:
self.reload_plugins_action.setEnabled(plugins_enabled)
if self.configure_plugins_action is not None:
self.configure_plugins_action.setEnabled(True)
if hasattr(self, "pluginstatus") and self.pluginstatus is not None:
self.pluginstatus.update_status()
def __apply_plugins_enabled_setting(self) -> None:
"""Apply third-party plugin enablement without manual user intervention."""
plugins_enabled = Conf.plugins_enabled.get()
if plugins_enabled:
self.reload_plugins()
return
self.__unregister_plugins()
for panel in (self.signalpanel, self.imagepanel):
panel.acthandler.clear_plugin_actions()
self.__update_actions(update_other_data_panel=True)
self.__update_plugins_availability()
def _create_global_actions(self) -> None:
"""Create standard and DataLab-specific global actions."""
super()._create_global_actions()
self.settings_action = create_action(
self,
_("Settings..."),
icon=get_icon("libre-gui-settings.svg"),
tip=_("Open settings dialog"),
triggered=self.__edit_settings,
)
self.command_palette_action = create_action(
self,
_("Command palette..."),
shortcut=QG.QKeySequence("Ctrl+Shift+P"),
icon=get_icon("command_palette.svg"),
tip=_("Search and run any command by its menu path"),
triggered=self.show_command_palette,
)
self.addAction(self.command_palette_action)
# View menu actions
self.autorefresh_action = create_action(
self,
_("Auto-refresh"),
icon=get_icon("refresh-auto.svg"),
tip=_("Auto-refresh plot when object is modified, added or removed"),
toggled=self.handle_autorefresh_action,
)
self.showfirstonly_action = create_action(
self,
_("Show first object only"),
icon=get_icon("show_first.svg"),
tip=_("Show only the first selected object (signal or image)"),
toggled=self.toggle_show_first_only,
)
self.showlabel_action = create_action(
self,
_("Show graphical object titles"),
icon=get_icon("show_titles.svg"),
tip=_("Show or hide ROI and other graphical object titles or subtitles"),
toggled=self.toggle_show_titles,
)
# Plugins menu actions
self.reload_plugins_action = create_action(
self,
_("Reload plugins"),
icon=get_icon("refresh-auto.svg"),
tip=_("Reload third-party plugins from disk"),
triggered=self.reload_plugins,
)
self.configure_plugins_action = create_action(
self,
_("Configure plugins..."),
icon=get_icon("libre-gui-settings.svg"),
tip=_("Enable or disable plugins"),
triggered=self.__configure_plugins,
)
self.__update_plugins_availability()
def _get_main_toolbar_actions(self) -> list[QW.QAction | None]:
"""Return standard HDF5 actions followed by DataLab settings."""
return [
self.openh5_action,
self.saveh5_action,
self.browseh5_action,
None,
self.settings_action,
]
def __add_signal_panel(self) -> DockablePlotWidget:
"""Setup signal toolbar, widgets and panel"""
self.signalpanel_toolbar = self._add_toolbar(
_("Signal Panel Toolbar"), "left", "signalpanel_toolbar"
)
dpw = DockablePlotWidget(self, PlotType.CURVE)
self.signalpanel = signal.SignalPanel(self, dpw, self.signalpanel_toolbar)
self.signalpanel.SIG_STATUS_MESSAGE.connect(self.statusBar().showMessage)
plot = dpw.get_plot()
plot.add_item(make.legend("TR"))
plot.SIG_ITEM_PARAMETERS_CHANGED.connect(
self.signalpanel.plot_item_parameters_changed
)
plot.SIG_ITEM_MOVED.connect(self.signalpanel.plot_item_moved)
return dpw
def __add_image_panel(self) -> DockablePlotWidget:
"""Setup image toolbar, widgets and panel"""
self.imagepanel_toolbar = self._add_toolbar(
_("Image Panel Toolbar"), "left", "imagepanel_toolbar"
)
dpw = DockablePlotWidget(self, PlotType.IMAGE)
self.imagepanel = image.ImagePanel(self, dpw, self.imagepanel_toolbar)
# -----------------------------------------------------------------------------
# # Before eventually disabling the "peritem" mode by default, wait for the
# # plotpy bug to be fixed (peritem mode is not compatible with multiple image
# # items):
# for cspanel in (
# self.imagepanel.plotwidget.get_xcs_panel(),
# self.imagepanel.plotwidget.get_ycs_panel(),
# ):
# cspanel.peritem_ac.setChecked(False)
# -----------------------------------------------------------------------------
self.imagepanel.SIG_STATUS_MESSAGE.connect(self.statusBar().showMessage)
plot = dpw.get_plot()
plot.SIG_ITEM_PARAMETERS_CHANGED.connect(
self.imagepanel.plot_item_parameters_changed
)
plot.SIG_ITEM_MOVED.connect(self.imagepanel.plot_item_moved)
plot.SIG_LUT_CHANGED.connect(self.imagepanel.plot_lut_changed)
return dpw
def __update_tab_menu(self) -> None:
"""Update tab menu"""
current_panel: BaseDataPanel = self.tabwidget.currentWidget()
add_actions(self.tabmenu, current_panel.get_context_menu().actions())
def _setup_panels(self) -> None:
"""Create signal and image panels, with their views and docks"""
self.signalview = self.__add_signal_panel()
self.imageview = self.__add_image_panel()
self._add_dockwidget(
self.signalview,
_("Signal View"),
name="signal_view",
key=self.signalpanel,
)
self._add_dockwidget(
self.imageview,
_("Image View"),
name="image_view",
key=self.imagepanel,
tabify_with=self.signalpanel,
)
self.signalpanel.SIG_OBJECT_ADDED.connect(
lambda: self.set_current_panel("signal")
)
self.imagepanel.SIG_OBJECT_ADDED.connect(
lambda: self.set_current_panel("image")
)
for panel in (self.signalpanel, self.imagepanel):
# Selecting an object must bring its view back to the front, even when
# the macro or AI assistant dock is currently raised:
panel.objview.SIG_SELECTION_CHANGED.connect(
functools.partial(self.__raise_view_dock, panel)
)
panel.setup_panel()
def _setup_central_widget(self) -> None:
"""Setup central widget (main panel)"""
super()._setup_central_widget()
configure_menu_about_to_show(self.tabmenu, self.__update_tab_menu)
self.tabwidget.setMaximumWidth(600)
s_idx = self.tabwidget.addTab(
self.signalpanel, get_icon("signal.svg"), _("Signal Panel")
)
i_idx = self.tabwidget.addTab(
self.imagepanel, get_icon("image.svg"), _("Image Panel")
)
self.tabwidget.setTabToolTip(
s_idx, _("1D Signals: Manage and process one-dimensional data")
)
self.tabwidget.setTabToolTip(
i_idx, _("2D Images: Manage and process two-dimensional data")
)
self.tabwidget.currentChanged.connect(self.__tab_index_changed)
def _get_menubar_layout(self) -> list[tuple[str, str]]:
"""Insert the DataLab menus between the standard File and View menus."""
layout = super()._get_menubar_layout()
return (
layout[:1]
+ [
("create", _("&Create")),
("edit", _("&Edit")),
("roi", _("ROI")),
("operation", _("Operations")),
("processing", _("Processing")),
("analysis", _("Analysis")),
("plugins", _("Plugins")),
]
+ layout[1:]
)
def _get_help_doc_actions(self) -> list[QW.QAction | None]:
"""Append the tour and demo entries to the standard documentation actions."""
return super()._get_help_doc_actions() + [
create_action(
self,
_("Tour") + "...",
icon=get_icon("tour.svg"),
triggered=self.show_tour,
),
create_action(
self,
_("Demo") + "...",
icon=get_icon("play_demo.svg"),
triggered=self.play_demo,
),
None,
]
def _get_help_support_actions(self) -> list[QW.QAction | None]:
"""Append the installation and configuration viewer."""
return super()._get_help_support_actions() + [
create_action(
self,
_("Installation and configuration") + "...",
icon=get_icon("libre-toolbox.svg"),
triggered=lambda: instconfviewer.exec_datalab_installconfig_dialog(
self
),
),
]
def _get_help_menu_actions(self) -> list[QW.QAction | None]:
"""Prepend the command palette to the standard help actions."""
return [self.command_palette_action, None] + super()._get_help_menu_actions()
def _add_menus(self) -> None:
"""Adding menus"""
super()._add_menus()
# Make plugins menu scrollable to handle many plugins without overflow
self.plugins_menu.setStyleSheet("QMenu { menu-scrollable: 1; }")
for menu in (
self.create_menu,
self.edit_menu,
self.roi_menu,
self.operation_menu,
self.processing_menu,
self.analysis_menu,
self.plugins_menu,
):
configure_menu_about_to_show(menu, self.__update_generic_menu)
# Command palette launcher in the top-right corner of the menu bar:
# a search-box-styled field so the palette is discoverable at a
# glance (mirrors the DataLab-Web command palette trigger).
shortcut_text = self.command_palette_action.shortcut().toString(
QG.QKeySequence.NativeText
)
command_palette_field = CommandSearchField(
self, self.show_command_palette, shortcut_text
)
self.menuBar().setCornerWidget(command_palette_field, QC.Qt.TopRightCorner)
[docs]
def show_command_palette(self) -> None:
"""Show the command palette (searchable list of menu commands).
Lists every command available for the current panel by its menu
path and triggers the one chosen by the user.
"""
panel = self.__get_current_basedatapanel()
commands = collect_commands(self, panel)
dialog = CommandPaletteDialog(self, commands)
# Anchor near the top-center of the main window, VSCode-style.
geometry = self.geometry()
dialog.move(geometry.center().x() - dialog.width() // 2, geometry.top() + 80)
if dialog.exec():
action = dialog.get_selected_action()
if action is not None:
action.trigger()
def _get_console_namespace(self) -> dict[str, object]:
"""Return the DataLab internal-console namespace."""
return {
"dl": self,
"np": np,
"sps": sps,
"spi": spi,
"os": os,
"sys": sys,
"osp": osp,
"time": time,
}
def _get_console_message(self) -> str:
"""Return the DataLab internal-console welcome message."""
return _(
"Welcome to DataLab console!\n"
"---------------------------\n"
"You can access the main window with the 'dl' variable.\n"
"Example:\n"
" o = dl.get_object() # returns currently selected object\n"
" o = dl[1] # returns object number 1\n"
" o = dl['My image'] # returns object which title is 'My image'\n"
" o.data # returns object data\n"
"Modules imported at startup: "
"os, sys, os.path as osp, time, "
"numpy as np, scipy.signal as sps, scipy.ndimage as spi"
)
def _configure_console(self) -> None:
"""Connect DataLab-specific console refresh behavior."""
super()._configure_console()
self.console.interpreter.widget_proxy.sig_new_prompt.connect(
lambda txt: self.repopulate_panel_trees()
)
self.__flush_startup_errors()
def __add_macro_panel(self) -> None:
"""Add macro panel"""
self.macropanel = macro.MacroPanel(self)
self._add_dockwidget(
self.macropanel,
_("Macro Panel"),
name="macro_panel",
tabify_with=self.imagepanel,
)
def __add_history_panel(self) -> None:
"""Add history panel"""
self.historypanel = history.HistoryPanel(self)
self._add_dockwidget(
self.historypanel,
_("History Panel"),
name="history_panel",
tabify_with=self.macropanel,
)
def __add_aiassistant_panel(self) -> None:
"""Add AI Assistant panel"""
# Local import to keep AI assistant fully optional/loadable on demand
# pylint: disable-next=import-outside-toplevel
from datalab.aiassistant.widgets.chatpanel import ( # noqa: WPS433
AIAssistantPanel,
)
self.aiassistantpanel = AIAssistantPanel(self)
self._add_dockwidget(
self.aiassistantpanel,
_("AI Assistant"),
name="ai_assistant",
tabify_with=self.macropanel,
)
def __configure_panels(self) -> None:
"""Configure panels"""
# Connectings signals
for panel in self.panels:
panel.SIG_OBJECT_ADDED.connect(self.set_modified)
panel.SIG_OBJECT_REMOVED.connect(self.set_modified)
self.macropanel.SIG_OBJECT_MODIFIED.connect(self.set_modified)
# Initializing common panel actions
self.autorefresh_action.setChecked(Conf.auto_refresh.get(True))
self.showfirstonly_action.setChecked(Conf.show_first_only.get(False))
self.showlabel_action.setChecked(Conf.show_label.get(False))
# Restoring current tab from last session
tab_idx = Conf.current_tab.get(None)
if tab_idx is not None:
self.tabwidget.setCurrentIndex(tab_idx)
# Set focus on current panel, so that keyboard shortcuts work (Fixes #10)
self.tabwidget.currentWidget().setFocus()
[docs]
def set_process_isolation_enabled(self, state: bool) -> None:
"""Enable/disable process isolation
Args:
state: True to enable process isolation
"""
for processor in (self.imagepanel.processor, self.signalpanel.processor):
processor.set_process_isolation_enabled(state)
# ------Remote control
[docs]
@remote_controlled
def get_current_panel(self) -> str:
"""Return current panel name
Returns:
Panel name (valid values: "signal", "image", "macro")
"""
panel = self.tabwidget.currentWidget()
dock = self.docks[panel]
if panel is self.signalpanel and dock.isVisible():
return "signal"
if panel is self.imagepanel and dock.isVisible():
return "image"
return "macro"
[docs]
@remote_controlled
def set_current_panel(
self, panel: Literal["signal", "image", "macro"] | BaseDataPanel
) -> None:
"""Switch to panel.
Args:
panel: panel name or panel instance
Raises:
ValueError: unknown panel
"""
if not isinstance(panel, str):
if panel not in self.panels:
raise ValueError(f"Unknown panel {panel}")
panel = (
"signal"
if panel is self.signalpanel
else "image"
if panel is self.imagepanel
else "macro"
)
if self.get_current_panel() == panel:
if panel in ("signal", "image"):
# Force tab index changed event to be sure that the dock associated
# to the current panel is raised
self.__tab_index_changed(self.tabwidget.currentIndex())
return
if panel == "signal":
self.tabwidget.setCurrentWidget(self.signalpanel)
# setCurrentWidget emits currentChanged only on an actual change, so the
# dock is raised explicitly for the already-current tab:
self.__raise_view_dock(self.signalpanel)
elif panel == "image":
self.tabwidget.setCurrentWidget(self.imagepanel)
self.__raise_view_dock(self.imagepanel)
elif panel == "macro":
self.docks[self.macropanel].raise_()
else:
raise ValueError(f"Unknown panel {panel}")
[docs]
@remote_controlled
def calc(
self, name: str, param: gds.DataSet | None = None, edit: bool = True
) -> None:
"""Call computation feature ``name``
.. note::
This calls either the processor's ``compute_<name>`` method (if it exists),
or the processor's ``<name>`` computation feature (if it is registered,
using the ``run_feature`` method).
It looks for the function in all panels, starting with the current one.
Args:
name: Compute function name
param: Compute function parameter. Defaults to None.
edit: Whether to show parameter edit dialog. Defaults to True.
Set to False when calling from remote/API to avoid blocking dialogs.
Raises:
ValueError: unknown function
"""
panels = [self.tabwidget.currentWidget()]
panels.extend(self.panels)
for panel in panels:
if isinstance(panel, base.BaseDataPanel):
name = name.removeprefix("compute_")
panel: base.BaseDataPanel
# Some computation features are wrapped in a method with a
# "compute_" prefix, so we check for this first:
func = getattr(panel.processor, f"compute_{name}", None)
if func is not None:
if param is None:
func()
else:
func(param)
return
# If the function is not wrapped, we check if it is a
# registered feature:
try:
feature = panel.processor.get_feature(name)
panel.processor.run_feature(feature, param, edit=edit)
return
except ValueError:
continue
raise ValueError(f"Unknown computation function {name}")
# ------GUI refresh
[docs]
def has_objects(self) -> bool:
"""Return True if sig/ima panels have any object"""
return sum(len(panel) for panel in self.panels) > 0
def _normalize_modified_state(self, state: bool) -> bool:
"""Keep empty DataLab workspaces unmodified."""
return state and self.has_objects()
[docs]
def repopulate_panel_trees(self) -> None:
"""Repopulate all panel trees"""
for panel in self.panels:
if isinstance(panel, base.BaseDataPanel):
panel.objview.populate_tree()
def __update_actions(self, update_other_data_panel: bool = False) -> None:
"""Update selection dependent actions
Args:
update_other_data_panel: True to update other data panel actions
(i.e. if the current panel is the signal panel, also update the image
panel actions, and vice-versa)
"""
is_signal = self.tabwidget.currentWidget() is self.signalpanel
panel = self.signalpanel if is_signal else self.imagepanel
other_panel = self.imagepanel if is_signal else self.signalpanel
if update_other_data_panel:
other_panel.selection_changed()
panel.selection_changed()
self.signalpanel_toolbar.setVisible(is_signal)
self.imagepanel_toolbar.setVisible(not is_signal)
def __tab_index_changed(self, index: int) -> None:
"""Switch from signal to image mode, or vice-versa"""
dock = self.docks[self.tabwidget.widget(index)]
dock.raise_()
self.__update_actions()
def __raise_view_dock(self, panel: BaseDataPanel) -> None:
"""Bring the panel view dock to the front, if the panel is the current tab
Args:
panel: signal or image panel
"""
# The guard matters because selection changes are also emitted for the
# panel that is not currently shown (see __update_actions).
if self.tabwidget.currentWidget() is panel:
self.docks[panel].raise_()
def __update_generic_menu(self, menu: QW.QMenu | None = None) -> None:
"""Update menu before showing up -- Generic method"""
if menu is None:
menu = self.sender()
menu.clear()
panel = self.tabwidget.currentWidget()
category = {
self.file_menu: ActionCategory.FILE,
self.create_menu: ActionCategory.CREATE,
self.edit_menu: ActionCategory.EDIT,
self.roi_menu: ActionCategory.ROI,
self.view_menu: ActionCategory.VIEW,
self.operation_menu: ActionCategory.OPERATION,
self.processing_menu: ActionCategory.PROCESSING,
self.analysis_menu: ActionCategory.ANALYSIS,
self.plugins_menu: ActionCategory.PLUGINS,
}[menu]
actions = panel.get_category_actions(category)
# Always expose the reload action in the Plugins menu, even if
# no plugin has registered actions yet (so that new plugins can be
# discovered after they are added on disk).
if menu is self.plugins_menu:
actions = list(actions) + [
None,
self.configure_plugins_action,
self.reload_plugins_action,
]
add_actions(menu, actions)
def _get_file_menu_actions(self) -> list[QW.QAction | None]:
"""Append the settings action to the standard HDF5 actions."""
return super()._get_file_menu_actions() + [None, self.settings_action]
def _update_file_menu(self) -> None:
"""Update file menu before showing up"""
self.saveh5_action.setEnabled(self.has_objects())
self.__update_generic_menu(self.file_menu)
add_actions(self.file_menu, self._get_file_menu_actions())
# Add Web API submenu
if self.webapi_actions is not None:
self.file_menu.addSeparator()
self.webapi_actions.create_menu(self.file_menu)
if self.quit_action is not None:
add_actions(self.file_menu, [self.quit_action])
def _update_view_menu(self) -> None:
"""Update view menu before showing up"""
self.__update_generic_menu(self.view_menu)
super()._update_view_menu()
[docs]
@remote_controlled
def toggle_show_titles(self, state: bool) -> None:
"""Toggle show annotations option
Args:
state: state
"""
Conf.show_label.set(state)
for datapanel in (self.signalpanel, self.imagepanel):
for obj in datapanel.objmodel:
obj.set_metadata_option("showlabel", state)
datapanel.refresh_plot("selected", True, False)
[docs]
def handle_autorefresh_action(self, state: bool) -> None:
"""Handle auto-refresh action from UI (with confirmation dialog)
Args:
state: desired state
"""
# If disabling auto-refresh, show confirmation dialog
if not state:
txtlist = [
"<b>" + _("Disable auto-refresh?") + "</b>",
"",
_(
"When auto-refresh is disabled, the plot view will not "
"automatically update when objects are modified, added or removed."
),
"",
_(
"You will need to manually click the refresh button to update "
"the view."
),
"",
_("Are you sure you want to disable auto-refresh?"),
]
answer = QW.QMessageBox.question(
self,
APP_NAME,
"<br>".join(txtlist),
QW.QMessageBox.Yes | QW.QMessageBox.No,
QW.QMessageBox.No,
)
if answer == QW.QMessageBox.No:
# User cancelled, restore the action's checked state
self.autorefresh_action.blockSignals(True)
self.autorefresh_action.setChecked(True)
self.autorefresh_action.blockSignals(False)
return
# Apply the change
self.toggle_auto_refresh(state)
[docs]
@remote_controlled
def toggle_auto_refresh(self, state: bool) -> None:
"""Toggle auto refresh option
Args:
state: state
"""
Conf.auto_refresh.set(state)
for datapanel in (self.signalpanel, self.imagepanel):
datapanel.plothandler.set_auto_refresh(state)
[docs]
@remote_controlled
def toggle_show_first_only(self, state: bool) -> None:
"""Toggle show first only option
Args:
state: state
"""
Conf.show_first_only.set(state)
for datapanel in (self.signalpanel, self.imagepanel):
datapanel.plothandler.set_show_first_only(state)
# ------Common features
[docs]
@remote_controlled
def reset_all(self) -> None:
"""Reset all application data"""
for panel in self.panels:
if panel is not None and panel is not self.historypanel:
panel.remove_all_objects()
if self.historypanel is not None:
self.historypanel.start_new_session_after_workspace_reset()
[docs]
@remote_controlled
def remove_object(self, force: bool = False) -> None:
"""Remove current object from current panel.
Args:
force: if True, remove object without confirmation. Defaults to False.
"""
panel = self.__get_current_basedatapanel()
panel.remove_object(force)
[docs]
@remote_controlled
def save_to_h5_file(self, filename: str | None = None) -> None:
"""Save to a DataLab HDF5 file
Args:
filename: HDF5 filename. If None, a file dialog is opened.
Raises:
IOError: if filename is invalid or file cannot be saved.
"""
super().save_to_h5_file(filename)
[docs]
@remote_controlled
def open_h5_files(
self,
h5files: list[str] | None = None,
import_all: bool | None = None,
reset_all: bool | None = None,
) -> None:
"""Open a DataLab HDF5 file or import from any other HDF5 file.
Args:
h5files: HDF5 filenames (optionally with dataset name, separated by ",")
import_all: Import all datasets from HDF5 files
reset_all: Reset all application data before importing
"""
super().open_h5_files(h5files, import_all, reset_all)
def _on_h5_save_requested(self, filename: str) -> None:
"""Record the workspace save in the history session"""
self.historypanel.add_ui_entry(
_("Save to HDF5 file"),
target="mainwindow",
method_name="save_to_h5_file",
save_state=False,
filename=filename,
)
def _on_h5_open_requested(
self,
h5files: list[str],
import_all: bool | None,
reset_all: bool | None,
) -> None:
"""Record the HDF5 open/import in the history session"""
if len(h5files) > 1:
entry_title = _("Open %d HDF5 files") % len(h5files)
else:
entry_title = _("Open HDF5 file")
self.historypanel.add_ui_entry(
entry_title,
target="mainwindow",
method_name="open_h5_files",
save_state=False,
h5files=h5files,
import_all=import_all,
reset_all=reset_all,
)
def _is_workspace_empty(self) -> bool:
"""Return whether the signal and image panels hold no object"""
return not self.has_objects()
def _get_clear_workspace_message(
self, import_all: bool | None, reset_all: bool
) -> str:
"""Return the confirmation message shown before clearing the workspace"""
msg = _(
"Do you want to clear current workspace "
"(signals and images) before importing data from "
"HDF5 files?"
)
# Only show the UUID conflict note when importing native DataLab
# workspace files (import_all=True), not when using HDF5 browser
if import_all:
msg += "<br><br>" + _(
"<u>Note:</u> If you choose <i>No</i>, when importing "
"DataLab workspace files, objects with conflicting "
"identifiers will have their processing history lost "
"(features like 'Show source' and 'Recompute' will not "
"work for those objects). Non-conflicting objects will "
"preserve their processing history."
)
msg += "<br><br>" + _(
"Choosing to ignore this message will prevent it "
"from being displayed again, and will use the "
"current setting (%s)."
) % (_("Yes") if reset_all else _("No"))
return msg
[docs]
def import_dataset_from_file(
self,
filename: str,
dsetname: str | None,
import_all: bool | None,
reset_all: bool,
) -> None:
"""Open a DataLab workspace file, or import a single dataset from it
Args:
filename: Path to the HDF5 file (already validated)
dsetname: Dataset name to import, or ``None`` to import all
import_all: If ``True``, import all datasets without browsing
reset_all: If ``True``, clear workspace before importing
"""
if dsetname is None:
self.h5inputoutput.open_file(filename, import_all, reset_all)
else:
self.h5inputoutput.import_dataset_from_file(filename, dsetname)
[docs]
def import_all_from_h5_file(
self, filename: str, reset_all: bool | None = None
) -> None:
"""Import an HDF5 file through the DataLab HDF5 browser
Args:
filename: HDF5 filename
reset_all: Delete all DataLab signals/images before importing data
"""
self.import_h5_file(filename, reset_all)
[docs]
def browse_h5_files(
self, filenames: list[str], reset_all: bool | None = None
) -> None:
"""Browse HDF5 files
Args:
filenames: HDF5 filenames
reset_all: Reset all application data before importing
"""
for filename in filenames:
self._check_h5file(filename, "load")
self.h5inputoutput.import_files(filenames, False, bool(reset_all))
[docs]
@remote_controlled
def load_h5_workspace(self, h5files: list[str], reset_all: bool = False) -> None:
"""Load native DataLab HDF5 workspace files programmatically.
This method does not create file-selection widgets or progress bars. When
history recording is active and the new-session policy is ``"ask"``, a
history-session question may still be shown before loading.
.. warning::
This method only supports native DataLab HDF5 files. For importing
arbitrary HDF5 files (non-native), use the GUI menu or macros with
:class:`datalab.control.proxy.RemoteProxy`.
Args:
h5files: List of native DataLab HDF5 filenames
reset_all: Reset all application data before importing. Defaults to False.
Raises:
ValueError: If a file is not a valid native DataLab HDF5 file
"""
# Offer a fresh history session for this load *before* recording anything.
self.historypanel.maybe_start_session_for_input(load=True)
with self.historypanel.session_prompt_suppressed():
for idx, filename in enumerate(h5files):
filename = self._check_h5file(filename, "load")
success = self.h5inputoutput.open_file_headless(
filename, reset_all=(reset_all and idx == 0)
)
if not success:
raise ValueError(
f"File '{filename}' is not a native DataLab HDF5 file. "
f"Use the GUI menu or a macro with RemoteProxy to import "
f"arbitrary HDF5 files."
)
# Refresh panel trees after loading
self.repopulate_panel_trees()
[docs]
@remote_controlled
def save_h5_workspace(self, filename: str) -> None:
"""Save current workspace to a native DataLab HDF5 file without GUI elements.
This method can be safely called from the internal console as it does not
create any Qt widgets, dialogs, or progress bars. It is designed for
programmatic use when saving DataLab workspace files.
Args:
filename: HDF5 filename to save to
Raises:
IOError: If file cannot be saved
"""
filename = self._check_h5file(filename, "save")
self.h5inputoutput.save_file(filename)
self.set_modified(False)
[docs]
@remote_controlled
def import_h5_file(self, filename: str, reset_all: bool | None = None) -> None:
"""Import HDF5 file into DataLab
Args:
filename: HDF5 filename (optionally with dataset name,
separated by ",")
reset_all: Delete all DataLab signals/images before importing data
"""
# Offer a fresh history session for this load *before* importing anything.
self.historypanel.maybe_start_session_for_input(load=True)
with self.historypanel.session_prompt_suppressed():
with qth.qt_try_loadsave_file(self, filename, "load"):
filename = self._check_h5file(filename, "load")
self.h5inputoutput.import_files([filename], False, reset_all)
# This method is intentionally *not* remote controlled
# (see TODO regarding RemoteClient.add_object method)
# @remote_controlled
[docs]
def add_object(
self,
obj: SignalObj | ImageObj,
group_id: str = "",
set_current=True,
new_session_behavior: SessionBehavior | None = None,
) -> bool:
"""Add object - signal or image
Args:
obj: object to add (signal or image)
group_id: group ID (optional)
set_current: True to set the object as current object
new_session_behavior: Optional history session creation policy
Returns:
True if the object was added successfully, False otherwise
"""
if not self.confirm_memory_state():
return False
if isinstance(obj, SignalObj):
panel = self.signalpanel
panel_str = "signal"
elif isinstance(obj, ImageObj):
panel = self.imagepanel
panel_str = "image"
else:
raise TypeError(f"Unsupported object type {type(obj)}")
# A remote call can reach the main window before the History panel dock
# exists, since the XML-RPC server is started before the UI is built.
historypanel = self.historypanel
if historypanel is None:
panel.add_object(obj, group_id, set_current)
return True
historypanel.maybe_start_session_for_input(behavior=new_session_behavior)
panel.add_object(obj, group_id, set_current)
# Record a creation entry so objects added programmatically (plugins,
# macros, remote control) appear in the history. ``panel.add_object``
# deliberately does not record, so creations entering through this
# proxy boundary would otherwise be lost (notably the very first one).
with historypanel.session_prompt_suppressed():
action = historypanel.add_ui_entry(
_("New %s") % panel_str,
target=panel_str + "panel",
method_name="new_object",
save_state=False,
)
if action is not None:
historypanel.register_action_outputs(action, [get_uuid(obj)])
return True
[docs]
@remote_controlled
def set_object(self, obj: SignalObj | ImageObj) -> None:
"""Set object data - update an existing signal or image in-place.
The existing object is identified by UUID carried by ``obj``
(from a previous :meth:`get_object` call).
Args:
obj: object with updated data (signal or image)
Raises:
KeyError: if no object with matching UUID is found
TypeError: if object type is unsupported
"""
if isinstance(obj, SignalObj):
self.signalpanel.set_object(obj)
elif isinstance(obj, ImageObj):
self.imagepanel.set_object(obj)
else:
raise TypeError(f"Unsupported object type {type(obj)}")
[docs]
@remote_controlled
def load_from_files(self, filenames: list[str]) -> None:
"""Open objects from files in current panel (signals/images)
Args:
filenames: list of filenames
"""
panel = self.__get_current_basedatapanel()
panel.load_from_files(filenames)
[docs]
@remote_controlled
def load_from_directory(self, path: str) -> None:
"""Open objects from directory in current panel (signals/images).
Args:
path: directory path
"""
panel = self.__get_current_basedatapanel()
panel.load_from_directory(path)
# ------Other methods related to AbstractDLControl interface
[docs]
def get_version(self) -> str:
"""Return DataLab public version.
Returns:
DataLab version
"""
return datalab.__version__
[docs]
def add_signal(
self,
title: str,
xdata: np.ndarray,
ydata: np.ndarray,
xunit: str = "",
yunit: str = "",
xlabel: str = "",
ylabel: str = "",
group_id: str = "",
set_current: bool = True,
new_session_behavior: SessionBehavior | None = None,
) -> bool: # pylint: disable=too-many-arguments
"""Add signal data to DataLab.
Args:
title: Signal title
xdata: X data
ydata: Y data
xunit: X unit. Defaults to ""
yunit: Y unit. Defaults to ""
xlabel: X label. Defaults to ""
ylabel: Y label. Defaults to ""
group_id: group id in which to add the signal. Defaults to ""
set_current: if True, set the added signal as current
new_session_behavior: Optional history session creation policy
Returns:
True if signal was added successfully, False otherwise
Raises:
ValueError: Invalid xdata dtype
ValueError: Invalid ydata dtype
"""
obj = create_signal(
title,
xdata,
ydata,
units=(xunit, yunit),
labels=(xlabel, ylabel),
)
return self.add_object(obj, group_id, set_current, new_session_behavior)
# This API mirrors the image metadata accepted by create_image, so the
# argument count is part of the stable public interface rather than noise.
[docs]
def add_image( # pylint: disable=too-many-arguments
self,
title: str,
data: np.ndarray,
xunit: str = "",
yunit: str = "",
zunit: str = "",
xlabel: str = "",
ylabel: str = "",
zlabel: str = "",
group_id: str = "",
set_current: bool = True,
new_session_behavior: SessionBehavior | None = None,
) -> bool:
"""Add image data to DataLab.
Args:
title: Image title
data: Image data
xunit: X unit. Defaults to ""
yunit: Y unit. Defaults to ""
zunit: Z unit. Defaults to ""
xlabel: X label. Defaults to ""
ylabel: Y label. Defaults to ""
zlabel: Z label. Defaults to ""
group_id: group id in which to add the image. Defaults to ""
set_current: if True, set the added image as current
new_session_behavior: Optional history session creation policy
Returns:
True if image was added successfully, False otherwise
Raises:
ValueError: Invalid data dtype
"""
obj = create_image(
title,
data,
units=(xunit, yunit, zunit),
labels=(xlabel, ylabel, zlabel),
)
return self.add_object(obj, group_id, set_current, new_session_behavior)
# ------?
def _about(self) -> None: # pragma: no cover
"""About dialog box"""
self.check_stable_release()
if self.remote_server.port is None:
xrpcstate = '<font color="red">' + _("not started") + "</font>"
else:
xrpcstate = _("started (port %s)") % self.remote_server.port
xrpcstate = f"<font color='green'>{xrpcstate}</font>"
if Conf.process_isolation_enabled.get():
pistate = "<font color='green'>" + _("enabled") + "</font>"
else:
pistate = "<font color='red'>" + _("disabled") + "</font>"
adv_conf = "<br>".join(
[
"<i>" + _("Advanced configuration:") + "</i>",
"• " + _("XML-RPC server:") + " " + xrpcstate,
"• " + _("Process isolation:") + " " + pistate,
]
)
created_by = _("Created by")
dev_by = _("Developed and maintained by %s open-source project team") % APP_NAME
cprght = "2023 DataLab Platform Developers"
QW.QMessageBox.about(
self,
_("About") + " " + APP_NAME,
f"""<b>{APP_NAME}</b> v{datalab.__version__}<br>{APP_DESC}
<p>{created_by} Pierre Raybaut<br>{dev_by}<br>Copyright © {cprght}
<p>{adv_conf}""",
)
def _update_extra_color_mode(self) -> None:
"""Update the macro panel color mode"""
if self.macropanel is not None:
self.macropanel.update_color_mode()
# Settings changes are intentionally dispatched in one place because each
# option may trigger a specific live UI update or panel refresh.
def __edit_settings(self) -> None: # pylint: disable=too-many-branches,too-many-statements
"""Edit settings"""
changed_options = edit_settings(self)
sigima_options.fft_shift_enabled.set(Conf.fft_shift_enabled.get())
sigima_options.auto_normalize_kernel.set(Conf.auto_normalize_kernel.get())
refresh_signal_panel = refresh_image_panel = False
# Handling changes to shape/marker parameters
s_view_result_param = (
"sig_shape_param" in changed_options
or "sig_marker_param" in changed_options
) and have_geometry_results(self.signalpanel.objview.get_sel_objects(True))
i_view_result_param = (
"ima_shape_param" in changed_options
or "ima_marker_param" in changed_options
) and have_geometry_results(self.imagepanel.objview.get_sel_objects(True))
if (s_view_result_param or i_view_result_param) and (
QW.QMessageBox.question(
self,
_("Apply settings to existing results?"),
_(
"Visualization settings for annotated shapes and "
"markers have been modified.\n\n"
"Do you want to apply these settings to existing results "
"in the workspace?"
),
QW.QMessageBox.Yes | QW.QMessageBox.No,
QW.QMessageBox.No,
)
== QW.QMessageBox.Yes
):
if s_view_result_param:
self.signalpanel.plothandler.refresh_all_shape_items()
if i_view_result_param:
self.imagepanel.plothandler.refresh_all_shape_items()
for option in changed_options:
if option in (
"max_shapes_to_draw",
"max_cells_in_label",
"max_cols_in_label",
):
refresh_signal_panel = refresh_image_panel = True
if option == "show_result_label":
for panel in (self.signalpanel, self.imagepanel):
panel.acthandler.show_label_action.setChecked(
Conf.show_result_label.get()
)
if option == "color_mode":
self._update_color_mode()
if option == "show_console_on_error":
self._update_console_show_mode()
if option == "plot_toolbar_position":
for dock in self.docks.values():
widget = dock.widget()
if isinstance(widget, DockablePlotWidget):
widget.update_toolbar_position()
if option.startswith(("sig_autodownsampling", "sig_linewidth")):
refresh_signal_panel = True
if option == "sig_autoscale_margin_percent":
# Update signal plot widget autoscale margin
sig_margin = Conf.sig_autoscale_margin_percent.get()
for dock in self.docks.values():
widget: DockablePlotWidget | QW.QWidget = dock.widget()
if isinstance(widget, DockablePlotWidget):
plot = widget.get_plot()
if (
hasattr(plot, "options")
and plot.options.type == PlotType.CURVE
):
plot.set_autoscale_margin_percent(sig_margin)
if option == "ima_autoscale_margin_percent":
# Update image plot widget autoscale margin
ima_margin = Conf.ima_autoscale_margin_percent.get()
for dock in self.docks.values():
widget: DockablePlotWidget | QW.QWidget = dock.widget()
if isinstance(widget, DockablePlotWidget):
plot = widget.get_plot()
if (
hasattr(plot, "options")
and plot.options.type == PlotType.IMAGE
):
plot.set_autoscale_margin_percent(ima_margin)
if option == "ima_defaults" and len(self.imagepanel) > 0:
answer = QW.QMessageBox.question(
self,
_("Visualization settings"),
_(
"Default visualization settings have changed.<br><br>"
"Do you want to update all active %s objects?"
)
% _("image"),
QW.QMessageBox.Yes | QW.QMessageBox.No,
)
if answer == QW.QMessageBox.Yes:
self.imagepanel.update_metadata_view_settings()
if option == "ima_aspect_ratio_1_1":
refresh_image_panel = True
if refresh_signal_panel:
self.signalpanel.manual_refresh()
if refresh_image_panel:
self.imagepanel.manual_refresh()
# Invalidate the AI assistant controller if any AI option changed,
# so the next prompt rebuilds it with the updated configuration.
if self.aiassistantpanel is not None and any(
option in AI_OPTION_NAMES for option in changed_options
):
self.aiassistantpanel.invalidate_controller()
[docs]
def play_demo(self) -> None:
"""Play demo"""
# pylint: disable=import-outside-toplevel
# pylint: disable=cyclic-import
from datalab.tests.scenarios import demo
demo.play_demo(self)
[docs]
def show_tour(self) -> None:
"""Show tour"""
# pylint: disable=import-outside-toplevel
# pylint: disable=cyclic-import
from datalab.gui import tour
tour.start(self)
# ------Close window
def _get_save_before_quit_message(self) -> str:
"""Return the DataLab workspace save confirmation message."""
return _(
"Do you want to save all signals and images "
"to an HDF5 file before quitting DataLab?"
)
def _close_managed_widgets(self) -> None:
"""Close DataLab panels and generic shell widgets."""
for panel in self.panels:
if panel is not None:
panel.close()
super()._close_managed_widgets()
def _cleanup_before_reset(self) -> None:
"""Clean up DataLab services before clearing panel data."""
super()._cleanup_before_reset()
if self.webapi_actions is not None:
self.webapi_actions.cleanup()
def _cleanup_after_state_save(self) -> None:
"""Persist DataLab UI state and unregister plugins after shutdown."""
self.__unregister_plugins()
if not execenv.unattended and self.tabwidget is not None:
Conf.current_tab.set(self.tabwidget.currentIndex())
super()._cleanup_after_state_save()
