Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
No results found
Show changes
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 478 additions and 3900 deletions
dstat_interface/core/experiments/cal.py deleted 100755 → 0
View file @ 86048afb
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# DStat Interface - An interface for the open hardware DStat potentiostat
# Copyright (C) 2017 Michael D. M. Dryden -
# Wheeler Microfluidics Laboratory <http://microfluidics.utoronto.ca>
#
#
# This program 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.
#
# This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
import time
import struct
import logging
logger = logging.getLogger(__name__)
import serial
from ..errors import InputError, VarError
from ..dstat import state
from ..experiments.experiment_template import Experiment, dstat_logger
def measure_offset(time):
gain_trim_table = [None, 'r100_trim', 'r3k_trim', 'r30k_trim', 'r300k_trim',
'r3M_trim', 'r30M_trim', 'r100M_trim']
parameters = {}
parameters['time'] = time
gain_offset = {}
for i in range(1,8):
parameters['gain'] = i
state.ser.start_exp(CALExp(parameters))
logger.info("measure_offset: %s", state.ser.get_proc(block=True))
gain_offset[gain_trim_table[i]] = state.ser.get_data(block=True)
return gain_offset
class CALExp(Experiment):
id = 'cal'
"""Offset calibration experiment"""
def __init__(self, parameters):
self.parameters = parameters
self.databytes = 8
self.scan = 0
self.data = []
self.commands = ["EA2 3 1 ", "EG"]
self.commands[1] += str(self.parameters['gain'])
self.commands[1] += " "
self.commands[1] += "0 "
self.commands.append(
("ER1 0", ["32768", str(self.parameters['time'])])
)
def serial_handler(self):
"""Handles incoming serial transmissions from DStat. Returns False
if stop button pressed and sends abort signal to instrument. Sends
data to self.data_pipe as result of self.data_handler).
"""
try:
while True:
if self.ctrl_pipe.poll():
input = self.ctrl_pipe.recv()
logger.debug("serial_handler: %s", input)
if input == ('a' or "DISCONNECT"):
self.serial.write('a')
logger.info("serial_handler: ABORT pressed!")
return False
for line in self.serial:
if self.ctrl_pipe.poll():
if self.ctrl_pipe.recv() == 'a':
self.serial.write('a')
logger.info("serial_handler: ABORT pressed!")
return False
if line.startswith('B'):
self.data.append(self.data_handler(
self.serial.read(size=self.databytes)))
elif line.lstrip().startswith("#"):
dstat_logger.info(line.lstrip().rstrip())
elif line.lstrip().startswith("@DONE"):
dstat_logger.debug(line.lstrip().rstrip())
self.serial.flushInput()
self.experiment_done()
return True
except serial.SerialException:
return False
def data_handler(self, data):
"""Takes data_input as tuple -- (scan, data).
Returns:
current
"""
seconds, milliseconds, current = struct.unpack('<HHl', data)
return current
def experiment_done(self):
"""Averages data points
"""
try:
sum = 0
self.data[0] = 0 # Skip first point
except IndexError:
return
for i in self.data:
sum += i
sum /= len(self.data)
if (sum > 32767):
sum = 32767
elif (sum < -32768):
sum = -32768
self.data_pipe.send(sum)
\ No newline at end of file
dstat_interface/core/experiments/chronoamp.py
View file @ d5704e4f
import time
import struct
import numpy as np
import serial
from ..interface.plot import mean, plotSpectrum, findBounds
from .experiment_template import PlotBox, Experiment, exp_logger
class ChronoampBox(PlotBox):
def setup(self):
self.plot_format = {
'current_time': {'xlabel': "Time (s)",
'ylabel': "Current (A)"
}
}
def format_plots(self):
"""
Creates and formats subplots needed. Overrides superclass.
"""
self.subplots = {'current_time': self.figure.add_subplot(111)}
for key, subplot in self.subplots.items():
subplot.ticklabel_format(style='sci', scilimits=(0, 3),
useOffset=False, axis='y')
subplot.plot([],[])
subplot.set_xlabel(self.plot_format[key]['xlabel'])
subplot.set_ylabel(self.plot_format[key]['ylabel'])
class Chronoamp(Experiment):
id = 'cae'
"""Chronoamperometry experiment"""
def setup(self):
self.datatype = "linearData"
self.datalength = 2
self.databytes = 8
self.data = {'current_time' : [([],[])]}
self.columns = ['Time (s)', 'Current (A)']
self.total_time = sum(self.parameters['time'])
self.plotlims = {'current_time': {'xlims': (0, self.total_time)}
}
self.commands.append(
("ER" + str(len(self.parameters['potential'])) + " 0 ", [])
)
for i in self.parameters['potential']:
self.commands[-1][1].append(str(int(i*(65536./3000)+32768)))
for i in self.parameters['time']:
self.commands[-1][1].append(str(i))
plot = ChronoampBox(['current_time'])
plot.setlims('current_time', **self.plotlims['current_time'])
self.plots.append(plot)
def data_handler(self, data_input):
"""Overrides Experiment method to not convert x axis to mV."""
scan, data = data_input
# 2*uint16 + int32
seconds, milliseconds, current = struct.unpack('<HHl', data)
return (scan, (
seconds+milliseconds/1000.,
(current+self.gain_trim)*(1.5/self.gain/8388607)
)
)
def store_data(self, incoming, newline):
"""Stores data in data attribute. Should not be called from subprocess.
Can be overriden for custom experiments."""
line, data = incoming
if newline is True:
self.data['current_time'].append(deepcopy(self.line_data))
for i, item in enumerate(self.data['current_time'][line]):
item.append(data[i])
from datetime import datetime
from typing import Any
from ..dstat.experiment_handler import ExperimentHandler
from ..dstat.experiment_process import ExperimentProcess
from ..dstat.utility import abs_mv_to_dac, rel_mv_to_dac, param_test_uint16, param_test_non_zero_uint8
from ..experiments.experiment_container import ExperimentContainer
class CAExperimentHandler(ExperimentHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.data['time'] = []
def data_handler(self, data_input: tuple[datetime, int, bytes]):
unpacked: tuple[int, int, int]
date, scan, data = data_input
unpacked = self.struct.unpack(data)
self.data['scan'].append(scan)
self.data['timestamp'].append(date)
for n, i in enumerate(unpacked):
try:
self.data[self.data_cols[n]].append(self.data_convert[n](i))
except TypeError: # If no converter
self.data[self.data_cols[n]].append(i)
self.data['time'].append(self.data['time_s'][-1] + self.data['time_ms'][-1])
class CAExperimentContainer(ExperimentContainer):
experiment_id = 'ca'
display_name = 'Chronoamperometry'
process = ExperimentProcess
handler = CAExperimentHandler
data_bytes = 0
def __init__(self, params: dict[str, Any], mux: int = 1):
super().__init__(params, mux=mux)
self.data_cols = ['time_s', 'time_ms', 'current']
self.data_format = 'smA'
self.cmd_str += [('ER{n_steps} 0 ', ('voltages', 'times'))]
self.param_input |= {'times': list, 'voltages': list}
self.param_input_display_names |= {'times': 'Time (s)', 'voltages': 'Voltage (mV)'}
self.param_tables = {'Steps:': ['times', 'voltages']}
self.param_input_limits |= {'times': (0, 'time_max'), 'voltages': ('mv_min', 'mv_max'), }
self.param_converters |= {'times': lambda x: [param_test_uint16(i) for i in x],
'voltages': lambda x: [abs_mv_to_dac(i) for i in x], }
self.defaults |= {'times': 0, 'voltages': 0}
if params:
self.progress_end = sum(self.params['times'])
self.params['n_steps'] = len(self.params['times'])
else:
self.progress_max = 0
self.progress_iters = 1
self.data_bytes = self.calculate_data_bytes()
def get_progress(self):
def get_progress(self) -> float:
if self.handler_instance.done:
return 100
try:
return self.data['current_time'][-1][0][-1]/self.total_time
return (1 - abs(
self.progress_end - self.handler_instance.data['time_s'][-1]) / self.progress_end) * 100
except IndexError:
return 0
class PDExp(Chronoamp):
"""Photodiode/PMT experiment"""
id = 'pde'
def setup(self):
self.plots.append(ChronoampBox('current_time'))
self.datatype = "linearData"
self.datalength = 2
self.databytes = 8
self.data = {'current_time' : [([],[])]}
self.columns = ['Time (s)', 'Current (A)']
self.plot_format = {
'current_time' : {
'labels' : self.columns,
'xlims' : (0, int(self.parameters['time']))
}
}
self.total_time = int(self.parameters['time'])
if self.parameters['shutter_true']:
if self.parameters['sync_true']:
self.commands.append("EZ")
self.commands[-1] += str(self.parameters['sync_freq'])
self.commands[-1] += " "
else:
self.commands.append(("E2", []))
command = "ER1 "
params = []
if self.parameters['interlock_true']:
command += "1 "
else:
command += "0 "
if self.parameters['voltage'] == 0: # Special case where V=0
params.append("65535")
else:
params.append(str(int(
65535-(self.parameters['voltage']*(65536./3000))))
)
params.append(str(self.parameters['time']))
self.commands.append((command, params))
if self.parameters['shutter_true']:
if self.parameters['sync_true']:
self.commands.append("Ez")
else:
self.commands.append("E1")
class FT_Box(PlotBox):
def updateline(self, Experiment, line_number):
def search_value(data, target):
for i in range(len(data)):
if data[i] > target:
return i
y = Experiment.data['data'][line_number][1]
x = Experiment.data['data'][line_number][0]
freq = Experiment.parameters['adc_rate_hz']
i = search_value(x, float(Experiment.parameters['fft_start']))
y1 = y[i:]
x1 = x[i:]
avg = mean(y1)
min_index, max_index = findBounds(y1)
y1[min_index] = avg
y1[max_index] = avg
f, Y = plotSpectrum(y1[min_index:max_index],freq)
self.axe1.lines[line_number].set_ydata(Y)
self.axe1.lines[line_number].set_xdata(f)
Experiment.data['ft'] = [(f, Y)]
def changetype(self, Experiment):
"""Change plot type. Set axis labels and x bounds to those stored
in the Experiment instance. Stores class instance in Experiment.
"""
self.axe1.set_xlabel("Freq (Hz)")
self.axe1.set_ylabel("|Y| (A/Hz)")
self.axe1.set_xlim(0, Experiment.parameters['adc_rate_hz']/2)
Experiment.plots['ft'] = self
self.figure.canvas.draw()
\ No newline at end of file
dstat_interface/core/experiments/cv.py deleted 100644 → 0
View file @ 86048afb
import time
import struct
from .experiment_template import PlotBox, Experiment
class CVExp(Experiment):
id = 'cve'
"""Cyclic Voltammetry experiment"""
def setup(self):
self.plotlims['current_voltage']['xlims'] = tuple(
sorted((int(self.parameters['v1']), int(self.parameters['v2'])))
)
super(CVExp, self).setup()
self.datatype = "CVData"
self.xlabel = "Voltage (mV)"
self.ylabel = "Current (A)"
self.datalength = 2 * self.parameters['scans'] # x and y for each scan
self.databytes = 6 # uint16 + int32
self.commands += "E"
self.commands[2] += "C"
self.commands[2] += str(self.parameters['clean_s'])
self.commands[2] += " "
self.commands[2] += str(self.parameters['dep_s'])
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['clean_mV'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['dep_mV'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['v1'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['v2'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['start'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(self.parameters['scans'])
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['slope'])/
self.re_voltage_scale*
(65536./3000)
))
self.commands[2] += " "
def get_progress(self):
return (len(self.data['current_voltage'])-1) / float(self.parameters['scans'])
dstat_interface/core/experiments/experiment_container.py 0 → 100644
View file @ d5704e4f
import logging
import struct
from abc import ABC, abstractmethod, ABCMeta
from typing import Type, Callable, Union, Any, List
from ..dstat.comm import SerialConnection
from ..dstat.experiment_handler import ExperimentHandler
from ..dstat.experiment_process import BaseExperimentProcess
from ..dstat.state import DStatState
from ..dstat.boards import V1_2Board
logger = logging.getLogger(__name__)
state = DStatState()
board = V1_2Board()
class ExperimentContainer(ABC, metaclass=ABCMeta):
dstat_params = {'buffer_false': bool, 'adc_rate': str,
'adc_pga': str, 'gain': int,
'short_true': bool, 'mux_channel': int}
@abstractmethod
def __init__(self, params: dict[str, Any], mux: int = 1):
self.mux = mux
self.data_cols: Union[None, list[str]] = None
self.data_format: str = ''
self.cmd_str: list[str] = ["EA{buffer} {adc_rate:x} {adc_pga:x} ",
"EG{gain} {short_true:d} "]
self.adc_rates = {label: code for label, code
in zip(board.adc_rate_labels, board.adc_rate_codes)}
self.adc_pga = {label: code for label, code
in zip(board.adc_pga_labels, board.adc_pga_codes)}
self.param_input: dict[str, Union[Callable, List]] = {'buffer_false': bool,
'adc_rate': list(self.adc_rates.keys()),
'adc_pga': list(self.adc_pga.keys()),
'gain': int,
'short_true': bool,
}
self.param_input_limits: dict[str, tuple[Union[str, int], Union[str, int]]] = {}
self.param_input_display_names: dict[str, str] = {}
self.param_tables: dict[str, list[str]] = {}
self.param_docs: dict[str, str] = {'buffer_false': 'Disable ADC buffer',
'adc_rate': 'ADC sample rate',
'adc_pga': 'ADC PGA value',
'gain': 'IV converter gain',
'short_true': 'Short RE and CE internally'}
self.param_converters: dict[str, Callable] = {}
self.defaults: dict = {'buffer_false': False, 'short_true': False, 'adc_rate': '60', 'adc_pga': '2',
'gain': 3000}
self.params = params
self.progress_max = 0
self.progress_iters = 0
self.progress_scan = 0
self.handler_instance: Union[None, ExperimentHandler] = None
self.commands: List[str] = []
self.calculate_data_bytes()
if mux > 1:
self.cmd_str += ['EX{mux_channel} ']
self.param_input['mux_channel'] = list(range(mux))
self.param_docs['mux_channel'] = 'Multiplexer Channel'
self.defaults['mux_channel'] = 0
@property
@abstractmethod
def experiment_id(self) -> Type[str]:
raise NotImplementedError
@property
@abstractmethod
def display_name(self) -> Type[str]:
raise NotImplementedError
@property
@abstractmethod
def data_bytes(self) -> Type[ExperimentHandler]:
raise NotImplementedError
@property
@abstractmethod
def handler(self) -> Type[ExperimentHandler]:
raise NotImplementedError
@property
@abstractmethod
def process(self) -> Type[BaseExperimentProcess]:
raise NotImplementedError
@property
def plots(self) -> dict[str, dict[str, str]]:
return {'current': {'x': 'voltage', 'y': 'current'}}
def calculate_data_bytes(self) -> int:
struct_formats = ['H' if i.islower() else 'l' for i in self.data_format]
return struct.Struct(f'<{"".join(struct_formats)}').size
def add_missing_from_defaults(self):
params = self.params
self.params = self.defaults.copy()
self.params |= params
self.commands = []
for i in self.cmd_str:
try:
cmd, params = i
param_list = []
for param in params:
param_list += self.params[param]
self.commands += [(cmd.format(**self.params), param_list)]
except ValueError:
self.commands += [i.format(**self.params)]
def parse_params(self):
self.params |= {key: self.param_converters[key](value) for key, value in self.params.items()
if key in self.param_converters}
self.params['adc_rate'] = self.adc_rates[self.params['adc_rate']]
self.params['adc_pga'] = self.adc_pga[self.params['adc_pga']]
self.params['gain'] = state.board_instance.gain.index(self.params['gain'])
if self.params['buffer_false']:
self.params['buffer'] = '0'
else:
self.params['buffer'] = '2'
logger.info(self.params)
def get_proc(self) -> BaseExperimentProcess:
proc = self.process()
self.parse_params()
for i in self.cmd_str:
proc.parse_command_string(i, self.params)
proc.data_bytes = self.data_bytes
return proc
def start_handler(self, ser: SerialConnection):
self.handler_instance = self.handler(ser=ser, data_cols=self.data_cols, data_format=self.data_format)
self.handler_instance.adc_gain = state.board_instance.gain[self.params['gain']]
self.handler_instance.adc_pga = state.board_instance.adc_pga[
state.board_instance.adc_pga_codes.index(self.params['adc_pga'])]
@abstractmethod
def get_progress(self) -> float:
pass
dstat_interface/core/experiments/experiment_loops.py deleted 100644 → 0
View file @ 86048afb
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# DStat Interface - An interface for the open hardware DStat potentiostat
# Copyright (C) 2014 Michael D. M. Dryden -
# Wheeler Microfluidics Laboratory <http://microfluidics.utoronto.ca>
#
#
# This program 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.
#
# This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
from __future__ import division, absolute_import, print_function, unicode_literals
from ..dstat import state
import logging
try:
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, GObject
except ImportError:
print("ERR: GTK not available")
sys.exit(1)
logger = logging.getLogger(__name__)
class BaseLoop(GObject.GObject):
__gsignals__ = {
b'experiment_done': (GObject.SIGNAL_RUN_FIRST, None, tuple()),
b'progress_update': (GObject.SIGNAL_RUN_FIRST, None, (float,))
}
def __init__(self, experiment, callbacks=None):
GObject.GObject.__init__(self)
self.line = None
self.lastdataline = 0
self.current_exp = experiment
self.experiment_proc = None
for signal, cb in callbacks.items():
try:
self.connect(signal, cb)
except TypeError:
logger.warning("Invalid signal %s", signal)
def run(self):
self.experiment_proc = [
GObject.idle_add(self.experiment_running_data),
GObject.idle_add(self.experiment_running_proc),
GObject.timeout_add(100, self.update_progress)
]
def experiment_running_data(self):
"""Receive data from experiment process and add to
current_exp.data['data].
Run in GTK main loop.
Returns:
True -- when experiment is continuing to keep function in GTK's queue.
False -- when experiment process signals EOFError or IOError to remove
function from GTK's queue.
"""
try:
incoming = state.ser.get_data()
while incoming is not None:
try:
self.line = incoming[0]
if self.line > self.lastdataline:
newline = True
try:
logger.info("running scan_process()")
self.current_exp.scan_process(self.lastdataline)
except AttributeError:
pass
self.lastdataline = self.line
else:
newline = False
self.current_exp.store_data(incoming, newline)
except TypeError:
pass
incoming = state.ser.get_data()
return True
except EOFError as err:
logger.error(err)
self.experiment_done()
return False
except IOError as err:
logger.error(err)
self.experiment_done()
return False
def experiment_running_proc(self):
"""Receive proc signals from experiment process.
Run in GTK main loop.
Returns:
True -- when experiment is continuing to keep function in GTK's queue.
False -- when experiment process signals EOFError or IOError to remove
function from GTK's queue.
"""
try:
ctrl_buffer = state.ser.get_ctrl()
try:
if ctrl_buffer is not None:
self.current_exp.ctrl_loop(ctrl_buffer)
except AttributeError:
pass
proc_buffer = state.ser.get_proc()
if proc_buffer is not None:
if proc_buffer in ["DONE", "SERIAL_ERROR", "ABORT"]:
self.experiment_done()
if proc_buffer == "SERIAL_ERROR":
self.on_serial_disconnect_clicked()
else:
logger.warning("Unrecognized experiment return code: %s",
proc_buffer)
return False
return True
except EOFError as err:
logger.warning("EOFError: %s", err)
self.experiment_done()
return False
except IOError as err:
logger.warning("IOError: %s", err)
self.experiment_done()
return False
def experiment_done(self):
logger.info("Experiment done")
for proc in self.experiment_proc:
GObject.source_remove(proc)
self.current_exp.scan_process(self.lastdataline)
self.current_exp.experiment_done()
self.emit("experiment_done")
def update_progress(self):
try:
progress = self.current_exp.get_progress()
except AttributeError:
progress = -1
self.emit("progress_update", progress)
return True
class PlotLoop(BaseLoop):
def experiment_running_plot(self, force_refresh=False):
"""Plot all data in current_exp.data.
Run in GTK main loop. Always returns True so must be manually
removed from GTK's queue.
"""
if self.line is None:
return True
for plot in self.current_exp.plots:
if (plot.scan_refresh and self.line > self.lastdataline):
while self.line > self.lastline:
# make sure all of last line is added
plot.updateline(self.current_exp, self.lastdataline)
self.lastdataline += 1
plot.updateline(self.current_exp, self.line)
plot.redraw()
else:
while self.line > self.lastdataline:
# make sure all of last line is added
plot.updateline(self.current_exp, self.lastdataline)
self.lastdataline += 1
plot.updateline(self.current_exp, self.line)
if plot.continuous_refresh is True or force_refresh is True:
plot.redraw()
return True
def run(self):
super(PlotLoop, self).run()
self.experiment_proc.append(
GObject.timeout_add(200, self.experiment_running_plot)
)
def experiment_done(self):
logger.info("Experiment done")
for proc in self.experiment_proc:
GObject.source_remove(proc)
self.current_exp.scan_process(self.lastdataline)
self.current_exp.experiment_done()
self.experiment_running_plot(force_refresh=True)
self.emit("experiment_done")
\ No newline at end of file
dstat_interface/core/experiments/experiment_template.py deleted 100755 → 0
View file @ 86048afb
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# DStat Interface - An interface for the open hardware DStat potentiostat
# Copyright (C) 2017 Michael D. M. Dryden -
# Wheeler Microfluidics Laboratory <http://microfluidics.utoronto.ca>
#
#
# This program 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.
#
# This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
import logging
import struct
import time
from collections import OrderedDict
from copy import deepcopy
from datetime import datetime
from math import ceil
try:
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, GObject
except ImportError:
print "ERR: GTK not available"
sys.exit(1)
from matplotlib.figure import Figure
import matplotlib.gridspec as gridspec
from matplotlib.backends.backend_gtk3agg \
import FigureCanvasGTK3Agg as FigureCanvas
from pandas import DataFrame
import seaborn as sns
sns.set(context='paper', style='darkgrid')
import serial
from ..dstat import state, comm
from ..dstat.comm import TransmitError
from . import experiment_loops
logger = logging.getLogger(__name__)
dstat_logger = logging.getLogger("{}.DSTAT".format(comm.__name__))
exp_logger = logging.getLogger("{}.Experiment".format(__name__))
class Experiment(GObject.Object):
"""Store and acquire a potentiostat experiment. Meant to be subclassed
to by different experiment types and not used instanced directly. Subclass
must instantiate self.plotbox as the PlotBox class to use and define id as
a class attribute.
"""
id = None
Loops = experiment_loops.PlotLoop
__gsignals__ = {
b'exp_ready': (GObject.SIGNAL_RUN_FIRST, None, ()),
b'exp_done': (GObject.SIGNAL_RUN_FIRST, None, ()),
}
def __init__(self, parameters):
"""Adds commands for gain and ADC."""
super(Experiment, self).__init__()
self.current_command = None
self.parameters = parameters
self.databytes = 8
self.datapoint = 0
self.scan = 0
self.time = 0
self.plots = []
self.re_voltage_scale = state.board_instance.re_voltage_scale
self.gain = state.board_instance.gain[int(self.parameters['gain'])]
try:
self.gain_trim = int(
state.settings[
state.board_instance.gain_trim[int(self.parameters['gain'])]
][1]
)
except AttributeError:
logger.debug("No gain trim table.")
self.commands = ["EA", "EG"]
if self.parameters['buffer_true']:
self.commands[0] += "2"
else:
self.commands[0] += "0"
self.commands[0] += " {p[adc_rate]} {p[adc_pga]} ".format(
p=self.parameters)
self.commands[1] += "{p[gain]} {p[short_true]:d} ".format(
p=self.parameters)
self.plotlims = {'current_voltage' : {'xlims' : (0, 1)}
}
self.setup()
self.time = [datetime.utcnow()]
def setup_loops(self, callbacks):
self.loops = self.__class__.Loops(self, callbacks)
self.loops.run()
def setup(self):
self.data = OrderedDict(current_voltage=[([], [])])
self.columns = ['Voltage (mV)', 'Current (A)']
# list of scans, tuple of dimensions, list of data
self.line_data = ([], [])
plot = PlotBox(['current_voltage'])
plot.setlims('current_voltage', **self.plotlims['current_voltage'])
self.plots.append(plot)
def write_command(self, cmd, params=None, retry=5):
"""Write command to serial with optional number of retries."""
def get_reply(retries=3):
while True:
reply = self.serial.readline().rstrip()
if reply.startswith('#'):
dstat_logger.info(reply)
elif reply == "":
retries -= 1
if retries <= 0:
raise TransmitError
else:
return reply
n = len(cmd)
if params is not None:
n_params = len(params)
for _ in range(retry):
tries = 5
while True:
time.sleep(0.2)
self.serial.reset_input_buffer()
self.serial.write('!{}\n'.format(n))
time.sleep(.1)
try:
reply = get_reply()
except TransmitError:
if tries <= 0:
continue
tries -= 1
pass
else:
break
if reply != "@ACK {}".format(n):
logger.warning("Expected ACK got: {}".format(reply))
continue
tries = 5
while True:
self.serial.write('{}\n'.format(cmd))
try:
reply = get_reply()
except TransmitError:
if tries <= 0:
continue
tries -= 1
pass
else:
break
if reply != "@RCV {}".format(n):
logger.warning("Expected RCV got: {}".format(reply))
continue
if params is None:
return True
tries = 5
while True:
try:
reply = get_reply()
except TransmitError:
if tries <= 0:
break
tries -= 1
pass
else:
break
if reply != "@RQP {}".format(n_params):
logger.warning("Expected RQP got: {}".format(reply))
continue
tries = 5
for i in params:
while True:
self.serial.write(i + " ")
try:
reply = get_reply()
if reply == "@RCVC {}".format(i):
break
except TransmitError:
if tries <= 0:
continue
tries -= 1
pass
else:
break
return True
return False
def run(self, ser, ctrl_pipe, data_pipe):
"""Execute experiment. Connects and sends handshake signal to DStat
then sends self.commands.
"""
self.serial = ser
self.ctrl_pipe = ctrl_pipe
self.data_pipe = data_pipe
exp_logger.info("Experiment running")
try:
for i in self.commands:
self.current_command = i
status = "DONE"
if isinstance(i, (str, unicode)):
logger.info("Command: %s", i)
if not self.write_command(i):
status = "ABORT"
break
else:
cmd = i[0]
data = i[1]
logger.info("Command: {}".format(cmd))
if not self.write_command(cmd, params=data):
status = "ABORT"
break
if not self.serial_handler():
status = "ABORT"
break
time.sleep(0.5)
except serial.SerialException:
status = "SERIAL_ERROR"
finally:
while self.ctrl_pipe.poll():
self.ctrl_pipe.recv()
return status
def serial_handler(self):
"""Handles incoming serial transmissions from DStat. Returns False
if stop button pressed and sends abort signal to instrument. Sends
data to self.data_pipe as result of self.data_handler).
"""
scan = 0
def check_ctrl():
if self.ctrl_pipe.poll():
input = self.ctrl_pipe.recv()
logger.info("serial_handler: %s", input)
if input == "DISCONNECT":
self.serial.write('a')
self.serial.reset_input_buffer()
logger.info("serial_handler: ABORT pressed!")
time.sleep(.3)
return False
elif input == 'a':
self.serial.write('a')
else:
self.serial.write(input)
try:
while True:
check_ctrl()
for line in self.serial:
check_ctrl()
if line.startswith('B'):
data = self.data_handler(
(scan, self.serial.read(size=self.databytes)))
data = self.data_postprocessing(data)
if data is not None:
self.data_pipe.send(data)
try:
self.datapoint += 1
except AttributeError: #Datapoint counting is optional
pass
elif line.lstrip().startswith('S'):
scan += 1
elif line.lstrip().startswith("#"):
dstat_logger.info(line.lstrip().rstrip())
elif line.lstrip().startswith("@DONE"):
dstat_logger.debug(line.lstrip().rstrip())
time.sleep(.3)
return True
except serial.SerialException:
return False
def data_handler(self, data_input):
"""Takes data_input as tuple -- (scan, data).
Returns:
(scan number, (voltage, current)) -- voltage in mV, current in A
"""
scan, data = data_input
voltage, current = struct.unpack('<Hl', data) #uint16 + int32
return (scan, (
(voltage-32768)*(3000./65536)*self.re_voltage_scale,
(current+self.gain_trim)*(1.5/8388607)/self.gain
)
)
def store_data(self, incoming, newline):
"""Stores data in data attribute. Should not be called from subprocess.
Can be overriden for custom experiments."""
line, data = incoming
if newline is True:
self.data['current_voltage'].append(deepcopy(self.line_data))
for i, item in enumerate(self.data['current_voltage'][line]):
item.append(data[i])
def data_postprocessing(self, data):
"""Discards first data point (usually gitched) by default, can be overridden
in subclass.
"""
try:
if self.datapoint <= 1:
return None
except AttributeError: # Datapoint counting is optional
pass
return data
def scan_process(self, line):
pass
def experiment_done(self):
"""Runs when experiment is finished (all data acquired)"""
self.data_to_pandas()
self.time += [datetime.utcnow()]
def export(self):
"""Return a dict containing data for saving."""
output = {
"datatype" : self.datatype,
"xlabel" : self.xlabel,
"ylabel" : self.ylabel,
"xmin" : self.xmin,
"xmax" : self.xmax,
"parameters" : self.parameters,
"data" : self.data,
"commands" : self.commands
}
return output
def data_to_pandas(self):
"""Convert data to pandas DataFrame and set as member of .df
attribute."""
self.df = OrderedDict()
for name, data in self.data.items():
try:
df = DataFrame(
columns=['Scan'] + list(self.plot_format[name]['labels']))
for n, line in enumerate(data): # Each scan
df = df.append(
DataFrame(
OrderedDict(zip(
['Scan'] + list(self.plot_format[name]['labels']),
[n] + list(line))
)
), ignore_index = True
)
except (AttributeError, KeyError):
try:
df = DataFrame(
columns=['Scan'] + list(self.columns))
for n, line in enumerate(data): # Each scan
df = df.append(
DataFrame(
OrderedDict(zip(
['Scan'] + list(self.columns),
[n] + list(line))
)
), ignore_index = True
)
except AttributeError as e: # Fallback if no self.columns
df = DataFrame(
columns=['Scan'] + ["{}{}".format(name, n)
for n in range(len(data))]
)
for n, line in enumerate(data):
df = df.append(
DataFrame(
OrderedDict(zip(
['Scan'] + ["{}{}".format(name, n)
for n in range(len(data))],
[n] + list(line))
)
), ignore_index = True
)
self.df[name] = df
def get_info_text(self):
"""Return string of text to disply on Info tab."""
buf = "#Time: S{} E{}\n".format(self.time[0], self.time[1])
buf += "#Commands:\n"
for line in self.commands:
buf += '#{}\n'.format(line)
return buf
def get_save_strings(self):
"""Return dict of strings with experiment parameters and data."""
buf = {}
buf['params'] = self.get_info_text()
buf.update(
{exp : df.to_csv(sep='\t', encoding='utf-8')
for exp, df in self.df.items()}
)
return buf
class PlotBox(object):
"""Contains data plot and associated methods."""
def __init__(self, plots=None):
"""Initializes plots. self.box should be reparented."""
self.name = "Main"
self.continuous_refresh = True
self.scan_refresh = False
if plots is not None:
self.plotnames = plots
else:
self.plotnames = []
self.subplots = {}
self.figure = Figure()
# self.figure.subplots_adjust(left=0.07, bottom=0.07,
# right=0.96, top=0.96)
self.setup()
self.format_plots() # Should be overriden by subclass
self.figure.set_tight_layout(True)
self.canvas = FigureCanvas(self.figure)
self.canvas.set_vexpand(True)
self.box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
self.box.pack_start(self.canvas, expand=True, fill=True, padding=0)
def setup(self):
self.plot_format = {
'current_voltage': {'xlabel': "Voltage (mV)",
'ylabel': "Current (A)"
}
}
def format_plots(self):
"""
Creates and formats subplots needed. Should be overriden by subclass
"""
# Calculate size of grid needed
if len(self.plotnames) > 1:
gs = gridspec.GridSpec(int(ceil(len(self.plotnames)/2.)),2)
else:
gs = gridspec.GridSpec(1,1)
for n, i in enumerate(self.plotnames):
self.subplots[i] = self.figure.add_subplot(gs[n])
for key, subplot in self.subplots.items():
subplot.ticklabel_format(style='sci', scilimits=(0, 3),
useOffset=False, axis='y')
subplot.plot([], [])
subplot.set_xlabel(self.plot_format[key]['xlabel'])
subplot.set_ylabel(self.plot_format[key]['ylabel'])
def clearall(self):
"""Remove all lines on plot. """
for name, plot in self.subplots.items():
for line in reversed(plot.lines):
line.remove()
self.addline()
def clearline(self, subplot, line_number):
"""Remove a single line.
Arguments:
subplot -- key in self.subplots
line_number -- line number in subplot
"""
self.subplots[subplot].lines[line_number].remove()
def addline(self):
"""Add a new line to plot. (initialized with dummy data)))"""
for subplot in self.subplots.values():
subplot.plot([], [])
def updateline(self, Experiment, line_number):
"""Update a line specified with new data.
Arguments:
Experiment -- Experiment instance
line_number -- line number to update
"""
for subplot in Experiment.data:
while True:
try:
self.subplots[subplot].lines[line_number].set_xdata(
Experiment.data[subplot][line_number][0])
self.subplots[subplot].lines[line_number].set_ydata(
Experiment.data[subplot][line_number][1])
except IndexError:
self.addline()
except KeyError:
pass
else:
break
# logger.warning("Tried to set line %s that doesn't exist.", line_number)
def setlims(self, plot, xlims=None, ylims=None):
"""Sets x and y limits.
"""
if xlims is not None:
self.subplots[plot].set_xlim(xlims)
if ylims is not None:
self.subplots[plot].set_ylim(ylims)
self.figure.canvas.draw()
def redraw(self):
"""Autoscale and refresh the plot."""
for name, plot in self.subplots.items():
plot.relim()
plot.autoscale(True, axis = 'y')
self.figure.canvas.draw()
return True
\ No newline at end of file
dstat_interface/core/experiments/idle.py deleted 100644 → 0
View file @ 86048afb
import time
import struct
from .experiment_template import Experiment
from ..dstat import state
class OCPExp(Experiment):
"""Open circuit potential measumement in statusbar."""
id = 'ocp'
def __init__(self):
self.re_voltage_scale = state.board_instance.re_voltage_scale
self.databytes = 8
self.commands = ["EA", "EP0 0 "]
self.commands[0] += "2 " # input buffer
self.commands[0] += "3 " # 2.5 Hz sample rate
self.commands[0] += "1 " # 2x PGA
def data_handler(self, data_input):
"""Overrides Experiment method to only send ADC values."""
scan, data = data_input
# 2*uint16 + int32
seconds, milliseconds, voltage = struct.unpack('<HHl', data)
return voltage/5.592405e6*self.re_voltage_scale
class PMTIdle(Experiment):
"""PMT idle mode."""
id = "pmt_idle"
def __init__(self):
self.databytes = 8
self.commands = ["EA", "EM"]
self.commands[0] += "2 " # input buffer
self.commands[0] += "3 " # 2.5 Hz sample rate
self.commands[0] += "1 " # 2x PGA
dstat_interface/core/experiments/lsv.py
View file @ d5704e4f
import time
import struct
from .experiment_template import PlotBox, Experiment
class LSVExp(Experiment):
"""Linear Scan Voltammetry experiment"""
id = 'lsv'
def setup(self):
self.plotlims['current_voltage']['xlims'] = tuple(
sorted(
(int(self.parameters['start']),
int(self.parameters['stop']))
)
)
super(LSVExp, self).setup()
self.datatype = "linearData"
self.datalength = 2
self.databytes = 6 # uint16 + int32
self.stop_mv = int(self.parameters['stop'])
self.max_mv = abs(int(self.parameters['start'])-int(self.parameters['stop']))
self.commands += "E"
self.commands[2] += "L"
self.commands[2] += str(self.parameters['clean_s'])
self.commands[2] += " "
self.commands[2] += str(self.parameters['dep_s'])
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['clean_mV'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['dep_mV'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['start'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['stop'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['slope'])/
self.re_voltage_scale*
(65536./3000)
))
self.commands[2] += " "
from typing import Any
from ..dstat.experiment_handler import ExperimentHandler
from ..dstat.experiment_process import ExperimentProcess
from ..dstat.utility import abs_mv_to_dac, rel_mv_to_dac, param_test_uint16, param_test_non_zero_uint8
from ..experiments.experiment_container import ExperimentContainer
class LSVExperimentContainer(ExperimentContainer):
experiment_id = 'lsv'
display_name = 'Linear Scan Voltammetry'
process = ExperimentProcess
handler = ExperimentHandler
data_bytes = 0
def __init__(self, params: dict[str, Any], mux: int = 1):
super().__init__(params, mux=mux)
self.data_cols = ['voltage', 'current']
self.data_format = 'vA'
self.cmd_str += ['EL{clean_s} {dep_s} {clean_mv} {dep_mv} {start} {stop} {slope} ']
self.param_input |= {'clean_s': int, 'dep_s': int, 'clean_mv': int, 'dep_mv': int,
'start': int, 'stop': int, 'slope': int}
self.param_input_display_names |= {'clean_s': 't_clean (s)', 'dep_s': 't_dep (s)',
'clean_mv': 'V_clean (mV)', 'dep_mv': 'V_dep (mV)',
'start': 'V_start (mV)', 'stop': 'V_stop (mV)', 'slope': 'Slope (mV/s)', }
self.param_input_limits |= {'clean_s': (0, 'time_max'), 'dep_s': (0, 'time_max'),
'clean_mv': ('mv_min', 'mv_max'), 'dep_mv': ('mv_min', 'mv_max'),
'start': ('mv_min', 'mv_max'), 'stop': ('mv_min', 'mv_max'),
'slope': (1, 5000)}
self.param_converters |= {'clean_s': param_test_uint16, 'dep_s': param_test_uint16,
'clean_mv': abs_mv_to_dac, 'dep_mv': abs_mv_to_dac,
'start': abs_mv_to_dac, 'stop': abs_mv_to_dac, 'slope': rel_mv_to_dac}
self.defaults |= {'clean_s': 0, 'dep_s': 0, 'clean_mv': 0, 'dep_mv': 0}
if params:
self.progress_max = abs(self.params['stop'] - self.params['start'])
self.progress_end = self.params['stop']
else:
self.progress_max = 0
self.progress_iters = 1
self.data_bytes = self.calculate_data_bytes()
def get_progress(self) -> float:
if self.handler_instance.done:
return 100
try:
return (1 - abs(self.progress_end - self.handler_instance.data['voltage'][-1]) / self.progress_max) * 100
except IndexError:
return 0
class CVExperimentContainer(ExperimentContainer):
experiment_id = 'cv'
display_name = 'Cyclic Voltammetry'
process = ExperimentProcess
handler = ExperimentHandler
plots = {'current': {'x': 'voltage', 'y': 'current', 'hue': 'scan'}}
data_bytes = 0
def __init__(self, params: dict[str, Any], mux: int = 1):
super().__init__(params, mux=mux)
self.data_cols = ['voltage', 'current']
self.data_format = 'vA'
self.cmd_str += ['EC{clean_s} {dep_s} {clean_mv} {dep_mv} {v1} {v2} {start} {scans} {slope} ']
self.param_input |= {'clean_s': int, 'dep_s': int, 'clean_mv': int, 'dep_mv': int, 'v1': int, 'v2': int,
'start': int, 'scans': int, 'slope': int}
self.param_input_display_names |= {'clean_s': 't_clean (s)', 'dep_s': 't_dep (s)',
'clean_mv': 'V_clean (mV)', 'dep_mv': 'V_dep (mV)',
'start': 'V_start (mV)', 'v1': 'V_1 (mV)', 'v2': 'V_2 (mV)',
'scans': 'Scans', 'slope': 'Slope (mV/s)'}
self.param_input_limits |= {'clean_s': (0, 'time_max'), 'dep_s': (0, 'time_max'),
'clean_mv': ('mv_min', 'mv_max'), 'dep_mv': ('mv_min', 'mv_max'),
'start': ('mv_min', 'mv_max'), 'v1': ('mv_min', 'mv_max'),
'v2': ('mv_min', 'mv_max'),
'scans': (1, 'scans_max'), 'slope': (1, 5000)}
self.param_converters |= {'clean_s': param_test_uint16, 'dep_s': param_test_uint16,
'clean_mv': abs_mv_to_dac, 'dep_mv': abs_mv_to_dac,
'v1': abs_mv_to_dac, 'v2': abs_mv_to_dac, 'start': abs_mv_to_dac,
'scans': param_test_non_zero_uint8,
'slope': rel_mv_to_dac}
self.defaults |= {'clean_s': 0, 'dep_s': 0, 'clean_mv': 0, 'dep_mv': 0}
if params:
self.progress_max = 2 * abs(self.params['v1'] - self.params['v2'])
self.progress_start = self.params['start']
self.progress_v1 = self.params['v1']
self.progress_v2 = self.params['v2']
self.progress_iters = self.params['scans']
self.progress_scan = 0
self.progress_lastmv = self.params['start']
else:
self.progress_max = 0
self.data_bytes = self.calculate_data_bytes()
def get_progress(self):
try:
return 1 - (abs(self.stop_mv - self.data['current_voltage'][-1][0][-1])/self.max_mv)
if self.handler_instance.done:
return 100
if self.handler_instance.data['scan'][-1] > self.progress_scan:
self.progress_scan = self.handler_instance.data['scan'][-1]
self.progress_lastmv = self.progress_start
raise StopIteration
current_mv = self.handler_instance.data['voltage'][-1]
# if moving towards v1
if abs(self.progress_v1 - self.progress_lastmv) > abs(self.progress_v1 - current_mv):
# Between v2 and start v1-mv and start-mv have same sign
if (self.progress_v1 - current_mv > 0) == (self.progress_start - current_mv > 0):
progress = 100 * (1 - abs(self.progress_start - current_mv) / self.progress_max)
else:
progress = 100 * (.25 - abs(self.progress_v1 - current_mv) / self.progress_max)
else:
progress = 100 * (.75 - abs(self.progress_v2 - current_mv) / self.progress_max)
self.progress_lastmv = current_mv
return progress
except IndexError:
return 0
\ No newline at end of file
return 0
dstat_interface/core/experiments/pot.py deleted 100644 → 0
View file @ 86048afb
import time
import struct
from .experiment_template import PlotBox, Experiment
class PotBox(PlotBox):
def setup(self):
self.plot_format = {
'voltage_time': {'xlabel': "Time (s)",
'ylabel': "Voltage (V)"
}
}
def format_plots(self):
"""
Creates and formats subplots needed. Overrides superclass.
"""
self.subplots = {'voltage_time': self.figure.add_subplot(111)}
for key, subplot in self.subplots.items():
subplot.ticklabel_format(style='sci', scilimits=(0, 3),
useOffset=False, axis='y')
subplot.plot([],[])
subplot.set_xlabel(self.plot_format[key]['xlabel'])
subplot.set_ylabel(self.plot_format[key]['ylabel'])
class PotExp(Experiment):
id = 'pot'
"""Potentiometry experiment"""
def setup(self):
self.plots.append(PotBox(['voltage_time']))
self.datatype = "linearData"
self.datalength = 2
self.databytes = 8
self.data = {'voltage_time' : [([],[])]}
self.columns = ['Time (s)', 'Voltage (V)']
self.plotlims = {
'voltage_time': {
'xlims': (0, int(self.parameters['time']))
}
}
self.plots[-1].setlims('voltage_time', **self.plotlims['voltage_time'])
self.total_time = int(self.parameters['time'])
self.commands += "E"
self.commands[2] += "P"
self.commands[2] += str(self.parameters['time'])
self.commands[2] += " 1 " # potentiometry mode
def data_handler(self, data_input):
"""Overrides Experiment method to not convert x axis to mV."""
scan, data = data_input
# 2*uint16 + int32
seconds, milliseconds, voltage = struct.unpack('<HHl', data)
return (scan, (
seconds+milliseconds/1000.,
voltage*self.re_voltage_scale*(1.5/8388607.)
)
)
def store_data(self, incoming, newline):
"""Stores data in data attribute. Should not be called from subprocess.
Can be overriden for custom experiments."""
line, data = incoming
if newline is True:
self.data['voltage_time'].append(deepcopy(self.line_data))
for i, item in enumerate(self.data['voltage_time'][line]):
item.append(data[i])
def get_progress(self):
try:
return self.data['voltage_time'][-1][0][-1]/self.total_time
except IndexError:
return 0
\ No newline at end of file
dstat_interface/core/experiments/swv.py
View file @ d5704e4f
import time
import struct
from copy import deepcopy
from .experiment_template import PlotBox, Experiment
class SWVBox(PlotBox):
def setup(self):
self.plot_format = {
'swv': {'xlabel': "Voltage (mV)",
'ylabel': "Current (A)"
}
}
def format_plots(self):
"""
Creates and formats subplots needed. Overrides superclass.
"""
self.subplots = {'swv': self.figure.add_subplot(111)}
for key, subplot in self.subplots.items():
subplot.ticklabel_format(style='sci', scilimits=(0, 3),
useOffset=False, axis='y')
subplot.plot([],[])
subplot.set_xlabel(self.plot_format[key]['xlabel'])
subplot.set_ylabel(self.plot_format[key]['ylabel'])
class SWVExp(Experiment):
"""Square Wave Voltammetry experiment"""
id = 'swv'
def setup(self):
self.datatype = "SWVData"
self.xlabel = "Voltage (mV)"
self.ylabel = "Current (A)"
self.data = {
'swv' : [([], [], [], [])]
} # voltage, current, forwards, reverse
self.line_data = ([], [], [], [])
self.datalength = 2 * self.parameters['scans']
self.databytes = 10
self.columns = ['Voltage (mV)', 'Net Current (A)',
'Forward Current (A)', 'Reverse Current (A)']
self.plotlims = {
'swv': {
'xlims': tuple(sorted(
(int(self.parameters['start']),
int(self.parameters['stop']))
)
)
}
}
plot = SWVBox()
plot.setlims('swv', **self.plotlims['swv'])
self.plots.append(plot)
self.stop_mv = int(self.parameters['stop'])
self.max_mv = abs(int(self.parameters['start']) - int(self.parameters['stop']))
self.scan_points = self.max_mv * 2 / float(self.parameters['step'])
self.commands += "E"
self.commands[2] += "S"
self.commands[2] += str(self.parameters['clean_s'])
self.commands[2] += " "
self.commands[2] += str(self.parameters['dep_s'])
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['clean_mV'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['dep_mV'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['start'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['stop'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['step'])/
self.re_voltage_scale*
(65536./3000)
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['pulse'])/
self.re_voltage_scale*
(65536./3000)
))
self.commands[2] += " "
self.commands[2] += str(self.parameters['freq'])
self.commands[2] += " "
self.commands[2] += str(self.parameters['scans'])
self.commands[2] += " "
def data_handler(self, input_data):
"""Overrides Experiment method to calculate difference current"""
scan, data = input_data
# uint16 + int32
voltage, forward, reverse = struct.unpack('<Hll', data)
f_trim = forward+self.gain_trim
r_trim = reverse+self.gain_trim
return (scan, (
(voltage-32768)*3000./65536*self.re_voltage_scale,
(f_trim-r_trim)*(1.5/self.gain/8388607),
f_trim*(1.5/self.gain/8388607),
r_trim*(1.5/self.gain/8388607)
)
)
def store_data(self, incoming, newline):
"""Stores data in data attribute. Should not be called from subprocess.
Can be overriden for custom experiments."""
line, data = incoming
if newline is True:
self.data['swv'].append(deepcopy(self.line_data))
for i, item in enumerate(self.data['swv'][line]):
item.append(data[i])
from datetime import datetime
from typing import Any, Tuple
from ..dstat.experiment_handler import ExperimentHandler
from ..dstat.experiment_process import ExperimentProcess
from ..dstat.utility import abs_mv_to_dac, rel_mv_to_dac, param_test_non_zero_uint16, \
param_test_uint16, param_test_uint8
from ..experiments.experiment_container import ExperimentContainer
class SWVExperimentHandler(ExperimentHandler):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.data['current'] = []
def data_handler(self, data_input: Tuple[datetime, int, bytes]):
unpacked: Tuple[int, int, int]
date, scan, data = data_input
unpacked = self.struct.unpack(data)
self.data['scan'].append(scan)
self.data['timestamp'].append(date)
for n, i in enumerate(unpacked):
try:
self.data[self.data_cols[n]].append(self.data_convert[n](i))
except TypeError: # If no converter
self.data[self.data_cols[n]].append(i)
self.data['current'].append(self.adc_to_amps(unpacked[1] - unpacked[2])) # forward - reverse
class SWVExperimentContainer(ExperimentContainer):
experiment_id = 'swv'
display_name = 'Square Wave Voltammetry'
process = ExperimentProcess
handler = SWVExperimentHandler
plots = {'current': {'x': 'voltage', 'y': 'current', 'hue': 'scan'}}
data_bytes = 0
def __init__(self, params: dict[str, Any], mux: int = 1):
super().__init__(params, mux=mux)
self.data_cols = ['voltage', 'forward', 'reverse']
self.data_format = 'vAA'
self.cmd_str += ['ES{clean_s} {dep_s} {clean_mv} {dep_mv} {start} {stop} {step} {pulse} {freq} {scans} ']
self.param_input |= {'clean_s': int, 'dep_s': int, 'clean_mv': int, 'dep_mv': int,
'start': int, 'stop': int, 'step': int, 'pulse': int, 'freq': int,
'scans': int}
self.param_input_display_names |= {'clean_s': 't_clean (s)', 'dep_s': 't_dep (s)',
'clean_mv': 'V_clean (mV)', 'dep_mv': 'V_dep (mV)',
'start': 'V_start (mV)', 'stop': 'V_stop (mV)', 'step': 'V_step (mV)',
'pulse': 'V_pulse (mV)', 'freq': 'f (Hz)',
'scans': 'Scans'}
self.param_input_limits |= {'clean_s': (0, 'time_max'), 'dep_s': (0, 'time_max'),
'clean_mv': ('mv_min', 'mv_max'), 'dep_mv': ('mv_min', 'mv_max'),
'start': ('mv_min', 'mv_max'), 'stop': ('mv_min', 'mv_max'),
'step': (1, 1000), 'pulse': (1, 1000), 'freq': (1, 'freq_max'),
'scans': (0, 'scans_max')}
self.param_converters |= {'clean_s': param_test_uint16, 'dep_s': param_test_uint16,
'clean_mv': abs_mv_to_dac, 'dep_mv': abs_mv_to_dac,
'start': abs_mv_to_dac, 'stop': abs_mv_to_dac, 'step': rel_mv_to_dac,
'pulse': rel_mv_to_dac, 'freq': param_test_non_zero_uint16,
'scans': param_test_uint8}
self.defaults |= {'clean_s': 0, 'dep_s': 0, 'clean_mv': 0, 'dep_mv': 0}
if params:
if self.params['scans'] > 0:
self.progress_max = 2 * abs(self.params['stop'] - self.params['start'])
else:
self.progress_max = abs(self.params['stop'] - self.params['start'])
self.progress_start = self.params['start']
self.progress_stop = self.params['stop']
self.progress_iters = self.params['scans']
self.progress_scan = 0
self.progress_lastmv = self.params['start']
else:
self.progress_max = 0
self.data_bytes = self.calculate_data_bytes()
def get_progress(self):
if self.handler_instance.done:
return 100
try:
if int(self.parameters['scans']) != 0:
scans_prog = (len(self.data['swv'])-1) / float(self.parameters['scans'])
scan_prog = (len(self.data['swv'][-1][0])-1) / self.scan_points / float(self.parameters['scans'])
prog = scans_prog + scan_prog
if prog > 1:
prog = 1
return prog
if self.params['scans'] > 0:
if self.handler_instance.data['scan'][-1] > self.progress_scan:
self.progress_scan = self.handler_instance.data['scan'][-1]
self.progress_lastmv = self.progress_start
raise StopIteration
current_mv = self.handler_instance.data['voltage'][-1]
if abs(self.progress_stop - self.progress_lastmv) > abs(self.progress_stop - current_mv):
progress = 100 * (.5 - abs(self.progress_stop - current_mv) / self.progress_max)
else:
progress = 100 * (1 - abs(self.progress_start - current_mv) / self.progress_max)
self.progress_lastmv = current_mv
return progress
else:
return 1 - (abs(self.stop_mv - self.data['swv'][-1][0][-1])/self.max_mv)
return (1 - abs(
self.progress_stop - self.handler_instance.data['voltage'][-1]) / self.progress_max) * 100
except IndexError:
return 0
class DPVExp(SWVExp):
"""Diffential Pulse Voltammetry experiment."""
id = 'dpv'
def setup(self):
self.datatype = "SWVData"
self.xlabel = "Voltage (mV)"
self.ylabel = "Current (A)"
self.data = {
'swv' : [([], [], [], [])]
} # voltage, current, forwards, reverse
self.line_data = ([], [], [], [])
self.datalength = 2
self.databytes = 10
self.columns = ['Voltage (mV)', 'Net Current (A)',
'Forward Current (A)', 'Reverse Current (A)']
self.plotlims = {
'swv': {
'xlims': tuple(sorted(
(int(self.parameters['start']),
int(self.parameters['stop']))
)
)
}
}
plot = SWVBox()
plot.setlims('swv', **self.plotlims['swv'])
self.plots.append(plot)
self.stop_mv = int(self.parameters['stop'])
self.max_mv = abs(int(self.parameters['start']) - int(self.parameters['stop']))
self.commands += "E"
self.commands[2] += "D"
self.commands[2] += str(self.parameters['clean_s'])
self.commands[2] += " "
self.commands[2] += str(self.parameters['dep_s'])
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['clean_mV'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['dep_mV'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['start'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['stop'])/
self.re_voltage_scale*
(65536./3000)+32768
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['step'])/
self.re_voltage_scale*
(65536./3000)
))
self.commands[2] += " "
self.commands[2] += str(int(
int(self.parameters['pulse'])/
self.re_voltage_scale*
(65536./3000)
))
self.commands[2] += " "
self.commands[2] += str(self.parameters['period'])
self.commands[2] += " "
self.commands[2] += str(self.parameters['width'])
self.commands[2] += " "
class DPVExperimentContainer(ExperimentContainer):
experiment_id = 'dpv'
display_name = 'Differential Pulse Voltammetry'
process = ExperimentProcess
handler = SWVExperimentHandler
data_bytes = 0
def __init__(self, params: dict[str, Any], mux: int = 1):
super().__init__(params, mux=mux)
self.data_cols = ['voltage', 'forward', 'reverse']
self.data_format = 'vAA'
self.cmd_str += ['ED{clean_s} {dep_s} {clean_mv} {dep_mv} {start} {stop} {step} {pulse} {period} {width} ']
self.param_input |= {'clean_s': int, 'dep_s': int, 'clean_mv': int, 'dep_mv': int,
'start': int, 'stop': int, 'step': int, 'pulse': int, 'period': int, 'width': int}
self.param_input_display_names |= {'clean_s': 't_clean (s)', 'dep_s': 't_dep (s)',
'clean_mv': 'V_clean (mV)', 'dep_mv': 'V_dep (mV)',
'start': 'V_start (mV)', 'stop': 'V_stop (mV)', 'step': 'V_step (mV)',
'pulse': 'V_pulse (mV)', 'period': 'Pulse Period (ms)',
'width': 'Pulse Width (ms)'}
self.param_input_limits |= {'clean_s': (0, 'time_max'), 'dep_s': (0, 'time_max'),
'clean_mv': ('mv_min', 'mv_max'), 'dep_mv': ('mv_min', 'mv_max'),
'start': ('mv_min', 'mv_max'), 'stop': ('mv_min', 'mv_max'),
'step': (1, 1000), 'pulse': (1, 1000), 'period': (1, 1000),
'width': (1, 1000)}
self.param_converters |= {'clean_s': param_test_uint16, 'dep_s': param_test_uint16,
'clean_mv': abs_mv_to_dac, 'dep_mv': abs_mv_to_dac,
'start': abs_mv_to_dac, 'stop': abs_mv_to_dac, 'step': rel_mv_to_dac,
'pulse': rel_mv_to_dac, 'period': param_test_uint16,
'width': param_test_non_zero_uint16}
self.defaults |= {'clean_s': 0, 'dep_s': 0, 'clean_mv': 0, 'dep_mv': 0}
if params:
self.progress_max = abs(self.params['stop'] - self.params['start'])
self.progress_start = self.params['start']
self.progress_stop = self.params['stop']
self.progress_scan = 0
self.progress_iters = 1
else:
self.progress_max = 0
self.data_bytes = self.calculate_data_bytes()
def get_progress(self):
if self.handler_instance.done:
return 100
try:
return 1 - (abs(self.stop_mv - self.data['swv'][-1][0][-1])/self.max_mv)
return (1 - abs(
self.progress_stop - self.handler_instance.data['voltage'][-1]) / self.progress_max) * 100
except IndexError:
return 0
\ No newline at end of file
return 0
dstat_interface/core/interface/__init__.py deleted 100644 → 0
View file @ 86048afb
dstat_interface/core/interface/acv.glade deleted 100644 → 0
View file @ 86048afb
This diff is collapsed.
dstat_interface/core/interface/adc_pot.glade deleted 100644 → 0
View file @ 86048afb
This diff is collapsed.
dstat_interface/core/interface/adc_pot.py deleted 100755 → 0
View file @ 86048afb
This diff is collapsed.
dstat_interface/core/interface/calib.glade deleted 100644 → 0
View file @ 86048afb
This diff is collapsed.
dstat_interface/core/interface/chronoamp.glade deleted 100644 → 0
View file @ 86048afb
This diff is collapsed.
dstat_interface/core/interface/cv.glade deleted 100644 → 0
View file @ 86048afb
This diff is collapsed.
dstat_interface/core/interface/data_view.py deleted 100644 → 0
View file @ 86048afb
This diff is collapsed.
dstat_interface/core/interface/db.glade deleted 100644 → 0
View file @ 86048afb
This diff is collapsed.