Skip to content
GitLab
Commit 27ae3d16 authored by Arnaud WATLET's avatar Arnaud WATLET
Browse files

Updates test_mux and config 3 mux 2024

parent 23fc9eca
master 144-create-testing-framework-for-tests-automation code_refactor dev/dummy 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
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Showing
with 260 additions and 6 deletions
+260 -6
configs/config_mb_2024_0_2__3_mux_2024_dps5005.py 0 → 100644
+ 144
− 0
View file @ 27ae3d16
import logging
from ohmpi.utils import get_platform
from paho.mqtt.client import MQTTv31 # noqa
_, on_pi = get_platform()
# DEFINE THE ID OF YOUR OhmPi
ohmpi_id = '0001' if on_pi else 'XXXX'
# DEFINE YOUR MQTT BROKER (DEFAULT: 'localhost')
mqtt_broker = 'localhost' if on_pi else 'NAME_YOUR_BROKER_WHEN_IN_SIMULATION_MODE_HERE'
# DEFINE THE SUFFIX TO ADD TO YOUR LOGS FILES
logging_suffix = ''
# OhmPi configuration
OHMPI_CONFIG = {
'id': ohmpi_id, # Unique identifier of the OhmPi board (string)
'settings': 'ohmpi_settings.json', # INSERT YOUR FAVORITE SETTINGS FILE HERE
}
HARDWARE_CONFIG = {
'ctl': {'model': 'raspberry_pi'},
'pwr': {'model': 'pwr_dps5005', 'voltage': 3., 'interface_name': 'modbus'},
'tx': {'model': 'mb_2024_0_2',
'voltage_max': 50., # Maximum voltage supported by the TX board [V]
'current_max': 4800, # Maximum voltage read by the current ADC on the TX board [mA]
'r_shunt': 2, # Shunt resistance in Ohms
'interface_name': 'i2c'
},
'rx': {'model': 'mb_2024_0_2',
'coef_p2': 1.00, # slope for conversion for ADS, measurement in V/V
'latency': 0.010, # latency in seconds in continuous mode
'sampling_rate': 50, # number of samples per second
'interface_name': 'i2c'
},
'mux': {'boards':
{'mux_02':
{'model': 'mux_2024_0_X',
'tca_address': None,
'tca_channel': 0,
'addr2': 'up',
'addr1': 'up',
# 'mcp_0': '0x26',
# 'mcp_1': '0x27',
'roles': {'A': 'X', 'B': 'Y', 'M': 'XX', 'N': 'YY'},
'cabling': {(i+8, j): ('mux_02', i) for j in ['A', 'B', 'M', 'N'] for i in range(1, 9)},
'voltage_max': 50.},
'mux_03':
{'model': 'mux_2024_0_X',
'tca_address': None,
'tca_channel': 0,
'addr2': 'down',
'addr1': 'up',
# 'mcp_0': '0x26',
# 'mcp_1': '0x27',
'roles': {'A': 'X', 'B': 'Y', 'M': 'XX', 'N': 'YY'},
'cabling': {(i+24, j): ('mux_03', i) for j in ['A', 'B', 'M', 'N'] for i in range(1, 9)},
'voltage_max': 50.},
'mux_05':
{'model': 'mux_2024_0_X',
'tca_address': None,
'tca_channel': 0,
'addr2': 'up',
'addr1': 'down',
'roles': {'A': 'X', 'B': 'Y', 'M': 'XX', 'N': 'YY'},
'cabling': {(i+0, j): ('mux_05', i) for j in ['A', 'B', 'M', 'N'] for i in range(1, 9)},
'voltage_max': 50.},
},
'default': {'interface_name': 'i2c_ext',
'voltage_max': 100.,
'current_max': 3.}
}
}
# SET THE LOGGING LEVELS, MQTT BROKERS AND MQTT OPTIONS ACCORDING TO YOUR NEEDS
# Execution logging configuration
EXEC_LOGGING_CONFIG = {
'logging_level': logging.INFO,
'log_file_logging_level': logging.DEBUG,
'logging_to_console': True,
'file_name': f'exec{logging_suffix}.log',
'max_bytes': 262144,
'backup_count': 30,
'when': 'd',
'interval': 1
}
# Data logging configuration
DATA_LOGGING_CONFIG = {
'logging_level': logging.INFO,
'logging_to_console': True,
'file_name': f'data{logging_suffix}.log',
'max_bytes': 16777216,
'backup_count': 1024,
'when': 'd',
'interval': 1
}
# State of Health logging configuration (For a future release)
SOH_LOGGING_CONFIG = {
'logging_level': logging.INFO,
'logging_to_console': True,
'log_file_logging_level': logging.DEBUG,
'file_name': f'soh{logging_suffix}.log',
'max_bytes': 16777216,
'backup_count': 1024,
'when': 'd',
'interval': 1
}
# MQTT logging configuration parameters
MQTT_LOGGING_CONFIG = {
'hostname': mqtt_broker,
'port': 1883,
'qos': 2,
'retain': False,
'keepalive': 60,
'will': None,
'auth': {'username': 'mqtt_user', 'password': 'mqtt_password'},
'tls': None,
'protocol': MQTTv31,
'transport': 'tcp',
'client_id': f'{OHMPI_CONFIG["id"]}',
'exec_topic': f'ohmpi_{OHMPI_CONFIG["id"]}/exec',
'exec_logging_level': logging.DEBUG,
'data_topic': f'ohmpi_{OHMPI_CONFIG["id"]}/data',
'data_logging_level': DATA_LOGGING_CONFIG['logging_level'],
'soh_topic': f'ohmpi_{OHMPI_CONFIG["id"]}/soh',
'soh_logging_level': SOH_LOGGING_CONFIG['logging_level']
}
# MQTT control configuration parameters
MQTT_CONTROL_CONFIG = {
'hostname': mqtt_broker,
'port': 1883,
'qos': 2,
'retain': False,
'keepalive': 60,
'will': None,
'auth': {'username': 'mqtt_user', 'password': 'mqtt_password'},
'tls': None,
'protocol': MQTTv31,
'transport': 'tcp',
'client_id': f'{OHMPI_CONFIG["id"]}',
'ctrl_topic': f'ohmpi_{OHMPI_CONFIG["id"]}/ctrl'
}
dev/test_mb_2024_3_mux_2024.py 0 → 100644
+ 104
− 0
View file @ 27ae3d16
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2__3_mux_2024_dps5005.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
stand_alone = False
part_of_hardware_system = True
within_ohmpi = False
# Stand alone
if stand_alone:
ctl_module = importlib.import_module(f'ohmpi.hardware_components.{HARDWARE_CONFIG["ctl"].pop("model")}')
pwr_module = importlib.import_module(f'ohmpi.hardware_components.{HARDWARE_CONFIG["pwr"].pop("model")}')
tx_module = importlib.import_module(f'ohmpi.hardware_components.{HARDWARE_CONFIG["tx"].pop("model")}')
rx_module = importlib.import_module(f'ohmpi.hardware_components.{HARDWARE_CONFIG["rx"].pop("model")}')
ctl = ctl_module.Ctl()
HARDWARE_CONFIG['tx'].update({'ctl': ctl, 'exec_logger': ctl.exec_logger, 'soh_logger': ctl.soh_logger})
HARDWARE_CONFIG['rx'].update({'ctl': ctl, 'exec_logger': ctl.exec_logger, 'soh_logger': ctl.soh_logger})
HARDWARE_CONFIG['tx'].update({'connection': HARDWARE_CONFIG['tx'].pop('connection',
ctl.interfaces[
HARDWARE_CONFIG['tx'].pop(
'interface_name', 'i2c')])})
HARDWARE_CONFIG['rx'].update({'connection': HARDWARE_CONFIG['rx'].pop('connection',
ctl.interfaces[
HARDWARE_CONFIG['rx'].pop(
'interface_name', 'i2c')])})
HARDWARE_CONFIG['pwr'].update({'connection': HARDWARE_CONFIG['pwr'].pop('connection',
ctl.interfaces[
HARDWARE_CONFIG['pwr'].pop(
'interface_name', None)])})
rx = rx_module.Rx(**HARDWARE_CONFIG['rx'])
tx = tx_module.Tx(**HARDWARE_CONFIG['tx'])
pwr = pwr_module.Pwr(**HARDWARE_CONFIG['pwr'])
mux_ids = ['mux_02', 'mux_05']
for m,mux_id in enumerate(mux_ids):
mux_module = importlib.import_module(
f'ohmpi.hardware_components.{HARDWARE_CONFIG["mux"]["boards"][mux_id].pop("model")}')
MUX_CONFIG = HARDWARE_CONFIG['mux']['boards'][mux_id]
MUX_CONFIG.update({'ctl': ctl, 'connection': MUX_CONFIG.pop('connection', ctl.interfaces[
MUX_CONFIG.pop('interface_name', 'i2c_ext')]), 'exec_logger': ctl.exec_logger,
'soh_logger': ctl.soh_logger})
MUX_CONFIG.update({'id': mux_id})
mux = mux_module.Mux(**MUX_CONFIG)
# tx.polarity = 1
# time.sleep(1)
# tx.polarity = 0
# mux.switch(elec_dict={'A': [1], 'B': [4], 'M': [2], 'N': [3]}, state='on')
# time.sleep(1)
# voltage = rx.voltage
# current = tx.current
# mux.switch(elec_dict={'A': [1], 'B': [4], 'M': [2], 'N': [3]}, state='off')
# print(f'Resistance: {voltage / current :.2f} ohm, voltage: {voltage:.2f} mV, current: {current:.2f} mA')
mux.reset()
mux.test({'A': [i+8*m for i in range(1, 9)], 'B': [i+8*m for i in range(1, 9)],
'M': [i+8*m for i in range(1, 9)], 'N': [i+8*m for i in range(1, 9)]}, activation_time=.1)
mux.reset()
# mux as part of a OhmPiHardware system
if part_of_hardware_system:
from ohmpi.hardware_system import OhmPiHardware
print('Starting test of as part of an OhmPiHardware system.')
# mux_id = 'mux_03'
k = OhmPiHardware()
k.exec_logger.setLevel(logging.DEBUG)
# Test mux switching
k.reset_mux()
# k.switch_mux(electrodes=[1, 4, 2, 3], roles=['A', 'B', 'M', 'N'], state='on')
# time.sleep(1.)
# k.switch_mux(electrodes=[1, 4, 2, 3], roles=['A', 'B', 'M', 'N'], state='off')
# k.mux_boards[mux_id].test(activation_time=.4)
k.test_mux()
k.reset_mux()
if within_ohmpi:
from ohmpi.ohmpi import OhmPi
# from ohmpi.plots import plot_exec_log
print('Starting test with OhmPi.')
k = OhmPi()
# A, B, M, N = (32, 29, 31, 30)
k.reset_mux()
k.load_sequence('sequences/ABMN2.txt')
k.run_sequence(tx_volt=50, injection_duration=1., nb_stack=2, duty_cycle=0.5)
print('using OhmPi')
#d = k.run_measurement([A, B, M, N], injection_duration=1., nb_stack=2, duty_cycle=0.5)
# print(d)
# k._hw._plot_readings()
print(f'OhmPiHardware: Resistance: {k._hw.last_resistance() :.2f} ohm, dev. {k._hw.last_dev():.2f} %, sp: {k._hw.sp:.2f} mV, rx bias: {k._hw.rx._bias:.2f} mV')
print(f'OhmPi: Resistance: {d["R [ohm]"] :.2f} ohm, dev. {d["R_std [%]"]:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings(save_fig=False)
# plot_exec_log('ohmpi/logs/exec.log')
change_config('../configs/config_default.py', verbose=False)
ohmpi/hardware_system.py
+ 12
− 6
View file @ 27ae3d16
...@@ -546,7 +546,7 @@ class OhmPiHardware: ...@@ -546,7 +546,7 @@ class OhmPiHardware:
# Create a new thread to perform some work # Create a new thread to perform some work
self.mux_boards[mux].barrier = b self.mux_boards[mux].barrier = b
kwargs.update({'elec_dict': elec_dict, 'state': state}) kwargs.update({'elec_dict': elec_dict, 'state': state})
mux_workers[idx] = Thread(target=self.mux_boards[mux].switch, kwargs=kwargs) mux_workers[idx] = Thread(target=self.mux_boards[mux].switch, kwargs=kwargs) # TODO: handle minimum delay between two relays activation (to avoid lagging during test_mux at high speed)
mux_workers[idx].start() mux_workers[idx].start()
try: try:
self.mux_barrier.wait() self.mux_barrier.wait()
...@@ -585,11 +585,17 @@ class OhmPiHardware: ...@@ -585,11 +585,17 @@ class OhmPiHardware:
time.sleep(activation_time) time.sleep(activation_time)
self.switch_mux(electrodes, roles, state='off') self.switch_mux(electrodes, roles, state='off')
else: else:
for c in self._cabling.keys(): for m_id in [for i in self.mux_boards.keys()].sort():
self.exec_logger.info(f'Testing electrode {c[0]} with role {c[1]}.') for c in self.mux_boards[m_id].cabling.keys():
self.switch_mux(electrodes=[c[0]], roles=[c[1]], state='on') self.exec_logger.info(f'Testing electrode {c[0]} with role {c[1]}.')
time.sleep(activation_time) self.switch_mux(electrodes=[c[0]], roles=[c[1]], state='on')
self.switch_mux(electrodes=[c[0]], roles=[c[1]], state='off') time.sleep(activation_time)
self.switch_mux(electrodes=[c[0]], roles=[c[1]], state='off')
# for c in self._cabling.keys():
# self.exec_logger.info(f'Testing electrode {c[0]} with role {c[1]}.')
# self.switch_mux(electrodes=[c[0]], roles=[c[1]], state='on')
# time.sleep(activation_time)
# self.switch_mux(electrodes=[c[0]], roles=[c[1]], state='off')
self.exec_logger.info('Test finished.') self.exec_logger.info('Test finished.')
def reset_mux(self): def reset_mux(self):
......
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