Repository: bitsy-ai/octoprint-nanny-plugin (language: python)

Issue:
Add "active learning" plugin mode
**Describe your Request**
I'm still performance tuning the on-device ML system. Right now, it's a bit painful to use with other plugins. 

Add an "online mode" to settings. When this setting is enabled, frames will be sampled and inference performed in a Beam pipeline. 

This feeds into eventual support for an "offline" mode. When "online" mode is disabled, inference is performed on-device.  


Discussion / hints:

Fix patch:
diff --git a/octoprint_nanny/clients/flatbuffers.py b/octoprint_nanny/clients/flatbuffers.py
index dba65c5a..42d7c0a2 100644
--- a/octoprint_nanny/clients/flatbuffers.py
+++ b/octoprint_nanny/clients/flatbuffers.py
@@ -1,169 +1,102 @@
 import io
 import flatbuffers
 from typing import Optional
-from PrintNannyMessage.Telemetry import (
+from PrintNannyEvent.TelemetrySchema import (
     MonitoringFrame,
     Image,
     Box,
     BoundingBoxes,
-    PluginEvent,
-    TelemetryMessage,
-    MessageType,
+    TelemetryEventEnum,
+    TelemetryEvent,
     Metadata,
 )
 import octoprint_nanny.types
 
 
-def build_telemetry_message(builder, message, metadata, message_type):
-    # begin metadata
-    Metadata.MetadataStart(builder)
-    Metadata.MetadataAddUserId(builder, metadata.user_id)
-    Metadata.MetadataAddDeviceCloudiotId(builder, metadata.device_cloudiot_id)
-    Metadata.MetadataAddDeviceId(builder, metadata.device_id)
-    metadata = Metadata.MetadataEnd(builder)
-    # end metadata
-    # begin telemetry message
-    TelemetryMessage.TelemetryMessageStart(builder)
-    TelemetryMessage.TelemetryMessageAddMessageType(builder, message_type)
-    TelemetryMessage.TelemetryMessageAddMessage(builder, message)
-    TelemetryMessage.TelemetryMessageAddMetadata(builder, metadata)
-    message = TelemetryMessage.TelemetryMessageEnd(builder)
-    builder.Finish(message)
-    # end message
-    return builder.Output()
+def build_bounding_boxes_message(
+    builder, monitoring_frame: octoprint_nanny.types.MonitoringFrame
+) -> bytes:
+    if monitoring_frame.bounding_boxes is None:
+        return
 
+    boxes = monitoring_frame.bounding_boxes.detection_boxes
+    scores = monitoring_frame.bounding_boxes.detection_scores
+    classes = monitoring_frame.bounding_boxes.detection_classes
+    num_detections = monitoring_frame.bounding_boxes.num_detections
 
-def build_monitoring_frame_raw_message(
-    ts: int,
-    image: octoprint_nanny.types.Image,
-    metadata: octoprint_nanny.types.Metadata,
-) -> bytes:
-    builder = flatbuffers.Builder(1024)
+    # begin boxes builder
+    BoundingBoxes.BoundingBoxesStartBoxesVector(builder, len(boxes))
+    for box in boxes:
+        Box.CreateBox(builder, *box)
+    boxes = builder.EndVector(len(boxes))
+    # end boxes
 
-    # begin byte array
-    Image.ImageStartDataVector(builder, len(image.data))
-    builder.Bytes[builder.head : (builder.head + len(image.data))] = image.data
-    image.data = builder.EndVector(len(image.data))
-    # end byte array
+    # begin scores
+    scores = builder.CreateNumpyVector(scores)
+    # end scores
 
-    # begin image
-    Image.ImageStart(builder)
-    Image.ImageAddHeight(builder, image.height)
-    Image.ImageAddWidth(builder, image.width)
-    Image.ImageAddData(builder, image.data)
-    image = Image.ImageEnd(builder)
-    # end image
+    # begin classes builder
+    classes = builder.CreateNumpyVector(classes)
+    # end classes
 
-    # begin message body
-    MonitoringFrame.MonitoringFrameStart(builder)
-    MonitoringFrame.MonitoringFrameAddTs(builder, ts)
-    MonitoringFrame.MonitoringFrameAddImage(builder, image)
-    MonitoringFrame.MonitoringFrameAddEventType(
-        builder,
-        PluginEvent.PluginEvent.monitoring_frame_raw,
-    )
-    message = MonitoringFrame.MonitoringFrameEnd(builder)
-    # end message body
+    # begin bounding boxes
+    BoundingBoxes.BoundingBoxesStart(builder)
+    BoundingBoxes.BoundingBoxesAddBoxes(builder, boxes)
+    BoundingBoxes.BoundingBoxesAddScores(builder, scores)
+    BoundingBoxes.BoundingBoxesAddClasses(builder, classes)
+    BoundingBoxes.BoundingBoxesAddNumDetections(builder, num_detections)
+    bounding_boxes = BoundingBoxes.BoundingBoxesEnd(builder)
+    # end bounding boxes
 
-    return build_telemetry_message(
-        builder, message, metadata, MessageType.MessageType.MonitoringFrame
-    )
+    return bounding_boxes
 
 
-def build_monitoring_frame_post_message(
-    ts: int,
-    image: octoprint_nanny.types.Image,
+def build_telemetry_event_message(
+    event_type: int,
     metadata: octoprint_nanny.types.Metadata,
+    monitoring_frame: octoprint_nanny.types.MonitoringFrame,
 ) -> bytes:
     builder = flatbuffers.Builder(1024)
 
-    # begin byte array
-    Image.ImageStartDataVector(builder, len(image.data))
-    builder.Bytes[builder.head : (builder.head + len(image.data))] = image.data
-    image_data = builder.EndVector(len(image.data))
-    # end byte array
-
     # begin image
+    Image.ImageStartDataVector(builder, len(monitoring_frame.image.data))
+    builder.Bytes[
+        builder.head : (builder.head + len(monitoring_frame.image.data))
+    ] = monitoring_frame.image.data
+    image_data = builder.EndVector(len(monitoring_frame.image.data))
     Image.ImageStart(builder)
-    Image.ImageAddHeight(builder, image.height)
-    Image.ImageAddWidth(builder, image.width)
+    Image.ImageAddHeight(builder, monitoring_frame.image.height)
+    Image.ImageAddWidth(builder, monitoring_frame.image.width)
     Image.ImageAddData(builder, image_data)
     image = Image.ImageEnd(builder)
     # end image
 
-    # begin message body
+    # begin event data
+    # prediction is optional; if not provided, inference will run in beam pipeline
+    bounding_boxes = build_bounding_boxes_message(builder, monitoring_frame)
     MonitoringFrame.MonitoringFrameStart(builder)
-    MonitoringFrame.MonitoringFrameAddTs(builder, ts)
     MonitoringFrame.MonitoringFrameAddImage(builder, image)
-    MonitoringFrame.MonitoringFrameAddEventType(
-        builder,
-        PluginEvent.PluginEvent.monitoring_frame_post,
-    )
-    message = MonitoringFrame.MonitoringFrameEnd(builder)
-    # end message body
-    return build_telemetry_message(
-        builder, message, metadata, MessageType.MessageType.MonitoringFrame
-    )
+    if bounding_boxes:
+        MonitoringFrame.MonitoringFrameAddBoundingBoxes(builder, bounding_boxes)
+    MonitoringFrame.MonitoringFrameAddTs(builder, monitoring_frame.ts)
+    event_data = MonitoringFrame.MonitoringFrameEnd(builder)
 
+    # end event data
 
-def build_bounding_boxes_message(
-    ts: int,
-    prediction: octoprint_nanny.types.BoundingBoxPrediction,
-    metadata: octoprint_nanny.types.Metadata,
-    image: Optional[octoprint_nanny.types.Image] = None,
-) -> bytes:
-    builder = flatbuffers.Builder(1024)
-    boxes = prediction.detection_boxes
-    scores = prediction.detection_scores
-    classes = prediction.detection_classes
-    num_detections = prediction.num_detections
-
-    # begin byte array
-    if image:
-        Image.ImageStartDataVector(builder, len(image.data))
-        builder.Bytes[builder.head : (builder.head + len(image.data))] = image.data
-        image.data = builder.EndVector(len(image.data))
-        # end byte array
-
-        # begin image
-        Image.ImageStart(builder)
-        Image.ImageAddHeight(builder, image.height)
-        Image.ImageAddWidth(builder, image.width)
-        Image.ImageAddData(builder, image.data)
-        image = Image.ImageEnd(builder)
-        # end image
-
-    # begin boxes builder
-    BoundingBoxes.BoundingBoxesStartBoxesVector(builder, len(boxes))
-    for box in boxes:
-        Box.CreateBox(builder, *box)
-    boxes = builder.EndVector(len(boxes))
-    # end boxes
-
-    # begin scores
-    scores = builder.CreateNumpyVector(scores)
-    # end scores
+    # begin metadata
+    Metadata.MetadataStart(builder)
+    Metadata.MetadataAddUserId(builder, metadata.user_id)
+    Metadata.MetadataAddDeviceCloudiotId(builder, metadata.device_cloudiot_id)
+    Metadata.MetadataAddDeviceId(builder, metadata.device_id)
+    metadata = Metadata.MetadataEnd(builder)
+    # end metadata
 
-    # begin classes builder
-    classes = builder.CreateNumpyVector(classes)
-    # end classes
+    # begin telemetry event
+    TelemetryEvent.TelemetryEventStart(builder)
+    TelemetryEvent.TelemetryEventAddEventData(builder, event_data)
+    TelemetryEvent.TelemetryEventAddEventType(builder, event_type)
+    telemetry_event = TelemetryEvent.TelemetryEventEnd(builder)
+    builder.Finish(telemetry_event)
 
-    # begin message body
-    BoundingBoxes.BoundingBoxesStart(builder)
-    BoundingBoxes.BoundingBoxesAddTs(builder, ts)
-    BoundingBoxes.BoundingBoxesAddBoxes(builder, boxes)
-    BoundingBoxes.BoundingBoxesAddScores(builder, scores)
-    BoundingBoxes.BoundingBoxesAddClasses(builder, classes)
-    BoundingBoxes.BoundingBoxesAddNumDetections(builder, num_detections)
-    BoundingBoxes.BoundingBoxesAddEventType(
-        builder, PluginEvent.PluginEvent.bounding_box_predict
-    )
-
-    if image:
-        BoundingBoxes.BoundingBoxesAddImage(builder, image)
-    message = BoundingBoxes.BoundingBoxesEnd(builder)
-    # end message body
-
-    return build_telemetry_message(
-        builder, message, metadata, MessageType.MessageType.BoundingBoxes
-    )
+    # end tlemetry event
+    return builder.Output()
diff --git a/octoprint_nanny/predictor.py b/octoprint_nanny/predictor.py
index 7330843a..5700eb37 100644
--- a/octoprint_nanny/predictor.py
+++ b/octoprint_nanny/predictor.py
@@ -276,6 +276,7 @@ def predict(self, image: PIL.Image) -> octoprint_nanny.types.BoundingBoxPredicti
 def explode_prediction_df(
     ts: int, prediction: octoprint_nanny.types.BoundingBoxPrediction
 ) -> pd.DataFrame:
+
     data = {"frame_id": ts, **asdict(prediction)}
     df = pd.DataFrame.from_records([data])
 
@@ -296,6 +297,8 @@ def explode_prediction_df(
 
 
 def print_is_healthy(df: pd.DataFrame, degree: int = 1) -> float:
+    if df.empty:
+        return True
     df = pd.concat(
         {
             "unhealthy": df[df["detection_classes"].isin(NEGATIVE_LABELS)],
@@ -325,12 +328,8 @@ def print_is_healthy(df: pd.DataFrame, degree: int = 1) -> float:
 
 
 def predict_threadsafe(
-    image_bytes: bytes, **kwargs
-) -> Tuple[
-    octoprint_nanny.types.Image,
-    Optional[octoprint_nanny.types.Image],
-    Optional[octoprint_nanny.types.BoundingBoxPrediction],
-]:
+    ts: int, image_bytes: bytes, **kwargs
+) -> octoprint_nanny.types.MonitoringFrame:
 
     global PREDICTOR
     if PREDICTOR is None:
@@ -356,8 +355,6 @@ def predict_threadsafe(
     else:
         post_frame = None
 
-    return (
-        original_frame,
-        post_frame,
-        prediction,
+    return octoprint_nanny.types.MonitoringFrame(
+        ts=ts, image=post_frame or original_frame, bounding_boxes=prediction
     )
diff --git a/octoprint_nanny/types.py b/octoprint_nanny/types.py
index 4c749ff9..c40120b9 100644
--- a/octoprint_nanny/types.py
+++ b/octoprint_nanny/types.py
@@ -11,6 +11,7 @@
     OctoPrintEventEventTypeEnum as OctoPrintEventTypes,
 )
 from print_nanny_client.models.command_enum import CommandEnum
+from PrintNannyEvent.TelemetrySchema.TelemetryEventEnum import TelemetryEventEnum
 
 PLUGIN_PREFIX = "octoprint_nanny_"
 
@@ -38,6 +39,13 @@ class BoundingBoxPrediction:
     image: Image = None
 
 
+@dataclass
+class MonitoringFrame:
+    ts: int
+    image: Image
+    bounding_boxes: BoundingBoxPrediction = None
+
+
 class MonitoringModes(Enum):
     ACTIVE_LEARNING = "active_learning"
     LITE = "lite"
diff --git a/octoprint_nanny/workers/monitoring.py b/octoprint_nanny/workers/monitoring.py
index 4b0aaea7..ac73b9a0 100644
--- a/octoprint_nanny/workers/monitoring.py
+++ b/octoprint_nanny/workers/monitoring.py
@@ -24,7 +24,7 @@
 
 import beeline
 
-import PrintNannyMessage
+import PrintNannyEvent
 from octoprint.events import Events
 
 import octoprint_nanny.types
@@ -144,7 +144,7 @@ def _producer_worker(self):
         loop.run_until_complete(asyncio.ensure_future(self._producer()))
         loop.close()
 
-    @beeline.traced(name="MonitoringWorker._create_active_learning_msgs")
+    @beeline.traced(name="MonitoringWorker._create_active_learning_json_msgs")
     def _create_active_learning_json_msgs(
         self, now, image
     ) -> PrintNannyMonitoringFrameMessage:
@@ -154,17 +154,19 @@ def _create_active_learning_json_msgs(
 
         ws_msg = PrintNannyMonitoringFrameMessage(
             ts=now,
-            event_type=PluginEvents.MONITORING_FRAME_RAW,
+            event_type=PluginEvents.MONITORING_FRAME_POST,
             image=b64_image,
         )
         return ws_msg
 
     @beeline.traced(name="MonitoringWorker._create_active_learning_msg")
     def _create_active_learning_flatbuffer_msg(
-        self, ts: int, image: octoprint_nanny.types.Image
+        self, monitoring_frame: octoprint_nanny.types.MonitoringFrame
     ) -> bytes:
-        msg = octoprint_nanny.clients.flatbuffers.build_monitoring_frame_raw_message(
-            ts, image, self._plugin.settings.metadata
+        msg = octoprint_nanny.clients.flatbuffers.build_telemetry_event_message(
+            event_type=octoprint_nanny.types.TelemetryEventEnum.monitoring_frame_raw,
+            metadata=self._plugin.settings.metadata,
+            monitoring_frame=monitoring_frame,
         )
         return msg
 
@@ -190,36 +192,26 @@ def _create_lite_json_msgs(
 
         return ws_msg, mqtt_msg
 
-    @beeline.traced(name="MonitoringWorker._create_lite_fb_ws_msg")
-    def _create_lite_fb_ws_msg(
-        self,
-        ts: int,
-        image: octoprint_nanny.types.Image,
-    ) -> Tuple[bytes, Optional[bytes]]:
-
-        return octoprint_nanny.clients.flatbuffers.build_monitoring_frame_post_message(
-            ts, image, self._plugin.settings.metadata
-        )
-
     @beeline.traced(name="MonitoringWorker._create_lite_fb_mqtt_msg")
-    def _create_lite_fb_mqtt_msg(
-        self,
-        ts: int,
-        image: octoprint_nanny.types.Image,
-        prediction: octoprint_nanny.types.BoundingBoxPrediction,
+    def _create_lite_fb_msg(
+        self, monitoring_frame: octoprint_nanny.types.MonitoringFrame
     ) -> Tuple[bytes, Optional[bytes]]:
-        return octoprint_nanny.clients.flatbuffers.build_bounding_boxes_message(
-            ts, prediction, self._plugin.settings.metadata
+        return octoprint_nanny.clients.flatbuffers.build_telemetry_event_message(
+            event_type=octoprint_nanny.types.TelemetryEventEnum.monitoring_frame_post,
+            metadata=self._plugin.settings.metadata,
+            monitoring_frame=monitoring_frame,
         )
 
     @beeline.traced(name="MonitoringWorker._active_learning_loop")
-    async def _active_learning_loop(self, loop, pool):
+    async def _active_learning_loop(self):
         ts = int(datetime.now(pytz.utc).timestamp())
         image_bytes = await self.load_url_buffer()
 
         (w, h) = PIL.Image.open(io.BytesIO(image_bytes)).size
         image = octoprint_nanny.types.Image(height=h, width=w, data=image_bytes)
-        msg = self._create_active_learning_flatbuffer_msg(ts, image)
+        monitoring_frame = octoprint_nanny.types.MonitoringFrame(ts=ts, image=image)
+
+        msg = self._create_active_learning_flatbuffer_msg(monitoring_frame)
 
         self._plugin._event_bus.fire(
             Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_RAW,
@@ -228,12 +220,14 @@ async def _active_learning_loop(self, loop, pool):
         self._pn_ws_queue.put_nowait(msg)
         self._mqtt_send_queue.put_nowait(msg)
 
-    @beeline.traced(name="MonitoringWorker._lite_loop")
-    async def _lite_loop(self, loop, pool):
-        ts = int(datetime.now(pytz.utc).timestamp())
+    @beeline.traced(name="MonitoringWorker._lite_predict_and_calc_health")
+    async def _lite_predict_and_calc_health(
+        self, ts
+    ) -> octoprint_nanny.types.MonitoringFrame:
+
         image_bytes = await self.load_url_buffer()
         func = functools.partial(
-            predict_threadsafe, image_bytes, **self._predictor_kwargs
+            predict_threadsafe, ts, image_bytes, **self._predictor_kwargs
         )
 
         # trace predict latency, including serialization in/out of process pool
@@ -242,12 +236,16 @@ async def _lite_loop(self, loop, pool):
                 "name": "predict_pooled",
             }
         )
-        raw_frame, post_frame, prediction = await loop.run_in_executor(pool, func)
-        self._honeycomb_tracer.add_context({"prediction": prediction})
+
+        monitoring_frame = await self.loop.run_in_executor(self.pool, func)
+        self._honeycomb_tracer.add_context(
+            {"bounding_boxes": monitoring_frame.bounding_boxes}
+        )
         self._honeycomb_tracer.finish_span(span)
 
         # trace metrics calculations in/out of process pool
-        self.update_dataframe(ts, prediction)
+        if monitoring_frame.bounding_boxes is not None:
+            self.update_dataframe(ts, monitoring_frame.bounding_boxes)
 
         # trace polyfit curve
         span = self._honeycomb_tracer.start_span(
@@ -255,40 +253,44 @@ async def _lite_loop(self, loop, pool):
                 "name": "print_is_healthy",
             }
         )
-        func = functools.partial(print_is_healthy)
-        healthy = await loop.run_in_executor(pool, func)
+
+        func = functools.partial(print_is_healthy, self._df)
+        healthy = await self.loop.run_in_executor(self.pool, func)
         if healthy is False:
+            octoprint_device = self._plugin.settings.device_id
+            dataframe = io.BytesIO(name=f"{octoprint_device}_{ts}.parquet")
+            self._df.to_parquet(dataframe, engine="pyarrow")
             alert = await self._plugin.settings.rest_client.create_defect_alert(
-                octoprint_device=self._plugin.settings.device_id
+                octoprint_device=octoprint_device, dataframe=self._df
             )
             logger.warning(f"Created DefectAlert with id={alert.id}")
 
-        video_frame = post_frame if post_frame is not None else raw_frame
-        ws_msg = self._create_lite_fb_ws_msg(ts=ts, image=video_frame)
+        return monitoring_frame
+
+    @beeline.traced(name="MonitoringWorker._lite_loop")
+    async def _lite_loop(self):
+        ts = int(datetime.now(pytz.utc).timestamp())
+
+        monitoring_frame = await self._lite_predict_and_calc_health(ts)
 
+        msg = self._create_lite_fb_msg(monitoring_frame=monitoring_frame)
         self._plugin._event_bus.fire(
-            Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_RAW,
-            payload=base64.b64encode(video_frame.data),
+            Events.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_POST,
+            payload=base64.b64encode(monitoring_frame.image.data),
         )
         if self._plugin.settings.webcam_upload:
-            self._pn_ws_queue.put_nowait(ws_msg)
+            self._pn_ws_queue.put_nowait(msg)
 
-        if prediction and post_frame:
-            mqtt_msg = self._create_lite_fb_mqtt_msg(
-                ts=ts, image=post_frame, prediction=prediction
-            )
-            octoprint_event = PluginEvents.to_octoprint_event(
-                PluginEvents.BOUNDING_BOX_PREDICT
-            )
-            self._mqtt_send_queue.put_nowait(mqtt_msg)
+        if monitoring_frame.bounding_boxes is not None:
+            self._mqtt_send_queue.put_nowait(msg)
 
     @beeline.traced(name="MonitoringWorker._loop")
-    async def _loop(self, loop, pool):
+    async def _loop(self):
 
         if self._monitoring_mode == MonitoringModes.ACTIVE_LEARNING:
-            await self._active_learning_loop(loop, pool)
+            await self._active_learning_loop()
         elif self._monitoring_mode == MonitoringModes.LITE:
-            await self._lite_loop(loop, pool)
+            await self._lite_loop()
         else:
             logger.error(f"Unsupported monitoring_mode={self._monitoring_mode}")
             return
@@ -299,15 +301,18 @@ async def _producer(self):
         """
         logger.info("Started MonitoringWorker.consumer thread")
         loop = asyncio.get_running_loop()
+        self.loop = loop
         with concurrent.futures.ProcessPoolExecutor() as pool:
+            self.pool = pool
             while not self._halt.is_set():
                 await asyncio.sleep(self._sleep_interval)
-                await self._loop(loop, pool)
+                await self._loop()
 
             logger.warning("Halt event set, worker will exit soon")
 
     def run(self):
         loop = asyncio.new_event_loop()
+        self.loop = loop
         asyncio.set_event_loop(loop)
         return loop.run_until_complete(self._producer())
 
diff --git a/setup.py b/setup.py
index 78a147a8..c1791747 100644
--- a/setup.py
+++ b/setup.py
@@ -87,28 +87,34 @@ class CPUNotSupported(Exception):
 ###
 
 # hardware layer : software layer : wheel
-tensorflow_wheel_map = {
+third_party_wheel_map = {
     "armv7l": {
-        # "32bit": "tensorflow @ https://github.com/bitsy-ai/tensorflow-arm-bin/releases/download/v2.4.0/tensorflow-2.4.0-cp37-none-linux_armv7l.whl"
-        "32bit": "tflite_runtime @ https://github.com/google-coral/pycoral/releases/download/v1.0.1/tflite_runtime-2.5.0-cp37-cp37m-linux_armv7l.whl"
+        "32bit": [
+            "tflite_runtime @ https://github.com/google-coral/pycoral/releases/download/v1.0.1/tflite_runtime-2.5.0-cp37-cp37m-linux_armv7l.whl",
+            "pyarrow @ https://github.com/bitsy-ai/pyarrow-arm-bin/releases/download/apache-arrow-3.0.0/pyarrow-3.0.0-cp37-cp37m-linux_armv7l.whl",
+        ]
     },
     "aarch64": {
-        # "32bit": "tensorflow @ https://github.com/bitsy-ai/tensorflow-arm-bin/releases/download/v2.4.0/tensorflow-2.4.0-cp37-none-linux_armv7l.whl",
-        # "64bit": "tensorflow @ https://github.com/bitsy-ai/tensorflow-arm-bin/releases/download/v2.4.0/tensorflow-2.4.0-cp37-none-linux_aarch64.whl",
-        "32bit": "tflite_runtime @ https://github.com/google-coral/pycoral/releases/download/v1.0.1/tflite_runtime-2.5.0-cp37-cp37m-linux_armv7l.whl",
-        "64bit": "tflite_runtime @ https://github.com/google-coral/pycoral/releases/download/v1.0.1/tflite_runtime-2.5.0-cp37-cp37m-linux_aarch64.whl",
+        "32bit": [
+            "tflite_runtime @ https://github.com/google-coral/pycoral/releases/download/v1.0.1/tflite_runtime-2.5.0-cp37-cp37m-linux_armv7l.whl"
+        ],
+        "64bit": [
+            "tflite_runtime @ https://github.com/google-coral/pycoral/releases/download/v1.0.1/tflite_runtime-2.5.0-cp37-cp37m-linux_aarch64.whl"
+        ],
     },
     "x86_64": {
-        # "64bit": "tensorflow==2.4.0"
-        "64bit": "tflite_runtime @ https://github.com/google-coral/pycoral/releases/download/v1.0.1/tflite_runtime-2.5.0-cp37-cp37m-linux_x86_64.whl"
+        "64bit": [
+            "tflite_runtime @ https://github.com/google-coral/pycoral/releases/download/v1.0.1/tflite_runtime-2.5.0-cp37-cp37m-linux_x86_64.whl",
+            "pyarrow==3.0.0",
+        ]
     },
 }
 
 hardware_arch = os.uname().machine
 software_arch, _ = platform.architecture()
 
-if hardware_arch in tensorflow_wheel_map.keys():
-    tensorflow = tensorflow_wheel_map[hardware_arch][software_arch]
+if hardware_arch in third_party_wheels.keys():
+    third_party_wheels = third_party_wheel_map[hardware_arch][software_arch]
 else:
     raise CPUNotSupported(
         "Sorry, OctoPrint Nanny does not support {} architechture. Please open a Github issue for support. https://github.com/bitsy-ai/octoprint-nanny-plugin/issues/new".format(
@@ -118,7 +124,6 @@ class CPUNotSupported(Exception):
     sys.exit(1)
 
 plugin_requires = [
-    tensorflow,
     "numpy",
     "pillow",
     "typing_extensions ; python_version < '3.8'",
@@ -134,7 +139,7 @@ class CPUNotSupported(Exception):
     "aiofiles>=0.6.0",
     "flatbuffers>=1.12",
     "pandas>=1.2.2",
-]
+] + third_party_wheels
 
 extra_requires = {
     "dev": ["pytest", "pytest-cov", "pytest-mock", "pytest-asyncio", "twine"]


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 1b973b9f..fb3233ae 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,6 +1,7 @@
 import pytest
 import octoprint_nanny.types
 from octoprint_nanny.workers.monitoring import MonitoringWorker
+from octoprint_nanny.plugins import OctoPrintNannyPlugin
 
 
 class MockResponse(object):
diff --git a/tests/test_benchmarks.py b/tests/test_benchmarks.py
index 43ecb46f..97bf9305 100644
--- a/tests/test_benchmarks.py
+++ b/tests/test_benchmarks.py
@@ -10,7 +10,10 @@
 from octoprint_nanny.utils.encoder import NumpyEncoder
 import sys
 import PIL
-from PrintNannyMessage.Telemetry import TelemetryMessage, MonitoringFrame, MessageType
+from PrintNannyEvent.TelemetrySchema import (
+    TelemetryEvent,
+    MonitoringFrame,
+)
 from datetime import datetime
 
 
@@ -72,10 +75,8 @@ def serialize(ts, buffer, image_height, image_width):
 
     ws_msg = benchmark(serialize, ts, buffer, image_height, image_width)
 
-    deserialized_ws_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
-        ws_msg, 0
-    )
-    ws_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_ws_msg)
+    deserialized_ws_msg = TelmetryEvent.TelmetryEvent.GetRootAsTelmetryEvent(ws_msg, 0)
+    ws_msg_obj = TelmetryEvent.TelmetryEventT.InitFromObj(deserialized_ws_msg)
 
     assert ws_msg_obj.message.ts == ts
     assert ws_msg_obj.message.image.width == 640
@@ -161,18 +162,16 @@ def serialize(ts, oh, ow, buffer, vh, vw, viz_buffer, prediction):
     benchmark.extra_info["mqtt_msg_size"] = sys.getsizeof(mqtt_msg)
     benchmark.extra_info["serializer"] = "fb"
 
-    deserialized_ws_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
-        ws_msg, 0
-    )
-    ws_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_ws_msg)
+    deserialized_ws_msg = TelmetryEvent.TelmetryEvent.GetRootAsTelmetryEvent(ws_msg, 0)
+    ws_msg_obj = TelmetryEvent.TelmetryEventT.InitFromObj(deserialized_ws_msg)
     assert ws_msg_obj.message.ts == ts
     assert ws_msg_obj.message.image.width == 640
     assert ws_msg_obj.message.image.height == 480
 
-    deserialized_mqtt_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
+    deserialized_mqtt_msg = TelmetryEvent.TelmetryEvent.GetRootAsTelmetryEvent(
         mqtt_msg, 0
     )
-    mqtt_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_mqtt_msg)
+    mqtt_msg_obj = TelmetryEvent.TelmetryEventT.InitFromObj(deserialized_mqtt_msg)
     assert mqtt_msg_obj.message.ts == ts
 
     assert mqtt_msg_obj.message.numDetections == len(mqtt_msg_obj.message.scores)
@@ -220,18 +219,16 @@ def serialize(ts, oh, ow, buffer, vh, vw, viz_buffer, prediction):
     benchmark.extra_info["mqtt_msg_size"] = sys.getsizeof(mqtt_msg)
     benchmark.extra_info["serializer"] = "fb"
 
-    deserialized_ws_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
-        ws_msg, 0
-    )
-    ws_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_ws_msg)
+    deserialized_ws_msg = TelmetryEvent.TelmetryEvent.GetRootAsTelmetryEvent(ws_msg, 0)
+    ws_msg_obj = TelmetryEvent.TelmetryEventT.InitFromObj(deserialized_ws_msg)
     assert ws_msg_obj.message.ts == ts
     assert ws_msg_obj.message.image.width == 640
     assert ws_msg_obj.message.image.height == 480
 
-    deserialized_mqtt_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
+    deserialized_mqtt_msg = TelmetryEvent.TelmetryEvent.GetRootAsTelmetryEvent(
         mqtt_msg, 0
     )
-    mqtt_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_mqtt_msg)
+    mqtt_msg_obj = TelmetryEvent.TelmetryEventT.InitFromObj(deserialized_mqtt_msg)
     assert mqtt_msg_obj.message.ts == ts
 
     assert mqtt_msg_obj.message.numDetections == len(mqtt_msg_obj.message.scores)
diff --git a/tests/workers/test_monitoring.py b/tests/workers/test_monitoring.py
index 75a476cc..503df066 100644
--- a/tests/workers/test_monitoring.py
+++ b/tests/workers/test_monitoring.py
@@ -4,6 +4,7 @@
 import threading
 from asynctest import CoroutineMock, patch
 import unittest.mock
+
 from octoprint_nanny.types import PluginEvents, MonitoringModes
 from octoprint_nanny.workers.monitoring import (
     MonitoringWorker,
@@ -11,11 +12,10 @@
     MonitoringModes,
 )
 from octoprint_nanny.predictor import predict_threadsafe
-from PrintNannyMessage.Telemetry import (
-    TelemetryMessage,
+from PrintNannyEvent.TelemetrySchema import (
     MonitoringFrame,
-    MessageType,
-    PluginEvent,
+    TelemetryEvent,
+    TelemetryEventEnum,
 )
 
 
@@ -45,14 +45,13 @@ async def test_lite_mode_webcam_enabled_with_prediction_results_uncalibrated(
     predict_worker = MonitoringWorker(pn_ws_queue, mqtt_send_queue, halt, plugin)
 
     loop = asyncio.get_running_loop()
+    predict_worker.loop = loop
     with concurrent.futures.ProcessPoolExecutor() as pool:
-        await predict_worker._loop(loop, pool)
+        predict_worker.pool = pool
+        await predict_worker._loop()
 
-    octoprint_event = PluginEvents.to_octoprint_event(
-        PluginEvents.MONITORING_FRAME_POST
-    )
     predict_worker._plugin._event_bus.fire.assert_called_once_with(
-        octoprint_event,
+        mock_events_enum.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_POST,
         payload=mock_base64.b64encode.return_value,
     )
     predict_worker._pn_ws_queue.put_nowait.assert_called_once()
@@ -61,13 +60,11 @@ async def test_lite_mode_webcam_enabled_with_prediction_results_uncalibrated(
     kall = predict_worker._mqtt_send_queue.put_nowait.mock_calls[0]
     _, args, kwargs = kall
 
-    mqtt_msg = args[0]
-    deserialized_mqtt_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
-        mqtt_msg, 0
-    )
-    mqtt_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_mqtt_msg)
+    msg = args[0]
+    deserialized_msg = TelemetryEvent.TelemetryEvent.GetRootAsTelemetryEvent(msg, 0)
+    msg_obj = TelemetryEvent.TelemetryEventT.InitFromObj(deserialized_msg)
     assert (
-        mqtt_msg_obj.message.eventType == PluginEvent.PluginEvent.bounding_box_predict
+        msg_obj.eventType == TelemetryEventEnum.TelemetryEventEnum.monitoring_frame_post
     )
 
 
@@ -103,14 +100,13 @@ async def test_lite_mode_webcam_enabled_with_prediction_results_calibrated(
     predict_worker = MonitoringWorker(pn_ws_queue, mqtt_send_queue, halt, plugin)
 
     loop = asyncio.get_running_loop()
+    predict_worker.loop = loop
     with concurrent.futures.ProcessPoolExecutor() as pool:
-        await predict_worker._loop(loop, pool)
+        predict_worker.pool = pool
+        await predict_worker._loop()
 
-    octoprint_event = PluginEvents.to_octoprint_event(
-        PluginEvents.MONITORING_FRAME_POST
-    )
     predict_worker._plugin._event_bus.fire.assert_called_once_with(
-        octoprint_event,
+        mock_events_enum.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_POST,
         payload=mock_base64.b64encode.return_value,
     )
     predict_worker._pn_ws_queue.put_nowait.assert_called_once()
@@ -119,13 +115,11 @@ async def test_lite_mode_webcam_enabled_with_prediction_results_calibrated(
     kall = predict_worker._mqtt_send_queue.put_nowait.mock_calls[0]
     _, args, kwargs = kall
 
-    mqtt_msg = args[0]
-    deserialized_mqtt_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
-        mqtt_msg, 0
-    )
-    mqtt_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_mqtt_msg)
+    msg = args[0]
+    deserialized_msg = TelemetryEvent.TelemetryEvent.GetRootAsTelemetryEvent(msg, 0)
+    msg_obj = TelemetryEvent.TelemetryEventT.InitFromObj(deserialized_msg)
     assert (
-        mqtt_msg_obj.message.eventType == PluginEvent.PluginEvent.bounding_box_predict
+        msg_obj.eventType == TelemetryEventEnum.TelemetryEventEnum.monitoring_frame_post
     )
 
 
@@ -156,14 +150,13 @@ async def test_lite_mode_webcam_enabled_zero_prediction_results_uncalibrated(
     predict_worker = MonitoringWorker(pn_ws_queue, mqtt_send_queue, halt, plugin)
 
     loop = asyncio.get_running_loop()
+    predict_worker.loop = loop
     with concurrent.futures.ProcessPoolExecutor() as pool:
-        await predict_worker._loop(loop, pool)
+        predict_worker.pool = pool
+        await predict_worker._loop()
 
-    octoprint_event = PluginEvents.to_octoprint_event(
-        PluginEvents.MONITORING_FRAME_POST
-    )
     predict_worker._plugin._event_bus.fire.assert_called_once_with(
-        octoprint_event,
+        mock_events_enum.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_POST,
         payload=mock_base64.b64encode.return_value,
     )
     predict_worker._pn_ws_queue.put_nowait.assert_called_once()
@@ -203,14 +196,13 @@ async def test_lite_mode_webcam_enabled_zero_prediction_results_calibrated(
     predict_worker = MonitoringWorker(pn_ws_queue, mqtt_send_queue, halt, plugin)
 
     loop = asyncio.get_running_loop()
+    predict_worker.loop = loop
     with concurrent.futures.ProcessPoolExecutor() as pool:
-        await predict_worker._loop(loop, pool)
+        predict_worker.pool = pool
+        await predict_worker._loop()
 
-    octoprint_event = PluginEvents.to_octoprint_event(
-        PluginEvents.MONITORING_FRAME_POST
-    )
     predict_worker._plugin._event_bus.fire.assert_called_once_with(
-        octoprint_event,
+        mock_events_enum.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_POST,
         payload=mock_base64.b64encode.return_value,
     )
     predict_worker._pn_ws_queue.put_nowait.assert_called_once()
@@ -245,14 +237,13 @@ async def test_lite_mode_webcam_disabled(
     predict_worker = MonitoringWorker(pn_ws_queue, mqtt_send_queue, halt, plugin)
 
     loop = asyncio.get_running_loop()
+    predict_worker.loop = loop
     with concurrent.futures.ProcessPoolExecutor() as pool:
-        await predict_worker._loop(loop, pool)
+        predict_worker.pool = pool
+        await predict_worker._loop()
 
-    octoprint_event = PluginEvents.to_octoprint_event(
-        PluginEvents.MONITORING_FRAME_POST
-    )
     predict_worker._plugin._event_bus.fire.assert_called_once_with(
-        octoprint_event,
+        mock_events_enum.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_POST,
         payload=mock_base64.b64encode.return_value,
     )
     predict_worker._pn_ws_queue.put_nowait.assert_not_called()
@@ -261,13 +252,11 @@ async def test_lite_mode_webcam_disabled(
     kall = predict_worker._mqtt_send_queue.put_nowait.mock_calls[0]
     _, args, kwargs = kall
 
-    mqtt_msg = args[0]
-    deserialized_mqtt_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
-        mqtt_msg, 0
-    )
-    mqtt_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_mqtt_msg)
+    msg = args[0]
+    deserialized_msg = TelemetryEvent.TelemetryEvent.GetRootAsTelemetryEvent(msg, 0)
+    msg_obj = TelemetryEvent.TelemetryEventT.InitFromObj(deserialized_msg)
     assert (
-        mqtt_msg_obj.message.eventType == PluginEvent.PluginEvent.bounding_box_predict
+        msg_obj.eventType == TelemetryEventEnum.TelemetryEventEnum.monitoring_frame_post
     )
 
 
@@ -299,12 +288,13 @@ async def test_active_learning_mode(
     predict_worker = MonitoringWorker(pn_ws_queue, mqtt_send_queue, halt, plugin)
 
     loop = asyncio.get_running_loop()
+    predict_worker.loop = loop
     with concurrent.futures.ProcessPoolExecutor() as pool:
-        await predict_worker._loop(loop, pool)
+        predict_worker.pool = pool
+        await predict_worker._loop()
 
-    octoprint_event = PluginEvents.to_octoprint_event(PluginEvents.MONITORING_FRAME_RAW)
     predict_worker._plugin._event_bus.fire.assert_called_once_with(
-        octoprint_event,
+        mock_events_enum.PLUGIN_OCTOPRINT_NANNY_MONITORING_FRAME_RAW,
         payload=mock_base64.b64encode.return_value,
     )
     predict_worker._pn_ws_queue.put_nowait.assert_called_once()
@@ -313,12 +303,10 @@ async def test_active_learning_mode(
     kall = predict_worker._mqtt_send_queue.put_nowait.mock_calls[0]
     _, args, kwargs = kall
 
-    mqtt_msg = args[0]
-    deserialized_mqtt_msg = TelemetryMessage.TelemetryMessage.GetRootAsTelemetryMessage(
-        mqtt_msg, 0
-    )
-    mqtt_msg_obj = TelemetryMessage.TelemetryMessageT.InitFromObj(deserialized_mqtt_msg)
+    msg = args[0]
+    deserialized_msg = TelemetryEvent.TelemetryEvent.GetRootAsTelemetryEvent(msg, 0)
+    msg_obj = TelemetryEvent.TelemetryEventT.InitFromObj(deserialized_msg)
 
     assert (
-        mqtt_msg_obj.message.eventType == PluginEvent.PluginEvent.monitoring_frame_raw
+        msg_obj.eventType == TelemetryEventEnum.TelemetryEventEnum.monitoring_frame_raw
     )