Skip to content
GitLab
Commit 90349c1e authored by Clement Remi's avatar Clement Remi
Browse files

Delete Ohmpi.py

parent 76cbb34a
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 0 additions and 278 deletions
+0 -278
Ohmpi.py deleted 100644 → 0
+ 0
− 278
View file @ 76cbb34a
"""
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.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]
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=860) # 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
t_gain=[0,0]
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:
ads = ADS.ADS1115(i2c, gain=2/3,data_rate=860)
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))]
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
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(2)
print(output.to_string())
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)
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