# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Module for managing workspace serialization and deserialization.
This module provides functionality for serializing and deserializing workspace
states, including image specifications, tool workspaces, and service workspaces.
It supports both JSON and HDF5 formats for workspace persistence.
The core components include:
- :class:`Workspace`: The neutral representation of an analysis workspace
- :class:`WorkspaceSerializer`: Abstract base class for workspace serializers
- :class:`JSONWorkspaceSerializer`: JSON-based workspace serializer
- :class:`HDF5WorkspaceSerializer`: HDF5-based workspace serializer
- :class:`WorkspaceReferenceManager`: Runtime context for workspace reconstruction
"""
from __future__ import annotations
import json
import sys
from abc import ABC, abstractmethod
from copy import deepcopy
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Protocol, TypeVar, cast, runtime_checkable
if sys.version_info >= (3, 11):
from enum import StrEnum
else:
from radioviz.models.legacy import StrEnum
from pathlib import Path
from uuid import UUID
import h5py
import numpy as np
from PySide6.QtCore import QObject, Signal
from radioviz.models.data_model import DataSource, DataStorage, SpatialDerivation
from radioviz.models.roi_model import ROI, ROIEnclosure, ROIType
from radioviz.models.transform_model import Transform, TransformType
SpecT = TypeVar('SpecT')
if TYPE_CHECKING:
from radioviz.tools.base_controller import ToolController
[docs]
@runtime_checkable
class WorkspaceSpec(Protocol):
"""Marker protocol for all workspace specs.
This protocol serves as a marker interface for workspace specification objects
that can be serialized and deserialized within the workspace management system.
"""
id: UUID
[docs]
@runtime_checkable
class WorkspaceSerializable(Protocol[SpecT]):
"""Protocol for objects that can be serialized to workspace specifications.
Objects implementing this protocol can be converted to and from workspace
specifications, enabling persistence and reconstruction of workspace state.
:param SpecT: Type parameter representing the workspace specification type
"""
id: UUID
[docs]
def to_workspace(self, *args: Any, **kwargs: Any) -> SpecT:
"""Convert this object to a workspace specification.
:param args: Additional positional arguments
:param kwargs: Additional keyword arguments
:return: Workspace specification representation
:rtype: SpecT
"""
...
[docs]
@classmethod
def from_workspace(cls, spec: SpecT, context: 'WorkspaceReferenceManager') -> object:
"""Create an instance from a workspace specification.
:param spec: Workspace specification to reconstruct from
:type spec: SpecT
:param context: Runtime context for workspace reconstruction
:type context: WorkspaceReferenceManager
:return: Reconstructed object instance
:rtype: object
"""
...
[docs]
class WorkspaceReferenceManager:
"""
Runtime context used during workspace reconstruction.
This manager keeps track of objects during workspace reconstruction to
resolve references between different workspace components.
:ivar _objects: Dictionary mapping UUIDs to registered objects
:vartype _objects: dict[UUID, object]
"""
def __init__(self) -> None:
"""Initialize the workspace reference manager."""
self._objects: dict[UUID, object] = {}
[docs]
def register(self, obj_id: UUID | str, obj: object) -> None:
"""Register an object with a given ID.
:param obj_id: Unique identifier for the object (UUID or string)
:type obj_id: UUID | str
:param obj: Object to register
:type obj: object
"""
self._objects[enforce_uuid(obj_id)] = obj
[docs]
def resolve(self, obj_id: UUID | str) -> object:
"""Resolve an object by its ID.
:param obj_id: Unique identifier for the object (UUID or string)
:type obj_id: UUID | str
:return: Registered object
:rtype: object
:raise KeyError: if the object ID is not found
"""
try:
return self._objects[enforce_uuid(obj_id)]
except KeyError:
raise KeyError(f'Unresolved workspace reference: {obj_id}')
[docs]
def reset(self) -> None:
"""Clear all registered objects."""
self._objects.clear()
[docs]
@dataclass
class ImageWorkspaceSpec:
"""Specification for image workspace components.
This data class represents the specification of an image within a workspace,
including its data source, storage, display parameters, and metadata.
:ivar id: Unique identifier for the image specification
:vartype id: UUID
:ivar source: Data source information for the image
:vartype source: DataSource
:ivar storage: Data storage information for the image
:vartype storage: DataStorage
:ivar display_parameters: Display parameters for the image
:vartype display_parameters: dict[str, Any]
:ivar metadata: Metadata associated with the image
:vartype metadata: dict[str, Any]
:ivar data: Image data array (optional)
:vartype data: Optional[np.ndarray]
"""
id: UUID
source: DataSource
storage: DataStorage
display_parameters: dict[str, Any]
metadata: dict[str, Any]
data: Optional[np.ndarray] = None
[docs]
@dataclass
class ServiceWorkspace:
"""Workspace state for a service component.
This data class represents the workspace state for a specific service,
including its identifier, version, and state information.
:ivar service_id: Identifier for the service
:vartype service_id: str
:ivar version: Version of the service (optional)
:vartype version: str | None
:ivar state: State information for the service
:vartype state: dict[str, Any]
"""
service_id: str
version: str | None = None
state: dict[str, Any] = field(default_factory=dict)
[docs]
def workspace_spec_as_mapping(spec: WorkspaceSpec) -> Mapping[str, Any]:
"""
Normalize a workspace specification to a mapping.
"""
if isinstance(spec, Mapping):
return spec
return vars(spec)
[docs]
class WindowState(StrEnum):
"""Enumeration of window states.
This enum defines the possible states a window can be in within the UI workspace.
These states control how windows are displayed and managed in the application interface.
:cvar Normal: Window is in normal size and position
:vartype Normal: str
:cvar Minimized: Window is minimized to the taskbar/system tray
:vartype Minimized: str
:cvar Maximized: Window is maximized to fill the screen
:vartype Maximized: str
:cvar FullScreen: Window is in full-screen mode
:vartype FullScreen: str
"""
Normal = 'Normal'
Minimized = 'Minimized'
Maximized = 'Maximized'
FullScreen = 'FullScreen'
[docs]
@dataclass
class UIWindowState:
"""State information for a single UI window.
This data class represents the state of a specific UI window within the workspace,
including its unique identifier, current state, and z-order for layering.
:ivar window_id: Unique identifier for the window
:vartype window_id: UUID
:ivar state: Current state of the window (normal, minimized, maximized, etc.)
:vartype state: WindowState
:ivar z_order: Z-order value for window layering (higher values appear on top)
:vartype z_order: int
"""
window_id: UUID
state: WindowState # normal, minimized, maximized
z_order: int
[docs]
@dataclass
class UIWorkspace:
"""Workspace state for UI configuration.
This data class represents the complete UI workspace state, including
information about the active window and all managed windows.
:ivar active_window_id: Unique identifier of the currently active window (optional)
:vartype active_window_id: Optional[UUID]
:ivar windows: List of window state configurations
:vartype windows: list[UIWindowState]
"""
active_window_id: Optional[UUID] = None
windows: list[UIWindowState] = field(default_factory=list)
[docs]
@dataclass
class Workspace:
"""Neutral representation of an analysis workspace.
This data class represents the complete workspace state, including images,
tools, services, and UI configuration.
:ivar version: Version of the workspace format
:vartype version: str
:ivar images: Dictionary mapping UUIDs to image workspace specifications
:vartype images: dict[UUID, ImageWorkspaceSpec]
:ivar tools: Dictionary mapping tool IDs to tool workspace states
:vartype tools: dict[str, ToolWorkspace]
:ivar services: Dictionary mapping service IDs to service workspace states
:vartype services: dict[str, ServiceWorkspace]
:ivar ui: UI configuration (optional)
:vartype ui: dict[str, Any] | None
"""
version: str
images: dict[UUID, ImageWorkspaceSpec] = field(default_factory=dict)
tools: dict[str, ToolWorkspace] = field(default_factory=dict)
services: dict[str, ServiceWorkspace] = field(default_factory=dict)
ui: Optional[UIWorkspace] = None
[docs]
class SerializerType(StrEnum):
"""Enumeration of supported serializer types.
This enum defines the available serialization formats for workspace data.
"""
JSON = 'json'
HDF5 = 'hdf5'
WORKSPACE_SUFFIX_BY_SERIALIZER: dict[SerializerType, str] = {
SerializerType.JSON: 'json',
SerializerType.HDF5: 'h5',
}
[docs]
class WorkspaceSerializer(ABC):
"""Abstract base class for workspace serializers.
This abstract base class defines the interface for workspace serializers
that can save and load workspace state to/from various formats.
:cvar serializer_type: The type of serializer (JSON or HDF5)
:vartype serializer_type: SerializerType
"""
[docs]
@abstractmethod
def save(self, workspace: Workspace, path: Path) -> None:
"""Save workspace to a file.
:param workspace: Workspace to save
:type workspace: Workspace
:param path: File path to save to
:type path: Path
"""
...
[docs]
@abstractmethod
def load(self, path: Path) -> Workspace:
"""Load workspace from a file.
:param path: File path to load from
:type path: Path
:return: Loaded workspace
:rtype: Workspace
"""
...
[docs]
class RadiovizEncoder(json.JSONEncoder):
"""Custom JSON encoder for Radioviz data types.
This encoder handles serialization of custom data types like NumPy arrays,
UUIDs, and other non-standard JSON types.
"""
[docs]
def default(self, obj: Any) -> Any:
"""Encode objects to JSON-serializable format.
:param obj: Object to encode
:return: JSON-serializable representation
"""
if isinstance(obj, np.integer):
return int(obj)
if isinstance(obj, np.floating):
return float(obj)
if isinstance(obj, np.ndarray):
return obj.tolist()
if isinstance(obj, UUID):
return str(obj)
return super().default(obj)
[docs]
class JSONWorkspaceSerializer(WorkspaceSerializer):
"""Light workspace serializer (JSON).
This serializer implements workspace serialization using JSON format,
suitable for lightweight workspace persistence.
:cvar FORMAT_VERSION: Format version string
:vartype FORMAT_VERSION: str
:cvar serializer_type: Serializer type (JSON)
:vartype serializer_type: SerializerType
"""
FORMAT_VERSION = '1.0'
serializer_type = SerializerType.JSON
[docs]
def save(self, workspace: Workspace, path: Path) -> None:
"""Save workspace to JSON file.
:param workspace: Workspace to save
:type workspace: Workspace
:param path: File path to save to
:type path: Path
"""
payload = {
'format': 'workspace-light',
'version': self.FORMAT_VERSION,
'workspace': self._encode_workspace(workspace),
}
path.write_text(json.dumps(payload, indent=2, cls=RadiovizEncoder))
[docs]
def load(self, path: Path) -> Workspace:
"""Load workspace from JSON file.
:param path: File path to load from
:type path: Path
:return: Loaded workspace
:rtype: Workspace
:raise ValueError: if the file is not a light workspace file
"""
payload = json.loads(path.read_text())
self._validate(payload)
return self._decode_workspace(payload['workspace'])
# -------------------------
# internals
# -------------------------
[docs]
def _encode_workspace(self, ws: Workspace) -> dict[str, Any]:
"""Encode workspace to dictionary format.
:param ws: Workspace to encode
:type ws: Workspace
:return: Encoded workspace dictionary
:rtype: dict
"""
return {
'version': ws.version,
'images': self._encode_image_spec(ws.images),
'tools': self._encode_tools(ws.tools),
'services': self._encode_services(ws.services),
'ui': self._encode_ui(ws.ui or UIWorkspace()),
}
[docs]
def _decode_workspace(self, data: dict[str, Any]) -> Workspace:
"""Decode workspace from dictionary format.
:param data: Workspace data dictionary
:type data: dict
:return: Decoded workspace
:rtype: Workspace
"""
return Workspace(
version=data['version'],
images=self._decode_image_specs(data['images']),
tools=self._decode_tools(data['tools']),
services=self._decode_services(data['services']),
ui=self._decode_ui(data.get('ui')),
)
[docs]
def _encode_data_source(self, data_source: DataSource) -> dict[str, Any]:
"""Encode data source to dictionary format.
:param data_source: Data source to encode
:type data_source: DataSource
:return: Encoded data source dictionary
:rtype: dict
"""
return {
'label': data_source.label,
'origin': str(data_source.origin),
'parent': self._encode_data_source(data_source.parent) if data_source.parent else None,
'derivation': self._encode_derivation(data_source.derivation) if data_source.derivation else None,
}
[docs]
def _encode_derivation(self, derivation: SpatialDerivation) -> dict[str, Any]:
"""Encode spatial derivation to dictionary format.
This internal method converts a spatial derivation object into a dictionary
representation suitable for serialization. It handles encoding of parent ID,
region of interest, and transformation information.
:param derivation: Spatial derivation to encode
:type derivation: SpatialDerivation
:return: Encoded derivation dictionary
:rtype: dict
"""
if derivation:
return {
'parent_id': str(derivation.parent_id),
'roi': self._encode_roi(derivation.roi),
'transform': self._encode_transform(derivation.transform),
}
else:
return {}
[docs]
def _decode_derivation(self, d: dict[str, Any]) -> SpatialDerivation:
"""Decode spatial derivation from dictionary format.
This internal method reconstructs a spatial derivation object from a dictionary
representation. It handles decoding of parent ID, region of interest, and
transformation information.
:param d: Derivation data dictionary
:type d: dict[str, Any]
:return: Decoded spatial derivation
:rtype: SpatialDerivation
:raise ValueError: when the ROI payload is missing
"""
roi_payload = d.get('roi')
if roi_payload is None:
raise ValueError('Missing ROI payload for spatial derivation.')
return SpatialDerivation(
parent_id=UUID(d['parent_id']),
roi=self._decode_roi(roi_payload),
transform=self._decode_transform(d.get('transform')),
)
[docs]
def _encode_roi(self, roi: ROI) -> dict[str, Any]:
"""Encode region of interest to dictionary format.
This internal method converts a region of interest object into a dictionary
representation suitable for serialization. It handles encoding of ROI type,
enclosure type, and geometric data.
:param roi: Region of interest to encode
:type roi: ROI
:return: Encoded ROI dictionary
:rtype: dict
"""
return {
'type': str(roi.type),
'enclosure': str(roi.enclosure),
'geometry': roi.geometry,
}
[docs]
def _decode_roi(self, r: dict[str, Any]) -> ROI:
"""Decode region of interest from dictionary format.
This internal method reconstructs a region of interest object from a dictionary
representation. It handles decoding of ROI type, enclosure type, and geometric data.
:param r: ROI data dictionary
:type r: dict[str, Any]
:return: Decoded region of interest
:rtype: ROI
"""
return ROI(
type=ROIType(r['type']),
enclosure=ROIEnclosure(r['enclosure']),
geometry=r['geometry'],
)
[docs]
def _decode_data_source(self, data: dict[str, Any]) -> DataSource:
"""Decode data source from dictionary format.
:param data: Data source data dictionary
:type data: dict[str, Any]
:return: Decoded data source
:rtype: DataSource
"""
ds = DataSource(
label=data['label'],
origin=data['origin'],
parent=self._decode_data_source(data['parent']) if data['parent'] else None,
derivation=self._decode_derivation(data['derivation']) if data.get('derivation') else None,
)
return ds
[docs]
def _encode_storage(self, storage: DataStorage) -> dict[str, Any]:
"""Encode data storage to dictionary format.
:param storage: Data storage to encode
:type storage: DataStorage
:return: Encoded data storage dictionary
:rtype: dict
"""
return {
'path': str(storage.path) if storage.path is not None else None,
}
[docs]
def _decode_storage(self, data: dict[str, Any]) -> DataStorage:
"""Decode data storage from dictionary format.
:param data: Data storage data dictionary
:type data: dict
:return: Decoded data storage
:rtype: DataStorage
"""
raw_path = data.get('path')
if raw_path in (None, '', 'None'):
return DataStorage(path=None)
return DataStorage(path=Path(str(raw_path)))
[docs]
def _encode_single_image_spec(self, spec: ImageWorkspaceSpec) -> dict[str, Any]:
"""Encode single image specification to dictionary format.
:param spec: Image workspace specification to encode
:type spec: ImageWorkspaceSpec
:return: Encoded image specification dictionary
:rtype: dict
"""
return {
'id': str(spec.id),
'source': self._encode_data_source(spec.source),
'storage': self._encode_storage(spec.storage),
'display_parameters': self._encode_display_paramters(spec.display_parameters),
'metadata': spec.metadata,
}
[docs]
def _encode_image_spec(self, spec_dict: dict[UUID, ImageWorkspaceSpec]) -> dict[str, Any]:
"""Encode image specifications dictionary to dictionary format.
:param spec_dict: Dictionary of image workspace specifications
:type spec_dict: dict[UUID, ImageWorkspaceSpec]
:return: Encoded image specifications dictionary
:rtype: dict
"""
result = {}
for key, spec in spec_dict.items():
result[str(key)] = self._encode_single_image_spec(spec)
return result
[docs]
def _decode_image_specs(self, data: dict[str, Any]) -> dict[UUID, ImageWorkspaceSpec]:
"""Decode image specifications from dictionary format.
:param data: Image specifications data dictionary
:type data: dict
:return: Decoded image specifications dictionary
:rtype: dict[UUID, ImageWorkspaceSpec]
"""
result = {}
for key, value in data.items():
result[enforce_uuid(key)] = self._decode_single_image_spec(value)
return result
[docs]
def _decode_single_image_spec(self, data: dict[str, Any]) -> ImageWorkspaceSpec:
"""Decode single image specification from dictionary format.
:param data: Single image specification data dictionary
:type data: dict
:return: Decoded image workspace specification
:rtype: ImageWorkspaceSpec
"""
return ImageWorkspaceSpec(
id=enforce_uuid(data['id']),
source=self._decode_data_source(data['source']),
storage=self._decode_storage(data['storage']),
display_parameters=data['display_parameters'],
metadata=data['metadata'],
)
[docs]
def _encode_services(self, services: dict[str, ServiceWorkspace]) -> dict[str, Any]:
"""Encode services to dictionary format.
:param services: Dictionary of service workspaces
:type services: dict[str, ServiceWorkspace]
:return: Encoded services dictionary
:rtype: dict
"""
encoded = {}
for service_id, service_ws in services.items():
encoded[service_id] = {
'service_id': service_ws.service_id,
'version': service_ws.version,
'state': service_ws.state,
}
return encoded
[docs]
def _encode_display_paramters(self, params: dict[str, Any]) -> dict[str, Any]:
"""Encode display parameters to dictionary format.
:param params: Display parameters dictionary
:type params: dict
:return: Encoded display parameters dictionary
:rtype: dict
"""
result = deepcopy(params)
if not isinstance(result['cmap'], str):
# it is a cmap, convert it
result['cmap'] = result['cmap'].name
return result
[docs]
def _encode_specs(self, specs: dict[UUID, WorkspaceSpec]) -> dict[str, Any]:
"""Encode workspace specifications to dictionary format.
:param specs: Dictionary of workspace specifications
:type specs: dict[UUID, WorkspaceSpec]
:return: Encoded specifications dictionary
:rtype: dict
"""
return {str(k): vars(v) for k, v in specs.items()}
[docs]
def _decode_specs(self, raw: dict[str, Any]) -> dict[UUID, WorkspaceSpec]:
"""Decode workspace specifications from dictionary format.
:param raw: Raw specifications dictionary
:type raw: dict
:return: Decoded specifications dictionary
:rtype: dict[UUID, WorkspaceSpec]
"""
return {UUID(k): v for k, v in raw.items()}
[docs]
def _decode_services(self, data: dict[str, dict[str, Any]]) -> dict[str, ServiceWorkspace]:
"""Decode services from dictionary format.
:param data: Services data dictionary
:type data: dict[str, dict]
:return: Decoded services dictionary
:rtype: dict[str, ServiceWorkspace]
"""
return {
sid: ServiceWorkspace(
service_id=sid,
version=payload.get('version'),
state=payload.get('state', {}),
)
for sid, payload in data.items()
}
[docs]
def _encode_ui(self, ui_ws: UIWorkspace) -> dict[str, Any]:
"""Encode UI workspace state to dictionary format.
This internal method converts the UI workspace state into a dictionary
representation suitable for serialization. It handles encoding of
active window ID and all managed windows with their state information.
:param ui_ws: UI workspace state to encode
:type ui_ws: UIWorkspace
:return: Encoded UI workspace dictionary
:rtype: dict
"""
return {
'active_window_id': str(ui_ws.active_window_id) if ui_ws.active_window_id else None,
'windows': [
{
'window_id': str(w.window_id),
'state': str(w.state),
'z_order': w.z_order,
}
for w in ui_ws.windows
],
}
[docs]
def _decode_ui(self, data: dict[str, Any] | None) -> UIWorkspace:
"""Decode UI workspace state from dictionary format.
This internal method reconstructs the UI workspace state from a dictionary
representation. It handles decoding of active window ID and all managed
windows with their respective state information.
:param data: UI workspace data dictionary
:type data: dict[str, Any] | None
:return: Decoded UI workspace state
:rtype: UIWorkspace
"""
if data is None:
return UIWorkspace()
ui = UIWorkspace(
active_window_id=enforce_uuid(data['active_window_id'])
if data.get('active_window_id') is not None
else None
)
for w in data.get('windows', []):
ui.windows.append(
UIWindowState(
window_id=enforce_uuid(w['window_id']),
state=WindowState(w['state']),
z_order=int(w['z_order']),
)
)
return ui
[docs]
def _validate(self, payload: dict[str, Any]) -> None:
"""Validate workspace payload format.
:param payload: Workspace payload to validate
:type payload: dict
:raise ValueError: if the payload is not a light workspace file
"""
if payload.get('format') != 'workspace-light':
raise ValueError('Not a light workspace file')
[docs]
class HDF5WorkspaceSerializer(WorkspaceSerializer):
"""Full workspace serializer (HDF5).
This serializer implements workspace serialization using HDF5 format,
suitable for storing large workspace data efficiently.
:cvar FORMAT_VERSION: Format version string
:vartype FORMAT_VERSION: str
:cvar serializer_type: Serializer type (HDF5)
:vartype serializer_type: SerializerType
"""
FORMAT_VERSION = '1.0'
serializer_type = SerializerType.HDF5
[docs]
def save(self, workspace: Workspace, path: Path) -> None:
"""Save workspace to HDF5 file.
:param workspace: Workspace to save
:type workspace: Workspace
:param path: File path to save to
:type path: Path
"""
with h5py.File(path, 'w') as h5:
h5.attrs['format'] = 'workspace-full'
h5.attrs['version'] = self.FORMAT_VERSION
h5.attrs['workspace_version'] = workspace.version
self._encode_image_specs(h5, 'images', workspace.images)
self._encode_tools(h5, workspace.tools)
self._encode_services(h5, workspace.services)
self._encode_ui(h5, workspace.ui or UIWorkspace())
[docs]
def load(self, path: Path) -> Workspace:
"""Load workspace from HDF5 file.
:param path: File path to load from
:type path: Path
:return: Loaded workspace
:rtype: Workspace
"""
with h5py.File(path, 'r') as h5:
self._validate(h5)
return Workspace(
version=h5.attrs['workspace_version'],
images=self._decode_image_specs(h5['images']),
tools=self._decode_tools(h5),
services=self._decode_services(h5),
ui=self._decode_ui(h5),
)
# -------------------------
# internals
# -------------------------
[docs]
def _save_specs(self, h5: h5py.File, name: str, specs: dict[UUID, WorkspaceSpec]) -> None:
"""Save workspace specifications to HDF5 file.
:param h5: HDF5 file handle
:type h5: h5py.File
:param name: Name of the group to create
:type name: str
:param specs: Dictionary of workspace specifications
:type specs: dict[UUID, WorkspaceSpec]
"""
grp = h5.create_group(name)
for uid, spec in specs.items():
obj_grp = grp.create_group(str(uid))
for k, v in vars(spec).items():
self._encode_value(obj_grp, k, v)
[docs]
def _decode_specs(self, grp: h5py.Group) -> dict[UUID, WorkspaceSpec]:
"""Decode workspace specifications from HDF5 group.
:param grp: HDF5 group containing specifications
:type grp: h5py.Group
:return: Decoded specifications dictionary
:rtype: dict[UUID, WorkspaceSpec]
"""
result: dict[UUID, WorkspaceSpec] = {}
for uid_str, obj_grp in grp.items():
result[enforce_uuid(uid_str)] = cast(WorkspaceSpec, cast(object, dict(obj_grp.attrs)))
return result
[docs]
def _decode_items(self, grp: h5py.Group) -> dict[UUID, Any]:
"""Main entry point to decode the workspace."""
result: dict[UUID, Any] = {}
for uid_str, obj_grp in grp.items():
# Root level keys are UUIDs
decoded_data = self._decode_value(obj_grp)
result[enforce_uuid(uid_str)] = decoded_data
return result
[docs]
def _decode_value(self, item: Any) -> Any:
"""
Recursive helper to decode HDF5 objects (Groups, Datasets, or Attributes)
back into Python/NumPy types.
"""
# CASE 1: The item is a Group -> Treat as a dictionary
if isinstance(item, h5py.Group):
# Start with the attributes (metadata)
decoded_dict = {}
for k, v in item.attrs.items():
decoded_dict[k] = self._process_attribute(v)
# Add members (Datasets or Subgroups)
for k, v in item.items():
decoded_dict[k] = self._decode_value(v)
return decoded_dict
# CASE 2: The item is a Dataset -> Return as NumPy array
elif isinstance(item, h5py.Dataset):
return item[:]
return item
[docs]
def _process_attribute(self, val: Any) -> Any:
"""Handles specific type conversions for attributes."""
if isinstance(val, str) and val == '__NONE__':
return None
if isinstance(val, (np.bool_, bool)):
return bool(val)
if isinstance(val, np.ndarray):
# convert the inner values if they are numpy type
list_val = []
for v in val:
if hasattr(v, 'item'):
list_val.append(v.item())
else:
list_val.append(v)
return list_val
# NumPy scalars (int64, float64) to Python native
if hasattr(val, 'item'):
return val.item()
return val
[docs]
def _encode_value(self, grp: h5py.Group, key: str, value: Any) -> None:
"""Encode a value to HDF5 group.
:param grp: HDF5 group to encode to
:type grp: h5py.Group
:param key: Key for the value
:type key: str
:param value: Value to encode
"""
if isinstance(value, dict):
subgroup = grp.create_group(key)
for sub_key, sub_value in value.items():
# Recursive call: each dict key becomes a member of the new subgroup
self._encode_value(subgroup, str(sub_key), sub_value)
elif isinstance(value, np.ndarray):
grp.create_dataset(key, data=value)
else:
if value is None:
grp.attrs[key] = '__NONE__'
else:
if isinstance(value, UUID):
grp.attrs[key] = str(value)
elif isinstance(value, (bool, np.bool_)):
grp.attrs[key] = bool(value)
elif isinstance(value, StrEnum):
grp.attrs[key] = str(value)
else:
grp.attrs[key] = value
[docs]
def _encode_attr(self, grp: h5py.Group, key: str, value: Any) -> None:
"""Encode attribute to HDF5 group.
:param grp: HDF5 group to encode to
:type grp: h5py.Group
:param key: Key for the attribute
:type key: str
:param value: Value to encode
"""
if value is None:
grp.attrs[key] = '__NONE__'
else:
grp.attrs[key] = value
[docs]
def _decode_attr(self, grp: h5py.Group, key: str) -> Any:
"""Decode attribute from HDF5 group.
:param grp: HDF5 group to decode from
:type grp: h5py.Group
:param key: Key for the attribute
:type key: str
:return: Decoded attribute value
:rtype: Any
"""
value = grp.attrs[key]
return None if value == '__NONE__' else value
[docs]
def _decode_group(self, grp: h5py.Group) -> dict[str, Any]:
"""Decode entire HDF5 group to dictionary.
:param grp: HDF5 group to decode
:type grp: h5py.Group
:return: Decoded dictionary
:rtype: dict[str, Any]
"""
return cast(Dict[str, Any], self._decode_value(grp))
[docs]
def _encode_data_source(self, grp: h5py.Group, ds: DataSource) -> None:
"""Encode data source to HDF5 group.
:param grp: HDF5 group to encode to
:type grp: h5py.Group
:param ds: Data source to encode
:type ds: DataSource
"""
grp.attrs['label'] = ds.label
grp.attrs['origin'] = str(ds.origin)
if ds.parent:
parent_grp = grp.create_group('parent')
self._encode_data_source(parent_grp, ds.parent)
if ds.derivation:
deriv_grp = grp.create_group('derivation')
self._encode_derivation(deriv_grp, ds.derivation)
[docs]
def _decode_data_source(self, grp: h5py.Group) -> DataSource:
"""Decode data source from HDF5 group.
:param grp: HDF5 group containing data source
:type grp: h5py.Group
:return: Decoded data source
:rtype: DataSource
"""
parent = None
if 'parent' in grp:
parent = self._decode_data_source(grp['parent'])
derivation = None
if 'derivation' in grp:
derivation = self._decode_derivation(grp['derivation'])
return DataSource(label=grp.attrs['label'], origin=grp.attrs['origin'], parent=parent, derivation=derivation)
[docs]
def _encode_derivation(self, grp: h5py.Group, derivation: SpatialDerivation) -> None:
"""Encode spatial derivation to HDF5 group.
This internal method converts a spatial derivation object into an HDF5 group
structure suitable for persistent storage. It handles encoding of parent ID,
region of interest, and transformation information.
:param grp: HDF5 group to encode to
:type grp: h5py.Group
:param derivation: Spatial derivation to encode
:type derivation: SpatialDerivation
"""
grp.attrs['parent_id'] = str(derivation.parent_id)
roi_grp = grp.create_group('roi')
self._encode_roi(roi_grp, derivation.roi)
if derivation.transform:
transform_grp = roi_grp.create_group('transform')
self._encode_transform(transform_grp, derivation.transform)
[docs]
def _encode_roi(self, grp: h5py.Group, roi: ROI) -> None:
"""Encode region of interest to HDF5 group.
This internal method converts a region of interest object into an HDF5 group
structure suitable for persistent storage. It handles encoding of ROI type,
enclosure type, and geometric data.
:param grp: HDF5 group to encode to
:type grp: h5py.Group
:param roi: Region of interest to encode
:type roi: ROI
"""
grp.attrs['type'] = str(roi.type)
grp.attrs['enclosure'] = str(roi.enclosure)
self._encode_value(grp, 'geometry', roi.geometry)
[docs]
def _decode_derivation(self, grp: h5py.Group) -> SpatialDerivation:
"""Decode spatial derivation from HDF5 group.
This internal method reconstructs a spatial derivation object from an HDF5 group
structure. It handles decoding of parent ID, region of interest, and
transformation information.
:param grp: HDF5 group containing derivation data
:type grp: h5py.Group
:return: Decoded spatial derivation
:rtype: SpatialDerivation
"""
return SpatialDerivation(
parent_id=UUID(grp.attrs['parent_id']),
roi=self._decode_roi(grp['roi']),
transform=self._decode_transform(grp['transform']) if 'transform' in grp else None,
)
[docs]
def _decode_roi(self, grp: h5py.Group) -> ROI:
"""Decode region of interest from HDF5 group.
This internal method reconstructs a region of interest object from an HDF5 group
structure. It handles decoding of ROI type, enclosure type, and geometric data.
:param grp: HDF5 group containing ROI data
:type grp: h5py.Group
:return: Decoded region of interest
:rtype: ROI
"""
return ROI(
type=ROIType(grp.attrs['type']),
enclosure=ROIEnclosure(grp.attrs['enclosure']),
geometry=self._decode_value(grp['geometry']),
)
[docs]
def _encode_data_storage(self, grp: h5py.Group, ds: DataStorage) -> None:
"""Encode data storage to HDF5 group.
:param grp: HDF5 group to encode to
:type grp: h5py.Group
:param ds: Data storage to encode
:type ds: DataStorage
"""
self._encode_attr(grp, 'path', str(ds.path) if ds.path else None)
[docs]
def _decode_data_storage(self, grp: h5py.Group) -> DataStorage:
"""Decode data storage from HDF5 group.
:param grp: HDF5 group containing data storage
:type grp: h5py.Group
:return: Decoded data storage
:rtype: DataStorage
"""
path = self._decode_attr(grp, 'path')
return DataStorage(path=Path(path) if path else None)
[docs]
def _encode_display_parameters(self, grp: h5py.Group, params: dict[str, Any]) -> None:
"""Encode display parameters to HDF5 group.
:param grp: HDF5 group to encode to
:type grp: h5py.Group
:param params: Display parameters to encode
:type params: dict
"""
for key, value in params.items():
if key == 'cmap' and not isinstance(value, str):
self._encode_attr(grp, key, value.name)
elif isinstance(value, (dict, list, tuple)):
self._encode_attr(grp, key, json.dumps(value))
else:
self._encode_attr(grp, key, value)
[docs]
def _encode_image_specs(self, h5: h5py.File, name: str, specs: dict[UUID, ImageWorkspaceSpec]) -> None:
"""Encode image specifications to HDF5 file.
:param h5: HDF5 file handle
:type h5: h5py.File
:param name: Name of the group to create
:type name: str
:param specs: Dictionary of image workspace specifications
:type specs: dict[UUID, ImageWorkspaceSpec]
"""
for key, value in specs.items():
self._encode_single_image_spec(h5, value)
[docs]
def _encode_single_image_spec(self, h5: h5py.File, spec: ImageWorkspaceSpec) -> None:
"""Encode single image specification to HDF5 file.
:param h5: HDF5 file handle
:type h5: h5py.File
:param spec: Image workspace specification to encode
:type spec: ImageWorkspaceSpec
"""
grp = h5.create_group(f'images/{spec.id}')
grp.attrs['id'] = str(spec.id)
# source
src_grp = grp.create_group('source')
self._encode_data_source(src_grp, spec.source)
# storage
st_grp = grp.create_group('storage')
self._encode_data_storage(st_grp, spec.storage)
# display parameters
disp_grp = grp.create_group('display_parameters')
self._encode_display_parameters(disp_grp, spec.display_parameters)
# metadata
meta_grp = grp.create_group('metadata')
for k, v in spec.metadata.items():
self._encode_value(meta_grp, str(k), v)
grp.create_dataset('data', data=spec.data)
[docs]
def _decode_image_specs(self, grp: h5py.Group) -> dict[UUID, ImageWorkspaceSpec]:
"""Decode image specifications from HDF5 group.
:param grp: HDF5 group containing image specifications
:type grp: h5py.Group
:return: Decoded image specifications dictionary
:rtype: dict[UUID, ImageWorkspaceSpec]
"""
results = {}
for key, value in grp.items():
results[enforce_uuid(key)] = self._decode_single_image_spec(value)
return results
[docs]
def _decode_single_image_spec(self, grp: h5py.Group) -> ImageWorkspaceSpec:
"""Decode single image specification from HDF5 group.
:param grp: HDF5 group containing single image specification
:type grp: h5py.Group
:return: Decoded image workspace specification
:rtype: ImageWorkspaceSpec
"""
return ImageWorkspaceSpec(
id=enforce_uuid(grp.attrs['id']),
source=self._decode_data_source(grp['source']),
storage=self._decode_data_storage(grp['storage']),
display_parameters=self._decode_display_parameters(grp['display_parameters'].attrs),
metadata=self._decode_value(grp['metadata']),
data=grp['data'][:],
)
[docs]
def _decode_display_parameters(self, attrs: h5py.AttributeManager) -> dict[str, Any]:
"""
Decode display parameters from HDF5 attributes.
:param attrs: HDF5 attributes mapping.
:type attrs: h5py.AttributeManager
:return: Display parameters dictionary.
:rtype: dict
"""
params = dict(attrs)
for key in ('scalebar',):
value = params.get(key)
if value is None:
continue
if isinstance(value, bytes):
try:
value = value.decode('utf-8')
except Exception:
continue
if isinstance(value, str):
try:
params[key] = json.loads(value)
except Exception:
params[key] = value
return params
[docs]
def _encode_order(self, grp: h5py.Group, key: str, value: list[UUID]) -> None:
"""Encode execution order to HDF5 group.
:param grp: HDF5 group to encode to
:type grp: h5py.Group
:param key: Key for the order
:type key: str
:param value: Order list to encode
:type value: list[UUID]
"""
str_value = [str(id_) for id_ in value]
self._encode_attr(grp, key, str_value)
[docs]
def _encode_services(self, h5: h5py.File, services: dict[str, ServiceWorkspace]) -> None:
"""Encode services to HDF5 file.
:param h5: HDF5 file handle
:type h5: h5py.File
:param services: Dictionary of service workspaces
:type services: dict[str, ServiceWorkspace]
"""
services_grp = h5.create_group('services')
for service_id, svc_ws in services.items():
svc_grp = services_grp.create_group(service_id)
if svc_ws.version is not None:
svc_grp.attrs['version'] = svc_ws.version
state_grp = svc_grp.create_group('state')
for k, v in svc_ws.state.items():
self._encode_value(state_grp, k, v)
[docs]
def _decode_services(self, h5: h5py.File) -> dict[str, ServiceWorkspace]:
"""Decode services from HDF5 file.
:param h5: HDF5 file handle
:type h5: h5py.File
:return: Decoded services dictionary
:rtype: dict[str, ServiceWorkspace]
"""
result: dict[str, ServiceWorkspace] = {}
if 'services' not in h5:
return result
services_grp = h5['services']
for service_id, svc_grp in services_grp.items():
version = svc_grp.attrs.get('version')
state = {}
if 'state' in svc_grp:
state = self._decode_group(svc_grp['state'])
result[service_id] = ServiceWorkspace(
service_id=service_id,
version=version,
state=state,
)
return result
[docs]
def _encode_ui(self, h5: h5py.File, ui: UIWorkspace) -> None:
"""Encode UI workspace state to HDF5 group.
This internal method converts the UI workspace state into an HDF5 group
structure suitable for persistent storage. It handles encoding of
active window ID and all managed windows with their state information.
:param h5: HDF5 file handle
:type h5: h5py.File
:param ui: UI workspace state to encode
:type ui: UIWorkspace
"""
ui_grp = h5.create_group('ui')
if ui.active_window_id is not None:
ui_grp.attrs['active_window_id'] = str(ui.active_window_id)
windows_grp = ui_grp.create_group('windows')
for w in ui.windows:
w_grp = windows_grp.create_group(str(w.window_id))
w_grp.attrs['state'] = str(w.state)
w_grp.attrs['z_order'] = w.z_order
[docs]
def _decode_ui(self, h5: h5py.File) -> UIWorkspace:
"""Decode UI workspace state from HDF5 group.
This internal method reconstructs the UI workspace state from an HDF5 group
structure. It handles decoding of active window ID and all managed windows
with their respective state information.
:param h5: HDF5 file handle
:type h5: h5py.File
:return: Decoded UI workspace state
:rtype: UIWorkspace
"""
if 'ui' not in h5:
return UIWorkspace()
ui_grp = h5['ui']
active_window_id = None
if 'active_window_id' in ui_grp.attrs:
active_window_id = enforce_uuid(ui_grp.attrs['active_window_id'])
ui = UIWorkspace(active_window_id=active_window_id)
if 'windows' in ui_grp:
windows_grp = ui_grp['windows']
for window_id_str, w_grp in windows_grp.items():
ui.windows.append(
UIWindowState(
window_id=enforce_uuid(window_id_str),
state=WindowState(w_grp.attrs['state']),
z_order=int(w_grp.attrs['z_order']),
)
)
return ui
[docs]
def _validate(self, h5: h5py.File) -> None:
"""Validate HDF5 workspace format.
:param h5: HDF5 file handle
:type h5: h5py.File
:raise ValueError: if the file is not a full workspace file
"""
if h5.attrs.get('format') != 'workspace-full':
raise ValueError('Not a full workspace file')
[docs]
class AsyncBarrier(QObject):
"""A synchronization primitive for managing asynchronous operations.
This class provides a mechanism to track multiple asynchronous operations
and emit a signal when all operations are completed. It is designed to work
with Qt's signal-slot mechanism for coordinating asynchronous tasks.
:ivar completed: Signal emitted when all pending operations are finished
:vartype completed: Signal
"""
completed = Signal()
def __init__(self, parent: Optional[QObject] = None) -> None:
"""Initialize the async barrier.
:param parent: Parent QObject for ownership management
:type parent: QObject | None
"""
super().__init__(parent=parent)
self._pending: set[object] = set()
self._closed = False
[docs]
def add(self, token: object) -> None:
"""Add a token to the barrier.
:param token: Token representing an asynchronous operation
:type token: object
:raise RuntimeError: if the barrier is already closed
"""
if self._closed:
raise RuntimeError('AsyncBarrier already closed')
self._pending.add(token)
[docs]
def done(self, token: object) -> None:
"""Mark an operation as completed.
:param token: Token representing the completed operation
:type token: object
"""
self._pending.remove(token)
if not self._pending:
self._closed = True
self.completed.emit()
[docs]
def is_completed(self) -> bool:
"""Check if all operations have completed.
:return: True if all operations are completed, False otherwise
:rtype: bool
"""
return not self._pending
[docs]
def enforce_uuid(value: UUID | str) -> UUID:
"""Convert a UUID or string representation to a UUID object.
This utility function ensures that a given value is returned as a UUID object,
converting from string representation if necessary.
:param value: UUID object or string representation of a UUID
:type value: UUID | str
:return: UUID object
:rtype: UUID
"""
if isinstance(value, str):
return UUID(value)
return value