diff --git a/config.py b/config.py
index 0767b2b4e0657a9cd9098d1878c9dfa8fdd58315..f0267d83f50872e7df020ddddc91888e807fa410 100644
--- a/config.py
+++ b/config.py
@@ -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
diff --git a/doc/source/V2_00.rst b/doc/source/V2_00.rst
index 06f7a30d0d0d3cc2e9d5a1687e44f6b43650b9ad..2276454a1b5913408c0696594023b7afc34f425d 100644
--- a/doc/source/V2_00.rst
+++ b/doc/source/V2_00.rst
@@ -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).
diff --git a/example_simple_measurement.py b/example_simple_measurement.py
index 2a5d50939cb20ac9506de31d4cd8bb451a119c2a..ee795b1808cf4fb2bf64484b566d58dc2275b984 100644
--- a/example_simple_measurement.py
+++ b/example_simple_measurement.py
@@ -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
diff --git a/http_interface.py b/http_interface.py
index bc0e71f3db458c1a3ba2cfab94ff949d2e3edab3..9c43bae70f13ba22942f9f5a0273d1a1c0f41d0b 100644
--- a/http_interface.py
+++ b/http_interface.py
@@ -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(),
diff --git a/index.html b/index.html
index db104d009f47532bc3ed3fcbdd304433b72c4496..70a2a2da71a4ede06ddc949424bdaa54a2a1a55d 100644
--- a/index.html
+++ b/index.html
@@ -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')
})
}
diff --git a/logging_setup.py b/logging_setup.py
index ad458f3df3b2a439e7086657fa143eb1692cf703..4c8162a4f6aeb6e578fd38d0fa811549d9e764df 100644
--- a/logging_setup.py
+++ b/logging_setup.py
@@ -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'])
diff --git a/ohmpi.py b/ohmpi.py
index 1eb110078099c2c1be379da3d06cc4f2ac1db6f6..ff36d078efe9a467b1246e6a41c8b47d2cd3ee0e 100644
--- a/ohmpi.py
+++ b/ohmpi.py
@@ -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()
diff --git a/ohmpi_settings.json b/ohmpi_settings.json
index 8d4907d7a2f3c971bec8dfd529cfbec10aa7bb7b..e839f5fb9a4ddef894a5a8b714757b71cd860c02 100644
--- a/ohmpi_settings.json
+++ b/ohmpi_settings.json
@@ -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"
}
diff --git a/test_ohmpi_mux.py b/test_ohmpi_mux.py
new file mode 100644
index 0000000000000000000000000000000000000000..ed3222f6b4cd2e79c282d71f9e025f1f57455181
--- /dev/null
+++ b/test_ohmpi_mux.py
@@ -0,0 +1,144 @@
+# 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()