# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Main controller for the RadioViz application.
This module contains the primary controller responsible for managing
the application's main window, handling image loading, managing tool
registrations, and coordinating between different components of the
RadioViz application.
Typical workflow for workspace save:
1. Save workspace (JSON/HDF5) using save_workspace()
2. For JSON:
- If there are nonpersistent images, the user is prompted to select which
ones to persist on disk before saving
- Unchecked images are excluded from the JSON workspace
- Tool items referencing excluded images are pruned
3. For HDF5:
- All image data are embedded in the workspace file
- No selection or exclusion is required
Typical workflow for workspace restoration:
1. Load workspace file using load_workspace() method
2. Workspace is deserialized and _restore_workspace() is called
3. Image windows are restored asynchronously:
- Images are loaded using a dedicated instance ImageLoaderService
- Each loaded image creates a new window via _on_workspace_image_ready()
- Failed image loads are recorded and emitted via image_load_failed signal
4. Once all images are processed, _on_workspace_image_complete() is called
5. Analysis tools are restored in phases:
- Tools are grouped by restore phase
- Phases are executed sequentially
- Within each phase, tools are restored concurrently
6. Tool items referencing missing images are skipped with a warning
7. Finally, UI workspace state is restored via _restore_ui_workspace()
"""
from __future__ import annotations
from collections import defaultdict
from copy import deepcopy
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, List, Optional
from uuid import UUID
import numpy as np
from PySide6.QtCore import QObject, QTimer, Signal
from radioviz.controllers.image_metadata_window_controller import ImageMetadataWindowController, build_metadata_source
from radioviz.controllers.image_window_controller import ImageWindowController
from radioviz.controllers.sub_window_controller import compare_flags
from radioviz.models.data_model import DataOrigin, DataSource, DataStorage
from radioviz.models.menu_spec import ToolMenuSpec
from radioviz.services.application_services import ApplicationServices
from radioviz.services.image_loader import ImageLoaderService
from radioviz.services.recent_files import RecentFilesManager
from radioviz.services.save_services import normalize_save_path
from radioviz.services.tiff_metadata import extract_xyz_dat_metadata
from radioviz.services.window_manager import WindowRequest
from radioviz.services.workspace_manager import (
WORKSPACE_SUFFIX_BY_SERIALIZER,
AsyncBarrier,
HDF5WorkspaceSerializer,
ImageWorkspaceSpec,
JSONWorkspaceSerializer,
SerializerType,
ToolWorkspace,
Workspace,
WorkspaceSerializable,
WorkspaceSerializableTool,
enforce_uuid,
)
from radioviz.tools.base_controller import ToolController
from radioviz.tools.base_dock import ToolDockWidget
from radioviz.tools.tool_api import ToolContext
from radioviz.typing.tool_types import ToolProtocol
if TYPE_CHECKING:
from radioviz.controllers.sub_window_controller import SubWindowController, T_SubWindow
[docs]
class MainController(QObject):
"""
Main controller for the RadioViz application.
This class serves as the central coordinator for the application,
managing windows, tools, image loading, and user interactions.
"""
#: Signal emitted when image loading fails
image_load_failed = Signal(Path, Exception)
#: Signal emitted when a new window view is requested
request_new_window_view = Signal(object) # it emits a WindowRequest
#: Signal emitted when JSON workspace save requires user selection
request_nonpersistent_image_selection = Signal(list)
#: Signal emitted when a XYZ DAT sidecar is missing
request_xyz_metadata = Signal(object)
#: Signal emitted when context menu is requested
context_menu_requested = Signal(object, object)
#: Signal emitted when tools are changed
tools_changed = Signal()
#: Signal emitted when a user-facing message is requested
message_requested = Signal(str, str, int)
def __init__(
self,
tool_list: Optional[list[ToolProtocol]] = None,
) -> None:
"""
Initialize the MainController.
:param tool_list: List of tools to register, defaults to None
:type tool_list: Optional[list[ToolProtocol]], optional
:param application_services: Application services instance, defaults to None
:type application_services: Optional[ApplicationServices], optional
"""
super().__init__()
self.application_services = ApplicationServices()
self.app_state = self.application_services.app_state
self.image_loader = self.application_services.image_loader
self.color_service = self.application_services.color_service
self.windows_manager = self.application_services.window_manager
self.window_factory = self.application_services.window_factory
self.overlay_factory = self.application_services.overlay_factory
self.recent_files = RecentFilesManager(max_files=5)
self.tool_context = ToolContext()
self.tool_register: list[ToolProtocol] = tool_list or [] # this is assigned by __init__
self.register_window_factory()
self.register_overlay_factory()
self.tool_controllers = self.register_tool_controllers() # this stores the global list of tool controllers
self.dock_register: dict[str, ToolDockWidget[Any]] = {}
self._pending_workspace_save: tuple[Path, SerializerType] | None = None
self._pending_xyz_metadata_overrides: dict[Path, dict[str, Any]] = {}
# Wire signals
self.image_loader.image_loaded.connect(self.create_new_image_view)
self.image_loader.image_load_failed.connect(self.image_load_failed)
[docs]
def register_window_factory(self) -> None:
"""
Register window factories for all registered tools.
This method iterates through all registered tools and registers
their window specifications with the window factory.
"""
for tool in self.tool_register:
for window_spec in tool.windows_to_be_registered:
self.window_factory.register(**window_spec.to_dict())
[docs]
def register_overlay_factory(self) -> None:
"""
Register overlay factories for all registered tools.
This method iterates through all registered tools and registers
their overlay specifications with the overlay factory.
"""
for tool in self.tool_register:
for overlay_spec in tool.overlays_to_be_registered:
self.overlay_factory.register(overlay_spec)
[docs]
def register_tool_controllers(self) -> list[ToolController[Any, Any]]:
"""
Register tool controllers for all registered tools.
:return: List of registered tool controllers
:rtype: list
"""
tool_controllers = []
self.tool_context.clear_tool_controllers()
for tool in self.tool_register:
tool_controller = tool.create_controller(self.tool_context)
tool_controllers.append(tool_controller)
self.tool_context.register_tool_controller(tool_controller)
tool_controller.request_new_window.connect(self._on_new_window_request)
return tool_controllers
[docs]
def set_tool_register(self, tool_list: list[ToolProtocol]) -> None:
"""
Set the tool register with a new list of tools.
:param tool_list: List of tools to register
:type tool_list: list[ToolProtocol]
"""
self.tool_register = tool_list
self.tool_controllers = self.register_tool_controllers()
[docs]
def tool_menu_specs(self) -> List[ToolMenuSpec]:
"""
Get menu specifications for all registered tools.
:return: List of tool menu specifications
:rtype: List[ToolMenuSpec]
"""
specs: List[ToolMenuSpec] = []
for tool in self.tool_controllers:
specs.extend(tool.menu_specs())
return specs
[docs]
def tool_menu_specs_for(
self, window_controller: Optional['SubWindowController[T_SubWindow]']
) -> List[ToolMenuSpec]:
"""
Get tool menu specifications filtered for the given window controller.
Tool menus are only included when the tool declares compatibility with the
active window type via :attr:`radioviz.tools.base_controller.ToolController.enable_for_window_type`.
:param window_controller: Active window controller or None
:type window_controller: Optional[SubWindowController]
:return: List of tool menu specifications
:rtype: List[ToolMenuSpec]
"""
if window_controller is None:
return []
active_type = window_controller.sub_window_type
specs: List[ToolMenuSpec] = []
for tool in self.tool_controllers:
if compare_flags(active_type, tool.enable_for_window_type):
specs.extend(tool.menu_specs())
return specs
[docs]
def context_menu_specs_for(
self, window_controller: Optional['SubWindowController[T_SubWindow]']
) -> List[ToolMenuSpec]:
"""
Get merged context menu specifications for the given window controller.
Combines tool-contributed menus with window-specific built-in menus.
Tool menus are filtered by window type, while window menus are always
included for the active window.
:param window_controller: Active window controller or None
:type window_controller: Optional[SubWindowController]
:return: List of menu specifications
:rtype: List[ToolMenuSpec]
"""
specs = self.tool_menu_specs_for(window_controller)
if window_controller is None:
return specs
return specs + window_controller.context_menu_specs()
[docs]
def sync_tool_menu_state(self, window_controller: Optional['SubWindowController[T_SubWindow]'] = None) -> None:
"""
Emit current tool menu states so on-demand menus reflect availability.
Only tools compatible with the active window type are synchronized.
"""
if window_controller:
for tool in self.tool_controllers:
tool.sync_menu_state_for(window_controller)
[docs]
def _on_new_window_request(self, request: WindowRequest) -> None:
"""
Handle new window requests.
:param request: Window request object
:type request: WindowRequest
"""
self.register_window(request.controller)
if request.requested_id:
request.requested_id = enforce_uuid(request.requested_id)
request.controller.id = request.requested_id
request.controller.context_menu_requested.connect(self.context_menu_requested)
self.request_new_window_view.emit(request)
[docs]
def open_image(self, path: Path) -> None:
"""
Open an image file.
High-level use case: open an image.
:param path: Path to the image file
:type path: Path
"""
path = Path(path)
if path.suffix.lower() == '.xyz':
dat_path = path.with_suffix('.dat')
if not dat_path.exists():
self.request_xyz_metadata.emit(
{
'xyz_path': path,
'expected_dat_path': dat_path,
}
)
return
self.image_loader.load(path)
[docs]
def open_xyz_with_dat(self, xyz_path: Path, dat_path: Path | None) -> None:
"""
Open a XYZ image, optionally using a DAT sidecar for metadata.
:param xyz_path: Path to the XYZ file.
:type xyz_path: Path
:param dat_path: Optional path to the DAT sidecar file.
:type dat_path: Path | None
"""
xyz_path = Path(xyz_path)
if dat_path is not None:
try:
metadata, _warnings = extract_xyz_dat_metadata(dat_path)
except Exception:
metadata = {}
self._pending_xyz_metadata_overrides[xyz_path] = metadata
self.image_loader.load(xyz_path)
[docs]
def create_new_image_view(self, path: Path, image_data: np.ndarray, metadata: dict[str, Any]) -> None:
"""
Create a new image view for the loaded image.
:param path: Path to the image file
:type path: Path
:param image_data: NumPy array containing image data
:type image_data: np.ndarray
:param metadata: Metadata associated with the image
:type metadata: dict[str, Any]
"""
metadata_override = self._pending_xyz_metadata_overrides.pop(path, None)
resolved_metadata = metadata_override if metadata_override is not None else metadata
if isinstance(resolved_metadata, dict) and '_warnings' in resolved_metadata:
warnings = resolved_metadata.pop('_warnings') or []
for warning in warnings:
self.message_requested.emit(str(warning), 'warning', 5000)
source = DataSource(label=path.name, origin=DataOrigin.FILE, parent=None)
storage = DataStorage(path=path)
image_controller = ImageWindowController(source, image_data, storage, metadata=resolved_metadata)
self.recent_files.add_file(path)
request = WindowRequest(window_type='image', controller=image_controller)
self._on_new_window_request(request)
[docs]
def show_image_metadata_window(self) -> None:
"""
Show a metadata window for the active image.
"""
active_window = self.app_state.active_window
if not isinstance(active_window, ImageWindowController):
return
title = f'Metadata - {active_window.name}'
source = build_metadata_source(label=title, parent=active_window.source)
metadata_controller = ImageMetadataWindowController(
source=source,
metadata=deepcopy(active_window.metadata),
title=title,
)
request = WindowRequest(window_type='image-metadata', controller=metadata_controller)
self._on_new_window_request(request)
[docs]
def register_global_dock(self, dock_id: str, dock: ToolDockWidget[Any]) -> None:
"""
Register a global dock widget.
:param dock_id: Identifier for the dock widget
:type dock_id: str
:param dock: Tool dock widget to register
:type dock: ToolDockWidget[Any]
"""
self.dock_register[dock_id] = dock
[docs]
def set_active_window(self, window_controller: 'SubWindowController[T_SubWindow]') -> None:
"""
Set the active window controller.
:param window_controller: Controller for the active window
:type window_controller: SubWindowController
"""
self.windows_manager.active_window = window_controller
[docs]
def register_window(self, window_controller: 'SubWindowController[T_SubWindow]') -> None:
"""
Register a window controller.
:param window_controller: Controller for the window to register
:type window_controller: SubWindowController
"""
if hasattr(window_controller, 'message_requested'):
window_controller.message_requested.connect(self.message_requested)
self.windows_manager.register(window_controller)
[docs]
def unregister_window(self, window_controller: 'SubWindowController[T_SubWindow]') -> None:
"""
Unregister a window controller.
:param window_controller: Controller for the window to unregister
:type window_controller: SubWindowController
"""
self.windows_manager.unregister(window_controller)
[docs]
def add_recent_file(self, path: Path) -> None:
"""
Add a file to recent files list.
:param path: Path to the file to add
:type path: Path
"""
self.recent_files.add_file(path)
[docs]
def open_recent(self, path: Path) -> None:
"""
Open a recently accessed file.
:param path: Path to the recent file
:type path: Path
"""
self.open_image(path)
[docs]
def clear_recent_files(self) -> None:
"""
Clear the recent files list.
"""
self.recent_files.clear_files()
[docs]
def get_recent_files(self) -> list[Path]:
"""
Get the list of recent files.
:return: List of recent file paths
:rtype: list[Path]
"""
return self.recent_files.get_files()
[docs]
def cancel_loading(self, path: Path) -> None:
"""
Cancel image loading for a specific path.
:param path: Path for which to cancel loading
:type path: Path
"""
self.image_loader.cancel(path)
[docs]
def save_workspace(self, path: Path, serializer_type: SerializerType) -> None:
"""
Save the current workspace to a file.
This method serializes the current application state and saves it
to the specified file using the given serializer type.
For JSON workspaces, if nonpersistent images are detected, the method
immediately emits a selection request and returns without writing the
workspace file. The UI will prompt the user to choose which images to
persist on disk. Once the selection completes, `on_nonpersistent_image_selection`
continues the save by persisting the selected images, excluding unchecked
images from the workspace, and pruning dependent tool items.
:param path: Path to the file where the workspace will be saved
:type path: Path
:param serializer_type: Type of serializer to use for saving
:type serializer_type: SerializerType
"""
path = normalize_save_path(path=path, selected_suffix=WORKSPACE_SUFFIX_BY_SERIALIZER[serializer_type])
if serializer_type == SerializerType.JSON:
nonpersistent = self._find_nonpersistent_images()
if nonpersistent:
dependency_counts = self._collect_tool_dependency_counts()
payload = []
for window in nonpersistent:
payload.append(
{
'image_id': window.id,
'name': window.name,
'dependency_counts': dependency_counts.get(window.id, {}),
}
)
self._pending_workspace_save = (path, serializer_type)
self.request_nonpersistent_image_selection.emit(payload)
return
ws: Workspace = self._create_workspace(serializer_type)
serializer = {SerializerType.JSON: JSONWorkspaceSerializer, SerializerType.HDF5: HDF5WorkspaceSerializer}[
serializer_type
]()
serializer.save(ws, path)
[docs]
def on_nonpersistent_image_selection(self, accepted: bool, excluded_ids: list[UUID]) -> None:
"""
Handle the result of the non‑persistent image selection dialog.
This method is called after the user has been prompted to choose which
non‑persistent images to keep when saving a JSON workspace. If the user
accepts the selection, a new workspace is created excluding the images
whose identifiers are provided in *excluded_ids* and the workspace is
saved using the serializer that was originally requested. If the user
cancels, the pending save operation is cleared.
:param accepted: ``True`` if the user confirmed the selection, ``False`` otherwise.
:type accepted: bool
:param excluded_ids: List of image UUIDs to exclude from the workspace.
:type excluded_ids: list[UUID]
"""
if not accepted or self._pending_workspace_save is None:
self._pending_workspace_save = None
return
path, serializer_type = self._pending_workspace_save
self._pending_workspace_save = None
excluded_set = set(excluded_ids)
ws: Workspace = self._create_workspace(serializer_type, excluded_image_ids=excluded_set)
serializer = {SerializerType.JSON: JSONWorkspaceSerializer, SerializerType.HDF5: HDF5WorkspaceSerializer}[
serializer_type
]()
serializer.save(ws, path)
[docs]
def load_workspace(self, path: Path, serializer_type: SerializerType) -> None:
"""
Load a workspace from a file.
This method reads and deserializes a workspace from the specified
file using the given serializer type, then restores the application
state from the loaded workspace.
:param path: Path to the file from which the workspace will be loaded
:type path: Path
:param serializer_type: Type of serializer to use for loading
:type serializer_type: SerializerType
"""
serializer = {SerializerType.JSON: JSONWorkspaceSerializer, SerializerType.HDF5: HDF5WorkspaceSerializer}[
serializer_type
]()
ws = serializer.load(path)
self._restore_workspace(ws)
[docs]
def _create_workspace(
self,
serializer_type: SerializerType,
excluded_image_ids: set[UUID] | None = None,
) -> Workspace:
"""
Create a workspace object from the current application state.
This internal method collects all serializable services from the
application services and creates a workspace object containing
their serialized state. If `excluded_image_ids` is provided, images
with those IDs are omitted and any tool items referencing those images
are pruned from the workspace.
:return: Workspace object containing the current application state
:rtype: Workspace
:param serializer_type: Type of serializer to use for saving
:type serializer_type: SerializerType
:param excluded_image_ids: Optional set of image IDs to exclude
:type excluded_image_ids: set[UUID] | None
"""
ws = Workspace(version='1.0')
include_data = serializer_type == SerializerType.HDF5
excluded_image_ids = excluded_image_ids or set()
for service_id, service in self.application_services.get_all_serializable_services().items():
ws.services[service_id] = service.to_workspace()
image_spec_dict = {}
for window in self.windows_manager.window_list:
if isinstance(window, WorkspaceSerializable):
if window.id in excluded_image_ids:
continue
image_spec_dict[window.id] = window.to_workspace(include_data=include_data)
ws.images = image_spec_dict
for tool in self.tool_register:
if isinstance(tool, WorkspaceSerializable) and hasattr(tool, 'tool_id'):
ws.tools[tool.tool_id] = tool.to_workspace(include_data=include_data)
if excluded_image_ids:
self._filter_tool_items_by_image_ids(
ws.tools,
should_keep=lambda image_id: image_id not in excluded_image_ids,
)
ws.ui = self.windows_manager.snapshot_ui()
return ws
[docs]
def _restore_workspace(self, workspace: Workspace) -> None:
"""
Restore the application state from a workspace object.
This internal method takes a workspace object and restores the
application state by deserializing each service and applying it
to the corresponding service in the application services. Images
with missing storage paths are skipped with a warning, and tool
items referencing missing images are pruned before tool restoration.
:param workspace: Workspace object containing the state to restore
:type workspace: Workspace
"""
self._workspace_being_restored = workspace
self.application_services.workspace_reference_manager.reset()
for service_id, saved_service in workspace.services.items():
service = self.application_services.get_service(service_id)
if service:
service.from_workspace(saved_service, self.application_services.workspace_reference_manager)
missing_image_ids = []
for image_id, image_spec in list(workspace.images.items()):
if image_spec.storage.path is None and image_spec.data is None:
missing_image_ids.append(image_id)
del workspace.images[image_id]
if missing_image_ids:
self.tool_context.show_message(
f'Skipped {len(missing_image_ids)} image(s) without storage paths.',
level='warning',
timeout_ms=5000,
)
# the reconstruction of the image windows is all async
self._workspace_image_loader = ImageLoaderService()
self._workspace_image_loader.image_loaded.connect(self._on_workspace_image_loaded)
self._workspace_image_loader.image_load_failed.connect(self._on_workspace_image_load_failed)
# a dictionary with path to specs
self._workspace_image_specs = {
spec.storage.path: spec
for spec in workspace.images.values()
if spec.storage.path is not None and spec.data is None
}
# let's make a set with all the pending loading images
self._workspace_pending_images: set[UUID] = set(workspace.images.keys())
self._workspace_failed_registry: dict[Path, Exception] = {}
if len(workspace.images) == 0:
# no images to restore.
# manually jump to the next step.
self._on_workspace_image_complete()
for image_id, image_spec in workspace.images.items():
if image_spec.data is not None:
QTimer.singleShot(0, lambda s=image_spec: self._on_workspace_image_ready(s, s.data, s.metadata))
elif image_spec.storage.path is not None:
self._workspace_image_loader.load(image_spec.storage.path)
# when finished with async image restoration, it goes ahead with _restore_analysis_tool
[docs]
def _on_workspace_image_loaded(self, path: Path, data: np.ndarray, metadata: dict[str, Any]) -> None:
"""
Handle successful loading of an image during workspace restoration.
This method retrieves the corresponding image specification using the
provided path and passes it along with the loaded data to the ready
handler for further processing.
:param path: Path to the loaded image file
:type path: Path
:param data: NumPy array containing the loaded image data
:type data: np.ndarray
:param metadata: Metadata associated with the image
:type metadata: dict[str, Any]
"""
# retrieve the spec using the path
spec = self._workspace_image_specs[path]
self._on_workspace_image_ready(spec, data, metadata)
[docs]
def _on_workspace_image_ready(
self, spec: ImageWorkspaceSpec, data: np.ndarray, metadata: dict[str, Any] | None = None
) -> None:
"""
Process an image that has been successfully loaded during workspace restoration.
This method creates an image controller using the provided specification
and image data, sets up the display properties, creates the corresponding
view, and registers it with the application. It also handles cleanup of
pending image loading operations.
:param spec: Specification for the image to be created
:type spec: ImageWorkspaceSpec
:param data: NumPy array containing the image data
:type data: np.ndarray
:param metadata: Optional metadata associated with the image
:type metadata: dict[str, Any] | None
"""
# create the image controller
resolved_metadata = metadata if metadata is not None else spec.metadata
if (
isinstance(resolved_metadata, dict)
and not resolved_metadata
and spec.storage.path is not None
and spec.storage.path.suffix.lower() == '.xyz'
and spec.metadata
):
resolved_metadata = spec.metadata
controller = ImageWindowController(
source=spec.source,
storage=spec.storage,
image_data=data,
metadata=resolved_metadata,
)
# restore other params
controller.id = enforce_uuid(spec.id)
controller.set_display_properties(spec.display_parameters)
# create the corresponding view
request = WindowRequest(window_type='image', controller=controller)
self._on_new_window_request(request)
# register the new controller in the context
self.application_services.workspace_reference_manager.register(controller.id, controller)
self._workspace_pending_images.remove(spec.id)
if not self._workspace_pending_images:
self._on_workspace_image_complete()
[docs]
def _on_workspace_image_load_failed(self, path: Path, error: Exception) -> None:
"""
Handle failed loading of an image during workspace restoration.
This method records the loading failure for the specified path and
removes the corresponding image from the pending loading operations.
If all images have been processed, it triggers the completion handler.
:param path: Path to the image file that failed to load
:type path: Path
:param error: Exception that occurred during image loading
:type error: Exception
"""
spec = self._workspace_image_specs[path]
self._workspace_failed_registry[path] = error
self._workspace_pending_images.remove(spec.id)
if not self._workspace_pending_images:
self._on_workspace_image_complete()
[docs]
def _on_workspace_image_complete(self) -> None:
"""
Handle completion of workspace image restoration.
This method processes any failed image loads and continues with
the restoration of analysis tools after all images have been
processed during workspace restoration.
If there were any failed image loads, they are emitted via the
image_load_failed signal. Then it proceeds to restore the analysis
tools from the workspace.
"""
if self._workspace_failed_registry:
for key, value in self._workspace_failed_registry.items():
self.image_load_failed.emit(key, value)
self._prune_tool_items_missing_images()
# continue here the workspace restoration
self._restore_analysis_tool()
[docs]
def _find_nonpersistent_images(self) -> list[ImageWindowController]:
"""
Find image windows that have no persistent storage path.
:return: List of nonpersistent image controllers
:rtype: list[ImageWindowController]
"""
nonpersistent: list[ImageWindowController] = []
for window in self.windows_manager.window_list:
if isinstance(window, ImageWindowController) and window.storage.path is None:
nonpersistent.append(window)
return nonpersistent
[docs]
def _collect_tool_dependency_counts(self) -> dict[UUID, dict[str, int]]:
"""
Collect counts of tool items per image.
This is used to inform users which tool items will be pruned if an
image is excluded from a JSON workspace.
:return: Mapping of image_id to tool_id counts
:rtype: dict[UUID, dict[str, int]]
"""
image_ids = [
window.id for window in self.windows_manager.window_list if isinstance(window, ImageWindowController)
]
return self.tool_context.collect_dependency_counts(image_ids)
[docs]
def _extract_image_controller_id(self, item: Any) -> UUID | None:
"""
Extract the image controller ID from a workspace item.
:param item: Workspace item spec (dict-like or object with attribute)
:type item: Any
:return: Image controller UUID or None if not present
:rtype: UUID | None
"""
image_id = None
if isinstance(item, dict):
image_id = item.get('image_controller_id')
elif hasattr(item, 'image_controller_id'):
image_id = getattr(item, 'image_controller_id')
if image_id is None:
return None
return enforce_uuid(image_id)
[docs]
def _clear_tool_selection_if_missing(self, tool_id: str, tool_ws: ToolWorkspace) -> None:
"""
Clear tool selection state if the selected item was removed.
:param tool_id: Tool identifier
:type tool_id: str
:param tool_ws: Tool workspace to update
:type tool_ws: ToolWorkspace
"""
if not tool_ws.state:
return
if tool_id == 'profile':
key = 'current_selected_profile'
elif tool_id == 'region':
key = 'current_selected_item'
else:
return
selected = tool_ws.state.get(key)
if selected in (None, '__NONE__'):
return
if TYPE_CHECKING:
assert isinstance(selected, (UUID, str))
if enforce_uuid(selected) not in tool_ws.items:
tool_ws.state[key] = None
[docs]
def _filter_tool_items_by_image_ids(
self,
tool_workspaces: dict[str, ToolWorkspace],
should_keep: Callable[[UUID], bool],
) -> dict[str, int]:
"""
Filter tool items based on the referenced image ID.
:param tool_workspaces: Tool workspace mapping to mutate
:type tool_workspaces: dict[str, ToolWorkspace]
:param should_keep: Predicate that returns True when the image_id is allowed
:type should_keep: Callable[[UUID], bool]
:return: Count of skipped items per tool_id
:rtype: dict[str, int]
"""
skipped_counts: dict[str, int] = defaultdict(int)
for tool_id, tool_ws in tool_workspaces.items():
if not tool_ws.items:
continue
new_items = {}
for item_id, item in tool_ws.items.items():
image_id = self._extract_image_controller_id(item)
if image_id is not None and not should_keep(image_id):
skipped_counts[tool_id] += 1
continue
new_items[item_id] = item
if len(new_items) != len(tool_ws.items):
tool_ws.items = new_items
allowed_ids = set(new_items.keys())
new_order = []
for item_id in tool_ws.order:
try:
uid = enforce_uuid(item_id)
except Exception:
continue
if uid in allowed_ids:
new_order.append(item_id)
tool_ws.order = new_order
self._clear_tool_selection_if_missing(tool_id, tool_ws)
return skipped_counts
[docs]
def _prune_tool_items_missing_images(self) -> None:
"""
Prune tool items referencing images that were not restored.
This prevents dangling tool items from causing errors during
restoration when images are missing or skipped.
:return: None
:rtype: None
"""
available_image_ids = {
window.id for window in self.windows_manager.window_list if isinstance(window, ImageWindowController)
}
skipped = self._filter_tool_items_by_image_ids(
self._workspace_being_restored.tools,
should_keep=lambda image_id: image_id in available_image_ids,
)
if any(skipped.values()):
total = sum(skipped.values())
self.tool_context.show_message(
f'Skipped {total} tool item(s) referencing missing images.',
level='warning',
timeout_ms=5000,
)
[docs]
def _restore_analysis_tool(self) -> None:
"""
Restore analysis tools from the workspace.
This method processes the tools stored in the workspace by grouping them
according to their restore phase, sorting the phases, and then restoring
tools in each phase sequentially. It manages the restoration process
through phase-based execution to ensure proper initialization order.
The method initializes the restoration process by:
1. Grouping tools by their restore phase
2. Sorting phases numerically
3. Starting the first phase of restoration
"""
tool_ws = self._workspace_being_restored.tools
self._tool_barriers: list[AsyncBarrier] = []
tools_by_phase: dict[int, list[WorkspaceSerializableTool[Any]]] = defaultdict(list)
# 1. group tools by phase
for tool in self.tool_register:
if tool.tool_id in tool_ws and isinstance(tool, WorkspaceSerializableTool):
phase = tool.restore_phase()
tools_by_phase[phase].append(tool)
# 2. sort phases
self._restore_phases = sorted(tools_by_phase.items())
self._current_phase_index = 0
# 3. start first phase
self._restore_next_phase()
[docs]
def _restore_next_phase(self) -> None:
"""
Restore the next phase of tools in the workspace restoration process.
This method handles the sequential restoration of tools by:
1. Checking if all phases have been processed
2. Creating an asynchronous barrier for the current phase
3. Adding all tools in the current phase to the barrier
4. Connecting tool restoration completion to the barrier
5. Starting the restoration of tools in the current phase
The method ensures that tools within the same phase are restored
concurrently while maintaining proper sequencing between phases.
"""
if self._current_phase_index >= len(self._restore_phases):
self._on_all_tools_restored()
return
phase, tools = self._restore_phases[self._current_phase_index]
self._current_phase_index += 1
barrier = AsyncBarrier(parent=self)
self._current_tool_barrier = barrier
# add all tools in this phase to the barrier
for tool in tools:
barrier.add(tool)
if TYPE_CHECKING:
assert hasattr(tool._controller, 'restore_completed')
tool._controller.restore_completed.connect(barrier.done)
barrier.completed.connect(self._restore_next_phase)
barrier.completed.connect(barrier.deleteLater)
# start restoring tools in this phase
for tool in tools:
tool.from_workspace(
spec=self._workspace_being_restored.tools[tool.tool_id],
context=self.application_services.workspace_reference_manager,
)
[docs]
def _on_all_tools_restored(self) -> None:
"""
Handle completion of workspace tool restoration.
This internal method is called when all tools have been successfully
restored from the workspace. It proceeds to restore the user interface
workspace state.
"""
self._restore_ui_workspace()
[docs]
def _restore_ui_workspace(self) -> None:
"""
Restore the user interface workspace state.
This internal method restores the application's user interface layout
and state from the workspace, including window positions, sizes, and
dock widget configurations.
"""
if self._workspace_being_restored.ui:
self.windows_manager.restore_ui_workspace(self._workspace_being_restored.ui)