PulseAnalyzer.full_settings_dict() - Code Metrics - Inspection of "Pulsed 3.0" - Ulm-IQO/qudi - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#384)

unknown

created 2018-06-26 12:52 UTC

PulseAnalyzer.full_settings_dict() A

↳ Parent: PulseAnalyzer

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	1
dl	0
loc	11
rs	9.85
c	2
b	0
f	0

# -*- coding: utf-8 -*-
"""
This file contains the Qudi logic for analysis of laser pulses.

Qudi is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

Qudi is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with Qudi. If not, see <http://www.gnu.org/licenses/>.

Copyright (c) the Qudi Developers. See the COPYRIGHT.txt file at the
top-level directory of this distribution and at <https://github.com/Ulm-IQO/qudi/>
"""

import os
import sys
import inspect
import importlib

from core.util.modules import get_main_dir


class PulseAnalyzerBase:

    """
    All analyzer classes to import from must inherit exclusively from this base class.
    This base class enables analyzer classes masked read-only access to settings from
    PulsedMeasurementLogic.

    See BasicPulseAnalyzer class for an example usage.
    """
    def __init__(self, pulsedmeasurementlogic):
        self.__pulsedmeasurementlogic = pulsedmeasurementlogic

    @property
    def is_gated(self):
        return self.__pulsedmeasurementlogic.fast_counter_settings.get('is_gated')

    @property
    def measurement_settings(self):
        return self.__pulsedmeasurementlogic.measurement_settings

    @property
    def sampling_information(self):
        return self.__pulsedmeasurementlogic.sampling_information

    @property
    def fast_counter_settings(self):
        return self.__pulsedmeasurementlogic.fast_counter_settings

    @property
    def log(self):
        return self.__pulsedmeasurementlogic.log


class PulseAnalyzer(PulseAnalyzerBase):
    """
    Management class to automatically combine and interface analysis methods and associated
    parameters from analyzer classes defined in several modules.

    Analyzer class to import from must comply to the following rules:
    1) Exclusive inheritance from PulseAnalyzerBase class
    2) No direct access to PulsedMeasurementLogic instance except through properties defined in
       base class (read-only access)
    3) Analysis methods must be bound instance methods
    4) Analysis methods must be named starting with "analyse_"
    5) Analysis methods must have as first argument "laser_data"
    6) Apart from "laser_data" analysis methods must have exclusively keyword arguments with
       default values of the right data type. (e.g. differentiate between 42 (int) and 42.0 (float))
    7) Make sure that no two analysis methods in any module share a keyword argument of different
       default data type.
    8) The keyword "method" must not be used in the analysis method parameters

    See BasicPulseAnalyzer class for an example usage.
    """

    def __init__(self, pulsedmeasurementlogic):
        # Init base class
        super().__init__(pulsedmeasurementlogic)

        # Dictionary holding references to all analysis methods
        self._analysis_methods = dict()
        # dictionary containing all possible parameters that can be used by the analysis methods
        self._parameters = dict()
        # Currently selected analysis method
        self._current_analysis_method = None

        # import path for analysis modules from default directory (logic.pulse_analysis_methods)
        path_list = [os.path.join(get_main_dir(), 'logic', 'pulsed', 'pulsed_analysis_methods')]
        # import path for analysis modules from non-default directory if a path has been given
        if isinstance(pulsedmeasurementlogic.analysis_import_path, str):
            path_list.append(pulsedmeasurementlogic.analysis_import_path)

        # Import analysis modules and get a list of analyzer classes
        analyzer_classes = self.__import_external_analyzers(paths=path_list)

        # create an instance of each class and put them in a temporary list
        analyzer_instances = [cls(pulsedmeasurementlogic) for cls in analyzer_classes]

        # add references to all analysis methods in each instance to a dict
        self.__populate_method_dict(instance_list=analyzer_instances)

        # populate "_parameters" dictionary from analysis method signatures
        self.__populate_parameter_dict()

        # Set default analysis method
        self._current_analysis_method = sorted(self._analysis_methods)[0]

        # Update from parameter_dict if handed over
        if isinstance(pulsedmeasurementlogic.analysis_parameters, dict):
            # Delete unused parameters
            params = [p for p in pulsedmeasurementlogic.analysis_parameters if
                      p not in self._parameters and p != 'method']
            for param in params:
                del pulsedmeasurementlogic.analysis_parameters[param]
            # Update parameter dict and current method
            self.analysis_settings = pulsedmeasurementlogic.analysis_parameters
        return

    @property
    def analysis_settings(self):
        """
        This property holds all parameters needed for the currently selected analysis_method as
        well as the currently selected method name.

        @return dict: dictionary with keys being the parameter name and values being the parameter
        """
        # Get reference to the extraction method
        method = self._analysis_methods.get(self._current_analysis_method)

        # Get keyword arguments for the currently selected method
        settings_dict = self._get_analysis_method_kwargs(method)

        # Attach current analysis method name
        settings_dict['method'] = self._current_analysis_method
        return settings_dict

    @analysis_settings.setter

    def analysis_settings(self, settings_dict):
        """
        Update parameters contained in self._parameters by values in settings_dict.
        Also sets the current analysis method by passing its name using key "method".
        Parameters not included in self._parameters (except "method") will be ignored.

        @param dict settings_dict: dictionary containing the parameters to set (name, value)
        """
        if not isinstance(settings_dict, dict):
            return

        # go through all key-value pairs in settings_dict and update self._parameters and
        # self._current_analysis_method accordingly. Ignore unknown parameters.
        for parameter, value in settings_dict.items():
            if parameter == 'method':
                if value in self._analysis_methods:
                    self._current_analysis_method = value
                else:
                    self.log.error('Analysis method "{0}" could not be found in PulseAnalyzer.'
                                   ''.format(value))
            elif parameter in self._parameters:
                self._parameters[parameter] = value
            else:
                self.log.warning('No analysis parameter "{0}" found in PulseAnalyzer.\n'
                                 'Parameter will be ignored.'.format(parameter))
        return

    @property
    def analysis_methods(self):
        """
        Return available analysis methods.

        @return dict: Dictionary with keys being the method names and values being the methods.
        """
        return self._analysis_methods

    @property
    def full_settings_dict(self):
        """
        Returns the full set of parameters for all methods as well as the currently selected method
        in order to store them in a StatusVar in PulsedMeasurementLogic.

        @return dict: full set of parameters and currently selected analysis method.
        """
        settings_dict = self._parameters.copy()
        settings_dict['method'] = self._current_analysis_method
        return settings_dict

    def analyse_laser_pulses(self, laser_data):
        """
        Wrapper method to call the currently selected analysis method with laser_data and the
        appropriate keyword arguments.

        @param numpy.ndarray laser_data: 2D numpy array (dtype='int64') containing the timetraces
                                         for all extracted laser pulses.
        @return (numpy.ndarray, numpy.ndarray): tuple of two numpy arrays containing the evaluated
                                                signal data (one data point for each laser pulse)
                                                and the measurement error corresponding to each
                                                data point.
        """
        analysis_method = self._analysis_methods[self._current_analysis_method]

        kwargs = self._get_analysis_method_kwargs(analysis_method)
        return analysis_method(laser_data=laser_data, **kwargs)

    def _get_analysis_method_kwargs(self, method):

        """
        Get the proper values for keyword arguments other than "laser_data" for <method>.
        Try to take the values from self._parameters. If the keyword is missing in the dictionary,
        take the default values from the method signature.

        @param method: reference to a callable analysis method
        @return dict: A dictionary containing the argument keywords for <method> and corresponding
                      values from self._parameters.
        """
        kwargs_dict = dict()
        method_signature = inspect.signature(method)
        for name in method_signature.parameters.keys():
            if name == 'laser_data':
                continue

            default = method_signature.parameters[name].default
            recalled = self._parameters.get(name)

            if recalled is not None and type(recalled) == type(default):
                kwargs_dict[name] = recalled
            else:
                kwargs_dict[name] = default
        return kwargs_dict

    def __import_external_analyzers(self, paths):

        """
        Helper method to import all modules from directories contained in paths.
        Find all classes in those modules that inherit exclusively from PulseAnalyzerBase class
        and return a list of them.

        @param iterable paths: iterable containing paths to import modules from
        @return list: A list of imported valid analyzer classes
        """
        class_list = list()
        for path in paths:
            if not os.path.exists(path):
                self.log.error('Unable to import analysis methods from "{0}".\n'
                               'Path does not exist.'.format(path))
                continue
            # Get all python modules to import from.
            # The assumption is that in the directory pulse_analysis_methods, there are
            # *.py files, which contain only analyzer classes!
            module_list = [name[:-3] for name in os.listdir(path) if
                           os.path.isfile(os.path.join(path, name)) and name.endswith('.py')]

            # append import path to sys.path
            sys.path.append(path)

            # Go through all modules and create instances of each class found.
            for module_name in module_list:
                # import module
                mod = importlib.import_module('{0}'.format(module_name))
                importlib.reload(mod)
                # get all analyzer class references defined in the module
                tmp_list = [m[1] for m in inspect.getmembers(mod, self.__is_analyzer_class)]
                # append to class_list
                class_list.extend(tmp_list)
        return class_list

    def __populate_method_dict(self, instance_list):
        """
        Helper method to populate the dictionaries containing all references to callable analysis
        methods contained in analyzer instances passed to this method.

        @param list instance_list: List containing instances of analyzer classes
        """
        self._analysis_methods = dict()
        for instance in instance_list:
            for method_name, method_ref in inspect.getmembers(instance, inspect.ismethod):
                if method_name.startswith('analyse_'):
                    self._analysis_methods[method_name[8:]] = method_ref
        return

    def __populate_parameter_dict(self):
        """
        Helper method to populate the dictionary containing all possible keyword arguments from all
        analysis methods.
        """
        self._parameters = dict()
        for method in self._analysis_methods.values():
            self._parameters.update(self._get_analysis_method_kwargs(method=method))
        return

    @staticmethod
    def __is_analyzer_class(obj):
        """
        Helper method to check if an object is a valid analyzer class.

        @param object obj: object to check
        @return bool: True if obj is a valid analyzer class, False otherwise
        """
        if inspect.isclass(obj):
            return PulseAnalyzerBase in obj.__bases__ and len(obj.__bases__) == 1
        return False


1		# -- coding: utf-8 --
2		"""
3		This file contains the Qudi logic for analysis of laser pulses.
4
5		Qudi is free software: you can redistribute it and/or modify
6		it under the terms of the GNU General Public License as published by
7		the Free Software Foundation, either version 3 of the License, or
8		(at your option) any later version.
9
10		Qudi is distributed in the hope that it will be useful,
11		but WITHOUT ANY WARRANTY; without even the implied warranty of
12		MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13		GNU General Public License for more details.
14
15		You should have received a copy of the GNU General Public License
16		along with Qudi. If not, see <http://www.gnu.org/licenses/>.
17
18		Copyright (c) the Qudi Developers. See the COPYRIGHT.txt file at the
19		top-level directory of this distribution and at <https://github.com/Ulm-IQO/qudi/>
20		"""
21
22		import os
23		import sys
24		import inspect
25		import importlib
26
27		from core.util.modules import get_main_dir
28
29
30	View Code Duplication	class PulseAnalyzerBase:
		0 ignored issues – show Duplication introduced 2018-06-01 10:04 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
31		"""
32		All analyzer classes to import from must inherit exclusively from this base class.
33		This base class enables analyzer classes masked read-only access to settings from
34		PulsedMeasurementLogic.
35
36		See BasicPulseAnalyzer class for an example usage.
37		"""
38		def __init__(self, pulsedmeasurementlogic):
39		self.__pulsedmeasurementlogic = pulsedmeasurementlogic
40
41		@property
42		def is_gated(self):
43		return self.__pulsedmeasurementlogic.fast_counter_settings.get('is_gated')
44
45		@property
46		def measurement_settings(self):
47		return self.__pulsedmeasurementlogic.measurement_settings
48
49		@property
50		def sampling_information(self):
51		return self.__pulsedmeasurementlogic.sampling_information
52
53		@property
54		def fast_counter_settings(self):
55		return self.__pulsedmeasurementlogic.fast_counter_settings
56
57		@property
58		def log(self):
59		return self.__pulsedmeasurementlogic.log
60
61
62		class PulseAnalyzer(PulseAnalyzerBase):
63		"""
64		Management class to automatically combine and interface analysis methods and associated
65		parameters from analyzer classes defined in several modules.
66
67		Analyzer class to import from must comply to the following rules:
68		1) Exclusive inheritance from PulseAnalyzerBase class
69		2) No direct access to PulsedMeasurementLogic instance except through properties defined in
70		base class (read-only access)
71		3) Analysis methods must be bound instance methods
72		4) Analysis methods must be named starting with "analyse_"
73		5) Analysis methods must have as first argument "laser_data"
74		6) Apart from "laser_data" analysis methods must have exclusively keyword arguments with
75		default values of the right data type. (e.g. differentiate between 42 (int) and 42.0 (float))
76		7) Make sure that no two analysis methods in any module share a keyword argument of different
77		default data type.
78		8) The keyword "method" must not be used in the analysis method parameters
79
80		See BasicPulseAnalyzer class for an example usage.
81		"""
82
83		def __init__(self, pulsedmeasurementlogic):
84		# Init base class
85		super().__init__(pulsedmeasurementlogic)
86
87		# Dictionary holding references to all analysis methods
88		self._analysis_methods = dict()
89		# dictionary containing all possible parameters that can be used by the analysis methods
90		self._parameters = dict()
91		# Currently selected analysis method
92		self._current_analysis_method = None
93
94		# import path for analysis modules from default directory (logic.pulse_analysis_methods)
95		path_list = [os.path.join(get_main_dir(), 'logic', 'pulsed', 'pulsed_analysis_methods')]
96		# import path for analysis modules from non-default directory if a path has been given
97		if isinstance(pulsedmeasurementlogic.analysis_import_path, str):
98		path_list.append(pulsedmeasurementlogic.analysis_import_path)
99
100		# Import analysis modules and get a list of analyzer classes
101		analyzer_classes = self.__import_external_analyzers(paths=path_list)
102
103		# create an instance of each class and put them in a temporary list
104		analyzer_instances = [cls(pulsedmeasurementlogic) for cls in analyzer_classes]
105
106		# add references to all analysis methods in each instance to a dict
107		self.__populate_method_dict(instance_list=analyzer_instances)
108
109		# populate "_parameters" dictionary from analysis method signatures
110		self.__populate_parameter_dict()
111
112		# Set default analysis method
113		self._current_analysis_method = sorted(self._analysis_methods)[0]
114
115		# Update from parameter_dict if handed over
116		if isinstance(pulsedmeasurementlogic.analysis_parameters, dict):
117		# Delete unused parameters
118		params = [p for p in pulsedmeasurementlogic.analysis_parameters if
119		p not in self._parameters and p != 'method']
120		for param in params:
121		del pulsedmeasurementlogic.analysis_parameters[param]
122		# Update parameter dict and current method
123		self.analysis_settings = pulsedmeasurementlogic.analysis_parameters
124		return
125
126		@property
127		def analysis_settings(self):
128		"""
129		This property holds all parameters needed for the currently selected analysis_method as
130		well as the currently selected method name.
131
132		@return dict: dictionary with keys being the parameter name and values being the parameter
133		"""
134		# Get reference to the extraction method
135		method = self._analysis_methods.get(self._current_analysis_method)
136
137		# Get keyword arguments for the currently selected method
138		settings_dict = self._get_analysis_method_kwargs(method)
139
140		# Attach current analysis method name
141		settings_dict['method'] = self._current_analysis_method
142		return settings_dict
143
144	View Code Duplication	@analysis_settings.setter
		0 ignored issues – show Duplication introduced 2018-06-01 10:04 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
145		def analysis_settings(self, settings_dict):
146		"""
147		Update parameters contained in self._parameters by values in settings_dict.
148		Also sets the current analysis method by passing its name using key "method".
149		Parameters not included in self._parameters (except "method") will be ignored.
150
151		@param dict settings_dict: dictionary containing the parameters to set (name, value)
152		"""
153		if not isinstance(settings_dict, dict):
154		return
155
156		# go through all key-value pairs in settings_dict and update self._parameters and
157		# self._current_analysis_method accordingly. Ignore unknown parameters.
158		for parameter, value in settings_dict.items():
159		if parameter == 'method':
160		if value in self._analysis_methods:
161		self._current_analysis_method = value
162		else:
163		self.log.error('Analysis method "{0}" could not be found in PulseAnalyzer.'
164		''.format(value))
165		elif parameter in self._parameters:
166		self._parameters[parameter] = value
167		else:
168		self.log.warning('No analysis parameter "{0}" found in PulseAnalyzer.\n'
169		'Parameter will be ignored.'.format(parameter))
170		return
171
172		@property
173		def analysis_methods(self):
174		"""
175		Return available analysis methods.
176
177		@return dict: Dictionary with keys being the method names and values being the methods.
178		"""
179		return self._analysis_methods
180
181		@property
182		def full_settings_dict(self):
183		"""
184		Returns the full set of parameters for all methods as well as the currently selected method
185		in order to store them in a StatusVar in PulsedMeasurementLogic.
186
187		@return dict: full set of parameters and currently selected analysis method.
188		"""
189		settings_dict = self._parameters.copy()
190		settings_dict['method'] = self._current_analysis_method
191		return settings_dict
192
193		def analyse_laser_pulses(self, laser_data):
194		"""
195		Wrapper method to call the currently selected analysis method with laser_data and the
196		appropriate keyword arguments.
197
198		@param numpy.ndarray laser_data: 2D numpy array (dtype='int64') containing the timetraces
199		for all extracted laser pulses.
200		@return (numpy.ndarray, numpy.ndarray): tuple of two numpy arrays containing the evaluated
201		signal data (one data point for each laser pulse)
202		and the measurement error corresponding to each
203		data point.
204		"""
205		analysis_method = self._analysis_methods[self._current_analysis_method]
206
207		kwargs = self._get_analysis_method_kwargs(analysis_method)
208		return analysis_method(laser_data=laser_data, **kwargs)
209
210	View Code Duplication	def _get_analysis_method_kwargs(self, method):
		0 ignored issues – show Duplication introduced 2018-06-01 10:04 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
211		"""
212		Get the proper values for keyword arguments other than "laser_data" for <method>.
213		Try to take the values from self._parameters. If the keyword is missing in the dictionary,
214		take the default values from the method signature.
215
216		@param method: reference to a callable analysis method
217		@return dict: A dictionary containing the argument keywords for <method> and corresponding
218		values from self._parameters.
219		"""
220		kwargs_dict = dict()
221		method_signature = inspect.signature(method)
222		for name in method_signature.parameters.keys():
223		if name == 'laser_data':
224		continue
225
226		default = method_signature.parameters[name].default
227		recalled = self._parameters.get(name)
228
229		if recalled is not None and type(recalled) == type(default):
230		kwargs_dict[name] = recalled
231		else:
232		kwargs_dict[name] = default
233		return kwargs_dict
234
235	View Code Duplication	def __import_external_analyzers(self, paths):
		0 ignored issues – show Duplication introduced 2018-06-01 10:04 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
236		"""
237		Helper method to import all modules from directories contained in paths.
238		Find all classes in those modules that inherit exclusively from PulseAnalyzerBase class
239		and return a list of them.
240
241		@param iterable paths: iterable containing paths to import modules from
242		@return list: A list of imported valid analyzer classes
243		"""
244		class_list = list()
245		for path in paths:
246		if not os.path.exists(path):
247		self.log.error('Unable to import analysis methods from "{0}".\n'
248		'Path does not exist.'.format(path))
249		continue
250		# Get all python modules to import from.
251		# The assumption is that in the directory pulse_analysis_methods, there are
252		# *.py files, which contain only analyzer classes!
253		module_list = [name[:-3] for name in os.listdir(path) if
254		os.path.isfile(os.path.join(path, name)) and name.endswith('.py')]
255
256		# append import path to sys.path
257		sys.path.append(path)
258
259		# Go through all modules and create instances of each class found.
260		for module_name in module_list:
261		# import module
262		mod = importlib.import_module('{0}'.format(module_name))
263		importlib.reload(mod)
264		# get all analyzer class references defined in the module
265		tmp_list = [m[1] for m in inspect.getmembers(mod, self.__is_analyzer_class)]
266		# append to class_list
267		class_list.extend(tmp_list)
268		return class_list
269
270		def __populate_method_dict(self, instance_list):
271		"""
272		Helper method to populate the dictionaries containing all references to callable analysis
273		methods contained in analyzer instances passed to this method.
274
275		@param list instance_list: List containing instances of analyzer classes
276		"""
277		self._analysis_methods = dict()
278		for instance in instance_list:
279		for method_name, method_ref in inspect.getmembers(instance, inspect.ismethod):
280		if method_name.startswith('analyse_'):
281		self._analysis_methods[method_name[8:]] = method_ref
282		return
283
284		def __populate_parameter_dict(self):
285		"""
286		Helper method to populate the dictionary containing all possible keyword arguments from all
287		analysis methods.
288		"""
289		self._parameters = dict()
290		for method in self._analysis_methods.values():
291		self._parameters.update(self._get_analysis_method_kwargs(method=method))
292		return
293
294		@staticmethod
295		def __is_analyzer_class(obj):
296		"""
297		Helper method to check if an object is a valid analyzer class.
298
299		@param object obj: object to check
300		@return bool: True if obj is a valid analyzer class, False otherwise
301		"""
302		if inspect.isclass(obj):
303		return PulseAnalyzerBase in obj.__bases__ and len(obj.__bases__) == 1
304		return False
305

Ulm-IQO / qudi

Pull Request — master (#384)

PulseAnalyzer.full_settings_dict() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like