Commit 95adb832 authored by Clement Remi's avatar Clement Remi
Browse files

Replace ohmpi.py

Showing with 96 additions and 69 deletions
+96 -69
...@@ -119,6 +119,9 @@ class OhmPi(object): ...@@ -119,6 +119,9 @@ class OhmPi(object):
# I2C connexion to MCP23008, for current injection # I2C connexion to MCP23008, for current injection
self.mcp = MCP23008(self.i2c, address=0x20) self.mcp = MCP23008(self.i2c, address=0x20)
self.pin4 = self.mcp.get_pin(4) # Ohmpi_run
self.pin4.direction = Direction.OUTPUT
self.pin4.value = True
# ADS1115 for current measurement (AB) # ADS1115 for current measurement (AB)
self.ads_current_address = 0x48 self.ads_current_address = 0x48
...@@ -130,6 +133,13 @@ class OhmPi(object): ...@@ -130,6 +133,13 @@ class OhmPi(object):
# current injection module # current injection module
if self.idps: if self.idps:
self.pin2 = self.mcp.get_pin(2) # dsp +
self.pin2.direction = Direction.OUTPUT
self.pin2.value = True
self.pin3 = self.mcp.get_pin(3) # dsp -
self.pin3.direction = Direction.OUTPUT
self.pin3.value = True
time.sleep(3)
self.DPS = minimalmodbus.Instrument(port='/dev/ttyUSB0', slaveaddress=1) # port name, address (decimal) self.DPS = minimalmodbus.Instrument(port='/dev/ttyUSB0', slaveaddress=1) # port name, address (decimal)
self.DPS.serial.baudrate = 9600 # Baud rate 9600 as listed in doc self.DPS.serial.baudrate = 9600 # Baud rate 9600 as listed in doc
self.DPS.serial.bytesize = 8 # self.DPS.serial.bytesize = 8 #
...@@ -137,7 +147,11 @@ class OhmPi(object): ...@@ -137,7 +147,11 @@ class OhmPi(object):
self.DPS.debug = False # self.DPS.debug = False #
self.DPS.serial.parity = 'N' # No parity self.DPS.serial.parity = 'N' # No parity
self.DPS.mode = minimalmodbus.MODE_RTU # RTU mode self.DPS.mode = minimalmodbus.MODE_RTU # RTU mode
self.DPS.write_register(0x0001, 40, 0) # max current allowed (36 mA for relays) self.DPS.write_register(0x0001, 100, 0) # max current allowed (100 mA for relays)
print(self.DPS.read_register(0x05,2 )) # max current allowed (100 mA for relays) #voltage
self.pin2.value = False
self.pin3.value = False
# (last number) 0 is for mA, 3 is for A # (last number) 0 is for mA, 3 is for A
# injection courant and measure (TODO check if it works, otherwise back in run_measurement()) # injection courant and measure (TODO check if it works, otherwise back in run_measurement())
...@@ -262,7 +276,6 @@ class OhmPi(object): ...@@ -262,7 +276,6 @@ class OhmPi(object):
Time in milliseconds for the half-cycle used to compute Rab. Time in milliseconds for the half-cycle used to compute Rab.
strategy : str, optional strategy : str, optional
Either: Either:
- vmin : compute Vab to reach a minimum Iab and Vmn
- vmax : compute Vab to reach a maximum Iab and Vmn - vmax : compute Vab to reach a maximum Iab and Vmn
- constant : apply given Vab - constant : apply given Vab
tx_volt : float, optional tx_volt : float, optional
...@@ -306,7 +319,7 @@ class OhmPi(object): ...@@ -306,7 +319,7 @@ class OhmPi(object):
vmn=0 vmn=0
count=0 count=0
while I < 1 or abs(vmn)<5 : # I supérieur à 1 mA et Vmn surpérieur while I < 3 or abs(vmn) < 10 : # I supérieur à 1 mA et Vmn surpérieur
if count >0 : if count >0 :
volt = volt + 2 volt = volt + 2
count=count+1 count=count+1
...@@ -316,7 +329,7 @@ class OhmPi(object): ...@@ -316,7 +329,7 @@ class OhmPi(object):
self.DPS.write_register(0x0000, volt, 2) self.DPS.write_register(0x0000, volt, 2)
self.DPS.write_register(0x09, 1) # DPS5005 on self.DPS.write_register(0x09, 1) # DPS5005 on
time.sleep(best_tx_injtime) # inject for given tx time time.sleep(best_tx_injtime) # inject for given tx time
# autogain # autogain
self.ads_current = ads.ADS1115(self.i2c, gain=2 / 3, data_rate=860, address=self.ads_current_address) 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) self.ads_voltage = ads.ADS1115(self.i2c, gain=2 / 3, data_rate=860, address=self.ads_voltage_address)
...@@ -324,31 +337,37 @@ class OhmPi(object): ...@@ -324,31 +337,37 @@ class OhmPi(object):
gain_voltage0 = self._gain_auto(AnalogIn(self.ads_voltage, 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_voltage2 = self._gain_auto(AnalogIn(self.ads_voltage, ads.P2))
gain_voltage = np.min([gain_voltage0, gain_voltage2]) 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_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) 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 # 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 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. # noqa measure voltage U0 = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000. # noqa measure voltage
U2 = AnalogIn(self.ads_voltage, ads.P2).voltage * 1000. # noqa U2 = AnalogIn(self.ads_voltage, ads.P2).voltage * 1000. # noqa
self.DPS.write_register(0x09, 0) # DPS5005 off
# print('I (mV)', I*50*self.r_shunt)
# print('I (mA)', I)
# print('U0 (mV)', U0)
# print('U2 (mV)', U2)
# check polarity # check polarity
polarity = 1 # by default, we guessed it right polarity = 1 # by default, we guessed it right
vmn = U0 vmn = U0
if U0 < 0: # we guessed it wrong, let's use a correction factor if U0 < 0: # we guessed it wrong, let's use a correction factor
polarity = -1 polarity = -1
vmn = U2 vmn = U2
if abs(vmn)>4500 or I> 45 :
volt = volt - 2
self.DPS.write_register(0x0000, volt, 2)
self.DPS.write_register(0x09, 1) # DPS5005 on
time.sleep(best_tx_injtime)
I = AnalogIn(self.ads_current, ads.P0).voltage * 1000. / 50 / self.r_shunt
U0 = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000.
U2 = AnalogIn(self.ads_voltage, ads.P2).voltage * 1000.
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
break
self.DPS.write_register(0x09, 0) # DPS5005 off
# print('polarity', polarity) # print('polarity', polarity)
self.pin0.value = False self.pin0.value = False
self.pin1.value = False self.pin1.value = False
...@@ -359,43 +378,16 @@ class OhmPi(object): ...@@ -359,43 +378,16 @@ class OhmPi(object):
self.exec_logger.debug(f'Rab = {Rab:.2f} Ohms') self.exec_logger.debug(f'Rab = {Rab:.2f} Ohms')
# implement different strategy # implement different strategy
if strategy == 'vmax_rc': if strategy == 'vmax':
factor_I = current_max / I factor_I = (current_max) / I
factor_vmn = voltage_max / Vmn factor_vmn = voltage_max / vmn
factor= factor_I factor = factor_I
if factor_I > factor_vmn: if factor_I > factor_vmn:
factor = factor_vmn factor = factor_vmn
Vab = factor * tx_volt vab = factor * volt * 0.95
if Vab > tx_max: if vab > tx_max:
Vab=tx_max vab = tx_max
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': elif strategy == 'constant':
vab = volt vab = volt
else: else:
...@@ -558,7 +550,6 @@ class OhmPi(object): ...@@ -558,7 +550,6 @@ class OhmPi(object):
self.exec_logger.info(f'Sequence {filename} of {sequence.shape[0]:d} quadrupoles loaded.') self.exec_logger.info(f'Sequence {filename} of {sequence.shape[0]:d} quadrupoles loaded.')
else: else:
self.exec_logger.warning(f'Unable to load sequence {filename}') self.exec_logger.warning(f'Unable to load sequence {filename}')
self.sequence = sequence self.sequence = sequence
def measure(self, **kwargs): def measure(self, **kwargs):
...@@ -573,7 +564,6 @@ class OhmPi(object): ...@@ -573,7 +564,6 @@ class OhmPi(object):
message : str message : str
message containing a command and arguments or keywords and arguments message containing a command and arguments or keywords and arguments
""" """
status = False status = False
cmd_id = '?' cmd_id = '?'
try: try:
...@@ -656,7 +646,7 @@ class OhmPi(object): ...@@ -656,7 +646,7 @@ class OhmPi(object):
self.version = OHMPI_CONFIG['version'] # hardware version self.version = OHMPI_CONFIG['version'] # hardware version
self.max_elec = OHMPI_CONFIG['max_elec'] # maximum number of electrodes self.max_elec = OHMPI_CONFIG['max_elec'] # maximum number of electrodes
self.board_addresses = OHMPI_CONFIG['board_addresses'] self.board_addresses = OHMPI_CONFIG['board_addresses']
self.board_version = OHMPI_CONFIG['measurement board_version'] self.board_version = OHMPI_CONFIG['board_version']
self.exec_logger.debug(f'OHMPI_CONFIG = {str(OHMPI_CONFIG)}') self.exec_logger.debug(f'OHMPI_CONFIG = {str(OHMPI_CONFIG)}')
def read_quad(self, **kwargs): def read_quad(self, **kwargs):
...@@ -710,9 +700,7 @@ class OhmPi(object): ...@@ -710,9 +700,7 @@ class OhmPi(object):
resolution of the reading. resolution of the reading.
strategy : str, optional strategy : str, optional
(V3.0 only) If we search for best voltage (tx_volt == 0), we can choose (V3.0 only) If we search for best voltage (tx_volt == 0), we can choose
different strategy: vmax strategy : find the highest voltage that stays in the range
- vmin: find the lowest voltage that gives us a signal
- vmax: find the highest voltage that stays in the range
For a constant value, just set the tx_volt. For a constant value, just set the tx_volt.
tx_volt : float, optional tx_volt : float, optional
(V3.0 only) If specified, voltage will be imposed. If 0, we will look (V3.0 only) If specified, voltage will be imposed. If 0, we will look
...@@ -751,7 +739,32 @@ class OhmPi(object): ...@@ -751,7 +739,32 @@ class OhmPi(object):
self.pin1 = self.mcp.get_pin(1) self.pin1 = self.mcp.get_pin(1)
self.pin1.direction = Direction.OUTPUT self.pin1.direction = Direction.OUTPUT
self.pin1.value = False self.pin1.value = False
self.pin7 = self.mcp.get_pin(7) #IHM on mesaurement
self.pin7.direction = Direction.OUTPUT
self.pin7.value = False
if self.sequence == None :
self.pin2 = self.mcp.get_pin(2) # dsp +
self.pin2.direction = Direction.OUTPUT
self.pin2.value = True
self.pin3 = self.mcp.get_pin(3) # dsp -
self.pin3.direction = Direction.OUTPUT
self.pin3.value = True
self.pin5 = self.mcp.get_pin(5) #IHM on mesaurement
self.pin5.direction = Direction.OUTPUT
self.pin5.value = True
self.pin6 = self.mcp.get_pin(6) #IHM on mesaurement
self.pin6.direction = Direction.OUTPUT
self.pin6.value = False
self.pin7 = self.mcp.get_pin(7) #IHM on mesaurement
self.pin7.direction = Direction.OUTPUT
self.pin7.value = False
if self.DPS.read_register(0x05,2) < 11:
self.pin7.value = True# max current allowed (100 mA for relays) #voltage
# get best voltage to inject AND polarity # get best voltage to inject AND polarity
if self.idps: if self.idps:
tx_volt, polarity = self._compute_tx_volt( tx_volt, polarity = self._compute_tx_volt(
...@@ -781,10 +794,11 @@ class OhmPi(object): ...@@ -781,10 +794,11 @@ class OhmPi(object):
if not out_of_range: # we found a Vab in the range so we measure if not out_of_range: # we found a Vab in the range so we measure
if autogain: if autogain:
if self.board_version == '22.11': if self.board_version == 'mb.2023.0.0':
# compute autogain # compute autogain
self.pin0.value = True self.pin0.value = True
self.pin1.value = False self.pin1.value = False
self.pin6.value = True # IHM current injection led on
time.sleep(injection_duration) time.sleep(injection_duration)
gain_current = self._gain_auto(AnalogIn(self.ads_current, ads.P0)) gain_current = self._gain_auto(AnalogIn(self.ads_current, ads.P0))
if polarity > 0: if polarity > 0:
...@@ -793,12 +807,13 @@ class OhmPi(object): ...@@ -793,12 +807,13 @@ class OhmPi(object):
gain_voltage = self._gain_auto(AnalogIn(self.ads_voltage, ads.P2)) gain_voltage = self._gain_auto(AnalogIn(self.ads_voltage, ads.P2))
self.pin0.value = False self.pin0.value = False
self.pin1.value = False self.pin1.value = False
self.pin6.value = False # IHM current injection led off
self.exec_logger.debug(f'Gain current: {gain_current:.3f}, gain voltage: {gain_voltage:.3f}') self.exec_logger.debug(f'Gain current: {gain_current:.3f}, gain voltage: {gain_voltage:.3f}')
self.ads_current = ads.ADS1115(self.i2c, gain=gain_current, data_rate=860, self.ads_current = ads.ADS1115(self.i2c, gain=gain_current, data_rate=860,
address=self.ads_current_address, mode=0) address=self.ads_current_address, mode=0)
self.ads_voltage = ads.ADS1115(self.i2c, gain=gain_voltage, data_rate=860, self.ads_voltage = ads.ADS1115(self.i2c, gain=gain_voltage, data_rate=860,
address=self.ads_voltage_address, mode=0) address=self.ads_voltage_address, mode=0)
elif self.board_version == 'mb.2023.0.0': elif self.board_version == '22.10':
gain_current = 2 / 3 gain_current = 2 / 3
gain_voltage = 2 / 3 gain_voltage = 2 / 3
self.ads_current = ads.ADS1115(self.i2c, gain=gain_current, data_rate=860, self.ads_current = ads.ADS1115(self.i2c, gain=gain_current, data_rate=860,
...@@ -830,9 +845,11 @@ class OhmPi(object): ...@@ -830,9 +845,11 @@ class OhmPi(object):
if (n % 2) == 0: if (n % 2) == 0:
self.pin0.value = True self.pin0.value = True
self.pin1.value = False self.pin1.value = False
self.pin6.value = True# IHM current injection led on
else: else:
self.pin0.value = False self.pin0.value = False
self.pin1.value = True # current injection nr2 self.pin1.value = True # current injection nr2
self.pin6.value = True# IHM current injection led on
self.exec_logger.debug(f'Stack {n} {self.pin0.value} {self.pin1.value}') self.exec_logger.debug(f'Stack {n} {self.pin0.value} {self.pin1.value}')
# measurement of current i and voltage u during injection # measurement of current i and voltage u during injection
...@@ -843,12 +860,12 @@ class OhmPi(object): ...@@ -843,12 +860,12 @@ class OhmPi(object):
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)
if self.board_version == '22.11': if self.board_version == 'mb.2023.0.0':
if pinMN == 0: if pinMN == 0:
meas[k, 1] = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000 meas[k, 1] = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000
else: else:
meas[k, 1] = -AnalogIn(self.ads_voltage, ads.P2).voltage * 1000 meas[k, 1] = -AnalogIn(self.ads_voltage, ads.P2).voltage * 1000
elif self.board_version == 'mb.2023.0.0': elif self.board_version == '22.10':
meas[k, 1] = -AnalogIn(self.ads_voltage, ads.P0, ads.P1).voltage * self.coef_p2 * 1000 meas[k, 1] = -AnalogIn(self.ads_voltage, ads.P0, ads.P1).voltage * self.coef_p2 * 1000
# else: # else:
# self.exec_logger.debug('Unknown board') # self.exec_logger.debug('Unknown board')
...@@ -861,6 +878,7 @@ class OhmPi(object): ...@@ -861,6 +878,7 @@ class OhmPi(object):
# stop current injection # stop current injection
self.pin0.value = False self.pin0.value = False
self.pin1.value = False self.pin1.value = False
self.pin6.value = False# IHM current injection led on
end_delay = time.time() end_delay = time.time()
# truncate the meas array if we didn't fill the last samples # truncate the meas array if we didn't fill the last samples
...@@ -873,12 +891,12 @@ class OhmPi(object): ...@@ -873,12 +891,12 @@ class OhmPi(object):
for k in range(0, measpp.shape[0]): for k in range(0, measpp.shape[0]):
# reading current value on ADS channels # reading current value on ADS channels
measpp[k, 0] = (AnalogIn(self.ads_current, ads.P0).voltage * 1000.) / (50 * self.r_shunt) measpp[k, 0] = (AnalogIn(self.ads_current, ads.P0).voltage * 1000.) / (50 * self.r_shunt)
if self.board_version == '22.11': if self.board_version == 'mb.2023.0.0':
if pinMN == 0: if pinMN == 0:
measpp[k, 1] = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000. measpp[k, 1] = AnalogIn(self.ads_voltage, ads.P0).voltage * 1000.
else: else:
measpp[k, 1] = AnalogIn(self.ads_voltage, ads.P2).voltage * 1000. * -1 measpp[k, 1] = AnalogIn(self.ads_voltage, ads.P2).voltage * 1000. * -1
elif self.board_version == 'mb.2023.0.0': elif self.board_version == '22.10':
measpp[k, 1] = -AnalogIn(self.ads_voltage, ads.P0, ads.P1).voltage * self.coef_p2 * 1000. measpp[k, 1] = -AnalogIn(self.ads_voltage, ads.P0, ads.P1).voltage * self.coef_p2 * 1000.
else: else:
self.exec_logger.debug('unknown board') self.exec_logger.debug('unknown board')
...@@ -944,7 +962,7 @@ class OhmPi(object): ...@@ -944,7 +962,7 @@ class OhmPi(object):
np.array([[]]) np.array([[]])
# create a dictionary and compute averaged values from all stacks # create a dictionary and compute averaged values from all stacks
if self.board_version == '22.11': if self.board_version == 'mb.2023.0.0':
d = { d = {
"time": datetime.now().isoformat(), "time": datetime.now().isoformat(),
"A": quad[0], "A": quad[0],
...@@ -962,7 +980,7 @@ class OhmPi(object): ...@@ -962,7 +980,7 @@ class OhmPi(object):
"Nb samples [-]": self.nb_samples, "Nb samples [-]": self.nb_samples,
"fulldata": fulldata, "fulldata": fulldata,
} }
elif self.board_version == 'mb.2023.0.0': elif self.board_version == '22.10':
d = { d = {
"time": datetime.now().isoformat(), "time": datetime.now().isoformat(),
"A": quad[0], "A": quad[0],
...@@ -1000,7 +1018,10 @@ class OhmPi(object): ...@@ -1000,7 +1018,10 @@ class OhmPi(object):
dd['cmd_id'] = str(cmd_id) dd['cmd_id'] = str(cmd_id)
self.data_logger.info(dd) self.data_logger.info(dd)
self.pin5.value = False #IHM led on measurement off
if self.sequence == None :
self.pin2.value = False # DSP + off
self.pin3.value = False # DSP - off
return d return d
def run_multiple_sequences(self, cmd_id=None, sequence_delay=None, nb_meas=None, **kwargs): def run_multiple_sequences(self, cmd_id=None, sequence_delay=None, nb_meas=None, **kwargs):
...@@ -1061,7 +1082,12 @@ class OhmPi(object): ...@@ -1061,7 +1082,12 @@ class OhmPi(object):
self.exec_logger.debug(f'Status: {self.status}') self.exec_logger.debug(f'Status: {self.status}')
self.exec_logger.debug(f'Measuring sequence: {self.sequence}') self.exec_logger.debug(f'Measuring sequence: {self.sequence}')
t0 = time.time() t0 = time.time()
self.pin2 = self.mcp.get_pin(2) # dsp -
self.pin2.direction = Direction.OUTPUT
self.pin2.value = True
self.pin3 = self.mcp.get_pin(3) # dsp -
self.pin3.direction = Direction.OUTPUT
self.pin3.value = True
# create filename with timestamp # create filename with timestamp
filename = self.settings["export_path"].replace('.csv', filename = self.settings["export_path"].replace('.csv',
f'_{datetime.now().strftime("%Y%m%dT%H%M%S")}.csv') f'_{datetime.now().strftime("%Y%m%dT%H%M%S")}.csv')
...@@ -1121,7 +1147,8 @@ class OhmPi(object): ...@@ -1121,7 +1147,8 @@ class OhmPi(object):
# save data and print in a text file # save data and print in a text file
self.append_and_save(filename, acquired_data) self.append_and_save(filename, acquired_data)
self.exec_logger.debug(f'quadrupole {i + 1:d}/{n:d}') self.exec_logger.debug(f'quadrupole {i + 1:d}/{n:d}')
self.pin2.value = True
self.pin3.value = True
self.status = 'idle' self.status = 'idle'
def run_sequence_async(self, cmd_id=None, **kwargs): def run_sequence_async(self, cmd_id=None, **kwargs):
......
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