"""
Providing the "Show Spectrum" experiment. It displays the (linearized)
intensities on the MCT detector averaged over the amount of samples that
should be acquired. The unit of the intensity, typically some kind of
energy, is determined by the linearization and in which unit the
reference data for that linearization was measured. Additionally it
displays the standard deviation of the pseudo shot to shot difference
signal for each pixel. (It is called pseudo because the sample - if
there is any - is not being pumped. This implies that there should the
resulting difference signal should be 0.)
The "Show Spectrum" experiment is mainly used to adjust the laser setup
and the OPAs. This includes optimizing for stability, moving the
wavelength of the OPAs and optimizing the alignment of the beams onto
the detector.
Note:
This is the only experiment that does not use a secondary processing
class. It would be unnecessarily complicated to pass data from
primary processing to secondary processing just to calculate
standard deviations.
#######################
Step by Step Algorithm:
#######################
**Acquisition:**
1. Preallocate dictionary (data container) which will contain data
and information about scan index, delay index etc.
2. Start the ADC task (in later experiments this was done
differently. See the source code for further details)
3. Read the data from the ADC
4. Place data into dictionary and hand over to primary processing
**Primary Processing:**
1. Preallocate arrays for data and weights
2. Subtract background from raw data (dark noise)
3. Linearize response of pixels
4. Calculate average intensity for each pixel
5. Since for show spectrum we are only interested in the
intensities, the data preparation for plotting is already done
here
6. Calculate the standard deviation of the intensities
7. Calculate transmission, or more precisely, relative intensity
(probe intensity / reference intensity) for each laser shot for
each pixel pair
8. Calculate the pseudo shot to shot difference signal (pseudo
because the sample - if there is any - is not being pumped and
thus no chopper is being used) and its statistical information
9. Put this information into data container and hand it over
to Pyqtplotting thread
10. Calculate the numbers which are displayed in the
"statistics box" on the GUI
11. Save data (and raw data) if respective checkboxes on GUI were
checked
**Pyqt Plotting:**
1. Remove old plots
2. Setup the plot that displays:
* Average intensities for each pixel
* Standard deviation of intensities times 5 as errorbars
* Standard deviation of pseudo shot to shot difference
spectrum
3. Plot the plots for the first time
4. Update plots
**Saving:**
.. code-block::
programming data dimension:
[(2 ([0] is current average scan, [1] is last single scan), n_pixels)]
saving dimension:
[(n_pixels)]
raw data dimension:
[(n_channels, samples_to_acquire)]
################
Folder Structure
################
**scans/** contains the collected data for each scan in the
dimensions which "saving dimension" suggests.
The data in **/averaged_data** is averaged equally. It contains the non
normalized averaged intensities.
The **/figure** folder is currently empty. It is possible to implement
plotting of figures which are saved here while the experiment is
running. This is however not implemented yet.
The **/hardware config** folder holds every file which contains hardware
configuration parameters. This folder is a compressed copy of the
hardware configuration folder in the software's directory. Here it is
possible to look up the ADC configuration to obtain what channel was
connected to which hardware element (e.g. chopper - ai78). It is also
possible to obtain the R-2R values, the FPAS configuration, the
linearization parameters, etc.
The **/raw data** folder is in the experiments directory only if the
"save raw data" checkbox on the GUI was checked. It contains the raw
(unedited... as raw as it can get) data. Basically, the voltages which
the ADC measured for each channel. The dimensions are as "raw data
dimensions" suggests.
The **background.npy** file is a copy of the most recently collected
dark noise background of the MCT detector array. It corrects for dark
noise. If no new background was collected before the experiment was
started the code will use the latest available background and display a
warning in the log.
**setupinfo.txt** is a ReadMe file that contains the most relevant
experimental parameters at first glance. Additionally, the user can
decide to write a comment in the readme editor of the GUI. The content
of this is written to the **notes.txt** file.
.. code-block::
username/
├── date1_experimentname1_000/
│ ├── averaged_data
│ │ └── date1_experimentname1_000.npy
│ ├── figures
│ ├── hardware config
│ ├── raw data
│ │ ├── s000000_date1_experimentname1_000_raw.npy
│ │ ├── ...
│ │ └── s000099_date1_experimentname1_000_raw.npy
│ ├── scans
│ │ ├── s000000_date1_experimentname1_000.npy
│ │ ├── ...
│ │ └── s000099_date1_experimentname1_000.npy
│ ├── setupinfo_date1_experimentname1_000.txt
│ ├── notes_date1_experimentname1_000.txt
│ └── background_date1_experimentname1_000.npy
├── date1_experimentname1_001/
├── date2_experimentname1_000/
└── date2_experimentname2_000/
"""
if __name__ == "__main__":
# Add directories to path for imports
import os, sys, inspect
currentdir = os.path.dirname(
os.path.abspath(inspect.getfile(inspect.currentframe()))
)
parentdir = os.path.dirname(currentdir)
sys.path.insert(0, parentdir)
sys.path.insert(0, os.path.join(parentdir, "hardware_interfaces"))
sys.path.insert(0, os.path.join(parentdir, "gui"))
import multiprocessing
from multiprocessing import Process, Queue
import threading
import numpy as np
from numpy import ndarray
# Matplotlib
import matplotlib
from matplotlib.backends.backend_qt5agg import (
FigureCanvasQTAgg,
NavigationToolbar2QT as NavigationToolbar,
)
from matplotlib.figure import Figure
# PyQTGraph
from pyqtgraph import PlotWidget, plot
import pyqtgraph as pg
from PyQt5 import QtWidgets, QtCore # , QtGui
from PyQt5.QtCore import QRunnable, QThreadPool, pyqtSignal, QObject
from qt_multithreading_wrapper import Worker
# Data processing
import data_processing as dp
from data_processing import PixelResponseLinearization as PRL
from analog_digital_converter import AnalogDigitalConverter as ADC
from triax import Triax as Spectrometer
from save_data import SaveData, Background
[docs]class ShowSpectrum:
"""
Args:
widget_pyqtgraph (QWidget): WidgetPyqtgraph object on which the
plots are going to be displayed. Has methods for plot
manipulation (i.e. removal of plots, autoscale).
adc (ADC): Analog to digital converter hardware object which is
used to communicate with and read data from the ADC.
prl (PRL): Pixel response linearization object which grants
the functionality to linearize raw data according to the
linearization parameters specified in the corresponding
*pixel_linearization_fit_parameters.json* file (for each
lab).
background_handler (Background): Instance of Background class
which can access the most recently collected background.
This background is later subtracted from the raw data as
dark noise correction.
spectrometer (Spectrometer): Spectrometer/Triax hardware class
which grants functionality to control the triax
spectrometer. Needed to obtain e.g. wavenumber axis etc.
info_queue (Queue): Multiprocessing queue object which is used
to transfer/hand over information to lineEdits on GUI.
Contains (if applicable) scan index, delay index, interleave
index, values for statistics groupBox etc.
saver (SaveData): Object that manages saving of data (including
counts, weights, probe wavenumber axis etc.) into their
respective directories. If None is passed, no data is going
to be saved. If the raw data checkbox was checked on the GUI
the savers raw data attribute is set to True and the raw
data is saved automatically. Defaults to None.
"""
def __init__(
self,
widget_pyqtgraph,
adc: ADC,
prl: PRL,
background_handler: Background,
spectrometer: Spectrometer,
info_queue: Queue,
saver: SaveData = None,
):
# Initialise Queues
self.acq_queue = (
Queue()
) # for data flow from acquisition to primary data processing
self.plot_queue = Queue() # for data flow from primary processing to plotting
self.processing_queue = (
Queue()
) # for data flow from primary to secondary processing
# in this experiment: relevant only for GUI
if saver:
saver.save_other(background_handler.load_background(), "background")
self.acquisition = Acquisition(adc, spectrometer, self.acq_queue)
self.primary_processing = PrimaryProcessing(
self.acq_queue,
self.plot_queue,
info_queue,
self.processing_queue,
adc.pixel_idx,
adc.probe_pixel_idx,
adc.reference_pixel_idx,
prl,
background_handler,
saver,
)
self.secondary_processing = SecondaryProcessing(self.processing_queue)
# self.plotting = Plotting(mpl_widget, adc, self.plot_queue)
self.plotting = PyqtPlotting(widget_pyqtgraph, adc, self.plot_queue)
[docs] def start(self):
# We start the the threading and processes
# in this order to ensure that the last
# called process/thread is ready for
# operation when the first are called
# for the first time
self.plotting.threadpool.start(self.plotting.work)
self.primary_processing.start()
self.secondary_processing.start()
self.acquisition.start()
[docs]class Acquisition(threading.Thread):
"""
Acquisition class (python multithreaded). This is where the actual
experiment is conducted.
This class is used to control the hardware devices required for the
experiment. The sole purpose of it is to collect the data according
to the parameters specified for the experiment by moving the delay
stages, opening and closing shutters etc.
A dictionary which will contain data and other information (scan
index etc.) is instantiated here. The acquisition class passes the
collected information (raw data, scan index etc.) to the primary
processing class.
Note:
**No data processing beyond what is required to conduct the
experiment should be implemented in this class.** The rationale
behind this is to minimize down time/ maximize laser time. Data
processing costs computation time and will, generally speaking,
slow down the measurement process because the computer is busy
while the rest of the hardware is idle. If implemented correctly
the data processing could be carried out while the data
acquisition is waiting for all data to become available. But
even in this scenario the problem that the data processing takes
longer than the acquisition time can occur and is thus best
avoided through parallelisation.
The reason why this class is a child of the threading module instead
of the multiprocessing module is that to use multiprocessing all
objects passed to the function must be picklable. This is not the
case for some of the objects interfacing with the hardware (e.g.
ADC). In an ideal scenario the acquisition too, would run in its own
process seperated from the GUI thread but this would only be
possible with major restructuring of the software.
Args:
adc (ADC): Analog to digital converter hardware object which is
used to communicate with and read data from the ADC.
spectrometer (Spectrometer): Spectrometer/Triax hardware class
which grants functionality to control the triax
spectrometer. Needed to obtain e.g. wavenumber axis etc.
acq_queue (Queue): Multiprocessing queue object that the
acquisition thread uses to pass data to the primary
processing process.
"""
def __init__(self, adc: ADC, spectrometer: Spectrometer, acq_queue):
threading.Thread.__init__(self)
self.adc = adc
self.spectrometer = spectrometer
# Multiprocessing Queue
self.acq_queue = acq_queue
# Initialize scan index
self.scan_idx = 0
# Create dictionary that will hold data that is passed
# to other queues
self.data_container = {}
# Multiprocessing event to stop experiment
self.exit = threading.Event()
[docs] def run(self):
# Read first set of data outside of loop
# see docstring of analog_digital_converter.py
self.adc.read()
# An exit/stop flag is used to indicate whether
# the experiment should stop excecuting
# setting exit stops the loop. When "Stop"
# button is pressed widget_main_window.py
# executes shutdown method
while not self.exit.is_set():
# Start acquisition without waiting
# for data to become available
self.adc.start()
# Put relevant information into data container
self.data_container["data"] = self.adc.data
self.data_container["scan index"] = self.scan_idx
self.data_container["probe axis"] = self.spectrometer.wn_axis
# Give data of prior acquisition to queue
# Hand over raw data because standard
# deviations needs to be calculated in
# the other process
self.acq_queue.put(self.data_container.copy())
# Update scan idx
self.scan_idx += 1
# Read data with given parameters
# (i.e. samples to acquire usually from GUI)
self.adc.read()
# Hand over the last collected data set
self.data_container["data"] = self.adc.data.copy()
self.data_container["scan index"] = self.scan_idx
self.data_container["probe axis"] = self.spectrometer.wn_axis.copy()
self.acq_queue.put(self.data_container.copy())
self.acq_queue.put("stop")
[docs] def shutdown(self):
# Setting exit will stop the loop
# within run
self.exit.set()
[docs]class PrimaryProcessing(multiprocessing.Process):
"""
Primary processing class (python multiprocess). This class' purpose
is to process the raw data to a state where it can be saved onto the
hard drive as npy (binary) files. Generally, background corrected
intensities are saved. This is different from all other experiments,
where background corrected transmissions (normalized intensities)
are saved. Saving raw data yields npy files which contain what the
ADC collects at each laser shot (intensities/voltages/ADC counts).
After this class was executed the important data is saved. Even if
the processes later on break, the data is secured and can be
analysed in post processing. The primary processed data is handed
over to the secondary processing process.
Args:
acq_queue (Queue): Multiprocessing queue object that the
acquisition thread uses to pass data to the primary
processing process.
processing_queue (Queue): Multiprocessing queue object that the
primary processing process uses to pass data to the secondary
processing process.
plot_queue (Queue): Multiprocessing queue object that the
primary processing process uses to pass data to the plot
thread.
info_queue (Queue): Multiprocessing queue object which is used
to transfer/hand over information to lineEdits on GUI.
Contains (if applicable) scan index, delay index, interleave
index, values for statistics groupBox etc.
pixel_idx (ndarray): Array that contains the indices of the rows
in the ADCs' data that correspond to pixel input channels.
These are specified in the "analog input configuration.json"
for each laboratory and can be easily accessed with the
attribute "pixel_idx" of the ADC.
* shape: 1D
* E.g.: (64) or (128)
probe_pixel_idx (ndarray): Array that contains the indices of
the rows in the ADCs' data that correspond to *probe* pixel
input channels. These are specified in the "analog input
configuration.json" for each laboratory and can be easily
accessed with the attribute "probe_pixel_idx" of the ADC. It
is highly relevant that the order the pixels are listed in
this array match the order of the array in the
reference_pixel_idx argument. This means that if reference
pixel 3 is listed first in the other array here probe pixel
3 needs to be listed first as well and so on. Otherwise the
intensities on the probe array are not going to be
normalized correctly. For the plotting to work correctly the
pixels also need to be listed in the order of the wavenumber
axis array of the spectrometer.
* shape: 1D
* E.g.: (32) or (64)
reference_pixel_idx (ndarray): Array that contains the indices
of the rows in the ADCs' data that correspond to *reference*
pixel input channels. These are specified in the "analog
input configuration.json" for each laboratory and can be
easily accessed with the attribute "reference_pixel_idx" of
the ADC. It is highly relevant that the order the pixels are
listed in this array match the order of the array in the
probe_pixel_idx argument. This means that if probe pixel 10
is listed first in the other array here reference pixel 10
needs to be listed first as well and so on. Otherwise the
intensities on the probe array are not going to be
normalized correctly. For the plotting to work correctly the
pixels also need to be listed in the order of the wavenumber
axis array of the spectrometer.
* shape: 1D
* E.g.: (32) or (64)
index_dict (dict): Dictionary that maps the names of the input
channels to their corresponding row in the ADCs' data as
they are specified in the "analog input configuration.json"
for each laboratory. I.e.: It contains the information which
entries of the ADCs' data array belong choppers, wobblers
etc. This dictionary can be easily accessed with the
attribute "index_dict" of the ADC.
prl (PRL): Pixel response linearization object which grants
the functionality to linearize raw data according to the
linearization parameters specified in the corresponding
*pixel_linearization_fit_parameters.json* file (for each
lab).
background_handler (Background): Instance of Background class
which can access the most recently collected background.
This background is later subtracted from the raw data as
dark noise correction.
saver (SaveData): Object that manages saving of data (including
counts, weights, probe wavenumber axis etc.) into their
respective directories. If None is passed, no data is going
to be saved. If the raw data checkbox was checked on the GUI
the savers raw data attribute is set to True and the raw
data is saved automatically. Defaults to None.
"""
def __init__(
self,
acq_queue: Queue,
plot_queue: Queue,
info_queue: Queue,
processing_queue: Queue,
pixel_idx: ndarray,
probe_pixel_idx: ndarray,
reference_pixel_idx: ndarray,
prl: PRL,
background_handler: Background,
saver: SaveData = None,
):
super(multiprocessing.Process, self).__init__()
# Multiprocessing Queue
# Gets data from acquisition class
self.acq_queue = acq_queue
# Gives data to secondary processing class
self.plot_queue = plot_queue
# Give experimental status and statistics to
# GUI
self.info_queue = info_queue
# Processing queue that is not used
# for any relevant operations
# It is just needed to end the experiment
# on the GUI
self.processing_queue = processing_queue
self.prl = prl
self.saver = saver
# Pixel index is an array which tells us which
# entries in our adc data are pixels
self.pixel_idx = pixel_idx
self.probe_pixel_idx = probe_pixel_idx
self.ref_pixel_idx = reference_pixel_idx
# Load background data from file
self.background = background_handler.load_background()
# Initialize array that holds both
# temporary and averaged data
# In this case the data is average
# intensity for each pixel
# By convention the averaged is the
# 0 th entry of the 0th axis of the array
# and the temp data is the 1st entry of the 0th
# axis of the array.
# dimensions: 2 = (current average scan, last single scan)
# dimensions: (2, n_probe_pixels)
self.data = np.zeros((2, self.pixel_idx.size))
self.counts = np.zeros(self.data.shape)
# Set temp part of the count array to be 1 constantly
# Is used for weighting lateron
self.counts[1] = 1
[docs] def run(self):
while True:
# Get data / information from acquisition process
data_container = self.acq_queue.get()
if type(data_container) == str:
if data_container == "stop":
break
# Write data in dictionary into variable
raw_data = data_container["data"]
scan_idx = data_container["scan index"]
# Subtract background from raw data
# of pixels and pixels only
background_corrected_data = (
raw_data[self.pixel_idx] - self.background[self.pixel_idx, np.newaxis]
)
# Linearize response of Pixels (and pixels only)
# The other data is disregarded in this routine.
intensities = self.prl.linearize(background_corrected_data)
# Calculate average intensity for each pixel
# Create / Average "temp" data by averaging raw data.
self.data[1] = np.average(intensities, axis=1)
# ----------------------------------------
# Calculate information for plotting and GUI
# For this special case it would be overkill
# to create a secondary processing class
# only to calculate the standard deviation
# We first calculate everything that is required
# for plotting and pass it to the plot queue because
# plotting can also cost time. Only then we
# calculate the values that we want to display on
# line edits on the GUI
# Calculate the standard deviation of the
# intensities
std_intensities = np.std(intensities, axis=1, ddof=1)
# Calculate transmission/ relative intensity
# (probe intensity / reference intensity)
transmission = (
intensities[self.probe_pixel_idx] / intensities[self.ref_pixel_idx]
)
# Calculate the pseudo difference signal
# (shot to shot) and its statistical information
_, amp_signal, std_signal, s2s_avg_std_signal = dp.shot_to_shot_signal(
transmission
)
# Add everything to data container
# and give to plot queue
data_container["average intensity"] = self.data[1]
data_container["std intensity"] = std_intensities
data_container["std s2s signal"] = std_signal
# Average standard deviation of pseudo signal over all wavenumbers
data_container[
"s2s signal average std"
] = s2s_avg_std_signal # s2s: shot to shot
# Relevant only for GUI
self.plot_queue.put(data_container)
# ----------------------------------------
# Calculate everything that is needed for statistical information
# on GUI and add it to data container and give it to info queue
# In this case there is only one state so we just pass
# the relative intensity (transmission)
data_container["mean state intensity"] = np.average(transmission, axis=1)
data_container["mean state std"] = np.std(transmission, axis=1, ddof=1)
# Average s2s signal amplitude
data_container["s2s signal amplitude"] = amp_signal # s2s: shot to shot
self.info_queue.put(data_container)
# ----------------------------------------
# Save data (if specified)
if self.saver:
# Create / Average "averaged" data by using weighted average
# between the temp data (weight = 1) and the already
# existing average data (weight = scan_idx)
self.data[0] = np.average(self.data, axis=0, weights=self.counts)
# Use saver class to save data
self.saver.save_scan(self.data[1], scan_idx)
self.saver.save_avg(self.data[0])
if self.saver.raw_data:
self.saver.save_raw_data(raw_data, scan_idx)
# Update counts
self.counts[0] = scan_idx
# Once stop signal was received the function breaks out of the loop.
# Now we need to tell the plotting and updating of info on GUI to stop.
self.plot_queue.put("stop")
self.info_queue.put("stop")
self.processing_queue.put("stop")
[docs]class SecondaryProcessing(multiprocessing.Process):
"""
In the Show Spectrum experiment this process is not used.
Secondary processing class (python multiprocess). This class is used
to process the data from the primary processing class such that it
can be displayed on the GUI within plots and lineEdits. This
generally implies (if applicable):
* calculation of signals
* calculation of statistics like standard deviation of
intensities and shot to shot standard deviation of signal
* interpolation for 2D / heatmap plots (see comments in code why
this is necessary)
* Fourier transform and phasing for time domain data
This data is handed over to the plotting thread.
Note:
The feature of saving figures/plots to the hard drive should be
implemented here if it is needed.
Args:
processing_queue (Queue): Multiprocessing queue object that the
primary processing process uses to pass data to the secondary
processing process.
"""
def __init__(self, processing_queue: Queue):
super(multiprocessing.Process, self).__init__()
# Multiprocessing Queue
# Gets data from acquisition class
self.processing_queue = processing_queue
[docs] def run(self):
while True:
# Get data / information from acquisition process
data_container = self.processing_queue.get()
# When data_container is a numpy array the
# first if is false. If we would
# exclude the first statement the code
# would raise an error in the "normal use mode".
# Directly asking if "stop" is contained raises
# an error. Try:
# a = np.arange(20)
# if a == "stop":
# print("blabla")
if type(data_container) == str:
if data_container == "stop":
break
[docs]class PyqtPlotting:
"""
Pyqt Plotting class (Qt multithreaded). This class is necessary for
displaying plots on the GUI. PyQtGraph is used as the plotting
engine. Generally the plots are set up first (type of plot, layout,
title etc.). On the first run, the plots are drawn for the first
time. Then the plots are updated every iteration. We update the same
plot references every time to make it more efficient.
Args:
widget_pyqtgraph (QWidget): WidgetPyqtgraph object on which the
plots are going to be displayed. Has methods for plot
manipulation (i.e. removal of plots, autoscale).
adc (ADC): Analog to digital converter hardware object which is
used to communicate with and read data from the ADC.
plot_queue (Queue): Multiprocessing queue object that the
primary processing process uses to pass data to the plot
thread.
"""
[docs] class Signals(QObject):
new_data = pyqtSignal(dict)
def __init__(self, widget_pyqtgraph, adc: ADC, plot_queue):
# Assign attributes
self.adc = adc
self.widget_pyqtgraph = widget_pyqtgraph
self.graphics_layout = widget_pyqtgraph.graphics_layout
self.plot_queue = plot_queue
# Setup signals and threadpool
self.threadpool = QThreadPool()
self.signals = self.Signals()
# Clear old plots
self.widget_pyqtgraph.remove_plots()
# Setup plot that displays intensities
self.widget_pyqtgraph.plots["intensities"] = self.graphics_layout.addPlot(
row=0, col=0, rowspan=2
)
self.widget_pyqtgraph.plots["intensities"].setTitle("Average intensities")
self.widget_pyqtgraph.plots["intensities"].setLabel(
"bottom", "wavenumber [cm<sup>-1</sup>]"
)
self.widget_pyqtgraph.plots["intensities"].setLabel("left", "intensity [a.u.]")
self.widget_pyqtgraph.set_style(self.widget_pyqtgraph.plots["intensities"])
# Setup plot that displays standard deviation of "pseudo" signal
# Meaning the difference spectrum actually does not exists here (show spectrum)
# because the chopper not running.
# We expect there to be 0 signal and if the beam are properly
# aligned on the detector the noise of this 0 line should be 0 too.
# Why this works is not entirely clear.
# Make the title of std signal so that the Average standard deviation
# of signal over all wavenumbers is displayed. HTML is
# used because setTitle works with it.
self.std_s2s_signal_title = """
<span style="color: #000000; font-size: 12pt;">
Standard deviation of pump-probe 'signal'</span>
<span style="color: #E93610 ; font-size: 30pt;">
{:0.3e} OD</span></div>
"""
self.widget_pyqtgraph.plots["std s2s signal"] = self.graphics_layout.addPlot(
row=2, col=0
)
self.widget_pyqtgraph.plots["std s2s signal"].setTitle(
self.std_s2s_signal_title
)
self.widget_pyqtgraph.plots["std s2s signal"].setLabel(
"bottom", "wavenumber [cm<sup>-1</sup>]"
)
self.widget_pyqtgraph.plots["std s2s signal"].setLabel(
"left", "difference signal [OD]"
)
self.widget_pyqtgraph.set_style(self.widget_pyqtgraph.plots["std s2s signal"])
# Create dictionary that holds reference to lines etc.
self.plot_ref = {}
# Connect signal that data has arrived to update the plot
self.signals.new_data.connect(
lambda data_container: self.update_plot(data_container)
)
# Start loop that gets data from queue in Qt Thread
self.work = Worker(self.run)
[docs] def run(self):
while True:
data_container = self.plot_queue.get()
if type(data_container) == str: # ? Why ask this?
if data_container == "stop":
break
self.signals.new_data.emit(data_container)
[docs] def update_plot(self, data_container):
# Plot for the first time to get line references
probe_axis = data_container["probe axis"]
avg_intensities = data_container["average intensity"]
std_intensities = data_container["std intensity"]
std_signal = data_container["std s2s signal"]
s2s_avg_std_signal = data_container["s2s signal average std"]
# Update the title of the std_signal plot to display the
# Average standard deviation of signal over all wavenumbers
self.widget_pyqtgraph.plots["std s2s signal"].setTitle(
self.std_s2s_signal_title.format(s2s_avg_std_signal)
)
if self.plot_ref:
# Update intensity error bars for probe array
self.plot_ref["probe error bars"].setData(
x=probe_axis,
y=avg_intensities[self.adc.probe_pixel_idx],
height=5 * std_intensities[self.adc.probe_pixel_idx],
)
# Update intensity error bars for reference array
self.plot_ref["ref error bars"].setData(
x=probe_axis,
y=avg_intensities[self.adc.reference_pixel_idx],
height=5 * std_intensities[self.adc.reference_pixel_idx],
)
# Update intensities
self.plot_ref["probe intensities"].setData(
x=probe_axis, y=avg_intensities[self.adc.probe_pixel_idx]
)
self.plot_ref["ref intensities"].setData(
x=probe_axis, y=avg_intensities[self.adc.reference_pixel_idx]
)
# Plot standard deviation of pseudo signal
self.plot_ref["std s2s signal"].setData(x=probe_axis, y=std_signal)
else:
# First time plotting
probe_pen = pg.mkPen(color="#1f77b4", width=2.5, style=QtCore.Qt.SolidLine)
reference_pen = pg.mkPen(
color="#ff7f0e", width=2.5, style=QtCore.Qt.SolidLine
)
std_intensity_pen = pg.mkPen(
color="#7f7f7f", width=1.5, style=QtCore.Qt.SolidLine
) # ? dashed lines?
std_signal_pen = pg.mkPen(
color="#2ca02c", width=2.5, style=QtCore.Qt.SolidLine
)
# Create intensity error bars for probe array
self.plot_ref["probe error bars"] = pg.ErrorBarItem(
x=probe_axis,
y=avg_intensities[self.adc.probe_pixel_idx],
height=5 * std_intensities[self.adc.probe_pixel_idx],
beam=0.3,
pen=std_intensity_pen,
)
self.widget_pyqtgraph.plots["intensities"].addItem(
self.plot_ref["probe error bars"]
)
# Create intensity error bars for reference array
self.plot_ref["ref error bars"] = pg.ErrorBarItem(
x=probe_axis,
y=avg_intensities[self.adc.reference_pixel_idx],
height=5 * std_intensities[self.adc.reference_pixel_idx],
beam=0.3,
pen=std_intensity_pen,
)
self.widget_pyqtgraph.plots["intensities"].addItem(
self.plot_ref["ref error bars"]
)
# Plot intensities
self.plot_ref["probe intensities"] = self.widget_pyqtgraph.plots[
"intensities"
].plot(
x=probe_axis,
y=avg_intensities[self.adc.probe_pixel_idx],
name="Average intensities on probe array",
pen=probe_pen,
)
self.plot_ref["ref intensities"] = self.widget_pyqtgraph.plots[
"intensities"
].plot(
x=probe_axis,
y=avg_intensities[self.adc.reference_pixel_idx],
name="Average intensities on reference array",
pen=reference_pen,
)
# Plot standard deviation of pseudo signal
self.plot_ref["std s2s signal"] = self.widget_pyqtgraph.plots[
"std s2s signal"
].plot(
x=probe_axis,
y=std_signal,
name="Standard deviation of pump-probe signal",
pen=std_signal_pen,
)
self.widget_pyqtgraph.disable_autoscale()
[docs]class MplPlotting(threading.Thread):
"""
Deprecated. Works but is too slow for our purposes.
"""
#! Deprecated but technically works
#! Turned out that matplotlib is too
#! slow for our purposes
def __init__(
self, mpl_widget, adc: ADC, plot_queue,
):
threading.Thread.__init__(self)
# Add attributes
self.adc = adc
self.mpl_widget = mpl_widget
self.canvas = self.mpl_widget.canvas
self.fig = self.canvas.figure
# Multiprocessing Queue
self.plot_queue = plot_queue
# Clear old plots (clear figure)
self.fig.clf()
# Create list that will hold references to axes
self.axes = []
# Create dictionary that holds references
# to all plotted lines
self._plot_ref = {}
# Setup Plotting ----------
self.axes.append(
self.fig.add_subplot(
211,
xlabel="wavenumber [cm$^{-1}$]",
ylabel="intensity [a.u.]",
# title="Average intensities"
)
)
self.axes.append(
self.fig.add_subplot(
212,
xlabel="wavenumber [cm$^{-1}$]",
ylabel="intensity [a.u.]",
# title="Standard deviation of intensities"
)
)
# Activate grid on all axes
[axis.grid() for axis in self.axes]
# -------------------------
[docs] def run(self):
while True:
data_container = self.plot_queue.get()
if type(data_container) == str:
if data_container == "stop":
break
# Assign data from queue to
probe_axis = data_container["probe axis"]
avg_intensities = data_container["average intensity"]
std_intensities = data_container["std intensity"]
# self.fig.suptitle("Show Spectrum\n Scan: {}".format(data_container["scan index"]))
if not self._plot_ref:
# First time we have no plot reference, so do a normal plot.
# .plot returns a list of line <reference>s, as we're
# only getting one we can take the first element.
# Plot average intensities in the upper subplot
#! Figure out how to quickly ERRORBAR update plot
#! Or check if "slow" plotting is fast is enough
# self._plot_ref["avg probe intensity"] = self.axes[0].errorbar(self.spectrometer.wn_axis,
# avg_intensities[self.adc.probe_pixel_idx],
# yerr = std_intensities[self.adc.probe_pixel_idx],
# label = "Average intensities on probe array")
# self._plot_ref["avg ref intensity"] = self.axes[0].errorbar(self.spectrometer.wn_axis,
# avg_intensities[self.adc.reference_pixel_idx],
# yerr = std_intensities[self.adc.reference_pixel_idx],
# label = "Average intensities on reference array")
(self._plot_ref["avg probe intensity"],) = self.axes[0].plot(
probe_axis,
avg_intensities[self.adc.probe_pixel_idx],
label="Average intensities on probe array",
)
(self._plot_ref["avg ref intensity"],) = self.axes[0].plot(
probe_axis,
avg_intensities[self.adc.reference_pixel_idx],
label="Average intensities on reference array",
)
(self._plot_ref["std probe intensity"],) = self.axes[1].plot(
probe_axis,
std_intensities, # [self.adc.probe_pixel_idx],
label="Standard deviation of intensities on probe array",
)
# self._plot_ref["std ref intensity"], = self.axes[1].plot(
# probe_axis,
# std_intensities[self.adc.reference_pixel_idx],
# label = "Standard deviation of intensities on reference array"
# )
[
axis.legend(
bbox_to_anchor=(0.0, 1.02, 1.0, 0.102),
loc="lower left",
ncol=2,
mode="expand",
borderaxespad=0.0,
)
for axis in self.axes
]
self.fig.tight_layout()
else:
# We have a reference, we can use it to update the data for that line.
self._plot_ref["avg probe intensity"].set_data(
probe_axis, avg_intensities[self.adc.probe_pixel_idx]
)
self._plot_ref["avg ref intensity"].set_data(
probe_axis, avg_intensities[self.adc.reference_pixel_idx]
)
self._plot_ref["std probe intensity"].set_data(
probe_axis, std_intensities
) # [self.adc.probe_pixel_idx])
# self._plot_ref["std ref intensity"].set_data(probe_axis, std_intensities[self.adc.reference_pixel_idx])
# Auto scale all graphs, if activated
if self.mpl_widget.auto_scale:
[axis.relim() for axis in self.axes]
[axis.autoscale() for axis in self.axes]
# Trigger the canvas to update and redraw.
self.canvas.draw_idle()
if __name__ == "__main__":
# ADC
adc_dev_name = "Dev1"
input_config_path = "hardware_config_files/H-Lab analog input configuration.json"
samps_to_acq = 12000 # samples to acquire
trig_src = "PFI4" # input trigger source
lsr_freq = 3000 # laser frequency
# Spectrometer
spectrometer_port = "COM1" #! ---------
turret = "Turret1" # ? Ask Erhan which one is appropriate
grating = 0 #!!!!!!!
# Initialise pixel linearisation
pixel_linearise_response_path = (
"hardware_config_files/pixel_no_linearization_fit_parameters_volts.json"
)
prl = dp.PixelResponseLinearization(pixel_linearise_response_path)
# Save path
path = r"C:\Users\Public\Music\Sample Music"
file_name = "JAJA"
username = "Grüner Powerranger"
saver = SaveData(path, file_name, username)
class MplCanvas(FigureCanvasQTAgg):
def __init__(self, parent=None, plot_queue=None, width=5, height=4, dpi=100):
fig = Figure(figsize=(width, height), dpi=dpi)
super(MplCanvas, self).__init__(fig)
class MainWindow(QtWidgets.QMainWindow):
def __init__(self, adc, prl, spectrometer, saver, *args, **kwargs):
super(MainWindow, self).__init__(*args, **kwargs)
# self.threadpool = QThreadPool()
# self.acq_queue = Queue()
self.info_queue = Queue()
self.canvas = MplCanvas(
self, plot_queue=self.plot_queue, width=5, height=4, dpi=100
)
# Create toolbar, passing canvas as first parament, parent (self, the MainWindow) as second.
toolbar = NavigationToolbar(self.canvas, self)
layout = QtWidgets.QVBoxLayout()
layout.addWidget(toolbar)
layout.addWidget(self.canvas)
# Create a placeholder widget to hold our toolbar and canvas.
widget = QtWidgets.QWidget()
widget.setLayout(layout)
self.setCentralWidget(widget)
# self.acq_process = Acquisition(self.acq_queue)
# self.data_processing_process = DataProcessing(self.acq_queue, self.plot_queue)
# self.plotting = Plotting(self.canvas, self.plot_queue)
self.show_spectrum = ShowSpectrum(
self.canvas, adc, prl, spectrometer, self.info_queue, saver
)
self.show_spectrum.primary_processing.daemon = True
self.show_spectrum.acquisition.daemon = True
self.show_spectrum.plotting.daemon = True
self.show_spectrum.acquisition.start()
self.show_spectrum.primary_processing.start()
self.show_spectrum.plotting.start()
self.show()
with ADC(
adc_dev_name, input_config_path, samps_to_acq, trig_src, lsr_freq
) as adc, Spectrometer(turret, port=spectrometer_port, init=True) as spectrometer:
app = QtWidgets.QApplication(sys.argv)
w = MainWindow(adc, prl, spectrometer, saver)
app.exec_()