Skip to content
GitLab
Commit 7587d63f authored by Olivier Kaufmann's avatar Olivier Kaufmann
Browse files

Merges changes on controller. Cleans up exec messages in info mode. Implements... 

Merges changes on controller. Cleans up exec messages in info mode. Implements a generic _process_commands.
parents eeb82292 c2c5863a
master EGU23 MUX_board_v2024 battery_charger board_v4 code_refactor dev2 mqtt revert-559ef012 v2023.0.0
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Showing
with 454 additions and 351 deletions
+454 -351
config.py
+ 3
− 2
View file @ 7587d63f
......@@ -2,11 +2,12 @@ import logging
from paho.mqtt.client import MQTTv31
mqtt_broker = 'mg3d-dev.umons.ac.be' # 'localhost'
ohmpi_id = '0001'
mqtt_broker = 'localhost'
logging_suffix = ''
# OhmPi configuration
OHMPI_CONFIG = {
'id': '0001', # Unique identifier of the OhmPi board (string)
'id': ohmpi_id, # Unique identifier of the OhmPi board (string)
'R_shunt': 2, # Shunt resistance in Ohms
'Imax': 4800/50/2, # Maximum current
'coef_p2': 2.50, # slope for current conversion for ADS.P2, measurement in V/V
......
doc/source/V2_00.rst
+ 49
− 49
View file @ 7587d63f
......@@ -114,58 +114,58 @@ files (.json and .py).
.. code-block:: python
:caption: Example of using the Python API to control OhmPi
from ohmpi import OhmPi
k = OhmPi(idps=True) # if V3.0 make sure to set idps to True
# the DPS5005 is used in V3.0 to inject higher voltage
# default parameters are stored in the pardict argument
# they can be manually changed
k.settings['injection_duration'] = 0.5 # injection time in seconds
k.settings['nb_stack'] = 1 # one stack is two half-cycles
k.settings['nbr_meas'] = 1 # number of time the sequence is repeated
# without multiplexer, one can simple measure using
out = k.run_measurement()
# out contains information about the measurement and can be save as
k.append_and_save('out.csv', out)
# custom or adaptative argument (see help of run_measurement())
k.run_measurement(nb_stack=4, # do 4 stacks (8 half-cycles)
injection_duration=2, # inject for 2 seconds
autogain=True, # adapt gain of ADS to get good resolution
strategy='vmin', # inject min voltage for Vab (v3.0)
tx_volt=5) # vab for finding vmin or vab injectected
# if 'strategy' is 'constant'
:caption: Example of using the Python API to control OhmPi
import os
import numpy as np
import time
os.chdir("/home/pi/OhmPi")
from ohmpi import OhmPi
### Define object from class OhmPi
k = OhmPi() # this load default parameters from the disk
### Default parameters can also be edited manually
k.settings['injection_duration'] = 0.5 # injection time in seconds
k.settings['nb_stack'] = 1 # one stack is two half-cycles
k.settings['nbr_meas'] = 1 # number of time the sequence is repeated
### Update settings if needed
k.update_settings({"injection_duration":0.2})
### Set or load sequence
k.sequence = np.array([[1,2,3,4]]) # set numpy array of shape (n,4)
# k.set_sequence('1 2 3 4\n2 3 4 5') # call function set_sequence and pass a string
# k.load_sequence('ABMN.txt') # load sequence from a local file
### Run contact resistance check
k.rs_check()
### Run sequence (synchronously - it will wait that all
# sequence is measured before returning the prompt
k.run_sequence()
# k.run_sequence_async() # sequence is run in a separate thread and the prompt returns immediately
# time.sleep(2)
# k.interrupt() # kill the asynchrone sequence
### Run multiple sequences at given time interval
k.settings['nb_meas'] = 3 # run sequence three times
k.settings['sequence_delay'] = 100 # every 100 s
k.run_multiple_sequences() # asynchrone
# k.interrupt() # kill the asynchrone sequence
### Single measurement can also be taken with
k.switch_mux_on([1, 4, 2, 3])
k.run_measurement() # use default acquisition parameters
k.switch_mux_off([1, 4, 2, 3]) # don't forget this! risk of short-circuit
### Custom or adaptative argument, see help(k.run_measurement)
k.run_measurement(nb_stack=4, # do 4 stacks (8 half-cycles)
injection_duration=2, # inject for 2 seconds
autogain=True) # adapt gain of ADS to get good resolution
# if a multiplexer is connected, we can also manually switch it
k.reset_mux() # check that all mux are closed (do this FIRST)
k.switch_mux_on([1, 4, 2, 3])
k.run_measurement()
k.switch_mux_off([1, 4, 2, 3]) # don't forget this! risk of short-circuit
# import a sequence
k.read_quad('sequence.txt') # four columns, no header, space as separator
print(k.sequence) # where the sequence is stored
# rs check
k.rs_check() # run an RS check (check contact resistances) for all
# electrodes of the given sequence
# run a sequence
k.measure() # measure accept same arguments as run_measurement()
# NOTE: this is an asynchronous command that runs in a separate thread
# after executing the command, the prompt will return immediately
# the asynchronous thread can be stopped during execution using
k.stop()
# otherwise, it will exit by itself at the end of the sequence
# if multiple measurement are to be taken, the sequence will be repeated
***MQTT interface***
Interface to communicate with the Pi designed for the Internet of Things (IoT).
......
example_simple_measurement.py
+ 5
− 3
View file @ 7587d63f
......@@ -16,8 +16,10 @@ k.sequence = np.array([[1,2,3,4]]) # set numpy array of shape (n,4)
# k.load_sequence('ABMN.txt') # load sequence from a local file
### Run contact resistance check
k.rs_check()
# k.rs_check()
### Run sequence
k.run_sequence()
# k.interrupt()
\ No newline at end of file
k.run_sequence()
# k.run_sequence_async()
# time.sleep(2)
# k.interrupt()
\ No newline at end of file
http_interface.py
+ 82
− 50
View file @ 7587d63f
......@@ -23,8 +23,56 @@ publisher_config.pop('ctrl_topic')
print(colored(f"Sending commands control topic {MQTT_CONTROL_CONFIG['ctrl_topic']} on {MQTT_CONTROL_CONFIG['hostname']} broker."))
cmd_id = None
received = False
rdic = {}
# set controller globally as __init__ seem to be called for each request and so we subscribe again each time (=overhead)
controller = mqtt_client.Client(f"ohmpi_{OHMPI_CONFIG['id']}_listener", clean_session=False) # create new instance
print(colored(f"Connecting to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']} on {MQTT_CONTROL_CONFIG['hostname']} broker", 'blue'))
trials = 0
trials_max = 10
broker_connected = False
while trials < trials_max:
try:
controller.username_pw_set(MQTT_CONTROL_CONFIG['auth'].get('username'),
MQTT_CONTROL_CONFIG['auth']['password'])
controller.connect(MQTT_CONTROL_CONFIG['hostname'])
trials = trials_max
broker_connected = True
except Exception as e:
print(f'Unable to connect control broker: {e}')
print('trying again to connect to control broker...')
time.sleep(2)
trials += 1
if broker_connected:
print(f"Subscribing to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']}")
controller.subscribe(MQTT_CONTROL_CONFIG['ctrl_topic'], MQTT_CONTROL_CONFIG['qos'])
else:
print(f"Unable to connect to control broker on {MQTT_CONTROL_CONFIG['hostname']}")
controller = None
# start a listener for acknowledgement
def _control():
def on_message(client, userdata, message):
global cmd_id, rdic, received
command = json.loads(message.payload.decode('utf-8'))
#print('++++', cmd_id, received, command)
if ('reply' in command.keys()) and (command['cmd_id'] == cmd_id):
print(f'Acknowledgement reception of command {command} by OhmPi')
# print('oooooooooook', command['reply'])
received = True
#rdic = command
controller.on_message = on_message
controller.loop_forever()
t = threading.Thread(target=_control)
t.start()
class MyServer(SimpleHTTPRequestHandler):
# because we use SimpleHTTPRequestHandler, we do not need to implement
# the do_GET() method (if we use the BaseHTTPRequestHandler, we would need to)
......@@ -42,60 +90,44 @@ class MyServer(SimpleHTTPRequestHandler):
def __init__(self, request, client_address, server):
super().__init__(request, client_address, server)
# set controller
self.controller = mqtt_client.Client(f"ohmpi_{OHMPI_CONFIG['id']}_listener", clean_session=False) # create new instance
print(colored(f"Connecting to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']} on {MQTT_CONTROL_CONFIG['hostname']} broker", 'blue'))
trials = 0
trials_max = 10
broker_connected = False
while trials < trials_max:
try:
self.controller.username_pw_set(MQTT_CONTROL_CONFIG['auth'].get('username'),
MQTT_CONTROL_CONFIG['auth']['password'])
self.controller.connect(MQTT_CONTROL_CONFIG['hostname'])
trials = trials_max
broker_connected = True
except Exception as e:
print(f'Unable to connect control broker: {e}')
print('trying again to connect to control broker...')
time.sleep(2)
trials += 1
if broker_connected:
print(f"Subscribing to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']}")
self.controller.subscribe(MQTT_CONTROL_CONFIG['ctrl_topic'], MQTT_CONTROL_CONFIG['qos'])
else:
print(f"Unable to connect to control broker on {MQTT_CONTROL_CONFIG['hostname']}")
self.controller = None
self.cmd_thread = threading.Thread(target=self._control)
def _control(self):
def on_message(client, userdata, message):
global cmd_id, rdic
command = message.payload.decode('utf-8')
print(f'Received command {command}')
# self.process_commands(command)
if 'reply' in command.keys and command['cmd_id'] == cmd_id :
rdic = command
self.controller.on_message = on_message
self.controller.loop_start()
while True:
time.sleep(.1)
# global controller, once # using global variable otherwise, we subscribe to client for EACH request
# if once:
# self.controller = controller
# self.cmd_thread = threading.Thread(target=self._control)
# self.cmd_thread.start()
# once = False
# we would like to listen to the ackn topic to check our message has been wel received
# by the OhmPi, however, this won't work as it seems an instance of MyServer is created
# each time (actually it's not a server but a requestHandler)
# def _control(self):
# def on_message(client, userdata, message):
# global cmd_id, rdic
# command = json.loads(message.payload.decode('utf-8'))
# print(f'Acknowledgement reception of command {command} by OhmPi')
# if 'reply' in command.keys() and command['cmd_id'] == cmd_id :
# print('oooooooooook', command['reply'])
# #rdic = command
# self.controller.on_message = on_message
# print('starting loop')
# self.controller.loop_forever()
# print('forever')
def do_POST(self):
global cmd_id, rdic
global cmd_id, rdic, received
received = False
cmd_id = uuid.uuid4().hex
# global socket
# global ohmpiThread, status, run
dic = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
#print('++', dic, cmd_id)
rdic = {} # response dictionary
if dic['cmd'] == 'run_sequence':
payload = json.dumps({'cmd_id': cmd_id, 'cmd': 'run_sequence'})
if dic['cmd'] == 'run_multiple_sequences':
payload = json.dumps({'cmd_id': cmd_id, 'cmd': 'run_multiple_sequences'})
publish.single(payload=payload, **publisher_config)
elif dic['cmd'] == 'interrupt':
# ohmpi.stop()
payload = json.dumps({'cmd_id': cmd_id, 'cmd': 'interrupt'})
publish.single(payload=payload, **publisher_config)
elif dic['cmd'] == 'getData':
......@@ -103,9 +135,9 @@ class MyServer(SimpleHTTPRequestHandler):
fnames = [fname for fname in os.listdir('data/') if fname[-4:] == '.csv']
ddic = {}
for fname in fnames:
if (fname.replace('.csv', '') not in dic['surveyNames']
and fname != 'readme.txt'
and '_rs' not in fname):
if ((fname != 'readme.txt')
and ('_rs' not in fname)
and (fname.replace('.csv', '') not in dic['surveyNames'])):
df = pd.read_csv('data/' + fname)
ddic[fname.replace('.csv', '')] = {
'a': df['A'].tolist(),
......
index.html
+ 26
− 23
View file @ 7587d63f
......@@ -26,7 +26,7 @@
<div class="form-check">
<input id="dataRetrievalCheck" class="form-check-input" type="checkbox" value="">
<label class="form-check-label" for="dataRetrievalCheck">
Automaticaly get data every 1 secondStart
Automaticaly get data every 1 second
</label>
</div>
<div id='output'>Status: idle</div>
......@@ -34,8 +34,8 @@
<!-- Pseudo section -->
<select id='surveySelect' class='custom-select'>
</select>
<input id="cmin" type="number" value="0"/>
<input id="cmax" type="number" value="150"/>
<input id="cmin" type="number" value=""/>
<input id="cmax" type="number" value=""/>
<button id="capplyBtn" type="button" class="btn btn-info">Apply</button>
<div id="gd"></div>
<div class="mb3 row">
......@@ -163,9 +163,9 @@
// run button
function runBtnFunc() {
sendCommand('{"cmd": "run_sequence"}', function(x) {
console.log(x['status'])
if (x['status'] == 'running') {
sendCommand('{"cmd": "run_multiple_sequences"}', function(x) {
console.log(x['ohmpi_status'])
if (x['ohmpi_status'] == 'running') {
output.innerHTML = 'Status: measuring...'
}
})
......@@ -176,7 +176,7 @@
// interrupt button
function stopBtnFunc() {
sendCommand('{"cmd": "interrupt"}', function(x) {
output.innerHTML = 'Status: ' + x['status']
output.innerHTML = 'Status: ' + x['ohmpi_status']
clearInterval(interv)
getData()
})
......@@ -375,13 +375,14 @@
// getData
function getData() {
let surveyNames = []
sendCommand(JSON.stringify({
'cmd': 'getData',
'surveyNames': Object.keys(data).slice(0, -1)
'surveyNames': surveyNames
// last survey is often partial so we download it again
}), function(ddic) {
// update status
output.innerHTML = 'Status: ' + ddic['status']
//output.innerHTML = 'Status: ' + ddic['status']
// update data dic with new data
data = { // destructuring assignement (magic! :o)
......@@ -389,8 +390,8 @@
...ddic['data'] // value from second dic are preferred
}
// dropdown with number of surveys and +++
let surveyNames = Object.keys(data).sort()
// dropdown with number of surveys
surveyNames = Object.keys(data).sort()
// remove listener as we will replace the choices
surveySelect.removeEventListener('change', surveySelectFunc)
......@@ -417,18 +418,20 @@
// update list of quadrupoles if any
if (quads.length == 0) {
console.log('updating list of quadrupoles')
let df = data[surveyNames[0]]
let quadSelect = document.getElementById('quadSelect')
quadSelect.innerHTML = ''
for (let i = 0; i < df['a'].length; i++) {
quad = [df['a'][i], df['b'][i], df['m'][i], df['n'][i]]
quads.push(quad)
let option = document.createElement('option')
option.value = quad.join(', ')
option.innerText = quad.join(', ')
quadSelect.appendChild(option)
if (surveyNames.length > 0) {
let df = data[surveyNames[0]]
let quadSelect = document.getElementById('quadSelect')
quadSelect.innerHTML = ''
for (let i = 0; i < df['a'].length; i++) {
quad = [df['a'][i], df['b'][i], df['m'][i], df['n'][i]]
quads.push(quad)
let option = document.createElement('option')
option.value = quad.join(', ')
option.innerText = quad.join(', ')
quadSelect.appendChild(option)
}
console.log('quads=', quads)
}
console.log('quads=', quads)
}
// update time-serie figure
......@@ -499,7 +502,7 @@
function removeDataBtnFunc() {
sendCommand('{"cmd": "removeData"}',function(x) {
data = {}
output.innerHTML = 'Status: ' + x['status'] + ' (all data cleared)'
output.innerHTML = 'Status: ' + x['ohmpi_status'] + ' (all data cleared)'
console.log('all data removed')
})
}
......
logging_setup.py
+ 1
− 1
View file @ 7587d63f
......@@ -76,7 +76,7 @@ def setup_loggers(mqtt=True):
interval=DATA_LOGGING_CONFIG['interval'])
data_formatter = logging.Formatter(log_format)
data_formatter.converter = gmtime
data_formatter.datefmt = '%Y/%m/%d %H:%M:%S UTC'
data_formatter.datefmt = '%Y-%m-%d %H:%M:%S UTC'
data_handler.setFormatter(exec_formatter)
data_logger.addHandler(data_handler)
data_logger.setLevel(DATA_LOGGING_CONFIG['logging_level'])
......
ohmpi.py
+ 143
− 222
View file @ 7587d63f
......@@ -4,7 +4,7 @@ created on January 6, 2020.
Updates May 2022, Oct 2022.
Ohmpi.py is a program to control a low-cost and open hardware resistivity meter OhmPi that has been developed by
Rémi CLEMENT (INRAE), Vivien DUBOIS (INRAE), Hélène GUYARD (IGE), Nicolas FORQUET (INRAE), Yannick FARGIER (IFSTTAR)
Olivier KAUFMANN (UMONS), Arnaud WATELET (UMONS) and Guillaume BLANCHY (ILVO).
Olivier KAUFMANN (UMONS), Arnaud WATELET (UMONS) and Guillaume BLANCHY (FNRS/ULiege).
"""
import os
......@@ -19,7 +19,6 @@ from io import StringIO
from datetime import datetime
from termcolor import colored
import threading
import paho.mqtt.client as mqtt_client
from logging_setup import setup_loggers
from config import MQTT_CONTROL_CONFIG, OHMPI_CONFIG, EXEC_LOGGING_CONFIG
from logging import DEBUG
......@@ -53,7 +52,7 @@ class OhmPi(object):
def __init__(self, settings=None, sequence=None, use_mux=False, mqtt=True, on_pi=None, idps=False):
"""Constructs the ohmpi object
Parameters
----------
settings:
......@@ -69,7 +68,7 @@ class OhmPi(object):
idps:
if true uses the DPS
"""
if on_pi is None:
_, on_pi = OhmPi._get_platform()
......@@ -86,40 +85,6 @@ class OhmPi(object):
self.soh_logger = None # TODO: Implement the SOH logger
print(msg)
# set controller
self.controller = mqtt_client.Client(f"ohmpi_{OHMPI_CONFIG['id']}_listener", clean_session=False)
trials = 0
trials_max = 10
broker_connected = False
self.exec_logger.debug(f"Connecting to control broker: {MQTT_CONTROL_CONFIG['hostname']}")
while trials < trials_max:
try:
self.controller.username_pw_set(MQTT_CONTROL_CONFIG['auth'].get('username'),
MQTT_CONTROL_CONFIG['auth']['password'])
self.controller.connect(MQTT_CONTROL_CONFIG['hostname'])
trials = trials_max
broker_connected = True
except Exception as e:
self.exec_logger.debug(f'Unable to connect control broker: {e}')
self.exec_logger.debug('Retrying to connect control broker...')
time.sleep(2)
trials += 1
if broker_connected:
self.exec_logger.debug(f"Subscribing to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']}")
try:
self.controller.subscribe(MQTT_CONTROL_CONFIG['ctrl_topic'], MQTT_CONTROL_CONFIG['qos'])
msg = f"\u2611 Subscribed to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']}" \
f" on {MQTT_CONTROL_CONFIG['hostname']} broker"
self.exec_logger.debug(msg)
print(colored(msg, 'blue'))
except Exception as e:
self.exec_logger.warning(f'Unable to subscribe to control topic : {e}')
self.controller = None
else:
self.exec_logger.error(f"Unable to connect to control broker on {MQTT_CONTROL_CONFIG['hostname']}")
self.controller = None
# read in hardware parameters (config.py)
self._read_hardware_config()
......@@ -179,14 +144,68 @@ class OhmPi(object):
self.pin1.direction = Direction.OUTPUT
self.pin1.value = False
# Starts the command processing thread
if self.controller is not None:
self.cmd_listen = True
self.cmd_thread = threading.Thread(target=self._control)
self.cmd_thread.start()
else:
self.exec_logger.warning('No connection to control broker.'
' Use python/ipython to interact with OhmPi object...')
# set controller
self.mqtt = mqtt
self.cmd_id = None
if self.mqtt:
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']}"
f" on {MQTT_CONTROL_CONFIG['hostname']} broker")
def connect_mqtt() -> mqtt_client:
def on_connect(client, userdata, flags, rc):
if rc == 0:
self.exec_logger.debug(f"Successfully connected to control broker:"
f" {MQTT_CONTROL_CONFIG['hostname']}")
else:
self.exec_logger.warning(f'Failed to connect to control broker. Return code : {rc}')
client = mqtt_client.Client("ohmpi_{OHMPI_CONFIG['id']}_listener", clean_session=False)
client.username_pw_set(MQTT_CONTROL_CONFIG['auth'].get('username'),
MQTT_CONTROL_CONFIG['auth']['password'])
client.on_connect = on_connect
client.connect(MQTT_CONTROL_CONFIG['hostname'], MQTT_CONTROL_CONFIG['port'])
return client
try:
self.exec_logger.debug(f"Connecting to control broker: {MQTT_CONTROL_CONFIG['hostname']}")
self.controller = connect_mqtt()
except Exception as e:
self.exec_logger.debug(f'Unable to connect control broker: {e}')
self.controller = None
if self.controller is not None:
self.exec_logger.debug(f"Subscribing to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']}")
try:
self.controller.subscribe(MQTT_CONTROL_CONFIG['ctrl_topic'], MQTT_CONTROL_CONFIG['qos'])
msg = f"Subscribed to control topic {MQTT_CONTROL_CONFIG['ctrl_topic']}" \
f" on {MQTT_CONTROL_CONFIG['hostname']} broker"
self.exec_logger.debug(msg)
print(colored(f'\u2611 {msg}', 'blue'))
except Exception as e:
self.exec_logger.warning(f'Unable to subscribe to control topic : {e}')
self.controller = None
publisher_config = MQTT_CONTROL_CONFIG.copy()
publisher_config['topic'] = MQTT_CONTROL_CONFIG['ctrl_topic']
publisher_config.pop('ctrl_topic')
def on_message(client, userdata, message):
print(message.payload.decode('utf-8'))
command = message.payload.decode('utf-8')
dic = json.loads(command)
if dic['cmd_id'] != self.cmd_id:
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
else:
self.controller = None
self.exec_logger.warning('No connection to control broker.'
' Use python/ipython to interact with OhmPi object...')
@property
def sequence(self):
......@@ -205,18 +224,6 @@ class OhmPi(object):
self.use_mux = False
self._sequence = sequence
def _control(self):
"""Gets commands from the controller(s) and execute them"""
def on_message(client, userdata, message):
command = message.payload.decode('utf-8')
self.exec_logger.debug(f'Received command {command}')
self._process_commands(command)
self.controller.on_message = on_message
self.controller.loop_start()
while True:
time.sleep(.5) # TODO: Check if this waiting time should be reduced...
def _update_acquisition_settings(self, config):
warnings.warn('This function is deprecated, use update_settings() instead.', DeprecationWarning)
self.update_settings(config)
......@@ -297,7 +304,7 @@ class OhmPi(object):
@staticmethod
def _get_platform():
"""Gets platform name and checks if it is a raspberry pi
Returns
-------
str, bool
......@@ -403,7 +410,8 @@ class OhmPi(object):
else:
mcp2.get_pin(relay_nr - 1).value = False
self.exec_logger.debug(f'Switching relay {relay_nr} {state} for electrode {electrode_nr}')
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}')
......@@ -456,11 +464,13 @@ class OhmPi(object):
Parameters
----------
channel:
channel : object
Instance of ADS where voltage is measured.
Returns
-------
float
gain : float
Gain to be applied on ADS1115.
"""
gain = 2 / 3
if (abs(channel.voltage) < 2.040) and (abs(channel.voltage) >= 1.023):
......@@ -553,9 +563,9 @@ class OhmPi(object):
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
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)
......@@ -571,7 +581,7 @@ class OhmPi(object):
# compute constant
c = vmn / I
Rab = (volt * 1000) / I
Rab = (volt * 1000.) / I
self.exec_logger.debug(f'Rab = {Rab:.2f} Ohms')
......@@ -585,9 +595,9 @@ class OhmPi(object):
iab = voltage_max / c
vab = iab * Rab
self.exec_logger.debug('target max voltage')
if vab > 25000:
vab = 25000
vab = vab / 1000 * 0.9
if vab > 25000.:
vab = 25000.
vab = vab / 1000. * 0.9
elif strategy == 'vmin':
vmn_min = c * current_min
......@@ -598,9 +608,9 @@ class OhmPi(object):
iab = voltage_min / c
vab = iab * Rab
self.exec_logger.debug('target min voltage')
if vab < 1000:
vab = 1000
vab = vab / 1000 * 1.1
if vab < 1000.:
vab = 1000.
vab = vab / 1000. * 1.1
elif strategy == 'constant':
vab = volt
......@@ -614,7 +624,8 @@ class OhmPi(object):
return vab, polarity
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,
cmd_id=None):
"""Measures on a quadrupole and returns transfer resistance.
Parameters
......@@ -742,7 +753,7 @@ class OhmPi(object):
else:
self.pin0.value = False
self.pin1.value = True # current injection nr2
self.exec_logger.debug(str(n) + ' ' + str(self.pin0.value) + ' ' + str(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
meas = np.zeros((self.nb_samples, 3)) * np.nan
......@@ -759,7 +770,7 @@ class OhmPi(object):
elif self.board_version == '22.10':
meas[k, 1] = -AnalogIn(self.ads_voltage, ads.P0, ads.P1).voltage * self.coef_p2 * 1000
# else:
# self.exec_logger.debug('Unknown board')
# self.exec_logger.debug('Unknown board')
time.sleep(sampling_interval / 1000)
dt = time.time() - start_delay # real injection time (s)
meas[k, 2] = time.time() - start_time
......@@ -873,17 +884,6 @@ class OhmPi(object):
d = {'time': datetime.now().isoformat(), 'A': quad[0], 'B': quad[1], 'M': quad[2], 'N': quad[3],
'R [ohm]': np.abs(np.random.randn(1)).tolist()}
# round number to two decimal for nicer string output
output = [f'{k}\t' for k in d.keys()]
output = str(output)[:-1] + '\n'
for k in d.keys():
if isinstance(d[k], float):
val = np.round(d[k], 2)
else:
val = d[k]
output += f'{val}\t'
output = output[:-1]
# to the data logger
dd = d.copy()
dd.pop('fulldata') # too much for logger
......@@ -891,7 +891,14 @@ class OhmPi(object):
dd.update({'B': str(dd['B'])})
dd.update({'M': str(dd['M'])})
dd.update({'N': str(dd['N'])})
self.data_logger.info(json.dumps(dd))
# round float to 2 decimal
for key in dd.keys():
if isinstance(dd[key], float):
dd[key] = np.round(dd[key], 3)
dd['cmd_id'] = str(cmd_id)
self.data_logger.info(dd)
return d
......@@ -962,11 +969,9 @@ class OhmPi(object):
# close mux path and put pin back to GND
self.switch_mux_off(quad)
self.reset_mux()
else:
pass
self.status = 'idle'
# self.run = False
#
# # TODO if interrupted, we would need to restore the values
......@@ -1024,10 +1029,12 @@ class OhmPi(object):
cmd_id = decoded_message.pop('cmd_id', None)
cmd = decoded_message.pop('cmd', None)
args = decoded_message.pop('args', '[]')
if args[0] != '[':
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
......@@ -1040,7 +1047,9 @@ class OhmPi(object):
output = getattr(self, cmd)(*args, **kwargs)
status = True
except Exception as e:
self.exec_logger.error(f'{e}\nUnable to execute {cmd}({args}, {kwargs}')
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}')
......@@ -1050,10 +1059,6 @@ class OhmPi(object):
reply = json.dumps(reply)
self.exec_logger.debug(f'Execution report: {reply}')
def measure(self, *args, **kwargs):
warnings.warn('This function is deprecated. Use load_sequence instead.', DeprecationWarning)
self.run_sequence(*args, **kwargs)
def set_sequence(self, sequence=sequence):
try:
self.sequence = np.loadtxt(StringIO(sequence)).astype('uint32')
......@@ -1063,7 +1068,8 @@ class OhmPi(object):
status = False
def run_sequence(self, cmd_id=None, **kwargs):
"""Runs sequence in sync mode
"""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}')
......@@ -1110,160 +1116,72 @@ class OhmPi(object):
acquired_data.update({'cmd_id': cmd_id})
# log data to the data logger
self.data_logger.info(f'{acquired_data}')
print(f'{acquired_data}')
# save data and print in a text file
self.append_and_save(filename, acquired_data)
self.exec_logger.debug(f'{i+1:d}/{n:d}')
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:
cmd_id:
"""
# self.run = True
self.status = 'running'
self.exec_logger.debug(f'Status: {self.status}')
self.exec_logger.debug(f'Measuring sequence: {self.sequence}')
def func():
# if self.status != 'running':
# self.exec_logger.warning('Data acquisition interrupted')
# break
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}')
print(f'{acquired_data}')
# save data and print in a text file
self.append_and_save(filename, acquired_data)
self.exec_logger.debug(f'{i+1:d}/{n:d}')
self.status = 'idle'
self.run_sequence(**kwargs)
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().
def run_multiple_sequences(self, *args, **kwargs):
"""Run multiple sequences in a separate thread for monitoring mode.
Can be stopped by 'OhmPi.interrupt()'.
Parameters
----------
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}')
cmd_id = kwargs.pop('cmd_id', None)
def func():
for g in range(0, self.settings["nb_meas"]): # for time-lapse monitoring
if self.status != 'running':
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()
# 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}')
print(f'{acquired_data}')
# save data and print in a text file
self.append_and_save(filename, acquired_data)
self.exec_logger.debug(f'{i+1:d}/{n:d}')
# compute time needed to take measurement and subtract it from interval
# between two sequence run (= sequence_delay)
measuring_time = time.time() - t0
sleep_time = self.settings["sequence_delay"] - measuring_time
if sleep_time < 0:
# it means that the measuring time took longer than the sequence delay
sleep_time = 0
self.exec_logger.warning('The measuring time is longer than the sequence delay. '
'Increase the sequence delay')
self.run_sequence(**kwargs)
# sleeping time between sequence
if self.settings["nb_meas"] > 1:
time.sleep(sleep_time) # waiting for next measurement (time-lapse)
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()
......@@ -1275,6 +1193,7 @@ class OhmPi(object):
"""Interrupts the acquisition. """
self.status = 'stopping'
if self.thread is not None:
self.exec_logger.debug('Joining tread...')
self.thread.join()
else:
self.exec_logger.debug('No sequence measurement thread to interrupt.')
......@@ -1320,3 +1239,5 @@ print(f'local date and time : {current_time.strftime("%Y-%m-%d %H:%M:%S")}')
# for testing
if __name__ == "__main__":
ohmpi = OhmPi(settings=OHMPI_CONFIG['settings'])
if ohmpi.controller is not None:
ohmpi.controller.loop_forever()
ohmpi_settings.json
+ 1
− 1
View file @ 7587d63f
......@@ -3,6 +3,6 @@
"injection_duration": 0.2,
"nb_stack": 1,
"nb_meas": 1,
"sequence_delay": 1,
"sequence_delay": 120,
"export_path": "data/measurement.csv"
}
test_ohmpi_mux.py 0 → 100644
+ 144
− 0
View file @ 7587d63f
# test ohmpi and multiplexer on test resistances
from ohmpi import OhmPi
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import os
import time
# configure testing
idps = False
board_version = '22.10' # v2.0
use_mux = True
start_elec = 17 # start elec
nelec = 16 # max elec in the sequence for testing
# nelec electrodes Wenner sequence
a = np.arange(nelec-3) + start_elec
b = a + 3
m = a + 1
n = a + 2
seq = np.c_[a, b, m, n]
# testing measurement board only
k = OhmPi(idps=idps, use_mux=use_mux)
k.sequence = seq # we need a sequence for possibly switching the mux
k.reset_mux() # just for safety
if use_mux:
k.switch_mux_on(seq[:, 0])
out1 = k.run_measurement(injection_duration=0.25, nb_stack=4, strategy='constant', tx_volt=12, autogain=True)
out2 = k.run_measurement(injection_duration=0.5, nb_stack=2, strategy='vmin', tx_volt=5, autogain=True)
out3 = k.run_measurement(injection_duration=1, nb_stack=1, strategy='vmax', tx_volt=5, autogain=True)
if use_mux:
k.switch_mux_off(seq[:, 0])
# visual figure of the full wave form
fig, axs = plt.subplots(2, 1, sharex=True)
ax = axs[0]
labels = ['constant', 'vmin', 'vmax']
for i, out in enumerate([out1, out2, out3]):
data = out['fulldata']
inan = ~(np.isnan(out['fulldata']).any(1))
ax.plot(data[inan,2], data[inan,0], '.-', label=labels[i])
ax.set_ylabel('Current AB [mA]')
ax.legend()
ax = axs[1]
for i, out in enumerate([out1, out2, out3]):
data = out['fulldata']
inan = ~(np.isnan(out['fulldata']).any(1))
ax.plot(data[inan,2], data[inan,1], '.-', label=labels[i])
ax.set_ylabel('Voltage MN [mV]')
ax.set_xlabel('Time [s]')
fig.savefig('check-fullwave.jpg')
# test a sequence
if use_mux:
# manually edit default settings
k.settings['injection_duration'] = 1
k.settings['nb_stack'] = 1
#k.settings['nbr_meas'] = 1
k.sequence = seq
k.reset_mux()
# set quadrupole manually
k.switch_mux_on(seq[0, :])
out = k.run_measurement(quad=[3, 3, 3, 3], nb_stack=1, tx_volt=12, strategy='constant', autogain=True)
k.switch_mux_off(seq[0, :])
print(out)
# run rs_check() and save data
k.rs_check() # check all electrodes of the sequence
# check values measured
fname = sorted(os.listdir('data/'))[-1]
print(fname)
dfrs = pd.read_csv('data/' + fname)
fig, ax = plt.subplots()
ax.hist(dfrs['RS [kOhm]'])
ax.set_xticks(np.arange(dfrs.shape[0]))
ax.set_xticklabels(dfrs['A'].astype(str) + ' - ' +
dfrs['B'].astype(str), rotation=90)
ax.set_ylabel('Contact resistances [kOhm]')
fig.tight_layout()
fig.savefig('check-rs.jpg')
# run sequence synchronously and save data to file
k.run_sequence(nb_stack=1, injection_duration=0.25)
# check values measured
fname = sorted(os.listdir('data/'))[-1]
print(fname)
df = pd.read_csv('data/' + fname)
fig, ax = plt.subplots()
ax.hist(df['R [ohm]'])
ax.set_ylabel('Transfer resistance [Ohm]')
ax.set_xticks(np.arange(df.shape[0]))
ax.set_xticklabels(df['A'].astype(str) + ','
+ df['B'].astype(str) + ','
+ df['M'].astype(str) + ','
+ df['N'].astype(str), rotation=90)
fig.tight_layout()
fig.savefig('check-r.jpg')
# run sequence asynchronously and save data to file
k.run_sequence_async(nb_stack=1, injection_duration=0.25)
time.sleep(2) # if too short, it will still be resetting the mux (= no effect)
k.interrupt() # will kill the asynchronous sequence running
# run a series of asynchronous sequences
k.settings['nb_meas'] = 2 # run the sequence twice
k.run_multiple_sequences(nb_stack=1, injection_duration=0.25)
time.sleep(5)
k.interrupt()
# look at the noise frequency with FFT
if False:
from numpy.fft import fft, ifft
x = data[inan, 1][10:300]
t = np.linspace(0, len(x)*4, len(x))
sr = 1/0.004
X = fft(x)
N = len(X)
n = np.arange(N)
T = N/sr
freq = n/T
plt.figure(figsize = (12, 6))
plt.subplot(121)
plt.stem(freq, np.abs(X), 'b', \
markerfmt=" ", basefmt="-b")
plt.xlabel('Freq (Hz)')
plt.ylabel('FFT Amplitude |X(freq)|')
#plt.xlim(0, 10)
plt.subplot(122)
plt.plot(t, ifft(X), 'r')
plt.xlabel('Time (s)')
plt.ylabel('Amplitude')
plt.tight_layout()
plt.show()
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