mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
2b1fe03159
The NAU7802 ADC returns a stale max-scale value (0x7FFFFF) on its first conversion after power-up, polluting the moving average and making the initial weight report wildly inaccurate. Flush the first reading during init(). Also extract both hardware drivers out of diagnostic scripts into proper daemon modules: - NAU7802 scale driver: scripts/scale_diag.py -> daemon/nau7802.py - PN5180 NFC driver: scripts/read_tag.py -> daemon/pn5180.py The production daemon was importing driver classes from test scripts since the original SpoolBuddy commit. Diagnostic scripts now import from the driver modules. Removed the sys.path hack from main.py.
101 lines
3.2 KiB
Python
101 lines
3.2 KiB
Python
"""Scale reader wrapper with stability detection and calibration."""
|
|
|
|
import logging
|
|
import time
|
|
from collections import deque
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MOVING_AVG_SIZE = 20
|
|
|
|
|
|
class ScaleReader:
|
|
def __init__(self, tare_offset: int = 0, calibration_factor: float = 1.0):
|
|
self._scale = None
|
|
self._tare_offset = tare_offset
|
|
self._calibration_factor = calibration_factor
|
|
self._samples: deque[float] = deque(maxlen=MOVING_AVG_SIZE)
|
|
self._stability_history: deque[tuple[float, float]] = deque(maxlen=20)
|
|
self._ok = False
|
|
self._last_raw = 0
|
|
|
|
try:
|
|
from .nau7802 import NAU7802
|
|
|
|
self._scale = NAU7802()
|
|
self._scale.init()
|
|
self._ok = True
|
|
bus_num = getattr(self._scale, "_bus_num", "?")
|
|
logger.info(
|
|
"Scale initialized on I2C bus %s (tare=%d, cal=%.6f)",
|
|
bus_num,
|
|
tare_offset,
|
|
calibration_factor,
|
|
)
|
|
except Exception as e:
|
|
logger.info("Scale not available: %s", e)
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return self._ok
|
|
|
|
@property
|
|
def last_raw(self) -> int:
|
|
return self._last_raw
|
|
|
|
def close(self):
|
|
try:
|
|
if self._scale:
|
|
self._scale.close()
|
|
except Exception:
|
|
pass
|
|
|
|
def update_calibration(self, tare_offset: int, calibration_factor: float):
|
|
self._tare_offset = tare_offset
|
|
self._calibration_factor = calibration_factor
|
|
logger.info("Calibration updated: tare=%d, factor=%.6f", tare_offset, calibration_factor)
|
|
|
|
def tare(self):
|
|
"""Set current raw reading as tare offset."""
|
|
if self._last_raw:
|
|
self._tare_offset = self._last_raw
|
|
self._samples.clear()
|
|
self._stability_history.clear()
|
|
logger.info("Tared at raw=%d", self._tare_offset)
|
|
return self._tare_offset
|
|
|
|
def read(self) -> tuple[float, bool, int] | None:
|
|
"""Read current weight. Returns (grams, stable, raw_adc) or None."""
|
|
try:
|
|
if not self._scale.data_ready():
|
|
return None
|
|
|
|
raw = self._scale.read_raw()
|
|
self._last_raw = raw
|
|
self._ok = True
|
|
|
|
grams = (raw - self._tare_offset) * self._calibration_factor
|
|
self._samples.append(grams)
|
|
|
|
# Moving average
|
|
avg_grams = sum(self._samples) / len(self._samples)
|
|
|
|
# Stability: track readings over time
|
|
now = time.monotonic()
|
|
self._stability_history.append((now, avg_grams))
|
|
|
|
# Stable if all readings within 1s window are within 2g of each other
|
|
stable = False
|
|
if len(self._stability_history) >= 5:
|
|
cutoff = now - 1.0
|
|
recent = [g for t, g in self._stability_history if t >= cutoff]
|
|
if len(recent) >= 3:
|
|
spread = max(recent) - min(recent)
|
|
stable = spread < 2.0
|
|
|
|
return round(avg_grams, 1), stable, raw
|
|
|
|
except Exception as e:
|
|
logger.debug("Scale read error: %s", e)
|
|
self._ok = False
|
|
return None
|