Source

Target

Commits (2619)
Showing with 1775 additions and 0 deletions
+1775 -0
import logging
from ohmpi.utils import *
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': 'settings/default.json', # INSERT YOUR FAVORITE SETTINGS FILE HERE
}
r_shunt = 2
# default properties of system components that will be
# overwritten by properties defined in each the board dict below.
# if bounds are defined in board specs, values out of specs will be bounded to remain in specs
# omitted properties in config will be set to board specs default values if they exist
HARDWARE_CONFIG = {
'ctl': {'model': 'raspberry_pi'},
'pwr': {'model': 'pwr_dph5005', 'voltage': 3., 'pwr_discharge_latency': 2.}, #, 'current_max':0.05,'current_overload':0.002}, # 'pwr_batt', 'voltage': 12.}, #
'tx': {'model': 'mb_2024_1_X',
'voltage_max': 50., # Maximum voltage supported by the TX board [V]
'current_max': 4.80/(50*r_shunt), # Maximum voltage read by the current ADC on the TX board [A]
'r_shunt': r_shunt, # Shunt resistance in Ohms
'interface_name': 'i2c',
'vmn_hardware_offset': 2500.
},
'rx': {'model': 'mb_2024_1_X',
'latency': 0.010, # latency in seconds in continuous mode
'sampling_rate': 800, # number of samples per second
'interface_name': 'i2c',
},
'mux': {'boards':
{'mux_01':
{'model': 'mux_2024_0_X',
'roles': ['A', 'B'],
'electrodes': mux_2023_to_mux_2024_takeouts(range(1, 17)),
'addr1': 'up',
'addr2': 'up',
'i2c_ext_tca_address': 0x70,
'i2c_ext_tca_channel': 0,
},
'mux_02':
{'model': 'mux_2024_0_X',
'roles': ['M', 'N'],
'electrodes': mux_2023_to_mux_2024_takeouts(range(1, 17)),
'addr1': 'down',
'addr2': 'down',
'i2c_ext_tca_address': 0x70,
'i2c_ext_tca_channel': 0,
}
},
'default': {'interface_name': 'i2c_ext',
'voltage_max': 50.,
'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.DEBUG,
'log_file_logging_level': logging.DEBUG,
'logging_to_console': True,
'file_name': f'exec{logging_suffix}.log',
'max_bytes': 3355443,
'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.INFO,
'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': EXEC_LOGGING_CONFIG['logging_level'],
'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'
}
# ohmpi integration test
# to get coverage use
# pip install coverage
# coverage run --source=ohmpi .dev/test_all.py
# coverage report --include=ohmpi/ohmpi.py,ohmpi/hardware_system.py
# coverage html
import time
import os
t0 = time.time()
from ohmpi.ohmpi import OhmPi
k = OhmPi()
print('---------- test ----------')
k.test(remote=True)
#k.test_mux(activation_time=0.1)) # already tested in k.test() > k._test_mux_ABMN()
print('---------- sequence generation ----------')
k.create_sequence(16, params=[('dpdp', 1, 8)])
k.interrupt()
k.create_sequence(16, params=[('schlum', 1, 8)])
k.create_sequence(16, params=[('multigrad', 1, 8, 2)])
k.load_sequence(os.path.join(os.path.dirname(__file__), '../sequences/ABMN.txt'))
k.create_sequence(16, params=[('wenner', 2)])
k.remove_data()
print('---------- rs_check ----------')
k._process_commands('{"cmd_id": "102392038", "cmd": "rs_check", "kwargs": {}}')
#k.rs_check()
#print('---------- async sequence ----------')
#k.run_sequence_async()
#time.sleep(3)
#k.interrupt()
#print('interrupting!!')
print('---------- vmax ----------')
vmax = k.run_measurement([1, 4, 2, 3], nb_stack=2, injection_duration=0.2, duty_cycle=0.5, strategy='vmax')
print('---------- vmin ----------')
vmin = k.run_measurement([1, 4, 2, 3], nb_stack=1, injection_duration=0.2, duty_cycle=0.3, strategy='vmin', vmn_req=0.02)
print('---------- safe ----------')
safe = k.run_measurement([1, 4, 2, 3], nb_stack=1, injection_duration=0.2, duty_cycle=0.7, strategy='safe', vab_req=40)
print('---------- fixed ----------')
fixed = k.run_measurement([1, 4, 2, 3], nb_stack=2, injection_duration=0.2, duty_cycle=1, strategy='fixed', vab_req=7.3)
print('---------- flex ----------')
flex = k.run_measurement([1, 4, 2, 3], nb_stack=2, injection_duration=0.2, duty_cycle=1, strategy='flex', pab_req=0.3)
dic = {
'vmin': vmin,
'vmax': vmax,
'safe': safe,
'fixed': fixed,
'flex': flex
}
for key in dic:
q = dic[key]
print('{:8s} >>> vab: {:10.2f} V, vmn: {:10.2f} mV, r: {:10.2f} Ohm'.format(
key, q['vab_[V]'], q['vmn_[mV]'], q['r_[Ohm]']))
#k.plot_last_fw()
print('---------- multiple sequences ----------')
k.run_multiple_sequences(sequence_delay=60, nb_meas=2)
time.sleep(20)
k.interrupt()
print('---------- inversion ----------')
k.remove_data()
k.create_sequence(8, params=[('wenner', 3)])
k.run_sequence(vab_req=5., duty_cycle=1, injection_duration=0.1, strategy='fixed', fw_in_zip=True)
datadir = os.path.join(os.path.dirname(__file__), '../data/')
fnames = sorted([f for f in os.listdir(datadir) if f[-4:] == '.csv'])
survey_names = [os.path.join(datadir, fnames[-1])]
k.get_data(survey_names=survey_names, full=True)
print(fnames[-1])
k.run_inversion(survey_names=survey_names)
print('---------- data download ----------')
k.download_data(ftype='ohmpi')
k.download_data(ftype='ohmpi', start_date='2022-01-01', end_date='2030-01-01')
k.download_data(ftype='pygimli')
k.download_data(ftype='protocol')
print('---------- find optimal vab ----------')
k.create_sequence(8, params=[('wenner', 3)])
k.find_optimal_vab_for_sequence(n_samples=3)
print('TOTAL time running: {:.1f}s'.format(time.time() - t0))
import os
import numpy as np
import time
os.chdir("/home/pi/OhmPi")
from ohmpi.ohmpi import OhmPi
from ohmpi.tests import contact_resistance_test_board
# Define object from class OhmPi
k = OhmPi()
# Set or load sequence
sequence = np.array([np.array([1,2,3,4])+k for k in range(29)]) # [[1,2,3,4],[2,3,4,5]...] but can actually make other combinations of AB to increase number of contact resistance tested
sequence = np.vstack([sequence,np.array([[30,31,1,2],[31,32,2,3]])])
sequence = sequence[-4:]
k.sequence = contact_resistance_test_board(sequence) # checks if sequence contains potential shortcut quads (AB odd odd or even even)
# Run contact resistance check
k.rs_check()
# Run sequence
k.run_sequence(strategy='vmax',vab=2., vab_max=5, nb_stack=2, injection_duration=.5, duty_cycle=1.)
import os
import numpy as np
import time
os.chdir("/home/pi/OhmPi")
from ohmpi.ohmpi import OhmPi
#from ohmpi.tests import contact_resistance_test_board
#######################
strategy = 'constant'
min_agg = True
r_ground = 5
vab_min = None
vab_req = 5.
vab_max = 12.
iab_min = 0.0025
iab_req = 0.005
iab_max = None
vmn_min = 0.75
vmn_req = 2.
vmn_max = None
pab_min = 0.0125
pab_req = 0.0625
pab_max = None
#######################
# Define object from class OhmPi
k = OhmPi()
k.reset_mux()
# Set or load sequence
sequence = np.array([np.array([1,2,3,4])+k for k in range(29)]) # [[1,2,3,4],[2,3,4,5]...] but can actually make other combinations of AB to increase number of contact resistance tested
sequence = np.vstack([sequence,np.array([[30,31,2,1],[31,32,3,2]])])
#sequence = sequence[:2]
#k.sequence = contact_resistance_test_board(sequence) # checks if sequence contains potential shortcut quads (AB odd odd or even even)
k.sequence = sequence
if strategy == 'vmin':
k.update_settings({'strategy': strategy, 'vmn_req': vmn_req})
k.update_settings({'export_path':f'data/r{r_ground}_{strategy}_{vmn_req:.1f}.csv'})
elif strategy == 'vmax':
k.update_settings({'strategy': strategy})
k.update_settings({'export_path':f'data/r{r_ground}_{strategy}.csv'})
elif strategy == 'flex':
k.update_settings({'strategy': strategy})
k.update_settings({'vab_min': vab_min, 'vab_req': vab_req, 'vab_max': vab_max, 'iab_min': iab_min, 'iab_req': iab_req, 'iab_max': iab_max,
'vmn_min': vmn_min, 'vmn_req': vmn_req, 'vmn_max': vmn_max, 'pab_min': pab_min, 'pab_req': pab_req, 'pab_max': pab_max, 'min_agg': min_agg})
k.update_settings({'export_path':f'data/r{r_ground}_{strategy}_vab-{vab_min}-{vab_req}-{vab_max}_iab-{iab_min}-{iab_req}-{iab_max}_vmn-{vmn_min}-{vmn_req}-{vmn_max}_pab-{pab_min}-{pab_req}-{pab_max}_{min_agg}.csv'})
elif strategy == "constant":
k.update_settings({'strategy': strategy})
k.update_settings({'vab_min': vab_min, 'vab_req': vab_req, 'vab_max': vab_max})
k.update_settings({'export_path':f'data/r{r_ground}_{strategy}_vab_req-{vab_req}-vab_max-{vab_max}.csv'})
print(f'Settings: {k.settings}')
# Run contact resistance check
#k.rs_check()
#print('TX',k._hw.tx.specs, 'PWR', k._hw.pwr.specs, 'RX', k._hw.rx.specs)
#print(k._hw.vab_min)
# Run sequence
#kwargs = {'vab_square_wave':{'append':False}}
k.run_sequence(nb_stack=2, injection_duration=.5, duty_cycle=1.,save_strategy_fw=True)
#k.plot_last_fw()
from ohmpi.hardware_components.raspberry_pi import Ctl
ctl = Ctl()
print(ctl.interfaces)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2__4_mux_2024_dph5005.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
from ohmpi.ohmpi import OhmPi
k = OhmPi()
k.pwr_state = 'on'
k._hw.tx.voltage = 50
k._hw.tx.pwr.pwr_state = 'on'
time.sleep(.5)
k._hw.tx.voltage = 5
time.sleep(.5)
k._hw.tx.pwr.pwr_state = 'off'
k.pwr_state = 'off'
import matplotlib
matplotlib.use('TkAgg')
import time
from ohmpi.ohmpi import OhmPi
k = OhmPi()
if True:
# from resistor board find max voltage that can be tried out (manually)
k._hw.pwr_state = 'on' # if you start with 'on' like in run_sequence()
# dps isn't turned off between run_measurment() so more likely to get decay issue
out1 = k.run_measurement([1, 4, 2, 3], injection_duration=0.5, vab_req=30, strategy='constant', duty_cycle=0.5)
out = k.run_measurement([1, 4, 2, 3], injection_duration=0.5, vab_req=5, strategy='constant', duty_cycle=0.5)
k._hw._plot_readings(save_fig=False)
# cannot use run_measurement as it doesn't modify the dps voltage
# trying a test at the ohmpi hardware level
# there is an unwanted delay between each pulse (maybe due to tx.sync_wait() ?)
# k._hw.pwr_state = 'on'
# k.switch_mux_on([1, 4, 2, 3])
# print(k._hw.tx._activation_delay, k._hw.tx._release_delay,
# k._hw.tx.latency, k._hw.rx.latency)
# k._hw._vab_pulse(0.1, duration=1, polarity=1, append=True)
# k._hw._vab_pulse(12, duration=1, polarity=1, append=True)
# k._hw._vab_pulse(12, duration=1, polarity=1, append=True)
# k._hw._vab_pulse(5, duration=1, polarity=1, append=True)
# k._hw._vab_pulse(0.1, duration=1, polarity=1, append=True)
# k._hw._vab_pulse(0.1, duration=1, polarity=0, append=True)
# k.switch_mux_off([1, 4, 2, 3])
# k._hw._plot_readings(save_fig=False)
# lower level
def read_values(t0=None):
if t0 is None:
t0 = time.time()
_readings = []
for i in range(1000):
r = [time.time() - t0, k._hw._pulse, k._hw.tx.polarity, k._hw.tx.current, k._hw.rx.voltage]
_readings.append(r)
time.sleep(0.01)
k._hw.readings = np.array(_readings)
from threading import Thread
import numpy as np
import matplotlib.pyplot as plt
# k._hw.pwr_state = 'on' # give current to dph
# k.switch_mux_on([1, 4, 2, 3])
# readings = Thread(target=read_values)
# readings.start()
# time.sleep(1)
# k._hw.tx.voltage = 5 # instruction but still not on
# # TODO k._hw.tx.pwr_state = 'on' doesn't to anything... just a flag?
# k._hw.tx.pwr.pwr_state = 'on' # dph sending current
# time.sleep(1)
# k._hw.tx.polarity = 1
# time.sleep(1)
# k._hw.tx.polarity = 0
# time.sleep(1)
# k._hw.tx.polarity = -1
# time.sleep(1)
# k._hw.tx.pwr.pwr_state = 'off' # dph stops
# time.sleep(1)
# readings.join()
# k.switch_mux_off([1, 4, 2, 3])
# k._hw._plot_readings(save_fig=False)
# rise and fall on polarity 1 for different Vab
if False:
vabs = [1, 5, 10, 15, 20, 25, 30]
traces = {}
k._hw.pwr_state = 'on'
k.switch_mux_on([1, 4, 2, 3])
for vab in vabs:
print('trying vab', vab)
readings = Thread(target=read_values)
k._hw.tx.voltage = vab
k._hw.tx.polarity = 1
readings.start()
k._hw.tx.pwr.pwr_state = 'on'
time.sleep(1)
#k._hw.tx.pwr.pwr_state = 'off'
k._hw.tx.voltage = 0.01
time.sleep(4)
readings.join()
traces['Vab=' + str(vab)] = {'data': k._hw.readings}
k._hw.pwr_state = 'off'
k.switch_mux_off([1, 4, 2, 3])
# consecutive injection from high to low vab
if False:
traces = {}
k._hw.pwr_state = 'on'
k.switch_mux_on([1, 4, 2, 3])
#for wait_time in [0, 0.2, 0.5, 2]:
for wait_time in [0.5]:
print('wait_time', wait_time)
tdic = {}
t0 = time.time()
readings = Thread(target=read_values, args=(t0,))
k._hw.tx.voltage = 30
k._hw.tx.polarity = 1
readings.start()
tdic['start'] = time.time() - t0
k._hw.tx.pwr.pwr_state = 'on'
tdic['turn on 1'] = time.time() - t0
time.sleep(1)
k._hw.tx.polarity = 0
tdic['polarity 0'] = time.time() - t0
#k._hw.tx.voltage = 5 # in this case, decay start from slightly lower
#tdic['vab set 2'] = time.time() - t0
k._hw.tx.pwr.pwr_state = 'off' # start discharging
tdic['turn off 1'] = time.time() - t0
for i in range(10):
k._hw.tx.pwr._retrieve_voltage()
print(k._hw.tx.pwr._voltage)
time.sleep(0.1)
time.sleep(wait_time)
k._hw.tx.pwr.pwr_state = 'on'
tdic['turn on 2'] = time.time() - t0
k._hw.tx.voltage = 5
tdic['vab set'] = time.time() - t0
k._hw.tx.polarity = 1
tdic['polarity 1 (2)'] = time.time() - t0
time.sleep(3) # if time here too short we don't reach target voltage
#k._hw.tx.polarity = 0
k._hw.tx.pwr.pwr_state = 'off'
tdic['turn off 2'] = time.time() - t0
readings.join()
tdic['end time 2'] = time.time() - t0
print(tdic)
traces['wait ' + str(wait_time) + 's'] = {
'data': k._hw.readings,
'tdic': tdic
}
k._hw.pwr_state = 'off'
k.switch_mux_off([1, 4, 2, 3])
if False:
# figure
fig, axs = plt.subplots(2, 1, figsize=(8, 4), sharex=True)
axs[0].set_ylabel('Current [mA]')
axs[1].set_ylabel('Voltage [mV]')
for key in traces:
tdic = traces[key]['tdic']
trace = traces[key]['data']
cax = axs[0].plot(trace[:, 0], trace[:, -2], '.-', label=key)
color = cax[0].get_color()
axs[1].plot(trace[:, 0], trace[:, -1], '.-', color=color)
for a in tdic:
axs[0].axvline(tdic[a], color=color)
axs[1].axvline(tdic[a], color=color)
axs[0].legend()
axs[1].set_xlabel('Time [s]')
plt.show()
# k._hw.tx.voltage = 5
# time.sleep(2)
# k._hw.tx.pwr.pwr_state = 'off'
# k._hw.pwr_state = 'off'
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('TkAgg')
import numpy as np
from ohmpi.utils import change_config
change_config('../configs/config_mb_2023.py', verbose=False)
import os
from ohmpi.ohmpi import OhmPi
k = OhmPi()
# single inversion
#k.run_inversion(['measurement_20220206T194552.csv'])
# batch inversion
xzv = k.run_inversion([
'measurements_20231130T215031.csv'
], reg_mode=0)
# make a contour figure with the output
fig, axs = plt.subplots(len(xzv), 1, sharex=True, sharey=True, figsize=(10, 6))
axs = [axs] if len(xzv) == 1 else axs
for i, dic in enumerate(xzv):
ax = axs[i]
x, z = np.meshgrid(dic['x'], dic['z'])
cax = ax.contourf(x, z, np.log10(dic['rho']))
fig.colorbar(cax, ax=ax, label=r'$\log_{10}(\rho)$ [$\Omega$.m]')
plt.show(block=True)
# restore default config
change_config('../configs/config_default.py', verbose=False)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2023.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
stand_alone = False
part_of_hardware_system = False
within_ohmpi = True
# 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.test_mux(mux_id='mux_03')
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12.,1., cycles=2)
# k._hw.switch_mux([A, B, M, N], state='off')
# k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
# k._hw.rx._bias = -1.38
# print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
A, B, M, N = (0, 0, 0, 0)
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
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)
import time
from ohmpi.utils import change_config
from ohmpi.plots import plot_exec_log
import logging
change_config('../configs/config_mb_2023_3_mux_2024.py', verbose=False)
from ohmpi.hardware_components.mux_2024_0_X import Mux
from ohmpi.hardware_components import raspberry_pi as ctl_module
from ohmpi.config import HARDWARE_CONFIG
MUX_CONFIG = HARDWARE_CONFIG['mux']
stand_alone_mux = False
part_of_hardware_system = False
within_ohmpi = True
# Stand alone mux
if stand_alone_mux:
mux_id = 'mux_04'
first = 24
print(MUX_CONFIG)
MUX_CONFIG.update(HARDWARE_CONFIG['mux']['boards'][mux_id])
MUX_CONFIG.update({'id': mux_id})
MUX_CONFIG['ctl'] = ctl_module.Ctl()
mux = Mux(**MUX_CONFIG)
mux.switch_one(elec=1+first, role='M', state='on')
time.sleep(1)
mux.switch_one(elec=1+first, role='M', state='off')
mux.switch({'A': [1], 'B': [2], 'M': [3], 'N': [4]}, state='on')
time.sleep(2)
# mux.switch({'A': [1], 'B': [4], 'M': [2], 'N': [3]}, state='off')
mux.reset()
mux.test({'A': [i+first for i in range(1, 9)], 'B': [i+first for i in range(1, 9)],
'M': [i+first for i in range(1, 9)], 'N': [i+first for i in range(1, 9)]}, activation_time=.1)
# mux as part of a OhmPiHardware system
if part_of_hardware_system:
from ohmpi.hardware_system import OhmPiHardware
print('Starting test of mux as part of a OhmPiHardware system.')
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')
if within_ohmpi:
from ohmpi.ohmpi import OhmPi
print('Starting test of mux within OhmPi.')
k = OhmPi()
#A, B, M, N = (32, 29, 31, 30)
k.reset_mux()
#k._hw.switch_mux([A, B, M, N], state='on')
#k._hw.vab_square_wave(12.,1., cycles=2)
#k._hw.switch_mux([A, B, M, N], state='off')
#k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
#k._hw.rx._bias = -1.38
#print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
A, B, M, N = (1, 4, 2, 3)
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
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_rho :.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)
import matplotlib
matplotlib.use('TkAgg')
import matplotlib.pyplot as plt
from ohmpi.utils import change_config
change_config('../configs/config_mb_2023__4_mux_2023.py', verbose=False)
import importlib
import os
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
import pandas as pd
stand_alone = False
part_of_hardware_system = False
within_ohmpi = True
# 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'])
role = 'A'
mux_id = f'mux_{role}'
mux_boards = []
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)
mux.reset()
mux.test({role: [i for i in range(1, 65)]}, activation_time=.5)
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()
# k.get_data()
k.load_sequence(os.path.join(os.path.dirname(__file__), '../sequences/wenner1-16.txt'))
k.reset_mux()
# k.run_multiple_sequences(sequence_delay=20, nb_meas=3)
k.run_sequence(injection_duration=0.2)
# k.rs_check(vab=4)
# k.test_mux(mux_id=None, activation_time=0.2)
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12.,1., cycles=2)
# k._hw.switch_mux([A, B, M, N], state='off')
# k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
# k._hw.rx._bias = -1.38
# print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
# A, B, M, N = (17, 20, 18, 19)
# A, B, M, N = (1, 4, 2, 3)
# d = k.run_measurement([A, B, M, N], injection_duration=0.5, 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)
# k.run_sequence(nb_stack=2, injection_duration=0.2, duty_cycle=0.5)
# ddir = os.path.join(os.path.dirname(__file__), '../data/')
# print(sorted(os.listdir(ddir)))
# df = pd.read_csv(os.path.join(ddir, sorted(os.listdir(ddir))[-3]))
# fig, ax = plt.subplots()
# df['R [Ohm]'].plot(ax=ax)
# fig.show(block=True)
# plot_exec_log('ohmpi/logs/exec.log')
change_config('../configs/config_default.py', verbose=False)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
stand_alone = False
part_of_hardware_system = False
within_ohmpi = True
# 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}) # HARDWARE_CONFIG['tx'].pop('ctl', ctl_module.Ctl())})
HARDWARE_CONFIG['rx'].update({'ctl': ctl}) # HARDWARE_CONFIG['rx'].pop('ctl', ctl_module.Ctl())})
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')])})
rx = rx_module.Rx(**HARDWARE_CONFIG['rx'])
tx = tx_module.Tx(**HARDWARE_CONFIG['tx'])
pwr = pwr_module.Pwr(**HARDWARE_CONFIG['pwr'])
tx.polarity = 1
time.sleep(1)
tx.polarity = 0
# 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.')
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')
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._hw.switch_mux([A, B, M, N], state='on')
#k._hw.vab_square_wave(12.,1., cycles=2)
#k._hw.switch_mux([A, B, M, N], state='off')
#k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
#k._hw.rx._bias = -1.38
#print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
A, B, M, N = (0, 0, 0, 0)
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
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)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2__1_mux_2024_4roles_dph5005.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:
mux_id = 'mux_03'
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")}')
mux_module = importlib.import_module(f'ohmpi.hardware_components.{HARDWARE_CONFIG["mux"]["boards"][mux_id].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)])})
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})
rx = rx_module.Rx(**HARDWARE_CONFIG['rx'])
tx = tx_module.Tx(**HARDWARE_CONFIG['tx'])
pwr = pwr_module.Pwr(**HARDWARE_CONFIG['pwr'])
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 for i in range(1, 9)], 'B': [i for i in range(1, 9)],
# 'M': [i for i in range(1, 9)], 'N': [i 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.test_mux(mux_id='mux_03')
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12.,1., cycles=2)
# k._hw.switch_mux([A, B, M, N], state='off')
# k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
# k._hw.rx._bias = -1.38
# print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
A, B, M, N = (1, 4, 2, 3)
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
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)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
#change_config('../configs/config_mb_2024_0_2__2_mux_2024_dph5005.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
import numpy as np
stand_alone = False
part_of_hardware_system = False
within_ohmpi = True
# 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.test_mux(mux_id='mux_03')
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12.,1., cycles=2)
# k._hw.switch_mux([A, B, M, N], state='off')
# k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
# k._hw.rx._bias = -1.38
# print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
abmn = np.array([1, 4, 2, 3]) + 0
k.switch_mux_on(abmn)
# k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
print('using OhmPi')
d = k.run_measurement(abmn, injection_duration=0.5, nb_stack=2, duty_cycle=0.5)
k.switch_mux_off(abmn)
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)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2__3_mux_2024_4roles_dph5005.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(vab=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)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2__4_mux_2023.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
stand_alone = False
part_of_hardware_system = False
within_ohmpi = True
# 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'])
role = 'A'
mux_id = f'mux_{role}'
mux_boards = []
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)
mux.reset()
mux.test({role: [i for i in range(1, 65)]}, 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.test_mux(mux_id='mux_03')
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12.,1., cycles=2)
# k._hw.switch_mux([A, B, M, N], state='off')
# k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
# k._hw.rx._bias = -1.38
# print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
A, B, M, N = (1, 4, 2, 3)
k._hw.switch_mux([A, B, M, N], state='on')
k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
k._hw._plot_readings()
# k.load_sequence('sequences/test_circuit_1423.txt')
#k.load_sequence('sequences/wenner1-16.txt')
# k.run_sequence(tx_volt=5, 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)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2__4_mux_2023_dph5005.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
stand_alone = False
part_of_hardware_system = False
within_ohmpi = True
# 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'])
role = 'A'
mux_id = f'mux_{role}'
mux_boards = []
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)
mux.reset()
mux.test({role: [i for i in range(1, 65)]}, 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()
time.sleep(2)
k.reset_mux()
# k.create_sequence(nelec=6, params=[('wenner', 1)], ireciprocal=True)
# k.rs_check()
# print(k.sequence)
# k.run_sequence(cycles=1, vab=5, strategy='constant', injection_duration=1)
# k.plot_last_fw()
# k.export(ftype='bert')
# k.export(fnames=['data/measurements_20240226T180635.csv'], ftype='bert')
# k.download_data()
# k.test_mux(mux_id='mux_03')
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12.,1., cycles=2)
# k._hw.switch_mux([A, B, M, N], state='off')
# k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
# k._hw.rx._bias = -1.38
# print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
# A, B, M, N = (1, 4, 2, 3)
# A, B, M, N = (5, 8, 6, 7)
# A, B, M, N = (32, 29, 31, 30)
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(5., cycle_duration=0.500, cycles=2, duty_cycle=0.5)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# if k._hw.sp is None:
# k._hw.sp = 0 # if not off-time, sp is None
# k._hw._plot_readings()
k.load_sequence('sequences/wenner1-16.txt')
k.run_multiple_sequences(nb_meas=2)
# k.run_sequence(vab=5, injection_duration=.2, nb_stack=1, duty_cycle=1)
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)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2__4_mux_2024_4roles_dph5005.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
stand_alone = False
part_of_hardware_system = False
within_ohmpi = True
# 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()
for pol in [1,0,-1,0]:
tx.polarity = pol
print(pol)
time.sleep(5)
# 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.test_mux(mux_id='mux_03')
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12.,1., cycles=2)
# k._hw.switch_mux([A, B, M, N], state='off')
# k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
# k._hw.rx._bias = -1.38
# print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
A, B, M, N = (1, 4, 2, 3)
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
# k.load_sequence('sequences/9991_GRAD_16_s1_a1.txt')
# k.run_sequence(vab=5, 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, vab=42., strategy='vmax', vab_max=40., delay=0.01)
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)
import matplotlib
matplotlib.use('TkAgg')
from ohmpi.utils import change_config
change_config('../configs/config_mb_2024_0_2__4_mux_2023__2_mux_2024_2roles_dph5005.py', verbose=False)
import importlib
import time
import logging
from ohmpi.config import HARDWARE_CONFIG
stand_alone = True
part_of_hardware_system = False
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'])
role = 'A'
id = '02'
m = 0
mux_id = f'mux_{role}'
#mux_id = f'mux_{id}'
mux_boards = []
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)
mux.reset()
mux.test({role: [i for i in range(1, 65)]}, activation_time=.1)
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 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.test_mux() #mux_id='mux_03')
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12.,1., cycles=2)
# k._hw.switch_mux([A, B, M, N], state='off')
# k._hw.calibrate_rx_bias() # electrodes 1 4 2 3 should be connected to a reference circuit
# k._hw.rx._bias = -1.38
# print(f'Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
A, B, M, N = (1, 4, 2, 3)
# k._hw.switch_mux([A, B, M, N], state='on')
# k._hw.vab_square_wave(12., cycle_duration=10., cycles=3)
# k._hw.switch_mux([A, B, M, N], state='off')
# print(f'OhmPiHardware Resistance: {k._hw.last_rho :.2f} ohm, dev. {k._hw.last_dev:.2f} %, rx bias: {k._hw.rx._bias:.2f} mV')
# k._hw._plot_readings()
# k.load_sequence('sequences/9991_GRAD_16_s1_a1.txt')
# k.run_sequence(vab=5, 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)
import time
from ohmpi.utils import change_config
import logging
change_config('../configs/config_mb_2023_mux_2024_2_roles_AB.py', verbose=False)
from ohmpi.hardware_components.mux_2024_0_X import Mux, MUX_CONFIG
from ohmpi.hardware_components import raspberry_pi as ctl_module
# from ohmpi.config import HARDWARE_CONFIG
stand_alone_mux = True # Testing hardware component alone
part_of_hardware_system = False # Testing hardware component as a part of the hardware system
within_ohmpi = False # Testing hardware component as a part of the hardware system through the ohmpi object
update_mux_config = False
# Stand alone mux
if stand_alone_mux:
MUX_CONFIG['ctl'] = ctl_module.Ctl()
MUX_CONFIG['id'] = 'mux_1'
MUX_CONFIG['cabling'] = {(i, j): ('mux_1', i) for j in ['A', 'B'] for i in range(1,17)}
MUX_CONFIG['roles'] = {'A': 'X', 'B': 'Y'}
MUX_CONFIG['mcp_0'] = '0x26'
MUX_CONFIG['mcp_1'] = '0x27'
mux = Mux(**MUX_CONFIG)
mux.switch_one(elec=2, role='A', state='on')
time.sleep(2)
mux.switch_one(elec=2, role='A', state='off')
mux.switch({'A': [2], 'B': [3]}, state='on')
time.sleep(8)
# mux.switch({'A': [1], 'B': [4], 'M': [2], 'N': [3]}, state='off')
mux.reset()
mux.test({'A': [i for i in range(1, 17)], 'B': [i for i in range(1, 17)]}, 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 mux as part of a OhmPiHardware system.')
h = OhmPiHardware()
h.exec_logger.setLevel(logging.DEBUG)
# Test mux switching
h.reset_mux()
h.switch_mux(electrodes=[1, 4], roles=['A', 'B'], state='on')
time.sleep(1.)
h.switch_mux(electrodes=[1, 4], roles=['A', 'B'], state='off')
if within_ohmpi:
from ohmpi.ohmpi import OhmPi
print('Starting test of mux within OhmPi.')
k = OhmPi()
k.switch_mux_on([1, 4, 2, 3])
time.sleep(1.)
k.reset_mux()
change_config('../configs/config_default.py', verbose=False)