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>

6 files changed, 112 insertions, 124 deletions

diff --git a/klippy/extras/adxl345.py b/klippy/extras/adxl345.py
index d618598a..1dfb6bc7 100644
--- a/klippy/extras/adxl345.py
+++ b/klippy/extras/adxl345.py
@@ -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,18 +211,19 @@ 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.name, {'header': hdr})
+        self.batch_bulk.add_mux_endpoint("adxl345/dump_adxl345", "sensor",
+                                         self.name, {'header': hdr})
     def _build_config(self):
         cmdqueue = self.spi.get_command_queue()
         self.query_adxl345_cmd = self.mcu.lookup_command(
@@ -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)
diff --git a/klippy/extras/angle.py b/klippy/extras/angle.py
index 229e2946..0fe053df 100644
--- a/klippy/extras/angle.py
+++ b/klippy/extras/angle.py
@@ -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)
diff --git a/klippy/extras/bulk_sensor.py b/klippy/extras/bulk_sensor.py
index 826bf74c..95f6201e 100644
--- a/klippy/extras/bulk_sensor.py
+++ b/klippy/extras/bulk_sensor.py
@@ -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 = []
diff --git a/klippy/extras/lis2dw.py b/klippy/extras/lis2dw.py
index 5590804e..739c3641 100644
--- a/klippy/extras/lis2dw.py
+++ b/klippy/extras/lis2dw.py
@@ -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,18 +57,19 @@ 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.name, {'header': hdr})
+        self.batch_bulk.add_mux_endpoint("lis2dw/dump_lis2dw", "sensor",
+                                         self.name, {'header': hdr})
 
     def _build_config(self):
         cmdqueue = self.spi.get_command_queue()
@@ -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)
diff --git a/klippy/extras/motion_report.py b/klippy/extras/motion_report.py
index b34fdc1f..25c1d5e6 100644
--- a/klippy/extras/motion_report.py
+++ b/klippy/extras/motion_report.py
@@ -12,11 +12,12 @@ 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",
-                                       mcu_stepper.get_name(), api_resp)
+        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
         res = []
@@ -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
diff --git a/klippy/extras/mpu9250.py b/klippy/extras/mpu9250.py
index 883f399e..82438ca0 100644
--- a/klippy/extras/mpu9250.py
+++ b/klippy/extras/mpu9250.py
@@ -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,18 +74,19 @@ 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.name, {'header': hdr})
+        self.batch_bulk.add_mux_endpoint("mpu9250/dump_mpu9250", "sensor",
+                                         self.name, {'header': hdr})
     def _build_config(self):
         cmdqueue = self.i2c.get_command_queue()
         self.mcu.add_config_cmd("config_mpu9250 oid=%d i2c_oid=%d"
@@ -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)