MatthiasQuintern/m-teng
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

restructured using backends

Browse Source
This commit is contained in:
Matthias@Dell 2023-06-24 12:28:21 +02:00
parent 0eed4d402b
commit b526bfb913
5 changed files with 95 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
m_teng/backends/arduino/arduino.py
View File

@ -9,13 +9,15 @@ TENG_SUUID = "00010000-9a74-4b30-9361-4a16ec09930f"
TENG_STATUS_CUUID = "00010001-9a74-4b30-9361-4a16ec09930f"
TENG_COMMAND_CUUID = "00010002-9a74-4b30-9361-4a16ec09930f"
TENG_READING_CUUID = "00010003-9a74-4b30-9361-4a16ec09930f"
TENG_SETTING_INTERVAL_CUUID = "00010004-9a74-4b30-9361-4a16ec09930f"
TENG_COUNT_CUUID = "00010004-9a74-4b30-9361-4a16ec09930f"
TENG_INTERVAL_CUUID = "00010005-9a74-4b30-9361-4a16ec09930f"
TENG_COMMANDS = {
"NOOP": int(0).to_bytes(1),
"MEASURE_BASELINE": int(1).to_bytes(1),
"STOP": int(0).to_bytes(1, signed=False),
"MEASURE_COUNT": int(1).to_bytes(1, signed=False),
"MEASURE": int(2).to_bytes(1, signed=False),
}
TENG_STATUS = ["ERROR", "BUSY", "WAIT_CONNECT", "MEASURING_BASELINE", "READING"]
TENG_STATUS = ["ERROR", "BUSY", "WAIT_CONNECT", "CONNECTED", "MEASURING"]
# TODO save measurements on device buffer, transfer later
@ -26,6 +28,10 @@ class Buffer:
self.data = None
_buffer = Buffer()
# class Runner:
# def __init__(self):
runner = asyncio.Runner()
def teng_status_callback(characteristic, data):
value = int.from_bytes(data, byteorder="big", signed=False)
if 0 <= value and value < len(TENG_STATUS):
@ -35,7 +41,7 @@ def teng_status_callback(characteristic, data):
def disconnect_callback(client):
raise Exception(f"The bluetooth device {client.name} was disconnected")
raise Exception(f"The Bluetooth device {client.name} was disconnected")
async def init_arduino_async(n_tries: int=5) -> b.BleakClient:
@ -44,8 +50,8 @@ async def init_arduino_async(n_tries: int=5) -> b.BleakClient:
try:
target_device = None
while target_device is None and (n_tries == "inf" or n_try < n_tries):
print(f"Searching for bluetooth device '{TARGET_NAME}' ({n_try}/{n_tries})", end="\r")
devices = await b.BleakScanner.discover(return_adv=True)
print(f"Searching for Bluetooth device '{TARGET_NAME}' ({n_try+1}/{n_tries})", end="\r")
devices = await b.BleakScanner.discover(return_adv=True, timeout=1.5)
# print(devices)
for adr, (device, adv_data) in devices.items():
if device.name == TARGET_NAME:
@ -54,10 +60,12 @@ async def init_arduino_async(n_tries: int=5) -> b.BleakClient:
break
n_try += 1
if target_device is None:
raise Exception(f"Could not find bluetooth device 'ArduinoTENG'")
raise Exception(f"Could not find Bluetooth device 'ArduinoTENG'")
# print(f"Found target device: {target_device.name}: {target_device.metadata}, {target_device.details}")
# print(target_device.name, target_device.details)
client = b.BleakClient(target_device, disconnect_callback=disconnect_callback)
await client.connect()
print(f"Connected to Bluetooth device '{TARGET_NAME}' at [{client.address}]")
return client
except asyncio.exceptions.CancelledError:
raise Exception(f"Cancelled")
@ -68,18 +76,42 @@ def init(beep_success=True, n_tries: int=5) -> b.BleakClient:
Connect to the arduino
@returns: BleakClient
"""
client = asyncio.run(init_arduino_async(n_tries=n_tries))
client = runner.run(init_arduino_async(n_tries=n_tries))
if beep_success: beep(client)
return client
def set_interval(client, interval: float):
def exit(client):
try:
runner.run(stop_measurement(client))
runner.run(client.disconnect())
except Exception:
pass
async def set_interval(client, interval: float):
"""
Set the measurement interval
@param interval: interval in seconds
"""
interval = int(interval * 1000) # convert to ms for arduinos delay)
await client.write_gatt_char(TENG_SETTING_INTERVAL_CUUID, interval.to_bytes(2, byteorder=LITTLEENDIAN, signed=False))
await client.write_gatt_char(TENG_INTERVAL_CUUID, interval.to_bytes(2, byteorder="little", signed=False))
async def set_count(client, count: int):
"""
Set the measurement count
@param count: number of measurements to take
"""
await client.write_gatt_char(TENG_COUNT_CUUID, count.to_bytes(2, byteorder="little", signed=False))
async def stop_measurement(client):
await client.write_gatt_char(TENG_COMMAND_CUUID, TENG_COMMANDS["STOP"])
async def start_measure_count(client):
await client.write_gatt_char(TENG_COMMAND_CUUID, TENG_COMMANDS["MEASURE_COUNT"])
async def start_measure(client):
await client.write_gatt_char(TENG_COMMAND_CUUID, TENG_COMMANDS["MEASURE"])
# async def main():
@ -115,9 +147,9 @@ def collect_buffer(instr, buffer_nr=1):
@param buffer_nr: 1 -> current, 2 -> voltage
"""
assert(buffer_nr in (1, 2))
return np.vstack((_buffer.data[:,0], _buffer._data[:,buffer_nr])).T
return np.vstack((_buffer.data[:,0], _buffer.data[:,buffer_nr])).T
def beep(client):
# TODO connect beeper to arduino?
print(beep)
print("beep")

66
m_teng/backends/arduino/measure.py
View File

@ -4,56 +4,72 @@ import numpy as np
import asyncio
import datetime
from .arduino.arduino import beep, set_interval, TENG_READING_CUUID, _buffer
from m_teng.backends.arduino.arduino import beep, set_interval, set_count, TENG_READING_CUUID, _buffer, start_measure, start_measure_count, stop_measurement, runner
# equivalent to internal keithley buffer: write to this value and collect afterwards
def measure_count(client, count=100, interval=0.05, update_func=None, update_interval=0.5, beep_done=True, verbose=True, testing=False):
async def _measure_count_async(client, count=100, interval=0.05, update_func=None, update_interval=0.5, beep_done=True, verbose=True):
global _buffer
_buffer.data = np.zeros((count, 3))
i = 0
t_start = datetime.datetime.now()
async def add_buffer.data(client):
async def add_reading(teng_reading_cr, reading: bytearray):
nonlocal i, count
if i >= count: return
_buffer.data[i][0] = float((datetime.datetime.now() - t_start).microseconds) / 1000
reading = await client.read_gatt_char(TENG_READING_CUUID)
_buffer.data[i][2] = int.from_bytes(reading, byteorder=LITTLEENDIAN, signed=False).
# reading = await client.read_gatt_char(TENG_READING_CUUID)
_buffer.data[i][2] = int.from_bytes(reading, byteorder="little", signed=False)
i += 1
set_interval(client, interval)
await set_interval(client, interval)
await set_count(client, count)
# TODO check if notify works when the same value is written again
client.start_notify(TENG_READING_CUUID, add_buffer.data)
await client.start_notify(TENG_READING_CUUID, add_reading)
await start_measure_count(client)
while i < count:
asyncio.sleep(update_interval)
if update_func is not None: # assume an update has occured
update_func(i, 0, _buffer.data[i, 2])
await asyncio.sleep(update_interval)
if update_func is not None and i > 0: # assume an update has occured
update_func(i-1, 0, _buffer.data[i-1, 2])
await client.stop_notify(TENG_READING_CUUID)
if beep_done: beep(client)
def measure_count(client, count=100, interval=0.05, update_func=None, update_interval=0.5, beep_done=True, verbose=True):
runner.run(_measure_count_async(client, count=count, interval=interval, update_func=update_func, update_interval=update_interval, beep_done=beep_done, verbose=verbose))
def measure(client, interval, update_func=None, max_measurements=None, testing=False):
async def _measure_async(client, interval, update_func=None, max_measurements=None):
global _buffer
_buffer.data = np.zeros((count, 3))
readings = []
timestamps = []
i = 0
t_start = datetime.datetime.now()
async def add_buffer.data(client):
if i >= count: return
_buffer.data[i][0] = datetime.datetime.now() - t_start
vval = await client.read_gatt_char(TENG_READING_CUUID)
vval = int.from_bytes(reading, byteorder=LITTLEENDIAN, signed=False).
_buffer.data[i][2] = vval
async def add_reading(teng_reading_cr, reading):
nonlocal i
timestamps.append(float((datetime.datetime.now() - t_start).microseconds) / 1000)
reading = int.from_bytes(reading, byteorder="little", signed=False)
readings.append(reading)
if update_func:
update_func(i, 0, vval)
try:
update_func(i, 0, reading)
except KeyboardInterrupt:
raise asyncio.exceptions.CancelledError("KeyboardInterrupt in update_func")
i += 1
set_interval(client, interval)
client.start_notify(TENG_READING_CUUID, add_buffer.data)
await set_interval(client, interval)
await client.start_notify(TENG_READING_CUUID, add_reading)
await start_measure(client)
try:
while max_measurements is None or i < max_measurements:
asyncio.sleep(interval / 2) #
await asyncio.sleep(0.1) #
except asyncio.exceptions.CancelledError:
pass
except KeyboardInterrupt:
pass
await client.stop_notify(TENG_READING_CUUID)
await stop_measurement(client)
_buffer.data = np.vstack((timestamps, np.zeros(len(timestamps)), readings)).T
print("Measurement stopped" + " "*50)
def measure(client, interval, update_func=None, max_measurements=None):
runner.run(_measure_async(client, interval=interval, update_func=update_func, max_measurements=max_measurements))

5
m_teng/backends/keithley/keithley.py
View File

@ -31,6 +31,11 @@ def init(beep_success=True):
return keithley
def exit(instr):
instr.close()
def run_lua(instr, script_path, verbose=False):
"""
Run a lua script from the host on the instrument

4
m_teng/backends/keithley/measure.py
View File

@ -3,8 +3,8 @@ import numpy as np
from matplotlib import pyplot as plt
import pyvisa
from .keithley import reset
from ..utility import testing as _testing
from m_teng.backends.keithley import reset
from m_teng.utility import testing as _testing
def measure_count(instr, count=100, interval=0.05, update_func=None, update_interval=0.5, beep_done=True, verbose=True):
"""