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/rest.py b/octoprint_nanny/clients/rest.py
index 036e8351..2a6a8af1 100644
--- a/octoprint_nanny/clients/rest.py
+++ b/octoprint_nanny/clients/rest.py
@@ -299,3 +299,20 @@ async def update_or_create_device_calibration(
                 request
             )
             return device_calibration
+
+    @beeline.traced("RestAPIClient.create_alert")
+    @backoff.on_exception(
+        backoff.expo,
+        aiohttp.ClientConnectionError,
+        logger=logger,
+        max_time=MAX_BACKOFF_TIME,
+    )
+    async def create_defect_alert(self, octoprint_device_id, **kwargs):
+        async with AsyncApiClient(self._api_config) as api_client:
+            api_instance = print_nanny_client.AlertsApi(api_client=api_client)
+
+            request = print_nanny_client.DefectAlertRequest(
+                octoprint_device=octoprint_device_id, **kwargs
+            )
+            defect_alert = await api_instance.defect_alerts_create(request)
+            return defect_alert
diff --git a/octoprint_nanny/predictor.py b/octoprint_nanny/predictor.py
index f95fb88d..7330843a 100644
--- a/octoprint_nanny/predictor.py
+++ b/octoprint_nanny/predictor.py
@@ -2,10 +2,12 @@
 import json
 import logging
 import numpy as np
+import pandas as pd
 import os
 import time
 import io
 import threading
+from dataclasses import asdict
 
 import PIL
 import tflite_runtime.interpreter as tflite
@@ -20,6 +22,25 @@
 
 logger = logging.getLogger("octoprint.plugins.octoprint_nanny.predictor")
 
+DETECTION_LABELS = {
+    1: "nozzle",
+    2: "adhesion",
+    3: "spaghetti",
+    4: "print",
+    5: "raft",
+}
+
+NEUTRAL_LABELS = {1: "nozzle", 5: "raft"}
+
+NEGATIVE_LABELS = {
+    2: "adhesion",
+    3: "spaghetti",
+}
+
+POSITIVE_LABELS = {
+    4: "print",
+}
+
 
 class ThreadLocalPredictor(threading.local):
     base_path = os.path.join(os.path.dirname(__file__), "data")
@@ -94,47 +115,42 @@ def write_image(self, outfile: str, image_np: np.ndarray):
     def percent_intersection(
         self,
         prediction: octoprint_nanny.types.BoundingBoxPrediction,
-        bb1: tuple,
+        area_of_interest: np,
     ) -> float:
         """
-        bb1 - boundary box
-        bb2 - detection box
+        Returns intersection-over-union area, normalized between 0 and 1
         """
-        detection_boxes = prediction.detection_boxes
-        detection_scores = prediction.detection_scores
-        detection_classes = prediction.detection_classes
 
+        detection_boxes = prediction.detection_boxes
+        # initialize array of zeroes
         aou = np.zeros(len(detection_boxes))
 
-        for i, bb2 in enumerate(detection_boxes):
-
-            # assert bb1[0] < bb1[2]
-            # assert bb1[1] < bb1[3]
-            # assert bb2[0] < bb2[2]
-            # assert bb2[1] < bb2[3]
-
+        # for each bounding box, calculate the intersection-over-area
+        for i, box in enumerate(detection_boxes):
             # determine the coordinates of the intersection rectangle
-            x_left = max(bb1[0], bb2[0])
-            y_top = max(bb1[1], bb2[1])
-            x_right = min(bb1[2], bb2[2])
-            y_bottom = min(bb1[3], bb2[3])
+            x_left = max(area_of_interest[0], box[0])
+            y_top = max(area_of_interest[1], box[1])
+            x_right = min(area_of_interest[2], box[2])
+            y_bottom = min(area_of_interest[3], box[3])
 
-            # boxes do not intersect
+            # boxes do not intersect, area is 0
             if x_right < x_left or y_bottom < y_top:
                 aou[i] = 0.0
                 continue
+
+            # calculate
             # The intersection of two axis-aligned bounding boxes is always an
             # axis-aligned bounding box
             intersection_area = (x_right - x_left) * (y_bottom - y_top)
 
             # compute the area of detection box
-            bb2_area = (bb2[2] - bb2[0]) * (bb2[3] - bb2[1])
+            box_area = (box[2] - box[0]) * (box[3] - box[1])
 
-            if (intersection_area / bb2_area) == 1.0:
+            if (intersection_area / box_area) == 1.0:
                 aou[i] = 1.0
                 continue
 
-            aou[i] = intersection_area / bb2_area
+            aou[i] = intersection_area / box_area
 
         return aou
 
@@ -257,8 +273,57 @@ def predict(self, image: PIL.Image) -> octoprint_nanny.types.BoundingBoxPredicti
 PREDICTOR = None
 
 
-@beeline.traced(name="predict_threadsafe")
-@beeline.traced_thread
+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])
+
+    df = df[["frame_id", "detection_classes", "detection_scores"]]
+    df = df.reset_index()
+
+    NUM_FRAMES = len(df)
+
+    # explode nested detection_scores and detection_classes series
+    df = df.set_index(["frame_id"]).apply(pd.Series.explode).reset_index()
+    assert len(df) == NUM_FRAMES * prediction.num_detections
+
+    # add string labels
+    df["label"] = df["detection_classes"].map(DETECTION_LABELS)
+
+    # create a hierarchal index from exploded data, append to dataframe state
+    return df.set_index(["frame_id", "label"])
+
+
+def print_is_healthy(df: pd.DataFrame, degree: int = 1) -> float:
+    df = pd.concat(
+        {
+            "unhealthy": df[df["detection_classes"].isin(NEGATIVE_LABELS)],
+            "healthy": df[df["detection_classes"].isin(POSITIVE_LABELS)],
+        }
+    ).reset_index()
+
+    mask = df.level_0 == "unhealthy"
+    healthy_cumsum = np.log10(
+        df[~mask].groupby("frame_id")["detection_scores"].sum().cumsum()
+    )
+
+    unhealthy_cumsum = np.log10(
+        df[mask].groupby("frame_id")["detection_scores"].sum().cumsum()
+    )
+
+    xy = healthy_cumsum.subtract(unhealthy_cumsum, fill_value=0)
+
+    if len(xy) == 1:
+        return True
+
+    trend = np.polynomial.polynomial.Polynomial.fit(xy.index, xy, degree)
+    slope, intercept = list(trend)
+    if slope < 0:
+        return False
+    return True
+
+
 def predict_threadsafe(
     image_bytes: bytes, **kwargs
 ) -> Tuple[
diff --git a/octoprint_nanny/workers/monitoring.py b/octoprint_nanny/workers/monitoring.py
index 463ebdfb..4b0aaea7 100644
--- a/octoprint_nanny/workers/monitoring.py
+++ b/octoprint_nanny/workers/monitoring.py
@@ -9,6 +9,7 @@
 import functools
 from datetime import datetime
 import numpy as np
+import pandas as pd
 import json
 import threading
 from enum import Enum
@@ -31,6 +32,9 @@
 from octoprint_nanny.predictor import (
     ThreadLocalPredictor,
     predict_threadsafe,
+    print_is_healthy,
+    DETECTION_LABELS,
+    explode_prediction_df,
 )
 from octoprint_nanny.clients.honeycomb import HoneycombTracer
 from octoprint.events import Events
@@ -89,10 +93,12 @@ def __init__(
         self._pn_ws_queue = pn_ws_queue
         self._mqtt_send_queue = mqtt_send_queue
 
+        self._trace_context = trace_context
         self._honeycomb_tracer = HoneycombTracer(service_name="octoprint_plugin")
         self._honeycomb_tracer.add_global_context(trace_context)
 
         self._halt = halt
+        self._df = pd.DataFrame()
 
     @beeline.traced(name="MonitoringWorker.load_url_buffer")
     async def load_url_buffer(self):
@@ -127,6 +133,11 @@ def calc_calibration(x0, y0, x1, y1, height=480, width=640):
 
         return {"mask": mask, "coords": (x0, y0, x1, y1)}
 
+    @beeline.traced(name="MonitoringWorker.update_dataframe")
+    def update_dataframe(self, ts, prediction):
+        self._df = self._df.append(explode_prediction_df(ts, prediction))
+        return self._df
+
     def _producer_worker(self):
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
@@ -225,7 +236,33 @@ async def _lite_loop(self, loop, pool):
             predict_threadsafe, image_bytes, **self._predictor_kwargs
         )
 
+        # trace predict latency, including serialization in/out of process pool
+        span = self._honeycomb_tracer.start_span(
+            context={
+                "name": "predict_pooled",
+            }
+        )
         raw_frame, post_frame, prediction = await loop.run_in_executor(pool, func)
+        self._honeycomb_tracer.add_context({"prediction": prediction})
+        self._honeycomb_tracer.finish_span(span)
+
+        # trace metrics calculations in/out of process pool
+        self.update_dataframe(ts, prediction)
+
+        # trace polyfit curve
+        span = self._honeycomb_tracer.start_span(
+            context={
+                "name": "print_is_healthy",
+            }
+        )
+        func = functools.partial(print_is_healthy)
+        healthy = await loop.run_in_executor(pool, func)
+        if healthy is False:
+            alert = await self._plugin.settings.rest_client.create_defect_alert(
+                octoprint_device=self._plugin.settings.device_id
+            )
+            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)
 
@@ -256,7 +293,6 @@ async def _loop(self, loop, pool):
             logger.error(f"Unsupported monitoring_mode={self._monitoring_mode}")
             return
 
-    @beeline.traced(name="MonitoringWorker._producer")
     async def _producer(self):
         """
         Calculates prediction and publishes result to subscriber queues
diff --git a/setup.py b/setup.py
index 831dd39c..78a147a8 100644
--- a/setup.py
+++ b/setup.py
@@ -133,6 +133,7 @@ class CPUNotSupported(Exception):
     "honeycomb-beeline",
     "aiofiles>=0.6.0",
     "flatbuffers>=1.12",
+    "pandas>=1.2.2",
 ]
 
 extra_requires = {


Test patch:
diff --git a/tests/test_predictor.py b/tests/test_predictor.py
index c0c75fbd..34a58261 100644
--- a/tests/test_predictor.py
+++ b/tests/test_predictor.py
@@ -2,11 +2,13 @@
 import os
 from PIL import Image as PImage
 import numpy as np
-
+import pandas as pd
 import octoprint_nanny.types
 from octoprint_nanny.predictor import (
     ThreadLocalPredictor,
     predict_threadsafe,
+    print_is_healthy,
+    explode_prediction_df,
 )
 
 
@@ -96,3 +98,77 @@ def test_area_of_intersection_prediction_contained_0():
     percent_area = predictor.percent_intersection(prediction, calibration_box)
     expected = 1.0
     np.testing.assert_almost_equal(percent_area[0], expected)
+
+
+def test_print_health_trend_increasing():
+    num_detections = 40
+    prediction1 = octoprint_nanny.types.BoundingBoxPrediction(
+        detection_classes=np.repeat(4, num_detections),
+        detection_scores=np.linspace(0, 0.6, num=num_detections),
+        num_detections=num_detections,
+        detection_boxes=np.repeat([0.1, 0.1, 0.8, 0.8], num_detections),
+    )
+    prediction2 = octoprint_nanny.types.BoundingBoxPrediction(
+        detection_classes=np.repeat(4, num_detections),
+        detection_scores=np.linspace(0.7, 1, num=num_detections),
+        num_detections=num_detections,
+        detection_boxes=np.repeat([0.1, 0.1, 0.8, 0.8], num_detections),
+    )
+    df = explode_prediction_df(1234, prediction1)
+    df = df.append(explode_prediction_df(2345, prediction2))
+
+    assert print_is_healthy(df) == True
+
+
+def test_print_health_trend_decreasing():
+    num_detections = 40
+    classes = np.concatenate(
+        [
+            np.repeat(4, num_detections // 2),  # print
+            np.repeat(3, num_detections // 2),  # spaghetti
+        ]
+    )
+
+    scores1 = np.concatenate(
+        [
+            np.linspace(0, 0.6, num=num_detections // 2),  # print
+            np.linspace(0.4, 0.8, num=num_detections // 2),  # spaghetti
+        ]
+    )
+
+    scores2 = np.concatenate(
+        [
+            np.linspace(0, 0.7, num=num_detections // 2),  # print
+            np.linspace(0.6, 0.9, num=num_detections // 2),  # spaghetti
+        ]
+    )
+
+    prediction1 = octoprint_nanny.types.BoundingBoxPrediction(
+        detection_classes=classes,
+        detection_scores=scores1,
+        num_detections=num_detections,
+        detection_boxes=np.repeat([0.1, 0.1, 0.8, 0.8], num_detections),
+    )
+    prediction2 = octoprint_nanny.types.BoundingBoxPrediction(
+        detection_classes=classes,
+        detection_scores=scores2,
+        num_detections=num_detections,
+        detection_boxes=np.repeat([0.1, 0.1, 0.8, 0.8], num_detections),
+    )
+    df = explode_prediction_df(1234, prediction1)
+    df = df.append(explode_prediction_df(2345, prediction2))
+
+    assert print_is_healthy(df) == False
+
+
+def test_print_health_trend_initial():
+    num_detections = 40
+    prediction = octoprint_nanny.types.BoundingBoxPrediction(
+        detection_classes=np.repeat(4, num_detections),
+        detection_scores=np.linspace(0, 1, num=num_detections),
+        num_detections=num_detections,
+        detection_boxes=np.repeat([0.1, 0.1, 0.8, 0.8], num_detections),
+    )
+    df = explode_prediction_df(1234, prediction)
+
+    assert print_is_healthy(df) == True