Initial commit

app/scale.py

@@ -0,0 +1,169 @@
+"""USB HID scale reader.
+
+Supports any scale that implements the USB HID Weighing Device specification
+(Usage Page 0x8D). This covers most common shipping scales including:
+  - Dymo / Stamps.com USB scales (VID 0x0922)
+  - Mettler Toledo postal scales
+  - Generic USB postal scales
+
+Gracefully no-ops when the `hid` package is not installed or no scale is found,
+so the rest of the app runs normally on machines without a scale connected.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from typing import Callable
+
+log = logging.getLogger(__name__)
+
+try:
+    import hid as _hid
+    _HID_AVAILABLE = True
+except ImportError:
+    _hid = None  # type: ignore[assignment]
+    _HID_AVAILABLE = False
+
+# USB HID Usage Page for Weighing Devices
+_WEIGHING_USAGE_PAGE = 0x8D
+
+# HID weight unit codes
+_UNIT_GRAM   = 2
+_UNIT_OUNCE  = 11   # avoirdupois oz — most US postal scales report this
+_UNIT_POUND  = 12
+
+# HID status codes
+_STATUS_FAULT      = 1
+_STATUS_STABLE     = 2
+_STATUS_IN_MOTION  = 3
+_STATUS_OVER       = 4
+_STATUS_UNDER_ZERO = 5
+_STATUS_CALIBRATE  = 6
+_STATUS_TARE       = 7
+
+
+def is_available() -> bool:
+    """True if the hid package is importable."""
+    return _HID_AVAILABLE
+
+
+def find_scale() -> tuple[int, int] | None:
+    """Return (vendor_id, product_id) of the first HID weighing device found, or None."""
+    if not _HID_AVAILABLE:
+        return None
+    try:
+        for dev in _hid.enumerate():
+            if dev.get("usage_page") == _WEIGHING_USAGE_PAGE:
+                log.info(
+                    "Scale found: %s (VID=%04x PID=%04x)",
+                    dev.get("product_string", "Unknown"),
+                    dev["vendor_id"],
+                    dev["product_id"],
+                )
+                return dev["vendor_id"], dev["product_id"]
+    except Exception as exc:
+        log.warning("Scale enumeration error: %s", exc)
+    return None
+
+
+def read_weight_oz(vendor_id: int, product_id: int) -> float | None:
+    """Open the scale, read one report, return weight in ounces (or None on failure).
+
+    Returns None when the scale is in motion, over capacity, or not responding.
+    """
+    if not _HID_AVAILABLE:
+        return None
+    try:
+        with _hid.Device(vendor_id, product_id) as dev:
+            data = dev.read(64, timeout=2000)
+    except Exception as exc:
+        log.debug("Scale read error: %s", exc)
+        return None
+
+    if not data or len(data) < 6:
+        return None
+
+    # Standard HID scale report:
+    # byte 0: report ID
+    # byte 1: status
+    # byte 2: weight unit
+    # byte 3: exponent (signed)
+    # byte 4: weight LSB
+    # byte 5: weight MSB
+    status   = data[1]
+    unit     = data[2]
+    exponent = data[3] if data[3] < 128 else data[3] - 256   # treat as signed
+    raw      = (data[5] << 8) | data[4]
+    weight   = raw * (10.0 ** exponent)
+
+    if status not in (_STATUS_STABLE, _STATUS_IN_MOTION):
+        return None
+    if weight < 0:
+        return None
+
+    if unit == _UNIT_OUNCE:
+        return weight
+    if unit == _UNIT_GRAM:
+        return round(weight * 0.035274, 2)
+    if unit == _UNIT_POUND:
+        return round(weight * 16, 2)
+
+    log.debug("Unknown scale unit code: %d", unit)
+    return None
+
+
+class ScalePoller:
+    """Background thread that polls the scale and fires a callback with weight in oz.
+
+    Usage:
+        poller = ScalePoller(on_weight=my_callback)
+        poller.start()   # begin polling
+        poller.stop()    # stop polling
+    """
+
+    POLL_INTERVAL = 0.4   # seconds between reads
+
+    def __init__(self, on_weight: Callable[[float], None]) -> None:
+        self._on_weight   = on_weight
+        self._running     = False
+        self._thread: threading.Thread | None = None
+        self._device: tuple[int, int] | None = None
+
+    def start(self) -> bool:
+        """Start polling. Returns False if no scale is found."""
+        if self._running:
+            return True
+        self._device = find_scale()
+        if self._device is None:
+            log.warning("No USB scale found — auto-weigh unavailable.")
+            return False
+        self._running = True
+        self._thread  = threading.Thread(target=self._run, daemon=True, name="ScalePoller")
+        self._thread.start()
+        return True
+
+    def stop(self) -> None:
+        self._running = False
+
+    def read_once(self) -> float | None:
+        """Single blocking read — used by the Weigh button."""
+        dev = self._device or find_scale()
+        if dev is None:
+            return None
+        self._device = dev
+        return read_weight_oz(*dev)
+
+    def _run(self) -> None:
+        last: float | None = None
+        while self._running:
+            if self._device:
+                oz = read_weight_oz(*self._device)
+                if oz is not None and oz != last:
+                    last = oz
+                    try:
+                        self._on_weight(oz)
+                    except Exception:
+                        pass
+            time.sleep(self.POLL_INTERVAL)