WIP: Manual mode (!11) · Merge requests · dstat / dstat-interface

dstat_interface/core/dstat/comm.py

+2 −2

File changed.

Contains only whitespace changes.

dstat_interface/core/dstat/dfu.py

+1 −0

Changes for dstat_interface/core/dstat/dfu.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -278,6 +278,7 @@ def test_firmware_version(current=None):
		logger.error('Unexpected git error. Git exited {}'.format(test))
		return False, None


		def dfu_program(path='./dstat-firmware.hex'):
		"""Tries to program DStat over USB with DFU with hex file at path."""
		try:

Conflict: This file was modified in both the source and target branches. Ask someone with write access to resolve it.

dstat_interface/core/experiments/chronoamp.py

+136 −2

Changes for dstat_interface/core/experiments/chronoamp.py: 136 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -2,9 +2,12 @@ import time
		import struct
		import numpy as np
		import serial
		from copy import deepcopy

		from .experiment_template import PlotBox, Experiment, exp_logger, dstat_logger
		from ..interface.plot import mean, plotSpectrum, findBounds
		from ..dstat import state

		from .experiment_template import PlotBox, Experiment, exp_logger

		class ChronoampBox(PlotBox):
		def format_plots(self):
		@@ -63,9 +66,140 @@ class Chronoamp(Experiment):

		if newline is True:
		self.data['current_time'].append(deepcopy(self.line_data))

		try:
		for i, item in enumerate(self.data['current_time'][line]):
		item.append(data[i])
		except IndexError:
		pass


		class ManExp(Chronoamp):
		id = 'man'
		"""Manual experiment"""
		def setup(self):
		self.handler_ids = [self.parameters['exp_window'].connect('new_voltage', self.new_voltage),
		self.parameters['exp_window'].connect('new_gain', self.new_gain)]

		self.plots.append(ManBox('current_time'))

		self.datatype = "linearData"
		self.datalength = 2
		self.databytes = 8
		self.data = {'current_time': [([], [], [])]}
		self.line_data = ([], [], [])
		self.columns = ['Time (s)', 'Current (A)', 'Voltage (mV)']
		self.plot_format = {
		'current_time': {
		'labels': self.columns,
		'xlims': (0, 10)
		}
		}

		self.commands.append("EN")

		def new_voltage(self, widget, voltage):
		state.ser.send_ctrl("v{}\n".format(int(voltage*(65536./3000)+32768)))

		def new_gain(self, widget):
		state.ser.send_ctrl("g{}\n".format(self.parameters['adc_pot'].params['gain']))

		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).
		"""

		def check_ctrl():
		if self.ctrl_pipe.poll():
		input = self.ctrl_pipe.recv()
		exp_logger.info("serial_handler: %s", input)
		if input == "DISCONNECT":
		self.serial.write('a')
		self.serial.reset_input_buffer()
		exp_logger.info("serial_handler: ABORT pressed!")
		time.sleep(.3)
		return False
		elif input == 'a':
		self.serial.write('a')
		else:
		self.serial.write(input)

		scan = 0
		voltage = 0

		try:
		while True:
		check_ctrl()

		for line in self.serial:
		check_ctrl()
		line_in = line.lstrip()
		if line_in.startswith('B'):
		data = self.data_handler(
		(scan, voltage, 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_in.startswith('S'):
		scan += 1

		elif line_in.startswith('G'):
		self.gain = self.parameters["gain_table"][int(line.split(" ")[1])]

		elif line_in.startswith('V'):
		voltage = int(line.split(" ")[1])

		elif line_in.startswith("#"):
		dstat_logger.info(line.lstrip().rstrip())

		elif line_in.startswith("R"):
		self.ctrl_pipe.send("R")

		elif line_in.startswith("@DONE"):
		dstat_logger.debug(line.lstrip().rstrip())
		time.sleep(.3)
		return True

		except serial.SerialException:
		return False

		def ctrl_loop(self, data):
		if data == 'R':
		self.emit('exp_ready')

		def data_handler(self, data_input):
		"""Overrides Experiment method to not convert x axis to mV."""
		scan, voltage, 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),
		(voltage-32768)(3000./65536)self.re_voltage_scale
		)
		)

		def experiment_done(self):
		"""Runs when experiment is finished (all data acquired)"""
		super(ManExp, self).experiment_done()
		for i in self.handler_ids:
		self.parameters['exp_window'].disconnect(i)
		self.emit('exp_done')

		class ManBox(ChronoampBox):
		def redraw(self):
		"""Autoscale and refresh the plot. Overriden to scale both axes."""
		for name, plot in self.subplots.items():
		plot.relim()
		plot.autoscale(True, axis='both')
		self.figure.canvas.draw()

		return True

		class PDExp(Chronoamp):
		"""Photodiode/PMT experiment"""

dstat_interface/core/experiments/experiment_template.py

+1 −0

Changes for dstat_interface/core/experiments/experiment_template.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -69,6 +69,7 @@ class Experiment(GObject.Object):
		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__()

dstat_interface/core/interface/exp_int.py

+75 −0

Changes for dstat_interface/core/interface/exp_int.py: 75 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -21,6 +21,7 @@ from __future__ import division, absolute_import, print_function, unicode_litera
		import os
		import sys
		import logging
		from collections import OrderedDict

		try:
		import gi
		@@ -191,6 +192,80 @@ class Chronoamp(ExpInterface):
		for i in table:
		self.model.append(i)


		class Manual(ExpInterface):
		"""Experiment class for Manual experiments."""
		__gsignals__ = {
		b'new_voltage': (GObject.SIGNAL_RUN_FIRST, None, (int, )),
		b'new_gain': (GObject.SIGNAL_RUN_FIRST, None, ())
		}
		id = 'man'
		experiment = chronoamp.ManExp

		def __init__(self):
		"""Adds entry listings to superclass's self.entry dict"""
		super(ExpInterface, self).__init__() # initialize GObject
		self.name = "Manual Mode"

		self.entry = {} # to be used only for str parameters
		self._params = None

		self.window = Gtk.ScrolledWindow()
		self.window.set_vexpand(True)

		grid = Gtk.Grid(orientation=Gtk.Orientation.VERTICAL)
		grid.set_column_homogeneous(True)

		entries = OrderedDict([
		('voltage', 'Voltage (mV)'),
		])

		for n, i in enumerate(entries.items()):
		key, value = i
		grid.attach(Gtk.Label(label=value), 0, n, 1, 1)
		self.entry[key] = Gtk.Entry()
		grid.attach(self.entry[key], 1, n, 1, 1)

		n += 1
		voltage_button = Gtk.Button.new_with_label("Change Voltage")
		grid.attach(voltage_button, 0, n, 2, 1)
		voltage_button.connect('clicked', self.new_voltage)

		n += 1
		gain_button = Gtk.Button.new_with_label("Change Gain")
		grid.attach(gain_button, 0, n, 2, 1)
		gain_button.connect('clicked', self.new_gain)

		self.buttons = [voltage_button, gain_button]
		for i in self.buttons:
		i.set_sensitive(False)

		self.window.add(grid)

		def exp_ready(self, widget):
		for i in self.buttons:
		i.set_sensitive(True)

		def exp_done(self, widget):
		for i in self.buttons:
		i.set_sensitive(False)

		def get_experiment(self, parameters):
		exp = super(Manual, self).get_experiment(parameters)
		exp.connect('exp_ready', self.exp_ready)
		exp.connect('exp_done', self.exp_done)
		return exp

		def new_voltage(self, widget):
		self.emit('new_voltage', int(self.entry['voltage'].get_text()))

		def new_gain(self, widget):
		self.emit('new_gain')

		def get_window(self):
		return self.window


		class LSV(ExpInterface):
		"""Experiment class for LSV."""
		id = 'lsv'

WIP: Manual mode

Merge request reports