Skip to content
GitLab
Commit 07adaa6a authored by Clement Remi's avatar Clement Remi
Browse files

some corrections of the measurement and removing auto gain function inadapted impedance

parent 19165115
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.35 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
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 4937 additions and 19 deletions
+4937 -19
Ohmpi.py
+ 31
− 18
View file @ 07adaa6a
......@@ -32,11 +32,11 @@ print(current_time.strftime("%Y-%m-%d %H:%M:%S"))
"""
hardware parameters
"""
R_ref = 47.2# reference resistance value in ohm
coef_p0 = 2.5009 # slope for current conversion for ADS.P0, measurement in V/V
coef_p1 = 2.4947 # slope for current conversion for ADS.P1, measurement in V/V
coef_p2 = 2.4985 # slope for current conversion for ADS.P2, measurement in V/V
coef_p3 = 2.4994 # slope for current conversion for ADS.P3, measurement in V/V
R_ref = 50.20# reference resistance value in ohm
coef_p0 = 2.47 # slope for current conversion for ADS.P0, measurement in V/V
coef_p1 = 2.47# slope for current conversion for ADS.P1, measurement in V/V
coef_p2 = 2.4748 # slope for current conversion for ADS.P2, measurement in V/V
coef_p3 = 2.4748 # slope for current conversion for ADS.P3, measurement in V/V
export_path = "/home/pi/Desktop/measurement.csv"
base_dir = '/sys/bus/w1/devices/'
device_folder = glob.glob(base_dir + '28*')[0]
......@@ -48,7 +48,7 @@ GPIO.setwarnings(False)
GPIO.setup(7, GPIO.OUT)
GPIO.setup(8, GPIO.OUT)
integer=5
integer=30
meas=numpy.zeros((4,integer))
"""
......@@ -57,10 +57,8 @@ import parameters
with open('ohmpi_param.json') as json_file:
pardict = json.load(json_file)
i2c = busio.I2C(board.SCL, board.SDA) # I2C protocol setup
ads = ADS.ADS1115(i2c, gain=16,data_rate=860) # I2C communication setup
ads = ADS.ADS1115(i2c, gain=2/3,data_rate=860) # I2C communication setup
"""
functions
......@@ -99,7 +97,7 @@ def gain_auto(channel):
gain=4
elif ((abs(channel.voltage)<0.508) and (abs(channel.voltage)>=0.250)):
gain=8
elif abs(channel.voltage)<0.256:
elif abs(channel.voltage)<0.250:
gain=16
#print(gain)
return gain
......@@ -139,6 +137,12 @@ def read_temp():
return temp_c
# perform a measurement
def run_measurement(nb_stack, injection_deltat, Rref, coefp0, coefp1, coefp2, coefp3, elec_array):
# print('Enter contact resistance:')
# Rc = float(input())
# print('Enter soil resistance:')
# Rsoil = float(input())
start_time = time.time()
# inner variable initialization
sum_I=0
......@@ -160,21 +164,27 @@ def run_measurement(nb_stack, injection_deltat, Rref, coefp0, coefp1, coefp2, co
Tx=AnalogIn(ads,ADS.P0).voltage
Tab=AnalogIn(ads,ADS.P1).voltage
t_gain=[gain_auto(AnalogIn(ads,ADS.P0,ADS.P1)),gain_auto(AnalogIn(ads,ADS.P2,ADS.P3))]
t_gain=numpy.sort(t_gain)
ads = ADS.ADS1115(i2c, gain=t_gain[1],data_rate=860)
print(t_gain)
ads = ADS.ADS1115(i2c, gain=t_gain[0],data_rate=860)
for k in range(0,integer):
ads = ADS.ADS1115(i2c, gain=t_gain[0],data_rate=860)
#ads = ADS.ADS1115(i2c, gain=2/3,data_rate=860)
meas[0,k] = AnalogIn(ads,ADS.P0,ADS.P1).voltage# reading current value on ADS channel A0
ads = ADS.ADS1115(i2c, gain=t_gain[1],data_rate=860)
#ads = ADS.ADS1115(i2c, gain=2/3,data_rate=860)
meas[2,k] = AnalogIn(ads,ADS.P2,ADS.P3).voltage # reading voltage value on ADS channel A2
GPIO.output(8, GPIO.LOW)# stop current injection
startdelay=time.time()
# std=statistics.stdev(meas[0,:])
# mean=statistics.mean(meas[0,:])
# print( mean, std)
sum_I=sum_I+numpy.mean(meas[0,:]) * ((coefp0+coefp1)/2)/Rref
print(sum_I*Rref)
Vmn1= numpy.mean(meas[2,:]) * ((coefp2+coefp3)/2)
if (n % 2) == 0:
sum_Vmn=sum_Vmn-Vmn1
......@@ -194,13 +204,16 @@ def run_measurement(nb_stack, injection_deltat, Rref, coefp0, coefp1, coefp2, co
"Vmn [mV]":[(sum_Vmn/(3+2*nb_stack-1))*1000],
"I [mA]":[(sum_I/(3+2*nb_stack-1))*1000],
"R [ohm]":[( (sum_Vmn/(3+2*nb_stack-1)/(sum_I/(3+2*nb_stack-1))))],
"Rab [Ohm]":[(Tab*coefp1)/(sum_I/(3+2*nb_stack-1))],
"Tx [V]":[Tx*coefp0],
"Rab [Ohm]":[(Tab*2.47)/(sum_I/(3+2*nb_stack-1))],
"Tx [V]":[Tx*2.47],
"Ps [mV]":[(sum_Ps/(3+2*nb_stack-1))*1000],
"nbStack":[nb_stack],
"CPU temp [°C]":[cpu.temperature],
"Hardware temp [°C]":[read_temp()],
"Time [S]":[(-start_time+time.time())]
# "Rcontact[ohm]":[Rc],
# "Rsoil[ohm]":[Rsoil],
# "Rab_theory [Ohm]":[(Rc*2+Rsoil)]
# Dead time equivalent to the duration of the current injection pulse
......
Ohmpi_single.py 0 → 100644
+ 280
− 0
View file @ 07adaa6a
"""
created on January 6, 2020
Update December 2021
Ohmpi.py is a program to control a low-cost and open hardward resistivity meter OhmPi that has been developed by Rémi CLEMENT(INRAE),Vivien DUBOIS(INRAE),Hélène GUYARD(IGE), Nicolas FORQUET (INRAE), and Yannick FARGIER (IFSTTAR).
"""
print('\033[1m'+'\033[31m'+' ________________________________')
print('| _ | | | || \/ || ___ \_ _|')
print('| | | | |_| || . . || |_/ / | |' )
print('| | | | _ || |\/| || __/ | |')
print('\ \_/ / | | || | | || | _| |_')
print(' \___/\_| |_/\_| |_/\_| \___/ ')
print('\033[0m')
print('OHMPI start' )
print('Vers: 1.50')
print('Import library')
import RPi.GPIO as GPIO
import time , board, busio, numpy, os, sys, json, glob, statistics
from datetime import datetime
import adafruit_ads1x15.ads1115 as ADS
from adafruit_ads1x15.analog_in import AnalogIn
import pandas as pd
import os.path
from gpiozero import CPUTemperature
"""
display start time
"""
current_time = datetime.now()
print(current_time.strftime("%Y-%m-%d %H:%M:%S"))
"""
hardware parameters
"""
R_ref = 50.0# reference resistance value in ohm
coef_p0 = 2.46524 # slope for current conversion for ADS.P0, measurement in V/V
coef_p1 = 2.46610# slope for current conversion for ADS.P1, measurement in V/V
coef_p2 = 2.46526 # slope for current conversion for ADS.P2, measurement in V/V
coef_p3 = 2.46542 # slope for current conversion for ADS.P3, measurement in V/V
offset_p0=-0.00078559
offset_p1= 0.00655417
offset_p2= 0.00729835
offset_p3= 0.00102393
export_path = "/home/pi/Desktop/measurement.csv"
base_dir = '/sys/bus/w1/devices/'
device_folder = glob.glob(base_dir + '28*')[0]
device_file = device_folder + '/w1_slave'
# GPIO initialization
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)
GPIO.setup(7, GPIO.OUT)
GPIO.setup(8, GPIO.OUT)
integer=30
meas=numpy.zeros((4,integer))
"""
import parameters
"""
with open('ohmpi_param.json') as json_file:
pardict = json.load(json_file)
i2c = busio.I2C(board.SCL, board.SDA) # I2C protocol setup
ads = ADS.ADS1115(i2c, gain=2/3,data_rate=475) # I2C communication setup
"""
functions
"""
# function swtich_mux select the right channels for the multiplexer cascade for electrodes A, B, M and N.
def switch_mux(quadripole):
path2elec = numpy.loadtxt("path2elec.txt", delimiter=" ", dtype=bool)
quadmux = numpy.loadtxt("quadmux.txt", delimiter=" ", dtype=int)
for i in range(0,4):
for j in range(0,5) :
GPIO.output(int(quadmux[i,j]), bool(path2elec[quadripole[i]-1,j]))
# function to find rows with identical values in different columns
def find_identical_in_line(array_object):
output = []
if array_object.ndim == 1:
temp = numpy.zeros(4)
for i in range(len(array_object)):
temp[i] = numpy.count_nonzero(array_object == array_object[i])
if any(temp > 1):
output.append(0)
else:
for i in range(len(array_object[:,1])):
temp = numpy.zeros(len(array_object[1,:]))
for j in range(len(array_object[1,:])):
temp[j] = numpy.count_nonzero(array_object[i,:] == array_object[i,j])
if any(temp > 1):
output.append(i)
return output
# def gain_auto(channel):
# 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.250:
# gain=16
# #print(gain)
# return gain
# read quadripole file and apply tests
def read_quad(filename, nb_elec):
output = numpy.loadtxt(filename, delimiter=" ",dtype=int) # load quadripole file
# locate lines where the electrode index exceeds the maximum number of electrodes
test_index_elec = numpy.array(numpy.where(output > 32))
# locate lines where an electrode is referred twice
test_same_elec = find_identical_in_line(output)
# if statement with exit cases (rajouter un else if pour le deuxième cas du ticket #2)
if test_index_elec.size != 0:
for i in range(len(test_index_elec[0,:])):
print("Error: An electrode index at line "+ str(test_index_elec[0,i]+1)+" exceeds the maximum number of electrodes")
sys.exit(1)
elif len(test_same_elec) != 0:
for i in range(len(test_same_elec)):
print("Error: An electrode index is used twice at line " + str(test_same_elec[i]+1))
sys.exit(1)
else:
return output
def read_temp():
f = open(device_file, 'r')
lines = f.readlines()
f.close()
while lines[0].strip()[-3:] != 'YES':
time.sleep(0.2)
lines = read_temp_raw()
equals_pos = lines[1].find('t=')
if equals_pos != -1:
temp_string = lines[1][equals_pos+2:]
temp_c = float(temp_string) / 1000.0
return temp_c
# perform a measurement
def run_measurement(nb_stack, injection_deltat, Rref, coefp0, coefp1, coefp2, coefp3, elec_array):
# print('Enter contact resistance:')
# Rc = float(input())
# print('Enter soil resistance:')
# Rsoil = float(input())
start_time = time.time()
# inner variable initialization
sum_I=0
sum_Vmn=0
sum_Ps=0
# injection courant and measure
for n in range(0,3+2*nb_stack-1) :
# current injection
if (n % 2) == 0:
GPIO.output(7, GPIO.HIGH) # polarity n°1
else:
GPIO.output(7, GPIO.LOW) # polarity n°2
GPIO.output(8, GPIO.HIGH) # current injection
time.sleep(injection_deltat) # delay depending on current injection duration
if n==0:
Tx=AnalogIn(ads,ADS.P0).voltage
Tab=AnalogIn(ads,ADS.P1).voltage
for k in range(0,integer):
meas[0,k] = AnalogIn(ads,ADS.P0).voltage# reading current value on ADS channel A0
meas[1,k] = AnalogIn(ads,ADS.P1).voltage
meas[2,k] = AnalogIn(ads,ADS.P2).voltage # reading voltage value on ADS channel A2
meas[3,k] = AnalogIn(ads,ADS.P3).voltage
GPIO.output(8, GPIO.LOW)# stop current injection
startdelay=time.time()
std=statistics.stdev(meas[0,:])
mean=statistics.mean(meas[0,:])
#print( mean, std)
sum_I=sum_I+(((numpy.mean(meas[0,:])*coefp0)+offset_p0)-(((numpy.mean(meas[1,:])*coefp1)+offset_p1)))/Rref
Vmn1= ((numpy.mean(meas[2,:])*(coefp2))+ offset_p2)-((numpy.mean(meas[3,:])*(coefp3))+offset_p3)
if (n % 2) == 0:
sum_Vmn=sum_Vmn-Vmn1
sum_Ps=sum_Ps+Vmn1
else:
sum_Vmn=sum_Vmn+Vmn1
sum_Ps=sum_Ps+Vmn1
time.sleep(injection_deltat - ((time.time() - startdelay) % injection_deltat))
# return averaged values
cpu= CPUTemperature()
output = pd.DataFrame({
"time":[datetime.now()],
"A":elec_array[0],
"B":elec_array[1],
"M":elec_array[2],
"N":elec_array[3],
"Vmn [mV]":[(sum_Vmn/(3+2*nb_stack-1))*1000],
"I [mA]":[(sum_I/(3+2*nb_stack-1))*1000],
"R [ohm]":[( (sum_Vmn/(3+2*nb_stack-1)/(sum_I/(3+2*nb_stack-1))))],
"Rab [Ohm]":[(Tab*2.47)/(sum_I/(3+2*nb_stack-1))],
"Tx [V]":[Tx*2.47],
"Ps [mV]":[(sum_Ps/(3+2*nb_stack-1))*1000],
"nbStack":[nb_stack],
"CPU temp [°C]":[cpu.temperature],
"Hardware temp [°C]":[read_temp()],
"Time [S]":[(-start_time+time.time())]
# "Rcontact[ohm]":[Rc],
# "Rsoil[ohm]":[Rsoil],
# "Rab_theory [Ohm]":[(Rc*2+Rsoil)]
# Dead time equivalent to the duration of the current injection pulse
})
output=output.round(1)
print(output.to_string())
time.sleep(1)
return output
# save data
def append_and_save(path, last_measurement):
if os.path.isfile(path):
# Load data file and append data to it
with open(path, 'a') as f:
last_measurement.to_csv(f, header=False)
else:
# create data file and add headers
with open(path, 'a') as f:
last_measurement.to_csv(f, header=True)
"""
Initialization of GPIO channels
"""
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)
"""
Initialization of multiplexer channels
"""
# pinList = [12,16,20,21,26,18,23,24,25,19,6,13,4,17,27,22,10,9,11,5] # List of GPIOs enabled for relay cards (electrodes)
# for i in pinList:
# GPIO.setup(i, GPIO.OUT)
# GPIO.output(i, GPIO.HIGH)
"""
Main loop
"""
N=read_quad("ABMN.txt",pardict.get("nb_electrodes")) # load quadripole file
if N.ndim == 1:
N = N.reshape(1, 4)
for g in range(0,pardict.get("nbr_meas")): # for time-lapse monitoring
for i in range(0,N.shape[0]): # loop over quadripoles
# call the switch_mux function to switch to the right electrodes
# switch_mux(N[i,])
# run a measurement
current_measurement = run_measurement(pardict.get("stack"), pardict.get("injection_duration"), R_ref, coef_p0, coef_p1, coef_p2, coef_p3, N[i,])
# save data and print in a text file
append_and_save(pardict.get("export_path"), current_measurement)
# reset multiplexer channels
# GPIO.output(12, GPIO.HIGH); GPIO.output(16, GPIO.HIGH); GPIO.output(20, GPIO.HIGH); GPIO.output(21, GPIO.HIGH); GPIO.output(26, GPIO.HIGH)
# GPIO.output(18, GPIO.HIGH); GPIO.output(23, GPIO.HIGH); GPIO.output(24, GPIO.HIGH); GPIO.output(25, GPIO.HIGH); GPIO.output(19, GPIO.HIGH)
# GPIO.output(6, GPIO.HIGH); GPIO.output(13, GPIO.HIGH); GPIO.output(4, GPIO.HIGH); GPIO.output(17, GPIO.HIGH); GPIO.output(27, GPIO.HIGH)
# GPIO.output(22, GPIO.HIGH); GPIO.output(10, GPIO.HIGH); GPIO.output(9, GPIO.HIGH); GPIO.output(11, GPIO.HIGH); GPIO.output(5, GPIO.HIGH)
time.sleep(pardict.get("sequence_delay")) #waiting next measurement (time-lapse)
calibration.py 0 → 100644
+ 31
− 0
View file @ 07adaa6a
import RPi.GPIO as GPIO
import time , board, busio, numpy, os, sys, json, glob, statistics
from datetime import datetime
import adafruit_ads1x15.ads1115 as ADS
from adafruit_ads1x15.analog_in import AnalogIn
import pandas as pd
import os.path
from gpiozero import CPUTemperature
i2c = busio.I2C(board.SCL, board.SDA) # I2C protocol setup
def gain_auto(channel):
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
#print(gain)
return gain
ads = ADS.ADS1115(i2c, gain=2/3,data_rate=860) # I2C communication setup
t_gain=[gain_auto(AnalogIn(ads,ADS.P3))]
ads = ADS.ADS1115(i2c, gain=t_gain[0],data_rate=860)
print(t_gain)
print(AnalogIn(ads,ADS.P3).voltage)
print(AnalogIn(ads,ADS.P2).voltage)
calibration_single.py 0 → 100644
+ 33
− 0
View file @ 07adaa6a
import RPi.GPIO as GPIO
import time , board, busio, numpy, os, sys, json, glob, statistics
from datetime import datetime
import adafruit_ads1x15.ads1115 as ADS
from adafruit_ads1x15.analog_in import AnalogIn
import pandas as pd
import os.path
from gpiozero import CPUTemperature
i2c = busio.I2C(board.SCL, board.SDA) # I2C protocol setup
ads = ADS.ADS1115(i2c, gain=2/3,data_rate=860)
GPIO.setmode(GPIO.BCM)
GPIO.setwarnings(False)
GPIO.setup(7, GPIO.OUT)
GPIO.setup(8, GPIO.OUT)
coef_p0 = 2.47 # slope for current conversion for ADS.P0, measurement in V/V
coef_p1 = 2.47# slope for current conversion for ADS.P1, measurement in V/V
coef_p2 = 2.4748 # slope for current conversion for ADS.P2, measurement in V/V
coef_p3 = 2.4748 # slope for current conversion for ADS.P3, measurement in V/V
integer=1000
meas=numpy.zeros((4,integer))
for i in range(0,100) :
sum_I=0
GPIO.output(8, GPIO.HIGH) # current injection
time.sleep(4)
for i in range(0,integer) :
meas[0,i] = AnalogIn(ads,ADS.P0).voltage
GPIO.output(8, GPIO.LOW)
std=statistics.stdev(meas[0,:])
mean=statistics.mean(meas[0,:])
print(mean,std)
#print(2.4653*mean+0.0073, std)
time.sleep(4)
measurement.csv
+ 4561
− 0
View file @ 07adaa6a
This diff is collapsed.
ohmpi_param.json
+ 1
− 1
View file @ 07adaa6a
{
"nb_electrodes": 32,
"injection_duration": 0.500,
"injection_duration": 0.5,
"nbr_meas": 10000,
"sequence_delay": 1,
"stack": 1,
......
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