Skip to content
GitLab
Commit 31dd3887 authored by Olivier Kaufmann's avatar Olivier Kaufmann
Browse files

Fixes issue #77. Improves args and kwargs handling by _process_command.... 

Fixes issue #77. Improves args and kwargs handling by _process_command. Reorders method in alphabetical order to ease navigating the code
parent 5ede4f60
master 144-create-testing-framework-for-tests-automation EGU23 MB.2024.ter_restored MUX_board_v2024 Refactor_mb_2023 battery_charger board_v4 code_refactor dev/dummy mqtt update_V2023 v2024_rc v2025_alpha v2024.0.34 v2024.0.33 v2024.0.32 v2024.0.31 v2024.0.30 v2024.0.29 v2024.0.28 v2024.0.27 v2024.0.26 v2024.0.25 v2024.0.24 v2024.0.23 v2024.0.22 v2024.0.21 v2024.0.20 v2024.0.19 v2024.0.18 v2024.0.17 v2024.0.16 v2024.0.15 v2024.0.14 v2024.0.13 v2024.0.12 v2024.0.11 v2024.0.10 v2024.0.9 v2024.0.8 v2024.0.7 v2024.0.0 v2023.0.1 v2023.0.0 archives/mqtt
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Showing
with 576 additions and 555 deletions
+576 -555
config.py
+ 11
− 10
View file @ 31dd3887
...@@ -13,15 +13,16 @@ logging_suffix = '' ...@@ -13,15 +13,16 @@ logging_suffix = ''
OHMPI_CONFIG = { OHMPI_CONFIG = {
'id': ohmpi_id, # Unique identifier of the OhmPi board (string) 'id': ohmpi_id, # Unique identifier of the OhmPi board (string)
'R_shunt': 2, # Shunt resistance in Ohms 'R_shunt': 2, # Shunt resistance in Ohms
'Imax': 4800/50/2, # Maximum current 'Imax': 4800 / 50 / 2, # Maximum current
'coef_p2': 2.50, # slope for current conversion for ADS.P2, measurement in V/V 'coef_p2': 2.50, # slope for current conversion for ADS.P2, measurement in V/V
'coef_p3': 2.50, # slope for current conversion for ADS.P3, measurement in V/V # 'coef_p3': 2.50, # slope for current conversion for ADS.P3, measurement in V/V
'offset_p2': 0, # 'offset_p2': 0,
'offset_p3': 0, # 'offset_p3': 0,
'integer': 2, # Max value 10 # TODO: Explain what this is... 'nb_samples': 2, # Max value 10 # was named integer before...
'version': 2, 'version': 2, # Is this still needed?
'max_elec': 64, 'max_elec': 64,
'board_addresses': {'A': 0x73, 'B': 0x72, 'M': 0x71, 'N': 0x70}, # def. {'A': 0x76, 'B': 0x71, 'M': 0x74, 'N': 0x70} 'board_addresses': {'A': 0x73, 'B': 0x72, 'M': 0x71, 'N': 0x70},
# def. {'A': 0x76, 'B': 0x71, 'M': 0x74, 'N': 0x70}
'settings': 'ohmpi_settings.json', 'settings': 'ohmpi_settings.json',
'board_version': '22.10' 'board_version': '22.10'
} # TODO: add a dictionary with INA models and associated gain values } # TODO: add a dictionary with INA models and associated gain values
...@@ -29,7 +30,7 @@ OHMPI_CONFIG = { ...@@ -29,7 +30,7 @@ OHMPI_CONFIG = {
# Execution logging configuration # Execution logging configuration
EXEC_LOGGING_CONFIG = { EXEC_LOGGING_CONFIG = {
'logging_level': logging.INFO, 'logging_level': logging.INFO,
'log_file_logging_level' : logging.DEBUG, 'log_file_logging_level': logging.DEBUG,
'logging_to_console': True, 'logging_to_console': True,
'file_name': f'exec{logging_suffix}.log', 'file_name': f'exec{logging_suffix}.log',
'max_bytes': 262144, 'max_bytes': 262144,
...@@ -67,7 +68,7 @@ MQTT_LOGGING_CONFIG = { ...@@ -67,7 +68,7 @@ MQTT_LOGGING_CONFIG = {
'retain': False, 'retain': False,
'keepalive': 60, 'keepalive': 60,
'will': None, 'will': None,
'auth': { 'username': 'mqtt_user', 'password': 'mqtt_password' }, 'auth': {'username': 'mqtt_user', 'password': 'mqtt_password'},
'tls': None, 'tls': None,
'protocol': MQTTv31, 'protocol': MQTTv31,
'transport': 'tcp', 'transport': 'tcp',
...@@ -85,7 +86,7 @@ MQTT_CONTROL_CONFIG = { ...@@ -85,7 +86,7 @@ MQTT_CONTROL_CONFIG = {
'retain': False, 'retain': False,
'keepalive': 60, 'keepalive': 60,
'will': None, 'will': None,
'auth': { 'username': 'mqtt_user', 'password': 'mqtt_password' }, 'auth': {'username': 'mqtt_user', 'password': 'mqtt_password'},
'tls': None, 'tls': None,
'protocol': MQTTv31, 'protocol': MQTTv31,
'transport': 'tcp', 'transport': 'tcp',
......
ohmpi.py
+ 565
− 545
View file @ 31dd3887
...@@ -35,7 +35,7 @@ try: ...@@ -35,7 +35,7 @@ try:
import digitalio # noqa import digitalio # noqa
from digitalio import Direction # noqa from digitalio import Direction # noqa
from gpiozero import CPUTemperature # noqa from gpiozero import CPUTemperature # noqa
import minimalmodbus import minimalmodbus # noqa
arm64_imports = True arm64_imports = True
except ImportError as error: except ImportError as error:
if EXEC_LOGGING_CONFIG['logging_level'] == DEBUG: if EXEC_LOGGING_CONFIG['logging_level'] == DEBUG:
...@@ -43,6 +43,7 @@ except ImportError as error: ...@@ -43,6 +43,7 @@ except ImportError as error:
arm64_imports = False arm64_imports = False
except Exception as error: except Exception as error:
print(colored(f'Unexpected error: {error}', 'red')) print(colored(f'Unexpected error: {error}', 'red'))
arm64_imports = None
exit() exit()
...@@ -50,7 +51,7 @@ class OhmPi(object): ...@@ -50,7 +51,7 @@ class OhmPi(object):
""" OhmPi class. """ OhmPi class.
""" """
def __init__(self, settings=None, sequence=None, use_mux=False, mqtt=True, on_pi=None, idps=False): def __init__(self, settings=None, sequence=None, use_mux=False, mqtt=True, onpi=None, idps=False):
"""Constructs the ohmpi object """Constructs the ohmpi object
Parameters Parameters
...@@ -63,19 +64,20 @@ class OhmPi(object): ...@@ -63,19 +64,20 @@ class OhmPi(object):
if True use the multiplexor to select active electrodes if True use the multiplexor to select active electrodes
mqtt: bool, defaut: True mqtt: bool, defaut: True
if True publish on mqtt topics while logging, otherwise use other loggers only if True publish on mqtt topics while logging, otherwise use other loggers only
on_pi: bool,None default: None onpi: bool,None default: None
if None, the platform on which the class is instantiated is determined to set on_pi to either True or False. if None, the platform on which the class is instantiated is determined to set on_pi to either True or False.
if False the behaviour of an ohmpi will be partially emulated and return random data. if False the behaviour of an ohmpi will be partially emulated and return random data.
idps: idps:
if true uses the DPS if true uses the DPS
""" """
if on_pi is None: if onpi is None:
_, on_pi = get_platform() _, onpi = get_platform()
self._sequence = sequence self._sequence = sequence
self.nb_samples = 0
self.use_mux = use_mux self.use_mux = use_mux
self.on_pi = on_pi # True if run from the RaspberryPi with the hardware, otherwise False for random data self.on_pi = onpi # True if run from the RaspberryPi with the hardware, otherwise False for random data
self.status = 'idle' # either running or idle self.status = 'idle' # either running or idle
self.thread = None # contains the handle for the thread taking the measurement self.thread = None # contains the handle for the thread taking the measurement
...@@ -150,7 +152,6 @@ class OhmPi(object): ...@@ -150,7 +152,6 @@ class OhmPi(object):
self.cmd_id = None self.cmd_id = None
if self.mqtt: if self.mqtt:
import paho.mqtt.client as mqtt_client import paho.mqtt.client as mqtt_client
import paho.mqtt.publish as publish
self.exec_logger.debug(f"Connecting to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']}" self.exec_logger.debug(f"Connecting to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']}"
f" on {MQTT_CONTROL_CONFIG['hostname']} broker") f" on {MQTT_CONTROL_CONFIG['hostname']} broker")
...@@ -192,15 +193,9 @@ class OhmPi(object): ...@@ -192,15 +193,9 @@ class OhmPi(object):
publisher_config.pop('ctrl_topic') publisher_config.pop('ctrl_topic')
def on_message(client, userdata, message): def on_message(client, userdata, message):
print(message.payload.decode('utf-8'))
command = message.payload.decode('utf-8') command = message.payload.decode('utf-8')
dic = json.loads(command) self.exec_logger.debug(f'Received command {command}')
if dic['cmd_id'] != self.cmd_id: self._process_commands(command)
self.cmd_id = dic['cmd_id']
self.exec_logger.debug(f'Received command {command}')
# payload = json.dumps({'cmd_id': dic['cmd_id'], 'reply': 'ok'})
# publish.single(payload=payload, **publisher_config)
self._process_commands(command)
self.controller.on_message = on_message self.controller.on_message = on_message
else: else:
...@@ -208,75 +203,181 @@ class OhmPi(object): ...@@ -208,75 +203,181 @@ class OhmPi(object):
self.exec_logger.warning('No connection to control broker.' self.exec_logger.warning('No connection to control broker.'
' Use python/ipython to interact with OhmPi object...') ' Use python/ipython to interact with OhmPi object...')
@property @staticmethod
def sequence(self): def append_and_save(filename, last_measurement):
"""Gets sequence""" """Appends and saves the last measurement dict.
if self._sequence is not None:
assert isinstance(self._sequence, np.ndarray)
return self._sequence
@sequence.setter Parameters
def sequence(self, sequence): ----------
"""Sets sequence""" filename : str
if sequence is not None: filename to save the last measurement dataframe
assert isinstance(sequence, np.ndarray) last_measurement : dict
self.use_mux = True Last measurement taken in the form of a python dictionary
else: """
self.use_mux = False last_measurement = deepcopy(last_measurement)
self._sequence = sequence if 'fulldata' in last_measurement:
d = last_measurement['fulldata']
n = d.shape[0]
if n > 1:
idic = dict(zip(['i' + str(i) for i in range(n)], d[:, 0]))
udic = dict(zip(['u' + str(i) for i in range(n)], d[:, 1]))
tdic = dict(zip(['t' + str(i) for i in range(n)], d[:, 2]))
last_measurement.update(idic)
last_measurement.update(udic)
last_measurement.update(tdic)
last_measurement.pop('fulldata')
def _update_acquisition_settings(self, config): if os.path.isfile(filename):
warnings.warn('This function is deprecated, use update_settings() instead.', DeprecationWarning) # Load data file and append data to it
self.update_settings(config) with open(filename, 'a') as f:
w = csv.DictWriter(f, last_measurement.keys())
w.writerow(last_measurement)
# last_measurement.to_csv(f, header=False)
else:
# create data file and add headers
with open(filename, 'a') as f:
w = csv.DictWriter(f, last_measurement.keys())
w.writeheader()
w.writerow(last_measurement)
def update_settings(self, config): def _compute_tx_volt(self, best_tx_injtime=0.1, strategy='vmax', tx_volt=5):
"""Updates acquisition settings from a json file or dictionary. """Estimates best Tx voltage based on different strategies.
Parameters can be: At first a half-cycle is made for a short duration with a fixed
- nb_electrodes (number of electrode used, if 4, no MUX needed) known voltage. This gives us Iab and Rab. We also measure Vmn.
- injection_duration (in seconds) A constant c = vmn/iab is computed (only depends on geometric
- nb_meas (total number of times the sequence will be run) factor and ground resistivity, that doesn't change during a
- sequence_delay (delay in second between each sequence run) quadrupole). Then depending on the strategy, we compute which
- nb_stack (number of stack for each quadrupole measurement) vab to inject to reach the minimum/maximum Iab current or
- export_path (path where to export the data, timestamp will be added to filename) min/max Vmn.
This function also compute the polarity on Vmn (on which pin
of the ADS1115 we need to measure Vmn to get the positive value).
Parameters Parameters
---------- ----------
config : str, dict best_tx_injtime : float, optional
Path to the .json settings file or dictionary of settings. Time in milliseconds for the half-cycle used to compute Rab.
strategy : str, optional
Either:
- vmin : compute Vab to reach a minimum Iab and Vmn
- vmax : compute Vab to reach a maximum Iab and Vmn
- constant : apply given Vab
tx_volt : float, optional
Voltage apply to try to guess the best voltage. 5 V applied
by default. If strategy "constant" is chosen, constant voltage
to applied is "tx_volt".
Returns
-------
vab : float
Proposed Vab according to the given strategy.
polarity : int
Either 1 or -1 to know on which pin of the ADS the Vmn is measured.
""" """
status = False
if config is not None: # hardware limits
try: voltage_min = 10. # mV
if isinstance(config, dict): voltage_max = 4500.
self.settings.update(config) current_min = voltage_min / (self.r_shunt * 50) # mA
else: current_max = voltage_max / (self.r_shunt * 50)
with open(config) as json_file: tx_max = 40. # volt
dic = json.load(json_file)
self.settings.update(dic) # check of volt
self.exec_logger.debug('Acquisition parameters updated: ' + str(self.settings)) volt = tx_volt
status = True if volt > tx_max:
except Exception as e: self.exec_logger.warning('Sorry, cannot inject more than 40 V, set it back to 5 V')
self.exec_logger.warning('Unable to update settings.') volt = 5.
status = False
# redefined the pin of the mcp (needed when relays are connected)
self.pin0 = self.mcp.get_pin(0)
self.pin0.direction = Direction.OUTPUT
self.pin0.value = False
self.pin1 = self.mcp.get_pin(1)
self.pin1.direction = Direction.OUTPUT
self.pin1.value = False
# select a polarity to start with
self.pin0.value = True
self.pin1.value = False
# set voltage for test
self.DPS.write_register(0x0000, volt, 2)
self.DPS.write_register(0x09, 1) # DPS5005 on
time.sleep(best_tx_injtime) # inject for given tx time
# autogain
self.ads_current = ads.ADS1115(self.i2c, gain=2/3, data_rate=860, address=self.ads_current_address)
self.ads_voltage = ads.ADS1115(self.i2c, gain=2/3, data_rate=860, address=self.ads_voltage_address)
# print('current P0', AnalogIn(self.ads_current, ads.P0).voltage)
# print('voltage P0', AnalogIn(self.ads_voltage, ads.P0).voltage)
# print('voltage P2', AnalogIn(self.ads_voltage, ads.P2).voltage)
gain_current = self._gain_auto(AnalogIn(self.ads_current, ads.P0))
gain_voltage0 = self._gain_auto(AnalogIn(self.ads_voltage, ads.P0))
gain_voltage2 = self._gain_auto(AnalogIn(self.ads_voltage, ads.P2))
gain_voltage = np.min([gain_voltage0, gain_voltage2])
# print('gain current: {:.3f}, gain voltage: {:.3f}'.format(gain_current, gain_voltage))
self.ads_current = ads.ADS1115(self.i2c, gain=gain_current, data_rate=860, address=self.ads_current_address)
self.ads_voltage = ads.ADS1115(self.i2c, gain=gain_voltage, data_rate=860, address=self.ads_voltage_address)
# we measure the voltage on both A0 and A2 to guess the polarity
I = AnalogIn(self.ads_current, ads.P0).voltage * 1000. / 50 / self.r_shunt # noqa measure current
U0 = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000. # measure voltage
U2 = AnalogIn(self.ads_voltage, ads.P2).voltage * 1000.
# print('I (mV)', I*50*self.r_shunt)
# print('I (mA)', I)
# print('U0 (mV)', U0)
# print('U2 (mV)', U2)
# check polarity
polarity = 1 # by default, we guessed it right
vmn = U0
if U0 < 0: # we guessed it wrong, let's use a correction factor
polarity = -1
vmn = U2
# print('polarity', polarity)
# compute constant
c = vmn / I
Rab = (volt * 1000.) / I
self.exec_logger.debug(f'Rab = {Rab:.2f} Ohms')
# implement different strategy
if strategy == 'vmax':
vmn_max = c * current_max
if voltage_max > vmn_max > voltage_min:
vab = current_max * Rab
self.exec_logger.debug('target max current')
else:
iab = voltage_max / c
vab = iab * Rab
self.exec_logger.debug('target max voltage')
if vab > 25000.:
vab = 25000.
vab = vab / 1000. * 0.9
elif strategy == 'vmin':
vmn_min = c * current_min
if voltage_min < vmn_min < voltage_max:
vab = current_min * Rab
self.exec_logger.debug('target min current')
else:
iab = voltage_min / c
vab = iab * Rab
self.exec_logger.debug('target min voltage')
if vab < 1000.:
vab = 1000.
vab = vab / 1000. * 1.1
elif strategy == 'constant':
vab = volt
else: else:
self.exec_logger.warning('Settings are missing...') vab = 5
return status
def _read_hardware_config(self): # self.DPS.write_register(0x09, 0) # DPS5005 off
"""Reads hardware configuration from config.py self.pin0.value = False
""" self.pin1.value = False
self.exec_logger.debug('Getting hardware config')
self.id = OHMPI_CONFIG['id'] # ID of the OhmPi return vab, polarity
self.r_shunt = OHMPI_CONFIG['R_shunt'] # reference resistance value in ohm
self.Imax = OHMPI_CONFIG['Imax'] # maximum current
self.exec_logger.debug(f'The maximum current cannot be higher than {self.Imax} mA')
self.coef_p2 = OHMPI_CONFIG['coef_p2'] # slope for current conversion for ads.P2, measurement in V/V
self.nb_samples = OHMPI_CONFIG['integer'] # number of samples measured for each stack
self.version = OHMPI_CONFIG['version'] # hardware version
self.max_elec = OHMPI_CONFIG['max_elec'] # maximum number of electrodes
self.board_addresses = OHMPI_CONFIG['board_addresses']
self.board_version = OHMPI_CONFIG['board_version']
self.exec_logger.debug(f'OHMPI_CONFIG = {str(OHMPI_CONFIG)}')
@staticmethod @staticmethod
def _find_identical_in_line(quads): def _find_identical_in_line(quads):
...@@ -302,9 +403,41 @@ class OhmPi(object): ...@@ -302,9 +403,41 @@ class OhmPi(object):
return output return output
def read_quad(self, filename): def _gain_auto(self, channel):
warnings.warn('This function is deprecated. Use load_sequence instead.', DeprecationWarning) """Automatically sets the gain on a channel
self.load_sequence(filename)
Parameters
----------
channel : ads.ADS1x15
Instance of ADS where voltage is measured.
Returns
-------
gain : float
Gain to be applied on ADS1115.
"""
gain = 2 / 3
if (abs(channel.voltage) < 2.040) and (abs(channel.voltage) >= 1.023):
gain = 2
elif (abs(channel.voltage) < 1.023) and (abs(channel.voltage) >= 0.508):
gain = 4
elif (abs(channel.voltage) < 0.508) and (abs(channel.voltage) >= 0.250):
gain = 8
elif abs(channel.voltage) < 0.256:
gain = 16
self.exec_logger.debug(f'Setting gain to {gain}')
return gain
def interrupt(self):
"""Interrupts the acquisition. """
self.status = 'stopping'
if self.thread is not None:
self.thread.join()
self.exec_logger.debug('Interrupted sequence acquisition...')
else:
self.exec_logger.debug('No sequence measurement thread to interrupt.')
self.exec_logger.debug(f'Status: {self.status}')
def load_sequence(self, filename: str): def load_sequence(self, filename: str):
"""Reads quadrupole sequence from file. """Reads quadrupole sequence from file.
...@@ -352,257 +485,105 @@ class OhmPi(object): ...@@ -352,257 +485,105 @@ class OhmPi(object):
self.sequence = sequence self.sequence = sequence
def _switch_mux(self, electrode_nr, state, role): def measure(self, *args, **kwargs):
"""Selects the right channel for the multiplexer cascade for a given electrode. warnings.warn('This function is deprecated. Use run_multiple_sequences() instead.', DeprecationWarning)
self.run_multiple_sequences(self, *args, **kwargs)
Parameters
----------
electrode_nr : int
Electrode index to be switched on or off.
state : str
Either 'on' or 'off'.
role : str
Either 'A', 'B', 'M' or 'N', so we can assign it to a MUX board.
"""
if not self.use_mux or not self.on_pi:
if not self.on_pi:
self.exec_logger.warning('Cannot reset mux while in simulation mode...')
else:
self.exec_logger.warning('You cannot use the multiplexer because use_mux is set to False.'
' Set use_mux to True to use the multiplexer...')
elif self.sequence is None:
self.exec_logger.warning('Unable to switch MUX without a sequence')
else:
# choose with MUX board
tca = adafruit_tca9548a.TCA9548A(self.i2c, self.board_addresses[role])
# find I2C address of the electrode and corresponding relay
# considering that one MCP23017 can cover 16 electrodes
i2c_address = 7 - (electrode_nr - 1) // 16 # quotient without rest of the division
relay_nr = electrode_nr - (electrode_nr // 16) * 16 + 1
if i2c_address is not None:
# select the MCP23017 of the selected MUX board
mcp2 = MCP23017(tca[i2c_address])
mcp2.get_pin(relay_nr - 1).direction = digitalio.Direction.OUTPUT
if state == 'on':
mcp2.get_pin(relay_nr - 1).value = True
else:
mcp2.get_pin(relay_nr - 1).value = False
self.exec_logger.debug(f'Switching relay {relay_nr} '
f'({str(hex(self.board_addresses[role]))}) {state} for electrode {electrode_nr}')
else:
self.exec_logger.warning(f'Unable to address electrode nr {electrode_nr}')
def switch_mux_on(self, quadrupole): def _process_commands(self, message):
"""Switches on multiplexer relays for given quadrupole. """Processes commands received from the controller(s)
Parameters
----------
quadrupole : list of 4 int
List of 4 integers representing the electrode numbers.
"""
roles = ['A', 'B', 'M', 'N']
# another check to be sure A != B
if quadrupole[0] != quadrupole[1]:
for i in range(0, 4):
if quadrupole[i] > 0:
self._switch_mux(quadrupole[i], 'on', roles[i])
else:
self.exec_logger.error('A == B -> short circuit risk detected!')
def switch_mux_off(self, quadrupole):
"""Switches off multiplexer relays for given quadrupole.
Parameters Parameters
---------- ----------
quadrupole : list of 4 int message : str
List of 4 integers representing the electrode numbers. message containing a command and arguments or keywords and arguments
""" """
roles = ['A', 'B', 'M', 'N']
for i in range(0, 4):
if quadrupole[i] > 0:
self._switch_mux(quadrupole[i], 'off', roles[i])
def reset_mux(self): status = False
"""Switches off all multiplexer relays.""" cmd_id = '?'
if self.on_pi and self.use_mux: try:
roles = ['A', 'B', 'M', 'N'] decoded_message = json.loads(message)
for i in range(0, 4): self.exec_logger.debug(f'Decoded message {decoded_message}')
for j in range(1, self.max_elec + 1): cmd_id = decoded_message.pop('cmd_id', None)
self._switch_mux(j, 'off', roles[i]) cmd = decoded_message.pop('cmd', None)
self.exec_logger.debug('All MUX switched off.') args = decoded_message.pop('args', None)
elif not self.on_pi: if args is not None:
self.exec_logger.warning('Cannot reset mux while in simulation mode...') if len(args) != 0:
else: if args[0] != '[':
self.exec_logger.warning('You cannot use the multiplexer because use_mux is set to False.' args = f'["{args}"]'
' Set use_mux to True to use the multiplexer...') self.exec_logger.debug(f'args to decode: {args}')
args = json.loads(args) if args != '[]' else None
def _gain_auto(self, channel): self.exec_logger.debug(f'Decoded args {args}')
"""Automatically sets the gain on a channel else:
args = None
Parameters kwargs = decoded_message.pop('kwargs', None)
---------- if kwargs is not None:
channel : object if len(kwargs) != 0:
Instance of ADS where voltage is measured. if kwargs[0] != '{':
kwargs = '{"' + kwargs + '"}'
Returns self.exec_logger.debug(f'kwargs to decode: {kwargs}')
------- kwargs = json.loads(kwargs) if kwargs != '' else None
gain : float self.exec_logger.debug(f'Decoded kwargs {kwargs}')
Gain to be applied on ADS1115. else:
""" kwargs = None
gain = 2 / 3 self.exec_logger.debug(f"Calling method {cmd}({str(args) + ', ' if args is not None else ''}"
if (abs(channel.voltage) < 2.040) and (abs(channel.voltage) >= 1.023): f"{str(kwargs) if kwargs is not None else ''})")
gain = 2 if cmd_id is None:
elif (abs(channel.voltage) < 1.023) and (abs(channel.voltage) >= 0.508): self.exec_logger.warning('You should use a unique identifier for cmd_id')
gain = 4 if cmd is not None:
elif (abs(channel.voltage) < 0.508) and (abs(channel.voltage) >= 0.250): try:
gain = 8 if args is None:
elif abs(channel.voltage) < 0.256: if kwargs is None:
gain = 16 output = getattr(self, cmd)()
self.exec_logger.debug(f'Setting gain to {gain}') else:
return gain output = getattr(self, cmd)(**kwargs)
else:
if kwargs is None:
output = getattr(self, cmd)(*args)
else:
output = getattr(self, cmd)(*args, **kwargs)
status = True
except Exception as e:
self.exec_logger.error(
f"Unable to execute {cmd}({str(args) + ', ' if args is not None else ''}"
f"{str(kwargs) if kwargs is not None else ''}): {e}")
status = False
except Exception as e:
self.exec_logger.warning(f'Unable to decode command {message}: {e}')
status = False
finally:
reply = {'cmd_id': cmd_id, 'status': status}
reply = json.dumps(reply)
self.exec_logger.debug(f'Execution report: {reply}')
def _compute_tx_volt(self, best_tx_injtime=0.1, strategy='vmax', tx_volt=5): @staticmethod
"""Estimates best Tx voltage based on different strategies. def quit(self):
At first a half-cycle is made for a short duration with a fixed """Quits OhmPi"""
known voltage. This gives us Iab and Rab. We also measure Vmn.
A constant c = vmn/iab is computed (only depends on geometric
factor and ground resistivity, that doesn't change during a
quadrupole). Then depending on the strategy, we compute which
vab to inject to reach the minimum/maximum Iab current or
min/max Vmn.
This function also compute the polarity on Vmn (on which pin
of the ADS1115 we need to measure Vmn to get the positive value).
Parameters exit()
----------
best_tx_injtime : float, optional
Time in milliseconds for the half-cycle used to compute Rab.
strategy : str, optional
Either:
- vmin : compute Vab to reach a minimum Iab and Vmn
- vmax : compute Vab to reach a maximum Iab and Vmn
- constant : apply given Vab
tx_volt : float, optional
Voltage apply to try to guess the best voltage. 5 V applied
by default. If strategy "constant" is chosen, constant voltage
to applied is "tx_volt".
Returns def _read_hardware_config(self):
------- """Reads hardware configuration from config.py
vab : float
Proposed Vab according to the given strategy.
polarity : int
Either 1 or -1 to know on which pin of the ADS the Vmn is measured.
""" """
self.exec_logger.debug('Getting hardware config')
self.id = OHMPI_CONFIG['id'] # ID of the OhmPi
self.r_shunt = OHMPI_CONFIG['R_shunt'] # reference resistance value in ohm
self.Imax = OHMPI_CONFIG['Imax'] # maximum current
self.exec_logger.debug(f'The maximum current cannot be higher than {self.Imax} mA')
self.coef_p2 = OHMPI_CONFIG['coef_p2'] # slope for current conversion for ads.P2, measurement in V/V
self.nb_samples = OHMPI_CONFIG['nb_samples'] # number of samples measured for each stack
self.version = OHMPI_CONFIG['version'] # hardware version
self.max_elec = OHMPI_CONFIG['max_elec'] # maximum number of electrodes
self.board_addresses = OHMPI_CONFIG['board_addresses']
self.board_version = OHMPI_CONFIG['board_version']
self.exec_logger.debug(f'OHMPI_CONFIG = {str(OHMPI_CONFIG)}')
# hardware limits def read_quad(self, filename):
voltage_min = 10 # mV warnings.warn('This function is deprecated. Use load_sequence instead.', DeprecationWarning)
voltage_max = 4500 self.load_sequence(filename)
current_min = voltage_min / (self.r_shunt * 50) # mA
current_max = voltage_max / (self.r_shunt * 50)
tx_max = 40 # volt
# check of volt
volt = tx_volt
if volt > tx_max:
print('sorry, cannot inject more than 40 V, set it back to 5 V')
volt = 5
# redefined the pin of the mcp (needed when relays are connected)
self.pin0 = self.mcp.get_pin(0)
self.pin0.direction = Direction.OUTPUT
self.pin0.value = False
self.pin1 = self.mcp.get_pin(1)
self.pin1.direction = Direction.OUTPUT
self.pin1.value = False
# select a polarity to start with
self.pin0.value = True
self.pin1.value = False
# set voltage for test
self.DPS.write_register(0x0000, volt, 2)
self.DPS.write_register(0x09, 1) # DPS5005 on
time.sleep(best_tx_injtime) # inject for given tx time
# autogain
self.ads_current = ads.ADS1115(self.i2c, gain=2/3, data_rate=860, address=self.ads_current_address)
self.ads_voltage = ads.ADS1115(self.i2c, gain=2/3, data_rate=860, address=self.ads_voltage_address)
# print('current P0', AnalogIn(self.ads_current, ads.P0).voltage)
# print('voltage P0', AnalogIn(self.ads_voltage, ads.P0).voltage)
# print('voltage P2', AnalogIn(self.ads_voltage, ads.P2).voltage)
gain_current = self._gain_auto(AnalogIn(self.ads_current, ads.P0))
gain_voltage0 = self._gain_auto(AnalogIn(self.ads_voltage, ads.P0))
gain_voltage2 = self._gain_auto(AnalogIn(self.ads_voltage, ads.P2))
gain_voltage = np.min([gain_voltage0, gain_voltage2])
# print('gain current: {:.3f}, gain voltage: {:.3f}'.format(gain_current, gain_voltage))
self.ads_current = ads.ADS1115(self.i2c, gain=gain_current, data_rate=860, address=self.ads_current_address)
self.ads_voltage = ads.ADS1115(self.i2c, gain=gain_voltage, data_rate=860, address=self.ads_voltage_address)
# we measure the voltage on both A0 and A2 to guess the polarity
I = AnalogIn(self.ads_current, ads.P0).voltage * 1000. / 50 / self.r_shunt # measure current
U0 = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000. # measure voltage
U2 = AnalogIn(self.ads_voltage, ads.P2).voltage * 1000.
# print('I (mV)', I*50*self.r_shunt)
# print('I (mA)', I)
# print('U0 (mV)', U0)
# print('U2 (mV)', U2)
# check polarity
polarity = 1 # by default, we guessed it right
vmn = U0
if U0 < 0: # we guessed it wrong, let's use a correction factor
polarity = -1
vmn = U2
# print('polarity', polarity)
# compute constant
c = vmn / I
Rab = (volt * 1000.) / I
self.exec_logger.debug(f'Rab = {Rab:.2f} Ohms')
# implement different strategy
if strategy == 'vmax':
vmn_max = c * current_max
if voltage_max > vmn_max > voltage_min:
vab = current_max * Rab
self.exec_logger.debug('target max current')
else:
iab = voltage_max / c
vab = iab * Rab
self.exec_logger.debug('target max voltage')
if vab > 25000.:
vab = 25000.
vab = vab / 1000. * 0.9
elif strategy == 'vmin':
vmn_min = c * current_min
if voltage_min < vmn_min < voltage_max:
vab = current_min * Rab
self.exec_logger.debug('target min current')
else:
iab = voltage_min / c
vab = iab * Rab
self.exec_logger.debug('target min voltage')
if vab < 1000.:
vab = 1000.
vab = vab / 1000. * 1.1
elif strategy == 'constant':
vab = volt
else:
vab = 5
# self.DPS.write_register(0x09, 0) # DPS5005 off
self.pin0.value = False
self.pin1.value = False
return vab, polarity def restart(self):
self.exec_logger.info('Restarting pi...')
os.system('reboot')
def run_measurement(self, quad=None, nb_stack=None, injection_duration=None, def run_measurement(self, quad=None, nb_stack=None, injection_duration=None,
autogain=True, strategy='constant', tx_volt=5, best_tx_injtime=0.1, autogain=True, strategy='constant', tx_volt=5, best_tx_injtime=0.1,
...@@ -634,6 +615,8 @@ class OhmPi(object): ...@@ -634,6 +615,8 @@ class OhmPi(object):
measurement will be taken and values will be NaN. measurement will be taken and values will be NaN.
best_tx_injtime : float, optional best_tx_injtime : float, optional
(V3.0 only) Injection time in seconds used for finding the best voltage. (V3.0 only) Injection time in seconds used for finding the best voltage.
cmd_id :
""" """
self.exec_logger.debug('Starting measurement') self.exec_logger.debug('Starting measurement')
self.exec_logger.info('Waiting for data') self.exec_logger.info('Waiting for data')
...@@ -679,6 +662,8 @@ class OhmPi(object): ...@@ -679,6 +662,8 @@ class OhmPi(object):
address=self.ads_voltage_address, mode=0) address=self.ads_voltage_address, mode=0)
# turn on the power supply # turn on the power supply
start_delay = None
end_delay = None
out_of_range = False out_of_range = False
if self.idps: if self.idps:
if not np.isnan(tx_volt): if not np.isnan(tx_volt):
...@@ -740,6 +725,7 @@ class OhmPi(object): ...@@ -740,6 +725,7 @@ class OhmPi(object):
meas = np.zeros((self.nb_samples, 3)) * np.nan meas = np.zeros((self.nb_samples, 3)) * np.nan
start_delay = time.time() # stating measurement time start_delay = time.time() # stating measurement time
dt = 0 dt = 0
k = 0
for k in range(0, self.nb_samples): for k in range(0, self.nb_samples):
# reading current value on ADS channels # reading current value on ADS channels
meas[k, 0] = (AnalogIn(self.ads_current, ads.P0).voltage * 1000) / (50 * self.r_shunt) meas[k, 0] = (AnalogIn(self.ads_current, ads.P0).voltage * 1000) / (50 * self.r_shunt)
...@@ -824,6 +810,7 @@ class OhmPi(object): ...@@ -824,6 +810,7 @@ class OhmPi(object):
sum_i = np.nan sum_i = np.nan
sum_vmn = np.nan sum_vmn = np.nan
sum_ps = np.nan sum_ps = np.nan
fulldata = None
if self.idps: if self.idps:
self.DPS.write_register(0x0000, 0, 2) # reset to 0 volt self.DPS.write_register(0x0000, 0, 2) # reset to 0 volt
...@@ -883,6 +870,121 @@ class OhmPi(object): ...@@ -883,6 +870,121 @@ class OhmPi(object):
return d return d
def run_multiple_sequences(self, cmd_id=None, sequence_delay=None, nb_meas=None, **kwargs):
"""Runs multiple sequences in a separate thread for monitoring mode.
Can be stopped by 'OhmPi.interrupt()'.
Additional arguments are passed to run_measurement().
Parameters
----------
cmd_id :
sequence_delay : int, optional
Number of seconds at which the sequence must be started from each others.
nb_meas : int, optional
Number of time the sequence must be repeated.
kwargs : dict, optional
See help(k.run_measurement) for more info.
"""
# self.run = True
if sequence_delay is None:
sequence_delay = self.settings['sequence_delay']
sequence_delay = int(sequence_delay)
if nb_meas is None:
nb_meas = self.settings['nb_meas']
self.status = 'running'
self.exec_logger.debug(f'Status: {self.status}')
self.exec_logger.debug(f'Measuring sequence: {self.sequence}')
def func():
for g in range(0, nb_meas): # for time-lapse monitoring
if self.status == 'stopping':
self.exec_logger.warning('Data acquisition interrupted')
break
t0 = time.time()
self.run_sequence(**kwargs)
# sleeping time between sequence
dt = sequence_delay - (time.time() - t0)
if dt < 0:
dt = 0
if nb_meas > 1:
time.sleep(dt) # waiting for next measurement (time-lapse)
self.status = 'idle'
self.thread = threading.Thread(target=func)
self.thread.start()
def run_sequence(self, cmd_id=None, **kwargs):
"""Runs sequence synchronously (=blocking on main thread).
Additional arguments are passed to run_measurement().
"""
self.status = 'running'
self.exec_logger.debug(f'Status: {self.status}')
self.exec_logger.debug(f'Measuring sequence: {self.sequence}')
t0 = time.time()
# create filename with timestamp
filename = self.settings["export_path"].replace('.csv',
f'_{datetime.now().strftime("%Y%m%dT%H%M%S")}.csv')
self.exec_logger.debug(f'Saving to {filename}')
# make sure all multiplexer are off
self.reset_mux()
# measure all quadrupole of the sequence
if self.sequence is None:
n = 1
else:
n = self.sequence.shape[0]
for i in range(0, n):
if self.sequence is None:
quad = np.array([0, 0, 0, 0])
else:
quad = self.sequence[i, :] # quadrupole
if self.status == 'stopping':
break
# call the switch_mux function to switch to the right electrodes
self.switch_mux_on(quad)
# run a measurement
if self.on_pi:
acquired_data = self.run_measurement(quad, **kwargs)
else: # for testing, generate random data
acquired_data = {
'A': [quad[0]], 'B': [quad[1]], 'M': [quad[2]], 'N': [quad[3]],
'R [ohm]': np.abs(np.random.randn(1))
}
# switch mux off
self.switch_mux_off(quad)
# add command_id in dataset
acquired_data.update({'cmd_id': cmd_id})
# log data to the data logger
self.data_logger.info(f'{acquired_data}')
# save data and print in a text file
self.append_and_save(filename, acquired_data)
self.exec_logger.debug(f'quadrupole {i + 1:d}/{n:d}')
self.status = 'idle'
def run_sequence_async(self, cmd_id=None, **kwargs):
"""Runs the sequence in a separate thread. Can be stopped by 'OhmPi.interrupt()'.
Additional arguments are passed to run_measurement().
Parameters
----------
cmd_id:
"""
def func():
self.run_sequence(**kwargs)
self.thread = threading.Thread(target=func)
self.thread.start()
self.status = 'idle'
def rs_check(self, tx_volt=12): def rs_check(self, tx_volt=12):
"""Checks contact resistances""" """Checks contact resistances"""
# create custom sequence where MN == AB # create custom sequence where MN == AB
...@@ -935,112 +1037,28 @@ class OhmPi(object): ...@@ -935,112 +1037,28 @@ class OhmPi(object):
self.exec_logger.debug(msg) self.exec_logger.debug(msg)
# if contact resistance = 0 -> we have a short circuit!! # if contact resistance = 0 -> we have a short circuit!!
if resist < 1e-5: if resist < 1e-5:
msg = f'!!!SHORT CIRCUIT!!! {str(quad):s}: {resist:.3f} kOhm' msg = f'!!!SHORT CIRCUIT!!! {str(quad):s}: {resist:.3f} kOhm'
self.exec_logger.warning(msg) self.exec_logger.warning(msg)
print(msg)
# save data in a text file
# save data and print in a text file self.append_and_save(export_path_rs, {
self.append_and_save(export_path_rs, { 'A': quad[0],
'A': quad[0], 'B': quad[1],
'B': quad[1], 'RS [kOhm]': resist,
'RS [kOhm]': resist, })
})
# close mux path and put pin back to GND
# close mux path and put pin back to GND self.switch_mux_off(quad)
self.switch_mux_off(quad) else:
else: pass
pass self.status = 'idle'
self.status = 'idle' #
# # TODO if interrupted, we would need to restore the values
# # # TODO or we offer the possibility in 'run_measurement' to have rs_check each time?
# # TODO if interrupted, we would need to restore the values
# # TODO or we offer the possibility in 'run_measurement' to have rs_check each time? def set_sequence(self, sequence=None):
@staticmethod
def append_and_save(filename, last_measurement):
"""Appends and saves the last measurement dict.
Parameters
----------
filename : str
filename to save the last measurement dataframe
last_measurement : dict
Last measurement taken in the form of a python dictionary
"""
last_measurement = deepcopy(last_measurement)
if 'fulldata' in last_measurement:
d = last_measurement['fulldata']
n = d.shape[0]
if n > 1:
idic = dict(zip(['i' + str(i) for i in range(n)], d[:, 0]))
udic = dict(zip(['u' + str(i) for i in range(n)], d[:, 1]))
tdic = dict(zip(['t' + str(i) for i in range(n)], d[:, 2]))
last_measurement.update(idic)
last_measurement.update(udic)
last_measurement.update(tdic)
last_measurement.pop('fulldata')
if os.path.isfile(filename):
# Load data file and append data to it
with open(filename, 'a') as f:
w = csv.DictWriter(f, last_measurement.keys())
w.writerow(last_measurement)
# last_measurement.to_csv(f, header=False)
else:
# create data file and add headers
with open(filename, 'a') as f:
w = csv.DictWriter(f, last_measurement.keys())
w.writeheader()
w.writerow(last_measurement)
def _process_commands(self, message):
"""Processes commands received from the controller(s)
Parameters
----------
message : str
message containing a command and arguments or keywords and arguments
"""
try:
decoded_message = json.loads(message)
print(f'decoded message: {decoded_message}')
cmd_id = decoded_message.pop('cmd_id', None)
cmd = decoded_message.pop('cmd', None)
args = decoded_message.pop('args', '')
if len(args) == 0:
args = f'["{args}"]'
args = json.loads(args)
kwargs = decoded_message.pop('kwargs', '')
if len(kwargs) == 0:
kwargs = '"{}"'
kwargs = json.loads(kwargs)
self.exec_logger.debug(f'Calling method {cmd}({args}, {kwargs})')
status = False
# e = None # NOTE: Why this?
print(cmd, args, kwargs)
if cmd_id is None:
self.exec_logger.warning('You should use a unique identifier for cmd_id')
if cmd is not None:
try:
output = getattr(self, cmd)(*args, **kwargs)
status = True
except Exception as e:
self.exec_logger.error(
f"{e}\nUnable to execute {cmd}({args + ', ' if args != '[]' else ''}"
f"{kwargs if kwargs != '{}' else ''})")
status = False
except Exception as e:
self.exec_logger.warning(f'Unable to decode command {message}: {e}')
status = False
finally:
reply = {'cmd_id': cmd_id, 'status': status}
reply = json.dumps(reply)
self.exec_logger.debug(f'Execution report: {reply}')
def set_sequence(self, sequence=sequence):
try: try:
self.sequence = np.loadtxt(StringIO(sequence)).astype('uint32') self.sequence = np.loadtxt(StringIO(sequence)).astype('uint32')
status = True status = True
...@@ -1048,150 +1066,152 @@ class OhmPi(object): ...@@ -1048,150 +1066,152 @@ class OhmPi(object):
self.exec_logger.warning(f'Unable to set sequence: {e}') self.exec_logger.warning(f'Unable to set sequence: {e}')
status = False status = False
def run_sequence(self, cmd_id=None, **kwargs): def stop(self):
"""Runs sequence synchronously (=blocking on main thread). warnings.warn('This function is deprecated. Use interrupt instead.', DeprecationWarning)
Additional arguments are passed to run_measurement(). self.interrupt()
"""
self.status = 'running'
self.exec_logger.debug(f'Status: {self.status}')
self.exec_logger.debug(f'Measuring sequence: {self.sequence}')
t0 = time.time()
# create filename with timestamp
filename = self.settings["export_path"].replace('.csv',
f'_{datetime.now().strftime("%Y%m%dT%H%M%S")}.csv')
self.exec_logger.debug(f'Saving to {filename}')
# make sure all multiplexer are off def _switch_mux(self, electrode_nr, state, role):
self.reset_mux() """Selects the right channel for the multiplexer cascade for a given electrode.
# measure all quadrupole of the sequence Parameters
if self.sequence is None: ----------
n = 1 electrode_nr : int
else: Electrode index to be switched on or off.
n = self.sequence.shape[0] state : str
for i in range(0, n): Either 'on' or 'off'.
if self.sequence is None: role : str
quad = np.array([0, 0, 0, 0]) Either 'A', 'B', 'M' or 'N', so we can assign it to a MUX board.
"""
if not self.use_mux or not self.on_pi:
if not self.on_pi:
self.exec_logger.warning('Cannot reset mux while in simulation mode...')
else: else:
quad = self.sequence[i, :] # quadrupole self.exec_logger.warning('You cannot use the multiplexer because use_mux is set to False.'
if self.status == 'stopping': ' Set use_mux to True to use the multiplexer...')
break elif self.sequence is None:
self.exec_logger.warning('Unable to switch MUX without a sequence')
# call the switch_mux function to switch to the right electrodes else:
self.switch_mux_on(quad) # choose with MUX board
tca = adafruit_tca9548a.TCA9548A(self.i2c, self.board_addresses[role])
# run a measurement
if self.on_pi:
acquired_data = self.run_measurement(quad, **kwargs)
else: # for testing, generate random data
acquired_data = {
'A': [quad[0]], 'B': [quad[1]], 'M': [quad[2]], 'N': [quad[3]],
'R [ohm]': np.abs(np.random.randn(1))
}
# switch mux off # find I2C address of the electrode and corresponding relay
self.switch_mux_off(quad) # considering that one MCP23017 can cover 16 electrodes
i2c_address = 7 - (electrode_nr - 1) // 16 # quotient without rest of the division
relay_nr = electrode_nr - (electrode_nr // 16) * 16 + 1
# add command_id in dataset if i2c_address is not None:
acquired_data.update({'cmd_id': cmd_id}) # select the MCP23017 of the selected MUX board
# log data to the data logger mcp2 = MCP23017(tca[i2c_address])
self.data_logger.info(f'{acquired_data}') mcp2.get_pin(relay_nr - 1).direction = digitalio.Direction.OUTPUT
# save data and print in a text file
self.append_and_save(filename, acquired_data)
self.exec_logger.debug(f'quadrupole {i+1:d}/{n:d}')
self.status = 'idle' if state == 'on':
mcp2.get_pin(relay_nr - 1).value = True
else:
mcp2.get_pin(relay_nr - 1).value = False
def run_sequence_async(self, cmd_id=None, **kwargs): self.exec_logger.debug(f'Switching relay {relay_nr} '
"""Runs the sequence in a separate thread. Can be stopped by 'OhmPi.interrupt()'. f'({str(hex(self.board_addresses[role]))}) {state} for electrode {electrode_nr}')
Additional arguments are passed to run_measurement(). else:
self.exec_logger.warning(f'Unable to address electrode nr {electrode_nr}')
Parameters def switch_mux_on(self, quadrupole):
---------- """Switches on multiplexer relays for given quadrupole.
cmd_id:
Parameters
----------
quadrupole : list of 4 int
List of 4 integers representing the electrode numbers.
""" """
roles = ['A', 'B', 'M', 'N']
# another check to be sure A != B
if quadrupole[0] != quadrupole[1]:
for i in range(0, 4):
if quadrupole[i] > 0:
self._switch_mux(quadrupole[i], 'on', roles[i])
else:
self.exec_logger.error('Not switching MUX : A == B -> short circuit risk detected!')
def func(): def switch_mux_off(self, quadrupole):
self.run_sequence(**kwargs) """Switches off multiplexer relays for given quadrupole.
self.thread = threading.Thread(target=func)
self.thread.start()
self.status = 'idle'
def measure(self, *args, **kwargs):
warnings.warn('This function is deprecated. Use run_multiple_sequences() instead.', DeprecationWarning)
self.run_multiple_sequences(self, *args, **kwargs)
def run_multiple_sequences(self, cmd_id=None, sequence_delay=None, nb_meas=None, **kwargs):
"""Runs multiple sequences in a separate thread for monitoring mode.
Can be stopped by 'OhmPi.interrupt()'.
Additional arguments are passed to run_measurement().
Parameters Parameters
---------- ----------
sequence_delay : int, optional quadrupole : list of 4 int
Number of seconds at which the sequence must be started from each others. List of 4 integers representing the electrode numbers.
nb_meas : int, optional
Number of time the sequence must be repeated.
kwargs : dict, optional
See help(k.run_measurement) for more info.
""" """
# self.run = True roles = ['A', 'B', 'M', 'N']
if sequence_delay is None: for i in range(0, 4):
sequence_delay = self.settings['sequence_delay'] if quadrupole[i] > 0:
sequence_delay = int(sequence_delay) self._switch_mux(quadrupole[i], 'off', roles[i])
if nb_meas is None:
nb_meas = self.settings['nb_meas']
self.status = 'running'
self.exec_logger.debug(f'Status: {self.status}')
self.exec_logger.debug(f'Measuring sequence: {self.sequence}')
def func(): def reset_mux(self):
for g in range(0, nb_meas): # for time-lapse monitoring """Switches off all multiplexer relays."""
if self.status == 'stopping': if self.on_pi and self.use_mux:
self.exec_logger.warning('Data acquisition interrupted') roles = ['A', 'B', 'M', 'N']
break for i in range(0, 4):
t0 = time.time() for j in range(1, self.max_elec + 1):
self.run_sequence(**kwargs) self._switch_mux(j, 'off', roles[i])
self.exec_logger.debug('All MUX switched off.')
elif not self.on_pi:
self.exec_logger.warning('Cannot reset mux while in simulation mode...')
else:
self.exec_logger.warning('You cannot use the multiplexer because use_mux is set to False.'
' Set use_mux to True to use the multiplexer...')
# sleeping time between sequence def _update_acquisition_settings(self, config):
dt = sequence_delay - (time.time() - t0) warnings.warn('This function is deprecated, use update_settings() instead.', DeprecationWarning)
if dt < 0: self.update_settings(config)
dt = 0
if nb_meas > 1:
time.sleep(dt) # waiting for next measurement (time-lapse)
self.status = 'idle'
self.thread = threading.Thread(target=func)
self.thread.start()
def stop(self): def update_settings(self, config):
warnings.warn('This function is deprecated. Use interrupt instead.', DeprecationWarning) """Updates acquisition settings from a json file or dictionary.
self.interrupt() Parameters can be:
- nb_electrodes (number of electrode used, if 4, no MUX needed)
- injection_duration (in seconds)
- nb_meas (total number of times the sequence will be run)
- sequence_delay (delay in second between each sequence run)
- nb_stack (number of stack for each quadrupole measurement)
- export_path (path where to export the data, timestamp will be added to filename)
def interrupt(self): Parameters
"""Interrupts the acquisition. """ ----------
self.status = 'stopping' config : str, dict
if self.thread is not None: Path to the .json settings file or dictionary of settings.
self.exec_logger.debug('Joining tread...') """
self.thread.join() status = False
if config is not None:
try:
if isinstance(config, dict):
self.settings.update(config)
else:
with open(config) as json_file:
dic = json.load(json_file)
self.settings.update(dic)
self.exec_logger.debug('Acquisition parameters updated: ' + str(self.settings))
status = True
except Exception as e:
self.exec_logger.warning('Unable to update settings.')
status = False
else: else:
self.exec_logger.debug('No sequence measurement thread to interrupt.') self.exec_logger.warning('Settings are missing...')
self.exec_logger.debug(f'Status: {self.status}') return status
def quit(self): # Properties
"""Quit OhmPi. @property
""" def sequence(self):
self.cmd_listen = False """Gets sequence"""
if self.cmd_thread is not None: if self._sequence is not None:
self.cmd_thread.join() assert isinstance(self._sequence, np.ndarray)
self.exec_logger.debug(f'Stopped listening to control topic.') return self._sequence
exit()
def restart(self): @sequence.setter
self.exec_logger.info('Restarting pi...') def sequence(self, sequence):
os.system('reboot') """Sets sequence"""
if sequence is not None:
assert isinstance(sequence, np.ndarray)
self.use_mux = True
else:
self.use_mux = False
self._sequence = sequence
VERSION = '2.1.5' VERSION = '2.1.5'
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment