# definition of hardware level functions
import numpy as np
import board # noqa
import busio # noqa
import adafruit_tca9548a # noqa
import adafruit_ads1x15.ads1115 as ads # noqa
from adafruit_ads1x15.analog_in import AnalogIn # noqa
from adafruit_mcp230xx.mcp23008 import MCP23008 # noqa
from adafruit_mcp230xx.mcp23017 import MCP23017 # noqa
import digitalio # noqa
from digitalio import Direction # noqa
from gpiozero import CPUTemperature # noqa
import minimalmodbus # noqa
import time
# global variable
i2c = busio.I2C(board.SCL, board.SDA)
from config import OHMPI_CONFIG
class Alimentation():
def __init__(self, address=0x20, tx_voltage=12):
self.mcp = MCP23017(i2c, address=address)
self.tx_voltage = tx_voltage
self.polarity = True
self.on = False
self.pinA = 0
self.pinB = 1
# setup DPS
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.bytesize = 8 #
self.DPS.serial.timeout = 1 # greater than 0.5 for it to work
self.DPS.debug = False #
self.DPS.serial.parity = 'N' # No parity
self.DPS.mode = minimalmodbus.MODE_RTU # RTU mode
self.DPS.write_register(0x0001, 40, 0) # max current allowed (36 mA for relays)
# (last number) 0 is for mA, 3 is for A
def turn_on(self):
if self.on is False:
self.DPS.write_register(0x09, 1) # DPS5005 on
self.on = True
def turn_off(self):
self.DPS.write_register(0x09, 0) # DPS5005 off
self.on = False
def start_injection(self, polarity=True):
# injection courant and measure (TODO check if it works, otherwise back in run_measurement())
self.polarity = polarity
if self.polarity:
self.pin0 = self.mcp.get_pin(self.pinA)
self.pin0.direction = Direction.OUTPUT
self.pin0.value = True
self.pin1 = self.mcp.get_pin(self.pinB)
self.pin1.direction = Direction.OUTPUT
self.pin1.value = False
else:
self.pin0 = self.mcp.get_pin(self.pinA)
self.pin0.direction = Direction.OUTPUT
self.pin0.value = False
self.pin1 = self.mcp.get_pin(self.pinB)
self.pin1.direction = Direction.OUTPUT
self.pin1.value = True
def stop_injection(self):
self.pin0 = self.mcp.get_pin(self.pinA)
self.pin0.direction = Direction.OUTPUT
self.pin0.value = False
self.pin1 = self.mcp.get_pin(self.pinB)
self.pin1.direction = Direction.OUTPUT
self.pin1.value = False
def set_polarity(self, polarity=True):
self.polarity = polarity
def set_tx_voltage(self, tx_voltage=12):
if tx_voltage >= 0:
self.tx_voltage = tx_voltage
# set voltage for test
self.DPS.write_register(0x0000, tx_voltage, 2)
self.DPS.write_register(0x09, 1) # DPS5005 on
else:
raise ValueError('Voltage needs to be >= 0 V')
class ADS(): # analog to digital converter ADS1115
def __init__(self, address=0x48, gain=2/3, data_rate=820, mode=1):
self.ads = ads.ADS1115(i2c, gain=gain, data_rate=data_rate, address=address, mode=mode)
self.gain = gain
self.data_rate = data_rate
self.mode = mode
self.pins = {
0: self.ads.P0,
1: self.ads.P1,
2: self.ads.P2,
3: self.ads.P3,
}
self.vmin = 0.01 # volts
self.vmax = 4.5 # volts
def read_single(self, pin=0):
return AnalogIn(self.ads, self.pins[pin]).voltage
def read_diff(self, pins='01'):
if pins == '01':
return AnalogIn(self.ads, self.ads.P0, self.ads.P1).voltage
elif pins == '23':
return AnalogIn(self.ads, self.ads.P2, self.ads.P3).voltage
def set_gain(self, gain=2/3):
self.gain = gain
# TODO maybe there is already a set_gain() function in the library? check that
self.ads = ads.ADS1115(
i2c, gain=self.gain, data_rate=self.data_rate,
address=self.address, mode=self.mode)
def get_best_gain(self, channel=0):
"""Automatically sets the gain on a channel
Parameters
----------
channel : ads.ADS1x15
Instance of ADS where voltage is measured.
Returns
-------
gain : float
Gain to be applied on ADS1115.
"""
voltage = self.read_singl(channel)
gain = 2 / 3
if (abs(voltage) < 2.040) and (abs(voltage) >= 1.023):
gain = 2
elif (abs(voltage) < 1.023) and (abs(voltage) >= 0.508):
gain = 4
elif (abs(voltage) < 0.508) and (abs(voltage) >= 0.250):
gain = 8
elif abs(voltage) < 0.256:
gain = 16
#self.exec_logger.debug(f'Setting gain to {gain}')
return gain
def set_best_gain(self, channel=0):
gain = self.get_best_gain(channel)
self.set_gain(gain)
class Voltage(ADS): # for MN
def __init__(self):
super().__init__()
def read(self, pin=0):
return self.read_single(self, pin=pin)
def read_all(self, pins=[0, 2]):
return [self.read_single(pin) for pin in pins]
class Current(ADS): # for AB
def __init__(self, address=0x48, gain=2/3, data_rate=820, mode=1, r_shunt=OHMPI_CONFIG['R_shunt']):
super().__init__(address=address, gain=gain, data_rate=data_rate, mode=mode)
self.r_shunt = r_shunt
self.imin = self.vmin / (self.r_shunt * 50)
self.imax = self.vmax / (self.r_shunt * 50)
def read(self):
U = self.read_single(pin=0)
return U / 50 / self.r_shunt
class Multiplexer():
def __init__(self, addresses={
'A': 0x70,
'B': 0x71,
'M': 0x72,
'N': 0x73
},
nelec=64):
#OHMPI_CONFIG['board_addresses']
self.addresses = addresses
self.nelec = nelec # max number of electrodes per board
def switch_one(self, elec, role, state='off'):
self.tca = adafruit_tca9548a.TCA9548A(i2c, self.addresses[role])
# find I2C address of the electrode and corresponding relay
# considering that one MCP23017 can cover 16 electrodes
i2c_address = 7 - (elec - 1) // 16 # quotient without rest of the division
relay = (elec-1) - ((elec-1) // 16) * 16
if i2c_address is not None:
# select the MCP23017 of the selected MUX board
mcp = MCP23017(self.tca[i2c_address])
mcp.get_pin(relay - 1).direction = digitalio.Direction.OUTPUT
if state == 'on':
mcp.get_pin(relay - 1).value = True
else:
mcp.get_pin(relay - 1).value = False
#exec_logger.debug(f'Switching relay {relay} '
# f'({str(hex(self.addresses[role]))}) on:{on} for electrode {elec}')
else:
raise ValueError('No I2C address found for the electrode'
' {:d} on board {:s}'.format(elec, self.addresses[role]))
#exec_logger.warning(f'Unable to address electrode nr {elec}')
def switch(self, elecdic={}, state='on'):
"""Switch a given list of electrodes with different roles.
Electrodes with a value of 0 will be ignored.
Parameters
----------
elecdic : dictionary, optional
Dictionnary of the form: role: [list of electrodes].
state : str, optional
Either 'on' or 'off'.
"""
# check to prevent A == B (SHORT-CIRCUIT)
if 'A' in elecdic and 'B' in elecdic:
out = np.in1d(elecdic['A'], elecdic['B'])
if out.any():
raise ValueError('Some electrodes have A == B -> SHORT-CIRCUIT')
return
# check none of M and N are the same A or B
# as to prevent burning the MN part which cannot take
# the full voltage of the DPS
if 'A' in elecdic and 'B' in elecdic and 'M' in elecdic and 'N' in elecdic:
if (np.in1d(elecdic['M'], elecdic['A']).any()
or np.in1d(elecdic['M'], elecdic['B']).any()
or np.in1d(elecdic['N'], elecdic['A']).any()
or np.in1d(elecdic['N'], elecdic['B']).any()):
raise ValueError('Some electrodes M and N are on A and B -> cannot be with DPS')
return
# if all ok, then switch the electrodes
for role in elecdic:
for elec in elecdic[role]:
if elec > 0:
self.switch_one(elec, role, state)
def reset(self):
for role in self.addresses:
for elec in range(self.nelec):
self.switch_one(elec, role, 'off')
def test(self, role, activation_time=1):
"""Interactive method to test the multiplexer.
Parameters
----------
activation_time : float, optional
Time in seconds during which the relays are activated.
address : hex, optional
Address of the multiplexer board to test (e.g. 0x70, 0x71, ...).
"""
self.reset()
# ask use some details on how to proceed
a = input('If you want try 1 channel choose 1, if you want try all channels choose 2!')
if a == '1':
print('run channel by channel test')
electrode = int(input('Choose your electrode number (integer):'))
electrodes = [electrode]
elif a == '2':
electrodes = range(1, 65)
else:
print('Wrong choice !')
return
# run the test
for elec in electrodes:
self.switch_one(elec, role, 'on')
print('electrode:', elec, ' activated...', end='', flush=True)
time.sleep(activation_time)
self.switch_one(elec, role, 'off')
print(' deactivated')
time.sleep(activation_time)
print('Test finished.')