# -*- coding: utf-8 -*-
"""
Created on Wed Jan 10 16:54:14 2018
@author: Weber Sébastien
"""
from __future__ import annotations
from abc import ABC
from importlib import import_module
import os
from pathlib import Path
import sys
from typing import List, Tuple, Union, Optional
import time
from easydict import EasyDict as edict
from qtpy import QtWidgets
from qtpy.QtCore import Qt, QObject, Slot, QThread, Signal
from pymodaq_data.data import DataToExport, Averaging
from pymodaq.utils.data import DataFromPlugins
from pymodaq_utils.logger import set_logger, get_module_name
from pymodaq.control_modules.utils import ParameterControlModule, HardwareWorkerBase
from pymodaq_gui.utils.file_io import select_file
from pymodaq_gui.utils.widgets.lcd import LCD
from pymodaq_utils.config import GlobalConfig as Config, get_set_local_dir
from pymodaq_gui.h5modules.browsing import browse_data
from pymodaq_gui.h5modules.saving import H5Saver
from pymodaq.utils.h5modules import module_saving
from pymodaq_data.h5modules.backends import Node, SaveType
from pymodaq_utils.utils import ThreadCommand, find_dict_in_list_from_key_val
from pymodaq_gui.parameter import Parameter
from pymodaq_gui.parameter import utils as putils
from pymodaq.control_modules.viewer_utility_classes import params as daq_viewer_params
from pymodaq_utils import utils
from pymodaq_utils.warnings import deprecation_msg
from pymodaq_gui.utils import DockArea, Dock
from pymodaq.utils.gui_utils import get_splash_sc
from pymodaq.control_modules.daq_viewer_ui.ui_base import DAQ_Viewer_UI
from pymodaq.control_modules.instruments import (DET_TYPES, DetectorError, get_viewer_plugins)
from pymodaq.control_modules.enums import DAQTypesEnum
from pymodaq.control_modules.thread_commands import (ThreadStatus, ThreadStatusViewer, ControlToHardware,
ControlToHardwareViewer, UiToMainViewer)
from pymodaq_gui.plotting.data_viewers.viewer import ViewerBase
from pymodaq_gui.plotting.data_viewers import ViewersEnum
from pymodaq_gui.plotting.items.roi import RoiInfo
from pymodaq_utils.enums import enum_checker
from pymodaq.control_modules.viewer_utility_classes import DAQ_Viewer_base
from pymodaq.utils.leco.pymodaq_listener import ViewerActorListener, LECOClientCommands, LECOViewerCommands
from pymodaq.control_modules.daq_viewer_ui.viewer_selector import SelectedModule
logger = set_logger(get_module_name(__file__))
config = Config()
local_path = get_set_local_dir()
[docs]
class DAQ_Viewer(ParameterControlModule):
""" Main PyMoDAQ class to drive detectors
Qt object and generic UI to drive actuators. The class is giving you full functionality to select (daq_detector),
initialize detectors (init_hardware), grab or snap data (grab_data) and save them (save_new, save_current). If
a DockArea is given as parent widget, the full User Interface (DAQ_Viewer_UI) is loaded allowing easy control of the
instrument.
Attributes
----------
grab_done_signal: Signal[DataToExport]
Signal emitted when the data from the plugin (and eventually from the data viewers) has been received. To be
used by connected objects.
custom_sig: Signal[ThreadCommand]
use this to propagate info/data coming from the hardware plugin to another object
See Also
--------
ControlModule, DAQ_Viewer_UI, ParameterManager
Notes
-----
The ROIselect and the crosshair of the data viewers are forwarded to the plugin: each time the user moves them, the
plugin methods `roi_select(roi_info, ind_viewer)` and `crosshair(crosshair_info, ind_viewer)` are called (they do
nothing by default and can be overridden in the plugin).
"""
settings_name = 'daq_viewer_settings'
_hw_kind = 'detector'
_ini_hw_cmd = ControlToHardware.INI_HARDWARE
custom_sig = Signal(ThreadCommand) # particular case where DAQ_Viewer is used for a custom module
grab_done_signal = Signal(DataToExport)
data_saved = Signal()
grab_status = Signal(bool)
params = daq_viewer_params + [
{'title': 'Saver Settings:', 'name': 'saver_settings', 'type': 'group',
'visible': True, 'children': H5Saver.get_params_for_save_type(SaveType.detector), 'expanded': False}]
listener_class = ViewerActorListener
ui: Optional[DAQ_Viewer_UI]
def __init__(
self,
parent: Optional[QtWidgets.QWidget] = None,
title: str = "Testing",
daq_type=config("pymodaq", "viewer", "daq_type"),
area: DockArea = None,
rois_dock: Dock = None,
**kwargs,
):
self.logger = set_logger(f'{logger.name}.{title}')
self.logger.info(f'Initializing DAQ_Viewer: {title}')
super().__init__(listener_class=ViewerActorListener, title=title, **kwargs)
self.rois_dock: Dock = rois_dock
daq_type = enum_checker(DAQTypesEnum, daq_type)
self._detector = SelectedModule(daq_type=daq_type)
self._viewer_types: List[ViewersEnum] = []
self._viewers: List[ViewerBase] = []
self._viewer_connections: List[Tuple[Signal, callable]] = [] # (signal, slot) to disconnect on viewers change
self.override_grab_from_extension = False # boolean allowing an extension to tell to init a grab or not
# (see DataMixer for reasons and use case in ModulesManager and dashboard method add_det_from_extension)
self.parent = parent
if parent is not None:
self.ui = DAQ_Viewer_UI(self,
parent, title,
area = area,
rois_dock=self.rois_dock,
settings_dock=kwargs.pop('settings_dock', None),)
else:
self.ui = None
if self.ui is not None:
QtWidgets.QApplication.processEvents()
self.ui.add_setting_tree(self.settings_tree)
self.ui.command_sig.connect(self.process_ui_cmds)
self.viewers = self.ui.viewers
self._viewer_types = self.ui.viewer_types
self.splash_sc = get_splash_sc()
self._module_and_data_saver: Union[None,
module_saving.DetectorSaver,
module_saving.DetectorTimeSaver] = None
self._h5saver_continuous: Optional[H5Saver] = None
self._ind_continuous_grab = 0
self.settings.child('main_settings', 'DAQ_type').setValue(self._detector.daq_type.name)
self.settings.child('main_settings', 'detector_type').setValue(self._detector.module_name)
self._grabing: bool = False
self._do_bkg: bool = False
self._take_bkg: bool = False
self._grab_done: bool = False
self._start_grab_time: float = 0. # used for the refreshing rate
self._received_data: int = 0
self._lcd: Optional[LCD] = None
self._bkg: Optional[DataToExport] = None # buffer to store background
self._save_file_pathname: Optional[Path] = None # to store last active path, will be an Path object
self._snapshot_pathname: Optional[Path] = None
self._data_to_save_export: Optional[DataToExport] = None
self._do_save_data: bool = False
self.detector = self._detector
self.grab_done_signal.connect(self._save_export_data)
def __repr__(self):
return f'{self.__class__.__name__}: {self.title} {self.detector}'
# -------------------------------------------------------------------------
# Properties
# -------------------------------------------------------------------------
@property
def bkg(self) -> DataToExport:
"""Get the background data object"""
return self._bkg
@property
def viewer_docks(self) -> List[Dock]:
"""list of Viewer Docks from the UI"""
if self.ui is not None:
return self.ui.viewer_docks
@property
def viewers_docks(self) -> List[Dock]:
"""list of Viewer Docks from the UI, for back compatibility"""
deprecation_msg('viewers_docks is a deprecated property use viewer_docks instead')
return self.viewer_docks
@property
def grab_state(self):
""":obj:`bool`: Get the current grabbing status"""
return self._grabing
@property
def do_bkg(self) -> bool:
""":obj:`bool`: Get/Set if background subtraction should be done"""
return self._do_bkg
@do_bkg.setter
def do_bkg(self, doit: bool):
self._do_bkg = doit
def _connect_viewer_signal(self, signal: Signal, slot: callable):
signal.connect(slot)
self._viewer_connections.append((signal, slot))
def _send_roi_select(self, roi_info: RoiInfo, ind_viewer: int):
"""Forward the ROIselect info of a viewer to the plugin roi_select method"""
self.command_hardware.emit(ThreadCommand(ControlToHardwareViewer.ROI_SELECT,
dict(roi_info=roi_info, ind_viewer=ind_viewer)))
def _send_crosshair(self, posx: float, posy: float, ind_viewer: int):
"""Forward the crosshair position of a viewer to the plugin crosshair method"""
self.command_hardware.emit(ThreadCommand(ControlToHardwareViewer.CROSSHAIR,
dict(crosshair_info=(posx, posy), ind_viewer=ind_viewer)))
@property
def viewers(self) -> List[ViewerBase]:
""":obj:`list`: Get/Set the Viewers (instances of real implementation of ViewerBase class) from the UI"""
if self.ui is not None:
return self._viewers
@viewers.setter
def viewers(self, viewers: List[ViewerBase]):
for viewer in self._viewers:
try:
viewer.data_to_export_signal.disconnect()
except TypeError as e:
pass
for signal, slot in self._viewer_connections:
try:
signal.disconnect(slot)
except (TypeError, RuntimeError):
pass
self._viewer_connections = []
for ind_viewer, viewer in enumerate(viewers):
viewer.data_to_export_signal.connect(self._get_data_from_viewer)
# ind_viewer bound as default argument (evaluated now, not when the slot is called)
self._connect_viewer_signal(
viewer.roi_select_signal,
lambda roi_info, ind_viewer=ind_viewer: self._send_roi_select(roi_info, ind_viewer))
self._connect_viewer_signal(
viewer.crosshair_dragged,
lambda posx, posy, ind_viewer=ind_viewer: self._send_crosshair(posx, posy, ind_viewer))
self._viewers = viewers
self.instrument_changed.emit()
@property
def Naverage(self):
return self.settings['main_settings', 'Naverage']
@Naverage.setter
def Naverage(self, ngrab: int):
if ngrab >= 1:
self.settings.child('main_settings', 'Naverage').setValue(ngrab)
@property
def detector(self) -> SelectedModule:
"""Get/Set the currently selected detector among available detectors"""
return self._detector
@detector.setter
def detector(self, det: SelectedModule | str):
if isinstance(det, str):
det = SelectedModule(self._detector.daq_type, det)
self._detector = det
self.settings.child('main_settings', 'DAQ_type').setValue(det.daq_type.name)
self.settings.child('main_settings', 'detector_type').setValue(det.module_name)
self._reload_plugin_settings()
if self.ui is not None:
self.ui.detector = det
@property
def daq_type(self) -> DAQTypesEnum:
"""Get/Set the daq_type as a DAQTypesEnum
Update the detector property with the list of available detectors of a given daq_type
"""
return self.detector.daq_type
@daq_type.setter
def daq_type(self, daq_type: DAQTypesEnum | str):
daq_type = enum_checker(DAQTypesEnum, daq_type)
self.detector = SelectedModule(daq_type=daq_type)
@property
def daq_types(self) -> List[str]:
"""List of available DAQ_TYPES as keys of the DAQTypesEnum"""
return DAQTypesEnum.names()
@property
def current_data(self) -> DataToExport:
""" Get the current data stored internally"""
return self._data_to_save_export
# -------------------------------------------------------------------------
# UI command processing
# -------------------------------------------------------------------------
[docs]
def process_ui_cmds(self, cmd: utils.ThreadCommand):
"""Process commands sent by actions done in the ui
See pymodaq.control_modules.thread_commands.UiToMainViewer
Parameters
----------
cmd: ThreadCommand
Possible values are:
* init
* grab
* snap
* detector_changed
* save_current
* do_bkg
* take_bkg
* viewers_changed
* quit
"""
if cmd.command == UiToMainViewer.INIT:
self.do_init_hardware_signal.emit(cmd.attribute[0]) # usually connected to ini_hardware method, but could be bypassed (see Dashboard)
elif cmd.command == UiToMainViewer.GRAB:
self.grab_data(cmd.attribute, snap_state=False)
elif cmd.command == UiToMainViewer.SNAP:
self.grab_data(False, snap_state=True)
elif cmd.command == UiToMainViewer.RESET_LIVE:
self.reset_live_averaging()
elif cmd.command == UiToMainViewer.SAVE_CURRENT:
self.save_current()
elif cmd.command == UiToMainViewer.DETECTOR_CHANGED:
if isinstance(cmd.attribute, SelectedModule):
self.detector_changed_from_ui(cmd.attribute)
elif cmd.command == UiToMainViewer.TAKE_BKG:
self.take_bkg()
elif cmd.command == UiToMainViewer.DO_BKG:
self.do_bkg = cmd.attribute
elif cmd.command == UiToMainViewer.VIEWERS_CHANGED:
self._viewer_types: List[ViewersEnum] = cmd.attribute['viewer_types']
self.viewers = cmd.attribute['viewers']
elif cmd.command == UiToMainViewer.QUIT:
self.quit_fun()
def detector_changed_from_ui(self, detector: SelectedModule):
self._detector = detector
self.settings.child('main_settings', 'DAQ_type').setValue(detector.daq_type.name)
self._reload_plugin_settings()
# -------------------------------------------------------------------------
# Hardware lifecycle hooks
# -------------------------------------------------------------------------
def _create_hardware(self):
return DetectorWorker(self._title, self.settings, self.detector)
def _connect_hardware_signals(self, hardware):
hardware.data_detector_sig[DataToExport].connect(self.show_data)
hardware.data_detector_temp_sig[DataToExport].connect(self.show_temp_data)
def _post_hardware_init(self):
if self.ui is not None:
for dock in self.ui.viewer_docks:
dock.setEnabled(True)
def _quit_cleanup(self):
if self._lcd is not None:
try:
self._lcd.parent.close()
except Exception as e:
self.logger.exception(str(e))
# -------------------------------------------------------------------------
# Acquisition API
# -------------------------------------------------------------------------
[docs]
def snap(self, send_to_leco=False):
""" Launch a single grab """
self.grab_data(False, snap_state=True, send_to_leco=send_to_leco)
[docs]
def grab(self, send_to_leco=False):
""" Launch a continuous grab """
if self.ui is not None:
self.manage_ui_actions('grab', 'setChecked', not self._grabing)
self.grab_data(not self._grabing, snap_state=False, send_to_leco=send_to_leco)
[docs]
def snapshot(self, pathname=None, dosave=False, send_to_leco=False):
"""Do one single grab (snap) and eventually save the data.
Parameters
----------
pathname: str or Path object
The path where to save data
dosave: bool
Do save or just grab data
send_to_leco: bool
If True, send the grabbed data through the LECO Coordinator
"""
try:
self._do_save_data = dosave
if pathname is None:
raise (ValueError("filepathanme has not been defined in snapshot"))
self._save_file_pathname = pathname
self.grab_data(grab_state=False, send_to_leco=send_to_leco, snap_state=True)
except Exception as e:
self.logger.exception(str(e))
[docs]
def grab_data(self, grab_state=False, send_to_leco=False, snap_state=False):
""" Generic method to grab or snap data from the selected (and initialized) detector
Parameters
----------
grab_state: bool
Defines the grab status: if True: do live grabbing if False stops the grab
send_to_leco: bool
If True, send the grabbed data through the LECO Coordinator
snap_state: bool
if True performs a single grab
"""
self._grabing = grab_state
self._send_to_leco = send_to_leco
self._grab_done = False
if self.ui is not None:
self.ui.data_ready = False
self._start_grab_time = time.perf_counter()
if snap_state:
self.update_status(f'{self._title}: Snap')
self.command_hardware.emit(
ThreadCommand(ControlToHardwareViewer.SINGLE,
dict(Naverage=self.settings['main_settings', 'Naverage'])))
else:
if not grab_state:
self.stop()
else:
self.thread_status(ThreadCommand(ThreadStatusViewer.UPDATE_CHANNELS))
self.update_status(f'{self._title}: Continuous Grab')
self.command_hardware.emit(
ThreadCommand(ControlToHardwareViewer.GRAB,
dict(Naverage=self.settings['main_settings', 'Naverage'])))
[docs]
def take_bkg(self):
""" Do a snap and store data to be used as background into an attribute: `self._bkg`
The content of the bkg will be saved if data is further saved with do_bkg property set to True
"""
self._take_bkg = True
self.grab_data(snap_state=True)
[docs]
def stop_module(self):
""" Programmatic entry to stop the Control module either moving, polling or grabbing"""
self.stop_grab()
[docs]
def stop_grab(self):
""" Stop the current continuous grabbing and unchecked the Grab button of the UI
See Also
--------
:meth:`stop`
"""
if self.ui is not None and self.ui.is_action_checked('grab'):
self.ui.get_action('grab').trigger()
else:
self.stop()
[docs]
def stop(self):
""" Stop the current continuous grabbing """
self.update_status(f'{self._title}: Stop Grab')
self.command_hardware.emit(ThreadCommand(ControlToHardwareViewer.STOP_GRAB))
self._grabing = False
def reset_live_averaging(self):
self._ind_continuous_grab = 0
# -------------------------------------------------------------------------
# Data handling
# -------------------------------------------------------------------------
@Slot()
def _raise_timeout(self):
""" Print the "timeout occurred" error message in the status bar via the update_status method.
"""
self.update_status("Timeout occurred", log_type="log")
[docs]
def save_current(self):
"""Save current data into a h5file"""
self._do_save_data = True
start_path = self._save_file_pathname or config('data', 'data_saving', 'h5file', 'save_path')
self._save_file_pathname = select_file(start_path=start_path, save=True, ext='h5')
self._save_export_data(self._data_to_save_export)
def _init_continuous_save(self):
""" Initialize the continuous saving H5Saver object
Update the module_and_data_saver attribute as :class:`DetectorTimeSaver` object
"""
if self._h5saver_continuous.settings.child('do_save').value():
self.settings.child('saver_settings', 'base_name').setValue('Data')
self.settings.child('saver_settings', 'N_saved').show()
self.settings.child('saver_settings', 'N_saved').setValue(0)
self.module_and_data_saver.h5saver = self._h5saver_continuous
self._h5saver_continuous.init_file(update_h5=True)
self.module_and_data_saver = module_saving.DetectorTimeSaver(self)
self.module_and_data_saver.h5saver = self._h5saver_continuous
self.module_and_data_saver.get_set_node()
self.grab_done_signal.connect(self.append_data)
else:
self._do_continuous_save = False
self.settings.child('saver_settings', 'N_saved').hide()
self.grab_done_signal.disconnect(self.append_data)
try:
self._h5saver_continuous.close()
except Exception as e:
self.logger.exception(str(e))
[docs]
def append_data(self, dte: DataToExport = None,
where: Union[Node, str] = None,
**kwargs):
"""Appends current DataToExport to a DetectorTimeSaver
Method to be used when performing continuous saving into a h5file (continuous mode or DAQ_Logger)
Parameters
----------
dte: DataToExport
not really used
where: Node or str
kwargs: dict
See Also
--------
:class:`DetectorTimeSaver`
"""
if dte is None:
dte = self._data_to_save_export
init_step = kwargs.pop('init_step', None)
if init_step is None:
init_step = self.settings['saver_settings', 'N_saved'] == 0
self._add_data_to_saver(dte,
init_step=init_step,
where=where,
**kwargs)
self.settings.child('saver_settings', 'N_saved').setValue(self.settings['saver_settings', 'N_saved'] + 1)
def _add_data_to_saver(self, dte: DataToExport, init_step=False, where=None, **kwargs):
"""Adds DataToExport data to the current node using the declared module_and_data_saver
Filters the data to be saved by DataSource as specified in the current H5Saver (see self.module_and_data_saver)
Parameters
----------
dte: DataToExport
The data to be saved
init_step: bool
If True, means this is the first step of saving (if multisaving), then save background if any and a png image
kwargs: dict
Other named parameters to be passed as is to the module_and_data_saver
See Also
--------
DetectorSaver, DetectorTimeSaver
"""
if dte is not None:
detector_node = self.module_and_data_saver.get_set_node(where)
dte = self.module_and_data_saver.filter_data(dte)
self.module_and_data_saver.add_data(detector_node, dte, **kwargs)
if init_step:
if self._do_bkg and self._bkg is not None:
self.module_and_data_saver.add_bkg(detector_node, self._bkg)
def _save_data(self, path=None, dte: DataToExport = None):
"""Private. Practical implementation to save data into a h5file altogether with metadata, axes, background...
Parameters
----------
path: Path
where to save the data as returned from browse_file for instance
dte: DataToExport
See Also
--------
browse_file, _get_data_from_viewers
"""
if path is not None:
path = Path(path)
h5saver = H5Saver(save_type='detector')
h5saver.init_file(update_h5=True, custom_naming=False, addhoc_file_path=path)
self.module_and_data_saver = module_saving.DetectorSaver(self)
self.module_and_data_saver.h5saver = h5saver
self._add_data_to_saver(dte, init_step=True)
if self.ui is not None:
(root, filename) = os.path.split(str(path))
filename, ext = os.path.splitext(filename)
image_path = os.path.join(root, filename + '.png')
self.parent.parent().grab().save(image_path)
h5saver.close_file()
self.data_saved.emit()
@Slot(DataToExport)
def _save_export_data(self, data: DataToExport):
"""Auxiliary method (Slot) to receive all data (raw and processed from rois) and save them
Parameters
----------
data: DataToExport
See Also
--------
_save_data
"""
if self._do_save_data:
self._save_data(self._save_file_pathname, data)
self._do_save_data = False
def _get_data_from_viewer(self, data: DataToExport):
"""Get all data emitted by the current viewers
Each viewer *data_to_export_signal* is connected to this slot. The collected data is stored in another
DataToExport `self._data_to_save_export` for further processing. All raw data are also stored in this attribute.
When all viewers have emitted this signal, the collected data are emitted with the
`grab_done_signal` signal.
Parameters
---------_
data: DataToExport
All data collected from the viewers
"""
if self._data_to_save_export is not None: # means that somehow data are not initialized so no further procsessing
self._received_data += 1
if len(data) != 0:
viewer_title = data.name
for dat in data:
dat.name = '/'.join([viewer_title, dat.origin, dat.name])
dat.origin = self.title
self._data_to_save_export.append(data)
if self._received_data == len(self.viewers):
self._grab_done = True
self.grab_done_signal.emit(self._data_to_save_export)
[docs]
@Slot(DataToExport)
def show_temp_data(self, data: DataToExport):
"""Send data to their dedicated viewers but those will not emit processed data signal
Slot receiving data from plugins emitted with the `data_grabed_signal_temp`
Parameters
----------
data: list of DataFromPlugins
"""
self._init_show_data(data)
if self.ui is not None:
self.set_data_to_viewers(data, temp=True)
[docs]
def show_data(self, dte: DataToExport):
""" Receive data from plugins and send them to their dedicated viewers
Slot receiving data from plugins emitted with the `data_grabed_signal`
Process the data as specified in the settings, display them into the dedicated data viewers depending on the
settings:
* create a container (DataToExport ``_data_to_save_export``) with info from this DAQ_Viewer (title), a
timestamp...
* call ``_process_data``
* do background subtraction if any
* check refresh time (if set in the settings) to send or not data to data viewers
* either send to the data viewers (if refresh time is ok and/or show data option in settings is set)
* either
* send grab_done_signal (to the slot _save_export_data ) to save the data
Parameters
----------
dte: DataToExport
See Also
--------
_init_show_data, _process_data
"""
try:
dte = dte.deepcopy()
if self.settings['main_settings', 'leco', 'leco_connected'] and self._send_to_leco:
self._leco_commands_signal.emit(ThreadCommand(LECOViewerCommands.DATA_READY, dte))
if self.ui is not None:
self.ui.data_ready = True
if self.settings['main_settings', 'live_averaging']:
self.settings.child('main_settings', 'N_live_averaging').setValue(self._ind_continuous_grab)
_current_data = dte.deepcopy()
if self._ind_continuous_grab == 0:
self._data_to_save_export = _current_data
else:
self._data_to_save_export = \
_current_data.average(self._data_to_save_export, self._ind_continuous_grab)
self._ind_continuous_grab += 1
else:
for dwa in dte:
dwa.origin = self._title
if self.settings['main_settings', 'dynamic'] != 'as_is':
arrays = []
for data_array in dwa:
arrays.append(data_array.astype(self.settings['main_settings', 'dynamic'], copy=False))
dwa.data = arrays
self._data_to_save_export = DataToExport(self._title, control_module='DAQ_Viewer', data=dte.data)
if self._take_bkg:
self._bkg = self._data_to_save_export.deepcopy()
self._take_bkg = False
if self._grabing: # if live
refresh_time = self.settings['main_settings', 'refresh_time']
refresh = time.perf_counter() - self._start_grab_time > refresh_time / 1000
if refresh:
self._start_grab_time = time.perf_counter()
else:
refresh = True # if single
if self.ui is not None and self.settings.child('main_settings', 'show_data').value() and refresh:
self._received_data = 0 # so that data send back from viewers can be properly counted
data_to_plot = self._data_to_save_export.get_data_from_attribute('do_plot', True, deepcopy=True)
data_to_plot.append(self._data_to_save_export.get_data_from_missing_attribute('do_plot', deepcopy=True))
# process bkg if needed
if self.do_bkg and self._bkg is not None:
data_to_plot -= self._bkg
self._init_show_data(data_to_plot)
self.set_data_to_viewers(data_to_plot)
else:
self._grab_done = True
self.grab_done_signal.emit(self._data_to_save_export)
except Exception as e:
self.logger.exception(str(e))
def _init_show_data(self, dte: DataToExport):
"""Processing before showing data
* check the data dimensionality to update the dedicated viewers
Parameters
----------
dte: DataToExport
"""
self._viewer_types = [ViewersEnum(dwa.dim.name) for dwa in dte if
('do_plot' not in dwa.extra_attributes) or
('do_plot' in dwa.extra_attributes and dwa.do_plot)]
if self.ui is not None:
if self.ui.viewer_types != self._viewer_types:
self.ui.update_viewers(self._viewer_types,
viewers_name=[f'{self.title} {dwa.name}' for dwa in dte],
)
[docs]
def set_data_to_viewers(self, dte: DataToExport, temp=False):
"""Process data dimensionality and send appropriate data to their data viewers
Parameters
----------
dte: DataToExport
temp: bool
if True notify the data viewers to display data as temporary (meaning not exporting processed data from roi)
See Also
--------
ViewerBase, Viewer0D, Viewer1D, Viewer2D
"""
for ind, dwa in enumerate(dte):
if ('do_plot' not in dwa.extra_attributes) or \
('do_plot' in dwa.extra_attributes and dwa.do_plot):
self.viewers[ind].title = f'{self._title} {dwa.name}'
name = (f'{dwa.name}_Averaged: {dwa.n_averaged}'
if dwa.averaged else dwa.name)
self.viewer_docks[ind].setTitle(f'{self._title} {name}')
if temp:
self.viewers[ind].show_data_temp(dwa)
else:
self.viewers[ind].show_data(dwa)
# -------------------------------------------------------------------------
# Saving
# -------------------------------------------------------------------------
[docs]
def setup_continuous_saving(self, init: bool = True):
"""Configure the objects dealing with the continuous saving mode"""
if init:
self.module_and_data_saver = module_saving.DetectorSaver(self)
self._h5saver_continuous = H5Saver(save_type='detector')
self._h5saver_continuous.settings.child('do_save').sigValueChanged.connect(self._init_continuous_save)
else:
self._h5saver_continuous.close_file()
# -------------------------------------------------------------------------
# Settings / Plugin management
# -------------------------------------------------------------------------
def get_detector_module(self, detector: SelectedModule):
detector_module = find_dict_in_list_from_key_val(
DET_TYPES[detector.daq_type.name],
'name', detector.module_name)['module']
return detector_module
def _load_plugin_params(self):
det_params, _class = get_viewer_plugins(self.detector.daq_type.name,
self.detector.module_name)
return det_params
def _module_value_changed(self, param: Parameter):
"""Handle detector-specific parameter changes."""
path = self.settings.childPath(param)
if param.name() == 'DAQ_type':
self.settings.child('saver_settings', 'do_save').setValue(False)
elif param.name() == 'show_averaging':
self.settings.child('main_settings', 'live_averaging').setValue(False)
# show_averaging is under main_settings so the base won't propagate it;
# forward it explicitly.
self._update_settings_signal.emit(edict(path=path, param=param, change='value'))
elif param.name() == 'live_averaging':
self.settings.child('main_settings', 'show_averaging').setValue(False)
self.ui.get_action('reset_live').setVisible(param.value())
if param.value():
self.settings.child('main_settings', 'N_live_averaging').show()
self.reset_live_averaging()
self.settings.child('main_settings', 'N_live_averaging').setValue(0)
else:
self.settings.child('main_settings', 'N_live_averaging').hide()
elif param.name() == 'wait_time':
self.command_hardware.emit(ThreadCommand(ControlToHardwareViewer.UPDATE_WAIT_TIME,
[param.value()]))
elif param.name() in putils.iter_children(self.settings.child('saver_settings'), []):
try:
if param.name() == 'do_save':
self.setup_continuous_saving(param.value())
self._h5saver_continuous.settings.child(*path[1:]).setValue(param.value())
except KeyError:
pass
# -------------------------------------------------------------------------
# Thread status handler
# -------------------------------------------------------------------------
[docs]
def thread_status(self, status: ThreadCommand):
"""Get back info (using the ThreadCommand object) from the hardware
And re-emit this ThreadCommand using the custom_sig signal if it should be used in a higher level module
Commands valid for all control modules are defined in the parent class, here are described only the specific
ones
Parameters
----------
status: ThreadCommand
The info returned from the hardware, the command (str) can be either:
* ini_hardware: update the status with "Viewer initialized" value and init state if attribute not null.
* grab : emit grab_status(True)
* grab_stopped: emit grab_status(False)
* init_lcd: display a LCD panel
* lcd: display on the LCD panel, the content of the attribute
* stop: stop the grab
"""
super().thread_status(status)
if status.command == ThreadStatus.INI_HARDWARE or status.command == ThreadStatusViewer.INI_DETECTOR:
self.update_status("Viewer initialized: " + str(status.attribute['initialized']))
if self.ui is not None:
self.ui.detector_init = status.attribute['initialized']
if status.attribute['initialized']:
self._controller_and_thread.controller = status.attribute["controller"]
self._controller_and_thread.initialized = status.attribute["initialized"]
self.init_signal.emit(self._controller_and_thread.initialized)
if self.ui is not None:
self.ui.set_init_color(self.get_color_from_status())
elif status.command == ThreadStatusViewer.GRAB:
self.grab_status.emit(True)
elif status.command == ThreadStatusViewer.GRAB_STOPPED:
self.grab_status.emit(False)
elif status.command == ThreadStatusViewer.INI_LCD:
if self._lcd is not None:
try:
self._lcd.parent.close()
except Exception as e:
self.logger.exception(str(e))
# lcd module
lcd = QtWidgets.QWidget()
self._lcd = LCD(lcd, **status.attribute)
lcd.setVisible(True)
QtWidgets.QApplication.processEvents()
elif status.command == ThreadStatusViewer.LCD:
"""status.attribute should be a list of arguments for the
LCD setvalues method. The first argument is a list of
numpy arrays of shape (1,), use np.atleast_1D"""
if isinstance(status.attribute, list):
#for backcompatibility
status.attribute = dict(values = status.attribute)
self._lcd.setvalues(**status.attribute)
elif status.command in (ThreadStatus.STOP, ThreadStatusViewer.STOP):
self.stop_grab()
# -------------------------------------------------------------------------
# LECO
# -------------------------------------------------------------------------
[docs]
@Slot(ThreadCommand)
def process_leco_commands(self, status: ThreadCommand) -> None:
"""Receive commands from the LECO network and process them.
Parameters
----------
status: ThreadCommand
Possible commands are:
* :attr:`LECOViewerCommands.GRAB`: start a continuous grab and later forward acquired data to LECO.
* :attr:`LECOViewerCommands.SNAP`: take a single snapshot and later forward the data to LECO.
* :attr:`LECOViewerCommands.STOP`: stop an ongoing grab.
"""
if status.command == LECOViewerCommands.GRAB:
self.grab(send_to_leco=True)
elif status.command ==LECOViewerCommands.SNAP:
self.snap(send_to_leco=True)
elif status.command == LECOViewerCommands.STOP:
self.stop()
else:
super().process_leco_commands(status=status)
[docs]
class DetectorWorker(HardwareWorkerBase):
""" Worker class to control the instrument plugin
Attributes
----------
plugin: real instance of the instrument plugin class
controller: DAQ_Viewer_base
wrapper object used to control a given instrument in the instrument plugin
controller_address: int
unique integer used to identify a controller shared among multiple instrument plugins
"""
data_detector_sig = Signal(DataToExport)
data_detector_temp_sig = Signal(DataToExport)
_kind = 'detector'
def __init__(self, title, settings_parameter, detector: SelectedModule):
super().__init__(title, detector.module_name)
self.waiting_for_data = False
self.controller = None
self.logger = set_logger(f'{logger.name}.{title}.detector')
self.plugin: DAQ_Viewer_base = None
self.grab_state = False
self.single_grab = False
self.datas: DataToExport = None
self.ind_average = 0
self.Naverage = 1
self.average_done = False
self.hardware_averaging = False
self.show_averaging = False
self.wait_time = settings_parameter['main_settings', 'wait_time']
self.daq_type = detector.daq_type
# --- deprecated aliases ---------------------------------------------------
@property
def detector(self):
deprecation_msg("detector is deprecated, use plugin")
return self.plugin
@property
def detector_name(self):
deprecation_msg("detector_name is deprecated, use plugin_name")
return self.plugin_name
@property
def controller_adress(self):
deprecation_msg("controller_adress is deprecated, use controller_address")
return self.controller_address
[docs]
def queue_command(self, command: ThreadCommand):
"""Transfer command from the main module to the hardware module.
Common commands (ini_hardware, close) are handled by the base class.
Detector-specific commands are handled here.
"""
if super().queue_command(command):
return
# Legacy alias: INI_DETECTOR → ini_hardware (base class already handles INI_HARDWARE)
if command.command == ControlToHardwareViewer.INI_DETECTOR:
status = self.ini_hardware(*command.attribute)
self.status_sig.emit(ThreadCommand(ThreadStatusViewer.INI_DETECTOR, status))
elif command.command == ControlToHardwareViewer.GRAB:
self.single_grab = False
self.grab_state = True
self.grab_data(**command.attribute)
elif command.command == ControlToHardwareViewer.SINGLE:
self.single_grab = True
self.grab_state = True
self.single(**command.attribute)
elif command.command == ControlToHardwareViewer.STOP_GRAB:
self.grab_state = False
self.plugin.stop()
self.status_sig.emit(ThreadCommand(ThreadStatus.UPDATE_STATUS, 'Stopping grab'))
elif command.command == ControlToHardwareViewer.UPDATE_SCANNER: # may be deprecated
self.plugin.update_scanner(command.attribute[0])
elif command.command == ControlToHardwareViewer.UPDATE_WAIT_TIME:
self.wait_time = command.attribute[0]
else: # custom commands for particular plugins
self._dispatch_custom_command(command)
[docs]
def ini_hardware(self, params_state=None, controller=None):
""" Initialize an instrument plugin class and tries to apply experiment settings
When the instrument is initialized from the Dashboard using an Experiment, tries to apply the experiment
settings to the instrument instance
Parameters
----------
params_state: dict
controller: wrapper
"""
try:
status = edict(initialized=False, info="", x_axis=None, y_axis=None)
det_params, class_ = get_viewer_plugins(self.daq_type.name, self.plugin_name)
self.plugin: DAQ_Viewer_base = class_(self, params_state)
try:
self.plugin.dte_signal.connect(self.data_ready)
self.plugin.dte_signal_temp.connect(self.emit_temp_data)
infos = self.plugin.ini_detector(controller)
status.controller = self.plugin.controller
self.controller_address = self.plugin.controller
except Exception as e:
self.logger.exception("Hardware couldn't be initialized", exc_info=e)
infos = str(e), False
status.controller = None
if isinstance(infos, edict):
status.update(infos)
else:
status.info = infos[0]
status.initialized = infos[1]
self.hardware_averaging = class_.hardware_averaging
return status
except Exception as e:
self.logger.exception(str(e))
return status
[docs]
def ini_detector(self, params_state=None, controller=None):
"""Deprecated: use ini_hardware instead."""
deprecation_msg("ini_detector is deprecated, use ini_hardware")
return self.ini_hardware(params_state, controller)
[docs]
def emit_temp_data(self, data: DataToExport):
""" Convenience method to export temporary data using the data_detector_temp_sig Signal
Parameters
----------
data: DataToExport
"""
self.data_detector_temp_sig.emit(data)
[docs]
def data_ready(self, data: DataToExport):
""" Process the data received from the instrument plugin class
Processing here is eventual software averaging if it was not possible in the instrument plugin class
Parameters
----------
data: DataToExport
"""
do_averaging = self.Naverage > 1 and not self.hardware_averaging
if do_averaging: # to execute if the averaging has to be done software wise
self.ind_average += 1
if self.ind_average == 1:
self.datas = data.deepcopy()
else:
self.datas = data.average(self.datas, self.ind_average - 1)
if self.show_averaging:
self.emit_temp_data(self.datas)
if self.ind_average == self.Naverage:
self.average_done = True
self.data_detector_sig.emit(self.datas)
self.ind_average = 0
else:
self.average_done = True # expected to make sure the single_grab stop by itself
self.data_detector_sig.emit(data)
self.waiting_for_data = False
[docs]
def single(self, Naverage=1, *args, **kwargs):
""" Convenience function to grab a single set of data
Parameters
----------
Naverage: int
The number of data to average before displaying
kwargs: optional named arguments
"""
self.grab_data(Naverage, live=False, **kwargs)
[docs]
def grab_data(self, Naverage=1, live=True, **kwargs):
""" General method to grab data from the instrument plugin class
Will check if the plugin class can do hardware averaging (if NAverage > 1) and and live_mode, otherwise
do both software wise here
Parameters
----------
Naverage: int
The number of data to average
live: bool
Try to run the instrument plugin class grabbing in live mode
kwargs: optional named arguments passed to the grab_data method of the instrument plugin class
"""
try:
self.ind_average = 0
self.Naverage = Naverage
if Naverage > 1:
self.average_done = False
self.waiting_for_data = False
# for live mode:two possibilities: either snap one data and regrab software wise
# (while True) or if self.detector.live_mode_available is True all data is continuously
# emitted from the plugin
if self.plugin.live_mode_available:
kwargs['wait_time'] = self.wait_time
else:
kwargs['wait_time'] = 0
self.status_sig.emit(ThreadCommand('grab'))
while True:
try:
if not self.waiting_for_data:
self.waiting_for_data = True
self.plugin.grab_data(Naverage, live=live, **kwargs)
QtWidgets.QApplication.processEvents()
if self.single_grab:
if self.hardware_averaging:
break
else:
if self.average_done:
break
else:
QThread.msleep(self.wait_time) # if in grab mode apply a waiting time
# after acquisition
if not self.grab_state:
break # if not in grab mode breaks the while loop
if self.plugin.live_mode_available and (not self.hardware_averaging and
self.average_done):
break # if live can be done in the plugin breaks the while loop except
# if average is asked but not done hardware wise
except Exception as e:
self.logger.exception(str(e))
self.status_sig.emit(ThreadCommand('grab_stopped'))
except Exception as e:
self.logger.exception(str(e))
[docs]
def close(self):
"""Call the close method of the instrument plugin class."""
if self.plugin is not None and self.plugin.controller is not None:
status = self.plugin.close()
return status
def prepare_docks(area, title):
""" Static method to init docks to be used within a DAQ_Viewer
Parameters
----------
area
title
Returns
-------
"""
dock_settings = Dock(title + " settings", size=(150, 250))
dock_viewer = Dock(title + " viewer", size=(350, 350))
area.addDock(dock_settings)
area.addDock(dock_viewer, 'right', dock_settings)
return dict(dock_settings=dock_settings, dock_viewer=dock_viewer)
def main(init_qt=True, init_det=False):
""" Method called to start the DAQ_Viewer in standalone mode"""
from pymodaq.utils.gui_utils.loader_utils import create_load_daq_viewer
from pymodaq_gui.qt_utils import mkQApp
if init_qt: # used for the test suite
app = mkQApp("PyMoDAQ Viewer")
shared_ui, daq_viewer = create_load_daq_viewer()
shared_ui.show()
daq_viewer.shared_ui = shared_ui
if init_det:
daq_viewer.init_hardware_ui(init_det)
if init_qt:
sys.exit(app.exec())
if __name__ == '__main__':
main(init_det=False)