mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 19:21:33 +02:00
maziggy
eec7793955
Adds a Failure Detection tab under Settings that wires Bambuddy to a
self-hosted Obico ml_api container — no cloud, no account, no WebSocket.
While a print is running, the detection service periodically hands the
printer's camera snapshot URL to the ML API and smooths scores over
time (30-frame warmup + EWM, alpha=2/13, short/long rolling means) so
one noisy frame can't trigger an action. When the smoothed score
crosses HIGH, the configured action fires exactly once per print:
notify, pause, or pause-and-cut-power (via linked smart plugs).
- Backend: new obico_detection + obico_smoothing + obico_actions
services, /obico/status and /obico/test-connection routes
(SETTINGS_READ / SETTINGS_UPDATE), six obico_* AppSettings fields
with validators for sensitivity/action/enabled_printers.
- Frontend: FailureDetectionSettings component (enable, ML URL + test,
sensitivity, action, poll interval, per-printer monitor list, live
status + detection history), new sidebar tab with service-active
bullet, toast on save.
- Tests: 17 detection unit tests + 15 smoothing unit tests + 4
frontend component tests.
- Docs: README bullet, CHANGELOG entry, wiki page under Analytics,
website features.html entry.
106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
"""Temporal smoothing for Obico ML detection scores.
|
|
|
|
Ports Obico's failure-detection math:
|
|
- per-frame `current_p` = sum of detection confidences
|
|
- `ewm_mean` = exponentially weighted mean (alpha = 2 / (span + 1), span = 12)
|
|
- `rolling_mean_short` = ~310 frames of recent activity (≈52 min at 10s/frame)
|
|
- `rolling_mean_long` = ~7200 frames of long-term baseline noise
|
|
- First `WARMUP_FRAMES` frames always report "safe" while the state settles
|
|
- Final score = max(ewm_mean, rolling_mean_short - rolling_mean_long)
|
|
- Thresholds: LOW < score < HIGH is "warning", >= HIGH is "failure"
|
|
"""
|
|
|
|
import math
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
|
|
EWM_SPAN = 12
|
|
EWM_ALPHA = 2.0 / (EWM_SPAN + 1)
|
|
ROLLING_SHORT = 310
|
|
ROLLING_LONG = 7200
|
|
WARMUP_FRAMES = 30
|
|
|
|
# Base thresholds; sensitivity multipliers adjust them
|
|
BASE_LOW = 0.38
|
|
BASE_HIGH = 0.78
|
|
|
|
SENSITIVITY_MULT = {
|
|
"low": 1.25, # harder to trigger — higher thresholds
|
|
"medium": 1.0,
|
|
"high": 0.75, # easier to trigger — lower thresholds
|
|
}
|
|
|
|
|
|
def thresholds(sensitivity: str) -> tuple[float, float]:
|
|
mult = SENSITIVITY_MULT.get(sensitivity, 1.0)
|
|
return BASE_LOW * mult, BASE_HIGH * mult
|
|
|
|
|
|
@dataclass
|
|
class PrintState:
|
|
"""Per-print smoothing state. Reset when a new print starts."""
|
|
|
|
frame_count: int = 0
|
|
ewm_mean: float = 0.0
|
|
short_sum: float = 0.0
|
|
long_sum: float = 0.0
|
|
short_buf: deque = field(default_factory=lambda: deque(maxlen=ROLLING_SHORT))
|
|
long_buf: deque = field(default_factory=lambda: deque(maxlen=ROLLING_LONG))
|
|
|
|
def update(self, current_p: float) -> float:
|
|
"""Feed a new per-frame score and return the smoothed score.
|
|
|
|
Returns 0.0 during warmup so early noise doesn't trigger actions.
|
|
"""
|
|
self.frame_count += 1
|
|
|
|
if self.frame_count == 1:
|
|
self.ewm_mean = current_p
|
|
else:
|
|
self.ewm_mean = EWM_ALPHA * current_p + (1 - EWM_ALPHA) * self.ewm_mean
|
|
|
|
if len(self.short_buf) == self.short_buf.maxlen:
|
|
self.short_sum -= self.short_buf[0]
|
|
self.short_buf.append(current_p)
|
|
self.short_sum += current_p
|
|
|
|
if len(self.long_buf) == self.long_buf.maxlen:
|
|
self.long_sum -= self.long_buf[0]
|
|
self.long_buf.append(current_p)
|
|
self.long_sum += current_p
|
|
|
|
if self.frame_count <= WARMUP_FRAMES:
|
|
return 0.0
|
|
|
|
short_mean = self.short_sum / len(self.short_buf)
|
|
long_mean = self.long_sum / len(self.long_buf)
|
|
return max(self.ewm_mean, short_mean - long_mean)
|
|
|
|
|
|
def classify(score: float, sensitivity: str) -> str:
|
|
"""Return 'safe', 'warning', or 'failure' for a smoothed score."""
|
|
low, high = thresholds(sensitivity)
|
|
if score >= high:
|
|
return "failure"
|
|
if score >= low:
|
|
return "warning"
|
|
return "safe"
|
|
|
|
|
|
def score_from_detections(detections: list) -> float:
|
|
"""Sum confidences from the ML API `detections` array.
|
|
|
|
Each detection is `[label, confidence, [x, y, w, h]]`. We only care about
|
|
the confidence column — label is always "failure" for the single-class model.
|
|
"""
|
|
total = 0.0
|
|
for det in detections or []:
|
|
try:
|
|
value = float(det[1])
|
|
except (IndexError, TypeError, ValueError):
|
|
continue
|
|
if math.isnan(value) or math.isinf(value):
|
|
continue
|
|
total += value
|
|
return total
|