bulk_sensor: Rework APIDumpHelper() to BatchBulkHelper()

The APIDumpHelper class is mainly intended to help process messages in
batches.  Rework the class methods to make that more clear.

Signed-off-by: Kevin O'Connor <kevin@koconnor.net>
This commit is contained in:
Kevin O'Connor 2023-12-16 23:26:42 -05:00
parent 95c753292d
commit 3370134593
6 changed files with 112 additions and 124 deletions

View File

@ -187,7 +187,7 @@ MIN_MSG_TIME = 0.100
BYTES_PER_SAMPLE = 5
SAMPLES_PER_BLOCK = 10
API_UPDATES = 0.100
BATCH_UPDATES = 0.100
# Printer class that controls ADXL345 chip
class ADXL345:
@ -211,17 +211,18 @@ class ADXL345:
mcu.register_config_callback(self._build_config)
self.bulk_queue = bulk_sensor.BulkDataQueue(mcu, "adxl345_data", oid)
# Clock tracking
chip_smooth = self.data_rate * API_UPDATES * 2
chip_smooth = self.data_rate * BATCH_UPDATES * 2
self.clock_sync = bulk_sensor.ClockSyncRegression(mcu, chip_smooth)
self.clock_updater = bulk_sensor.ChipClockUpdater(self.clock_sync,
BYTES_PER_SAMPLE)
self.last_error_count = 0
# API server endpoints
self.api_dump = bulk_sensor.APIDumpHelper(
self.printer, self._api_update, self._api_startstop, API_UPDATES)
# Process messages in batches
self.batch_bulk = bulk_sensor.BatchBulkHelper(
self.printer, self._process_batch,
self._start_measurements, self._finish_measurements, BATCH_UPDATES)
self.name = config.get_name().split()[-1]
hdr = ('time', 'x_acceleration', 'y_acceleration', 'z_acceleration')
self.api_dump.add_mux_endpoint("adxl345/dump_adxl345", "sensor",
self.batch_bulk.add_mux_endpoint("adxl345/dump_adxl345", "sensor",
self.name, {'header': hdr})
def _build_config(self):
cmdqueue = self.spi.get_command_queue()
@ -248,7 +249,10 @@ class ADXL345:
"This is generally indicative of connection problems "
"(e.g. faulty wiring) or a faulty adxl345 chip." % (
reg, val, stored_val))
# Measurement collection
def start_internal_client(self):
cconn = self.batch_bulk.add_internal_client()
return AccelQueryHelper(self.printer, cconn)
# Measurement decoding
def _extract_samples(self, raw_samples):
# Load variables to optimize inner loop below
(x_pos, x_scale), (y_pos, y_scale), (z_pos, z_scale) = self.axes_map
@ -294,6 +298,7 @@ class ADXL345:
else:
raise self.printer.command_error("Unable to query adxl345 fifo")
self.clock_updater.update_clock(params)
# Start, stop, and process message batches
def _start_measurements(self):
# In case of miswiring, testing ADXL345 device ID prevents treating
# noise or wrong signal as a correctly initialized device
@ -329,8 +334,7 @@ class ADXL345:
params = self.query_adxl345_end_cmd.send([self.oid, 0, 0])
self.bulk_queue.clear_samples()
logging.info("ADXL345 finished '%s' measurements", self.name)
# API interface
def _api_update(self, eventtime):
def _process_batch(self, eventtime):
self._update_clock()
raw_samples = self.bulk_queue.pull_samples()
if not raw_samples:
@ -340,14 +344,6 @@ class ADXL345:
return {}
return {'data': samples, 'errors': self.last_error_count,
'overflows': self.clock_updater.get_last_limit_count()}
def _api_startstop(self, is_start):
if is_start:
self._start_measurements()
else:
self._finish_measurements()
def start_internal_client(self):
cconn = self.api_dump.add_internal_client()
return AccelQueryHelper(self.printer, cconn)
def load_config(config):
return ADXL345(config)

View File

@ -410,6 +410,7 @@ BYTES_PER_SAMPLE = 3
SAMPLES_PER_BLOCK = 16
SAMPLE_PERIOD = 0.000400
BATCH_UPDATES = 0.100
class Angle:
def __init__(self, config):
@ -440,13 +441,14 @@ class Angle:
% (oid,), on_restart=True)
mcu.register_config_callback(self._build_config)
self.bulk_queue = bulk_sensor.BulkDataQueue(mcu, "spi_angle_data", oid)
# API server endpoints
self.api_dump = bulk_sensor.APIDumpHelper(
self.printer, self._api_update, self._api_startstop, 0.100)
# Process messages in batches
self.batch_bulk = bulk_sensor.BatchBulkHelper(
self.printer, self._process_batch,
self._start_measurements, self._finish_measurements, BATCH_UPDATES)
self.name = config.get_name().split()[1]
api_resp = {'header': ('time', 'angle')}
self.api_dump.add_mux_endpoint("angle/dump_angle", "sensor", self.name,
api_resp)
self.batch_bulk.add_mux_endpoint("angle/dump_angle",
"sensor", self.name, api_resp)
def _build_config(self):
freq = self.mcu.seconds_to_clock(1.)
while float(TCODE_ERROR << self.time_shift) / freq < 0.002:
@ -460,7 +462,9 @@ class Angle:
"spi_angle_end oid=%c sequence=%hu", oid=self.oid, cq=cmdqueue)
def get_status(self, eventtime=None):
return {'temperature': self.sensor_helper.last_temperature}
# Measurement collection
def start_internal_client(self):
return self.batch_bulk.add_internal_client()
# Measurement decoding
def _extract_samples(self, raw_samples):
# Load variables to optimize inner loop below
sample_ticks = self.sample_ticks
@ -516,19 +520,9 @@ class Angle:
self.last_angle = last_angle
del samples[count:]
return samples, error_count
# API interface
def _api_update(self, eventtime):
if self.sensor_helper.is_tcode_absolute:
self.sensor_helper.update_clock()
raw_samples = self.bulk_queue.pull_samples()
if not raw_samples:
return {}
samples, error_count = self._extract_samples(raw_samples)
if not samples:
return {}
offset = self.calibration.apply_calibration(samples)
return {'data': samples, 'errors': error_count,
'position_offset': offset}
# Start, stop, and process message batches
def _is_measuring(self):
return self.start_clock != 0
def _start_measurements(self):
logging.info("Starting angle '%s' measurements", self.name)
self.sensor_helper.start()
@ -548,13 +542,18 @@ class Angle:
self.bulk_queue.clear_samples()
self.sensor_helper.last_temperature = None
logging.info("Stopped angle '%s' measurements", self.name)
def _api_startstop(self, is_start):
if is_start:
self._start_measurements()
else:
self._finish_measurements()
def start_internal_client(self):
return self.api_dump.add_internal_client()
def _process_batch(self, eventtime):
if self.sensor_helper.is_tcode_absolute:
self.sensor_helper.update_clock()
raw_samples = self.bulk_queue.pull_samples()
if not raw_samples:
return {}
samples, error_count = self._extract_samples(raw_samples)
if not samples:
return {}
offset = self.calibration.apply_calibration(samples)
return {'data': samples, 'errors': error_count,
'position_offset': offset}
def load_config_prefix(config):
return Angle(config)

View File

@ -5,20 +5,23 @@
# This file may be distributed under the terms of the GNU GPLv3 license.
import threading
API_UPDATE_INTERVAL = 0.500
BATCH_INTERVAL = 0.500
# Helper to periodically transmit data to a set of API clients
class APIDumpHelper:
def __init__(self, printer, data_cb, startstop_cb=None,
update_interval=API_UPDATE_INTERVAL):
# Helper to process accumulated messages in periodic batches
class BatchBulkHelper:
def __init__(self, printer, batch_cb, start_cb=None, stop_cb=None,
batch_interval=BATCH_INTERVAL):
self.printer = printer
self.data_cb = data_cb
if startstop_cb is None:
startstop_cb = (lambda is_start: None)
self.startstop_cb = startstop_cb
self.batch_cb = batch_cb
if start_cb is None:
start_cb = (lambda: None)
self.start_cb = start_cb
if stop_cb is None:
stop_cb = (lambda: None)
self.stop_cb = stop_cb
self.is_started = False
self.update_interval = update_interval
self.update_timer = None
self.batch_interval = batch_interval
self.batch_timer = None
self.clients = {}
self.webhooks_start_resp = {}
# Periodic batch processing
@ -27,40 +30,40 @@ class APIDumpHelper:
return
self.is_started = True
try:
self.startstop_cb(True)
self.start_cb()
except self.printer.command_error as e:
logging.exception("API Dump Helper start callback error")
logging.exception("BatchBulkHelper start callback error")
self.is_started = False
self.clients.clear()
raise
reactor = self.printer.get_reactor()
systime = reactor.monotonic()
waketime = systime + self.update_interval
self.update_timer = reactor.register_timer(self._update, waketime)
waketime = systime + self.batch_interval
self.batch_timer = reactor.register_timer(self._proc_batch, waketime)
def _stop(self):
self.clients.clear()
self.printer.get_reactor().unregister_timer(self.update_timer)
self.update_timer = None
self.printer.get_reactor().unregister_timer(self.batch_timer)
self.batch_timer = None
if not self.is_started:
return
try:
self.startstop_cb(False)
self.stop_cb()
except self.printer.command_error as e:
logging.exception("API Dump Helper stop callback error")
logging.exception("BatchBulkHelper stop callback error")
self.clients.clear()
self.is_started = False
if self.clients:
# New client started while in process of stopping
self._start()
def _update(self, eventtime):
def _proc_batch(self, eventtime):
try:
msg = self.data_cb(eventtime)
msg = self.batch_cb(eventtime)
except self.printer.command_error as e:
logging.exception("API Dump Helper data callback error")
logging.exception("BatchBulkHelper batch callback error")
self._stop()
return self.printer.get_reactor().NEVER
if not msg:
return eventtime + self.update_interval
return eventtime + self.batch_interval
for cconn, template in list(self.clients.items()):
if cconn.is_closed():
del self.clients[cconn]
@ -71,7 +74,7 @@ class APIDumpHelper:
tmp = dict(template)
tmp['params'] = msg
cconn.send(tmp)
return eventtime + self.update_interval
return eventtime + self.batch_interval
# Internal clients
def add_internal_client(self):
cconn = InternalDumpClient()
@ -90,7 +93,7 @@ class APIDumpHelper:
wh = self.printer.lookup_object('webhooks')
wh.register_mux_endpoint(path, key, value, self._add_api_client)
# An "internal webhooks" wrapper for using APIDumpHelper internally
# An "internal webhooks" wrapper for using BatchBulkHelper internally
class InternalDumpClient:
def __init__(self):
self.msgs = []

View File

@ -35,7 +35,7 @@ MIN_MSG_TIME = 0.100
BYTES_PER_SAMPLE = 6
SAMPLES_PER_BLOCK = 8
API_UPDATES = 0.100
BATCH_UPDATES = 0.100
# Printer class that controls LIS2DW chip
class LIS2DW:
@ -57,17 +57,18 @@ class LIS2DW:
mcu.register_config_callback(self._build_config)
self.bulk_queue = bulk_sensor.BulkDataQueue(mcu, "lis2dw_data", oid)
# Clock tracking
chip_smooth = self.data_rate * API_UPDATES * 2
chip_smooth = self.data_rate * BATCH_UPDATES * 2
self.clock_sync = bulk_sensor.ClockSyncRegression(mcu, chip_smooth)
self.clock_updater = bulk_sensor.ChipClockUpdater(self.clock_sync,
BYTES_PER_SAMPLE)
self.last_error_count = 0
# API server endpoints
self.api_dump = bulk_sensor.APIDumpHelper(
self.printer, self._api_update, self._api_startstop, API_UPDATES)
# Process messages in batches
self.batch_bulk = bulk_sensor.BatchBulkHelper(
self.printer, self._process_batch,
self._start_measurements, self._finish_measurements, BATCH_UPDATES)
self.name = config.get_name().split()[-1]
hdr = ('time', 'x_acceleration', 'y_acceleration', 'z_acceleration')
self.api_dump.add_mux_endpoint("lis2dw/dump_lis2dw", "sensor",
self.batch_bulk.add_mux_endpoint("lis2dw/dump_lis2dw", "sensor",
self.name, {'header': hdr})
def _build_config(self):
@ -95,7 +96,10 @@ class LIS2DW:
"This is generally indicative of connection problems "
"(e.g. faulty wiring) or a faulty lis2dw chip." % (
reg, val, stored_val))
# Measurement collection
def start_internal_client(self):
cconn = self.bulk_batch.add_internal_client()
return adxl345.AccelQueryHelper(self.printer, cconn)
# Measurement decoding
def _extract_samples(self, raw_samples):
# Load variables to optimize inner loop below
(x_pos, x_scale), (y_pos, y_scale), (z_pos, z_scale) = self.axes_map
@ -136,6 +140,7 @@ class LIS2DW:
params = self.query_lis2dw_status_cmd.send([self.oid],
minclock=minclock)
self.clock_updater.update_clock(params)
# Start, stop, and process message batches
def _start_measurements(self):
# In case of miswiring, testing LIS2DW device ID prevents treating
# noise or wrong signal as a correctly initialized device
@ -177,8 +182,7 @@ class LIS2DW:
self.bulk_queue.clear_samples()
logging.info("LIS2DW finished '%s' measurements", self.name)
self.set_reg(REG_LIS2DW_FIFO_CTRL, 0x00)
# API interface
def _api_update(self, eventtime):
def _process_batch(self, eventtime):
self._update_clock()
raw_samples = self.bulk_queue.pull_samples()
if not raw_samples:
@ -188,15 +192,6 @@ class LIS2DW:
return {}
return {'data': samples, 'errors': self.last_error_count,
'overflows': self.clock_updater.get_last_limit_count()}
def _api_startstop(self, is_start):
if is_start:
self._start_measurements()
else:
self._finish_measurements()
def start_internal_client(self):
cconn = self.api_dump.add_internal_client()
return adxl345.AccelQueryHelper(self.printer, cconn)
def load_config(config):
return LIS2DW(config)

View File

@ -12,10 +12,11 @@ class DumpStepper:
def __init__(self, printer, mcu_stepper):
self.printer = printer
self.mcu_stepper = mcu_stepper
self.last_api_clock = 0
self.api_dump = bulk_sensor.APIDumpHelper(printer, self._api_update)
self.last_batch_clock = 0
self.batch_bulk = bulk_sensor.BatchBulkHelper(printer,
self._process_batch)
api_resp = {'header': ('interval', 'count', 'add')}
self.api_dump.add_mux_endpoint("motion_report/dump_stepper", "name",
self.batch_bulk.add_mux_endpoint("motion_report/dump_stepper", "name",
mcu_stepper.get_name(), api_resp)
def get_step_queue(self, start_clock, end_clock):
mcu_stepper = self.mcu_stepper
@ -42,15 +43,15 @@ class DumpStepper:
% (i, s.first_clock, s.start_position, s.interval,
s.step_count, s.add))
logging.info('\n'.join(out))
def _api_update(self, eventtime):
data, cdata = self.get_step_queue(self.last_api_clock, 1<<63)
def _process_batch(self, eventtime):
data, cdata = self.get_step_queue(self.last_batch_clock, 1<<63)
if not data:
return {}
clock_to_print_time = self.mcu_stepper.get_mcu().clock_to_print_time
first = data[0]
first_clock = first.first_clock
first_time = clock_to_print_time(first_clock)
self.last_api_clock = last_clock = data[-1].last_clock
self.last_batch_clock = last_clock = data[-1].last_clock
last_time = clock_to_print_time(last_clock)
mcu_pos = first.start_position
start_position = self.mcu_stepper.mcu_to_commanded_position(mcu_pos)
@ -71,12 +72,13 @@ class DumpTrapQ:
self.printer = printer
self.name = name
self.trapq = trapq
self.last_api_msg = (0., 0.)
self.api_dump = bulk_sensor.APIDumpHelper(printer, self._api_update)
self.last_batch_msg = (0., 0.)
self.batch_bulk = bulk_sensor.BatchBulkHelper(printer,
self._process_batch)
api_resp = {'header': ('time', 'duration', 'start_velocity',
'acceleration', 'start_position', 'direction')}
self.api_dump.add_mux_endpoint("motion_report/dump_trapq", "name", name,
api_resp)
self.batch_bulk.add_mux_endpoint("motion_report/dump_trapq",
"name", name, api_resp)
def extract_trapq(self, start_time, end_time):
ffi_main, ffi_lib = chelper.get_ffi()
res = []
@ -115,17 +117,17 @@ class DumpTrapQ:
move.start_z + move.z_r * dist)
velocity = move.start_v + move.accel * move_time
return pos, velocity
def _api_update(self, eventtime):
qtime = self.last_api_msg[0] + min(self.last_api_msg[1], 0.100)
def _process_batch(self, eventtime):
qtime = self.last_batch_msg[0] + min(self.last_batch_msg[1], 0.100)
data, cdata = self.extract_trapq(qtime, NEVER_TIME)
d = [(m.print_time, m.move_t, m.start_v, m.accel,
(m.start_x, m.start_y, m.start_z), (m.x_r, m.y_r, m.z_r))
for m in data]
if d and d[0] == self.last_api_msg:
if d and d[0] == self.last_batch_msg:
d.pop(0)
if not d:
return {}
self.last_api_msg = d[-1]
self.last_batch_msg = d[-1]
return {"data": d}
STATUS_REFRESH_TIME = 0.250

View File

@ -52,7 +52,7 @@ MIN_MSG_TIME = 0.100
BYTES_PER_SAMPLE = 6
SAMPLES_PER_BLOCK = 8
API_UPDATES = 0.100
BATCH_UPDATES = 0.100
# Printer class that controls MPU9250 chip
class MPU9250:
@ -74,17 +74,18 @@ class MPU9250:
mcu.register_config_callback(self._build_config)
self.bulk_queue = bulk_sensor.BulkDataQueue(mcu, "mpu9250_data", oid)
# Clock tracking
chip_smooth = self.data_rate * API_UPDATES * 2
chip_smooth = self.data_rate * BATCH_UPDATES * 2
self.clock_sync = bulk_sensor.ClockSyncRegression(mcu, chip_smooth)
self.clock_updater = bulk_sensor.ChipClockUpdater(self.clock_sync,
BYTES_PER_SAMPLE)
self.last_error_count = 0
# API server endpoints
self.api_dump = bulk_sensor.APIDumpHelper(
self.printer, self._api_update, self._api_startstop, API_UPDATES)
# Process messages in batches
self.batch_bulk = bulk_sensor.BatchBulkHelper(
self.printer, self._process_batch,
self._start_measurements, self._finish_measurements, BATCH_UPDATES)
self.name = config.get_name().split()[-1]
hdr = ('time', 'x_acceleration', 'y_acceleration', 'z_acceleration')
self.api_dump.add_mux_endpoint("mpu9250/dump_mpu9250", "sensor",
self.batch_bulk.add_mux_endpoint("mpu9250/dump_mpu9250", "sensor",
self.name, {'header': hdr})
def _build_config(self):
cmdqueue = self.i2c.get_command_queue()
@ -105,11 +106,12 @@ class MPU9250:
def read_reg(self, reg):
params = self.i2c.i2c_read([reg], 1)
return bytearray(params['response'])[0]
def set_reg(self, reg, val, minclock=0):
self.i2c.i2c_write([reg, val & 0xFF], minclock=minclock)
# Measurement collection
def start_internal_client(self):
cconn = self.batch_bulk.add_internal_client()
return adxl345.AccelQueryHelper(self.printer, cconn)
# Measurement decoding
def _extract_samples(self, raw_samples):
# Load variables to optimize inner loop below
(x_pos, x_scale), (y_pos, y_scale), (z_pos, z_scale) = self.axes_map
@ -148,6 +150,7 @@ class MPU9250:
params = self.query_mpu9250_status_cmd.send([self.oid],
minclock=minclock)
self.clock_updater.update_clock(params)
# Start, stop, and process message batches
def _start_measurements(self):
# In case of miswiring, testing MPU9250 device ID prevents treating
# noise or wrong signal as a correctly initialized device
@ -190,9 +193,7 @@ class MPU9250:
logging.info("MPU9250 finished '%s' measurements", self.name)
self.set_reg(REG_PWR_MGMT_1, SET_PWR_MGMT_1_SLEEP)
self.set_reg(REG_PWR_MGMT_2, SET_PWR_MGMT_2_OFF)
# API interface
def _api_update(self, eventtime):
def _process_batch(self, eventtime):
self._update_clock()
raw_samples = self.bulk_queue.pull_samples()
if not raw_samples:
@ -202,14 +203,6 @@ class MPU9250:
return {}
return {'data': samples, 'errors': self.last_error_count,
'overflows': self.clock_updater.get_last_limit_count()}
def _api_startstop(self, is_start):
if is_start:
self._start_measurements()
else:
self._finish_measurements()
def start_internal_client(self):
cconn = self.api_dump.add_internal_client()
return adxl345.AccelQueryHelper(self.printer, cconn)
def load_config(config):
return MPU9250(config)