Jonah
170 lines
4.9 KiB
Python
170 lines
4.9 KiB
Python
"""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)
|