# -*- coding: utf-8 -*-
"""
Created the 23/11/2022
@author: Sebastien Weber
"""
from __future__ import annotations
from typing import Union, List, Tuple, TYPE_CHECKING, Iterable
import time
import xml.etree.ElementTree as ET
import numpy as np
from pymodaq_utils.logger import set_logger, get_module_name
from pymodaq_utils.enums import BaseEnum
from pymodaq_utils.abstract import ABCMeta, abstract_attribute, abstractmethod
from pymodaq_utils.utils import capitalize
from pymodaq_data import DataDim, DataWithAxes
from pymodaq_data.data import Axis, DataToExport, DataDistribution, DataRaw, DataSource
from pymodaq_data.h5modules.data_saving import (
DataToExportSaver, DataToExportEnlargeableSaver,
DataToExportTimedSaver, DataToExportExtendedSaver, DataBundle)
from pymodaq_data.h5modules.backends import GROUP, Node
from pymodaq_gui.h5modules.saving import H5SaverBase
from pymodaq_gui.parameter import ioxml
if TYPE_CHECKING:
from pymodaq.extensions.scan.daq_scan import DAQScan
from pymodaq.control_modules.daq_viewer import DAQ_Viewer
from pymodaq.control_modules.daq_move import DAQ_Move
from pymodaq.utils.custom_ext import CustomExt
logger = set_logger(get_module_name(__file__))
class GroupModuleType(BaseEnum):
DETECTOR = 0
ACTUATOR = 1
SCAN = 2
DATALOGGER = 3
OPTIMIZER = 4
TIME = 5
[docs]
class ModuleSaver(metaclass=ABCMeta):
"""Abstract base class to save info and data from main modules (DAQScan, DAQViewer, DAQMove, ...)"""
group_type: GroupModuleType = abstract_attribute()
_module = abstract_attribute()
_h5saver: H5SaverBase = abstract_attribute()
_module_group: GROUP = abstract_attribute()
main_module = True
[docs]
def flush(self):
"""Flush the underlying file"""
self._h5saver.flush()
[docs]
def get_set_node(self, where: Union[Node, str] = None, name: str = None) -> GROUP:
"""Get or create the node corresponding to this particular Module instance
Parameters
----------
where: Union[Node, str]
the path of a given node or the node itself
new: bool
if True force the creation of a new indexed node of this class type
if False return the last node (or create one if None)
Returns
-------
GROUP: the Node associated with this module which should be a GROUP node
"""
if where is None:
where = self._h5saver.raw_group
if name is None:
name = self._module.title
group = self._h5saver.get_node_from_title(where, name)
if group is not None:
self._module_group = group
return group # if I got one I return it else I create one
self._module_group = self._add_module(where)
return self._module_group
[docs]
def get_last_node(self, where: Union[Node, str] = None):
"""Get the last node corresponding to this particular Module instance
Parameters
----------
where: Union[Node, str]
the path of a given node or the node itself
Returns
-------
GROUP: the Node associated with this module which should be a GROUP node
"""
if where is None:
where = self._h5saver.raw_group
group = self._h5saver.get_last_group(where, self.group_type)
self._module_group = group
return self._module_group
@abstractmethod
def _add_module(self, where: Union[Node, str] = None, metadata={}):
""" Add here the Node and metadata describing the Module """
raise NotImplementedError
@property
def module(self):
return self._module
@property
def module_group(self):
return self._module_group
@property
def h5saver(self):
return self._h5saver
@h5saver.setter
def h5saver(self, _h5saver: H5SaverBase):
self._h5saver = _h5saver
self.update_after_h5changed()
[docs]
@abstractmethod
def update_after_h5changed(self):
""" Propagate the h5saver to eventual child modules """
raise NotImplementedError
def get_last_node_index(self, where: Union[Node, str] = None):
node = self.get_last_node(where)
return int(node.name.split(capitalize(self.group_type.name.lower()))[1])
def get_next_node_name(self, where: Union[Node, str] = None):
index = self.get_last_node_index(where)
return f'{capitalize(self.group_type.name.lower())}{index+1:03d}'
[docs]
class DetectorSaver(ModuleSaver):
"""Implementation of the ModuleSaver class dedicated to DAQ_Viewer modules
Parameters
----------
module
"""
group_type = GroupModuleType.DETECTOR
def __init__(self, module: DAQ_Viewer):
self._datatoexport_saver: DataToExportSaver = None
self._module: 'DAQ_Viewer' = module
self._module_group: GROUP = None
self._h5saver = None
def update_after_h5changed(self):
self._datatoexport_saver = DataToExportSaver(self.h5saver)
def _add_module(self, where: Union[Node, str] = None, metadata={}) -> Node:
"""
Parameters
----------
where: Union[Node, str]
the path of a given node or the node itself
metadata: dict
Returns
-------
"""
if where is None:
where = self._h5saver.raw_group
settings_xml = ET.Element('All_settings', type='group')
settings_xml.append(ioxml.walk_parameters_to_xml(param=self._module.settings))
if self.main_module:
saver_xml = ET.SubElement(settings_xml, 'H5Saver', type='group')
saver_xml.append(ioxml.walk_parameters_to_xml(param=self._h5saver.settings))
if self._module.ui is not None:
for ind, viewer in enumerate(self._module.viewers):
if hasattr(viewer, 'roi_manager') and viewer.roi_manager is not None:
roi_xml = ET.SubElement(settings_xml, f'ROI_Viewer_{ind:02d}', type='group')
roi_xml.append(ioxml.walk_parameters_to_xml(param=viewer.roi_manager.settings))
return self._h5saver.add_det_group(where, title=self._module.title, settings_as_xml=ET.tostring(settings_xml),
metadata=metadata)
def add_data(self, where: Union[Node, str],
data: DataToExport,
*args,
**kwargs):
self._datatoexport_saver.add_data(where, self.filter_data(data), **kwargs)
[docs]
def filter_data(self, dte: DataToExport) -> DataToExport:
""" Filter Data to be saved depending on first the presence of the extra_attribute: *do_save* and then on
the H5Saver settings
"""
dte_filtered = DataToExport('Filtered')
for dwa in dte:
if 'do_save' in dwa.extra_attributes: # this is the main filter
if dwa.do_save:
dte_filtered.append(dwa)
elif self.filter_data_wrt_settings(dwa):
dte_filtered.append(dwa)
return dte_filtered
[docs]
def filter_data_wrt_settings(self, dwa: DataWithAxes) -> bool:
""" Check if this DataWithAxes should be saved depending on the H5Saver settings
Return True if it should be saved
"""
flag = True
if not self.h5saver.settings['save_2D']: # exclude 2D data and above
flag = flag and not (dwa.dim == DataDim.Data2D or dwa.dim == DataDim.DataND)
if self.h5saver.settings['save_raw_only']: # exclude Calculated data
flag = flag and dwa.source == DataSource.raw
return flag
[docs]
def add_bkg(self, where: Union[Node, str], data_bkg: DataToExport):
""" Adds a DataToExport as a background node in the h5file
Parameters
----------
where: Union[Node, str]
the path of a given node or the node itself
data_bkg: DataToExport
The data to be saved as background
Returns
-------
"""
self._datatoexport_saver.add_bkg(where, data_bkg)
def add_external_h5(self, other_h5data: H5SaverBase):
if other_h5data is not None:
external_group = self._h5saver.add_group('external_data', 'external_h5', self.module_group)
try:
if not other_h5data.isopen:
h5saver = H5SaverBase()
h5saver.init_file(addhoc_file_path=other_h5data.filename)
h5_file = h5saver.h5_file
else:
h5_file = other_h5data
h5_file.copy_children(h5_file.get_node('/'), external_group, recursive=True)
h5_file.flush()
h5_file.close()
except Exception as e:
self.logger.exception(str(e))
class DetectorTimeSaver(DetectorSaver):
"""Implementation of the ModuleSaver class dedicated to DAQ_Viewer modules in order to save enlargeable data
Parameters
----------
module
"""
group_type = GroupModuleType.DETECTOR
def __init__(self, module: DAQ_Viewer):
super().__init__(module)
self._datatoexport_saver: DataToExportTimedSaver = None
def update_after_h5changed(self):
self._datatoexport_saver = DataToExportTimedSaver(self.h5saver)
[docs]
class DetectorEnlargeableSaver(DetectorSaver):
"""Implementation of the ModuleSaver class dedicated to DAQ_Viewer modules in order to save enlargeable data
Parameters
----------
module
"""
group_type = GroupModuleType.DETECTOR
def __init__(self, module: DAQ_Viewer,
enl_axis_names: Iterable[str] = None,
enl_axis_units: Iterable[str] = None):
super().__init__(module)
self.enl_axis_names = enl_axis_names
self.enl_axis_units = enl_axis_units
self._datatoexport_saver: DataToExportEnlargeableSaver = None
def update_after_h5changed(self):
self._datatoexport_saver = DataToExportEnlargeableSaver(
self.h5saver, self.enl_axis_names, self.enl_axis_units)
def add_data(self, where: Union[Node, str], data: DataToExport,
axis_values: list[float | np.ndarray] = None,
**kwargs):
self._datatoexport_saver.add_data(where,
self.filter_data(data),
axis_values,
**kwargs)
[docs]
class DetectorExtendedSaver(DetectorSaver):
"""Implementation of the ModuleSaver class dedicated to DAQ_Viewer modules in order to save enlargeable data
Parameters
----------
module
"""
group_type = GroupModuleType.DETECTOR
def __init__(self, module: DAQ_Viewer, extended_shape: Iterable[int]):
super().__init__(module)
self._extended_shape = extended_shape
self._datatoexport_saver: DataToExportExtendedSaver = None
def update_after_h5changed(self):
self._datatoexport_saver = DataToExportExtendedSaver(self.h5saver,
self._extended_shape)
def add_data(self,
where: Union[Node, str],
data: DataToExport,
indexes: Iterable[int],
distribution=DataDistribution.uniform):
self._datatoexport_saver.add_data(where,
self.filter_data(data),
indexes,
distribution=distribution)
def add_nav_axes(self, where: Union[Node, str], axes: list[Axis]):
self._datatoexport_saver.add_nav_axes(where, axes)
[docs]
class ActuatorSaver(ModuleSaver):
"""Implementation of the ModuleSaver class dedicated to DAQ_Move modules
Parameters
----------
h5saver
module
"""
group_type = GroupModuleType.ACTUATOR
def __init__(self, module: DAQ_Move):
self._datatoexport_saver: DataToExportSaver = None
self._module_group: GROUP = None
self._module: DAQ_Move = module
self._h5saver = None
def update_after_h5changed(self):
self._datatoexport_saver = DataToExportSaver(self.h5saver)
def _add_module(self, where: Union[Node, str] = None, metadata=None):
if metadata is None:
metadata={}
if where is None:
where = self._h5saver.raw_group
settings_xml = ET.Element('All_settings')
settings_xml.append(ioxml.walk_parameters_to_xml(param=self._module.settings))
return self._h5saver.add_act_group(where, title=self._module.title, settings_as_xml=ET.tostring(settings_xml),
metadata=metadata)
def add_data(self, where: Union[Node, str], data: DataToExport):
self._datatoexport_saver.add_data(where, data)
class ActuatorTimeSaver(ActuatorSaver):
def __init__(self, module: DAQ_Move):
super().__init__(module)
self._datatoexport_saver: DataToExportTimedSaver = None
def update_after_h5changed(self):
self._datatoexport_saver = DataToExportTimedSaver(self.h5saver)
def add_data(self, where: Union[Node, str], data: DataToExport):
self._datatoexport_saver.add_data(where, data)
class ActuatorEnlargeableSaver(ActuatorTimeSaver):
"""Implementation of the ModuleSaver class dedicated to DAQ_Move modules
Parameters
----------
h5saver
module
"""
def __init__(self, module: DAQ_Move,
enl_axis_names: Iterable[str] = None,
enl_axis_units: Iterable[str] = None):
super().__init__(module)
self.enl_axis_names = enl_axis_names
self.enl_axis_units = enl_axis_units
self._datatoexport_saver: DataToExportEnlargeableSaver = None
def update_after_h5changed(self):
self._datatoexport_saver = DataToExportEnlargeableSaver(
self.h5saver, self.enl_axis_names, self.enl_axis_units)
def add_data(self, where: Union[Node, str], data: DataToExport,
axis_values: List[Union[float, np.ndarray]] = None,
**kwargs):
self._datatoexport_saver.add_data(where, data, axis_values, **kwargs)
class TimeModule:
"""Minimal module-like object used by TimeModuleSaver."""
title = 'Timestamps'
class TimeModuleSaver(ModuleSaver):
"""Saves per-scan-point elapsed time in a proper HDF5 group.
Creates a 'Timestamps/' group inside the scan node, with a Data00 CARRAY
of shape = scan_shape written via DataToExportExtendedSaver, so naming
conventions and attributes are handled identically to detector data.
Nav axes are NOT repeated here — they already live in each detector's
NavAxes group and share the same index convention.
"""
group_type = GroupModuleType.TIME
def __init__(self):
self._module = TimeModule()
self._module_group = None
self._h5saver = None
self._extended_shape: Tuple[int] = None
self._datatoexport_saver: DataToExportExtendedSaver = None
self._start_time: float = None
def update_after_h5changed(self):
if self._extended_shape is not None:
self._datatoexport_saver = DataToExportExtendedSaver(
self._h5saver, self._extended_shape, fill_value=np.nan)
def _add_module(self, where=None, metadata=None) -> Node:
if metadata is None:
metadata = {}
if where is None:
where = self._h5saver.raw_group
# Fixed group name 'Timestamps' — no incremental index
group = self._h5saver.get_set_group(where, 'Timestamps', title='Timestamps')
self._h5saver.set_attr(group, 'type', 'time')
self._h5saver.set_attr(
group, 'description',
'Elapsed time (seconds) between scan start and when each scan point '
'was saved, measured with a monotonic clock (time.perf_counter). '
'Not an absolute/epoch timestamp.')
settings_xml = ET.Element('All_settings', type='group')
self._h5saver.set_attr(group, 'settings', ET.tostring(settings_xml))
return group
def initialize(self, extended_shape: Iterable[int]):
"""Set up the extended saver and start the internal clock."""
self._extended_shape = extended_shape
self._start_time = time.perf_counter()
self._datatoexport_saver = DataToExportExtendedSaver(
self._h5saver, extended_shape, fill_value=np.nan)
def add_time(self, indexes: Iterable[int]):
"""Record elapsed seconds since initialize() was called at the given scan indexes."""
if self._datatoexport_saver is not None:
elapsed_time = float(np.float32(time.perf_counter() - self._start_time))
dte = DataToExport('Timestamps', data=[
DataRaw('ElapsedTime',
data=[np.array([elapsed_time], dtype=np.float32)],
labels=['Elapsed time since scan start'],
units='s'),
])
self._datatoexport_saver.add_data(
self._module_group, dte, indexes=indexes,
distribution=DataDistribution['uniform'])
class ExtensionSaver(ModuleSaver):
""" Implementation focusing on CustomExt that should save data and metadata from
Control Modules
To be reimplemented !!!
"""
group_type: GroupModuleType = abstract_attribute()
_module: 'CustomExt' = abstract_attribute()
def __init__(self, module):
self._module_group: GROUP = None
self._module: DAQScan = module
self._h5saver = None
self.detectors : dict[str, DetectorSaver |
DetectorExtendedSaver |
DetectorEnlargeableSaver |
DetectorTimeSaver ] = {}
self.actuators : dict[str, ActuatorSaver |
ActuatorEnlargeableSaver |
ActuatorTimeSaver ] = {}
self.current_nodes: dict[str, Node] = {}
self._time_saver: TimeModuleSaver = None
def update_after_h5changed(self):
""" To be updated depending on the actual Saver you want to use"""
for module in self._module.modules_manager.detectors_all:
self.detectors[module.title] = DetectorSaver(module)
self.detectors[module.title].h5saver = self.h5saver
for module in self._module.modules_manager.actuators_all:
self.actuators[module.title] = ActuatorSaver(module)
self.actuators[module.title].h5saver = self.h5saver
def create_module_group(self, where: str | Node = None):
if where is None:
where = self.get_last_node()
for det_name in self.detectors:
self.current_nodes[det_name] = self.detectors[det_name].get_set_node(where)
for act_name in self.actuators:
self.current_nodes[act_name] = self.actuators[act_name].get_set_node(where)
def get_set_node(self, where: Union[Node, str] = None, new=False) -> GROUP:
"""Get the last group scan node
Get the last Scan Group or create one
get the last Scan Group if:
* there is one already created
* new is False
Parameters
----------
where: Union[Node, str]
the path of a given node or the node itself
new: bool
Returns
-------
GROUP: the GROUP associated with this module
"""
self._module_group = self.get_last_node(where)
new = new or (self._module_group is None)
if new:
self._module_group = self._add_module(where)
self.create_module_group(self._module_group)
if self._time_saver is not None:
self._time_saver.get_set_node(self._module_group)
return self._module_group
def initialize_time_array(self, extended_shape: Iterable[int]):
self._time_saver.initialize(extended_shape)
def add_time(self, indexes: Iterable[int]):
self._time_saver.add_time(indexes)
def _add_module(self, where: Union[Node, str] = None, metadata=None) -> Node:
"""
Parameters
----------
where: Union[Node, str]
the path of a given node or the node itself
metadata: dict
Returns
-------
"""
if metadata is None:
metadata = {}
if where is None:
where = self._h5saver.raw_group
settings_xml = ET.Element('All_settings', type='group')
settings_xml.append(ioxml.walk_parameters_to_xml(param=self._module.settings))
if self.main_module:
saver_xml = ET.SubElement(settings_xml, 'H5Saver', type='group')
saver_xml.append(ioxml.walk_parameters_to_xml(param=self._h5saver.settings))
return self._h5saver.add_generic_group(where, title=self._module.title,
settings_as_xml=ET.tostring(settings_xml),
metadata=metadata,
group_type=self.group_type.name)
def add_data(self, dte: DataToExport, **kwargs):
""" To be reimplemented with the right module saver signature"""
raise NotImplementedError
def add_data_bundle(self, data: DataBundle):
""" To be reimplemented with the right module saver signature"""
raise NotImplementedError
[docs]
class ScanSaver(ExtensionSaver):
"""Implementation of the ModuleSaver class dedicated to DAQScan module
Parameters
----------
h5saver
module
"""
group_type = GroupModuleType.SCAN
def __init__(self, module):
self._module: DAQScan = module
super().__init__(module)
self.detectors : dict[str, DetectorExtendedSaver] = {}
self.actuators : dict[str, ActuatorSaver] = {}
self._scan_shape: Iterable[int] = ()
self._time_saver = TimeModuleSaver()
def set_scan_shape(self, scan_shape: Iterable[int]):
self._scan_shape = scan_shape
[docs]
def update_after_h5changed(self):
""" To be updated depending on the actual Saver you want to use"""
for module in self._module.modules_manager.detectors_all:
self.detectors[module.title] = DetectorExtendedSaver(module, self._scan_shape)
self.detectors[module.title].h5saver = self.h5saver
for module in self._module.modules_manager.actuators_all:
self.actuators[module.title] = ActuatorSaver(module)
self.actuators[module.title].h5saver = self.h5saver
self._time_saver.h5saver = self.h5saver
def add_nav_axes(self, axes: List[Axis]):
for det_name in self.detectors:
self.detectors[det_name].add_nav_axes(self._module_group, axes)
def add_data(self, dte: DataToExport, indexes: Iterable[int] = None,
distribution=DataDistribution.uniform, **kwargs):
for origin in dte.get_origins():
self.detectors[origin].add_data(self.current_nodes[origin],
dte.get_data_from_origins([origin]),
indexes=indexes,
distribution=distribution,
)
self.add_time(indexes)
def add_data_bundle(self, data: DataBundle):
self.add_data(data.dte, indexes=data.indexes, distribution=data.distribution)
[docs]
class LoggerSaver(ExtensionSaver):
"""Implementation of the ModuleSaver class dedicated to Logger module
H5Logger is the special logger to h5file of the DAQ_Logger extension
Parameters
----------
h5saver
module
"""
group_type = GroupModuleType.DATALOGGER
def __init__(self, module):
super().__init__(module)
self.detectors : dict[str, DetectorTimeSaver] = {}
self.actuators : dict[str, ActuatorTimeSaver ] = {}
def update_after_h5changed(self):
for module in self._module.modules_manager.detectors_all:
self.detectors[module.title] = DetectorTimeSaver(module)
self.detectors[module.title].h5saver = self.h5saver
for module in self._module.modules_manager.actuators_all:
self.actuators[module.title] = ActuatorTimeSaver(module)
self.actuators[module.title].h5saver = self.h5saver
[docs]
def add_data(self, dte: DataToExport, **kwargs):
"""Add data to it's corresponding control module
The name of the control module is the DataToExport name attribute
"""
if dte.name in self.detectors:
self.detectors[dte.name].add_data(self.current_nodes[dte.name],
dte,)
elif dte.name in self.actuators:
self.actuators[dte.name].add_data(self.current_nodes[dte.name],
dte, )
else: # try to see if this dte is composed of multiple DataActuators or dwa
# from multiple detectors
for dwa in dte:
if dwa.origin in list(self.actuators.keys()) + list(self.detectors.keys()):
dte_from_dwa = DataToExport(name=dwa.origin, data=[dwa])
self.add_data(dte_from_dwa)
else:
raise NameError(f"Cannot save this DataToExport named {dte.name} to one of the named saver")
def add_data_bundle(self, data: DataBundle):
self.add_data(data.dte)
class OptimizerSaver(ExtensionSaver):
"""Implementation of the ModuleSaver class dedicated to Optimizer based modules
Parameters
----------
h5saver
module
"""
group_type = GroupModuleType.OPTIMIZER
def __init__(self, module,
enl_axis_names: Iterable[str] = None,
enl_axis_units: Iterable[str] = None):
super().__init__(module)
self.detectors : dict[str, DetectorEnlargeableSaver] = {}
self.actuators : dict[str, ActuatorEnlargeableSaver ] = {}
self.enl_axis_names = enl_axis_names
self.enl_axis_units = enl_axis_units
def add_data(self,
dte: DataToExport,
axis_values: List[Union[float, np.ndarray]] = None,
):
self.detectors[dte.name].add_data(self.current_nodes[dte.name],
dte, axis_values)
def add_data_bundle(self, data: DataBundle):
self.add_data(data.dte, axis_values=data.axis_values)