Monitor output from Dymo S100 USB-connected scale

monitor_dymo_s100.py

#! /usr/bin/env python3

# notes:
# - origin: https://steventsnyder.com/reading-a-dymo-usb-scale-using-python/
# - device support: https://github.com/pyusb/pyusb#requirements-and-platform-support
# - device permissions: https://stackoverflow.com/a/73357412/455641
# - kernel driver busy: https://ubuntuforums.org/archive/index.php/t-2044904.html


import usb.core
import usb.util
import time

STATE_STABLE_ZERO = 2
STATE_UNSTABLE_POSITIVE = 3
STATE_STABLE_POSITIVE = 4
STATE_NEGATIVE = 5  # returned both for stable and unstable values

UNITS_KG = 3
UNITS_LB = 12

SCALE_TENTHS = 255
SCALE_HUNDREDTHS = 256

SLEEP_NO_DEVICE = 1
SLEEP_STABLE = 1
SLEEP_UNSTABLE = 0.1

# from `lsusb`
DEVICE_CFG = {
    'idVendor': 0x0922,
    'idProduct': 0x8009
}


def connect_to_device(params):

    device = usb.core.find(**params)

    # was it found?
    if device is None:
        return None

    if device.is_kernel_driver_active(0):
        # need to free device from kernel driver before we can read it
        try:
            device.detach_kernel_driver(0)
        except usb.core.USBError:
            # print("Could not detatch kernel driver: %s" % str(e))
            return None

    # use the first/default configuration
    try:
        device.set_configuration()
    except Exception:
        return None

    return device


def parse_reading(data):

    state_flag = data[1]
    stable_states = [STATE_STABLE_ZERO, STATE_STABLE_POSITIVE]
    is_stable = state_flag in stable_states
    is_negative = state_flag == STATE_NEGATIVE

    scale_flag = data[3]
    if scale_flag == SCALE_TENTHS:
        scale_factor = 0.1
    elif scale_flag == SCALE_HUNDREDTHS:
        scale_factor = 0.01
    else:
        scale_factor = 100  # want an obviously wrong value

    raw_weight = data[4]
    weight = round(raw_weight * scale_factor, 1)
    if is_negative:
        weight = weight * -1

    unit_flag = data[2]
    if unit_flag == UNITS_KG:
        unit = 'kg'
    elif unit_flag == UNITS_LB:
        unit = 'lb'
    else:
        unit = unit_flag

    return {
        'is_stable': is_stable,
        'weight': weight,
        'unit': unit
    }


device = None
previous_reading = None

while True:
    if not device:
        device = connect_to_device(DEVICE_CFG)

    try:
        # first endpoint
        endpoint = device[0][(0, 0)][0]

        data = device.read(endpoint.bEndpointAddress,
                           endpoint.wMaxPacketSize)
    except Exception:
        device = None
        previous_reading = None
        time.sleep(SLEEP_NO_DEVICE)
        continue

    if data:
        reading = parse_reading(data)
        if reading != previous_reading:
            # here's where we can output our reading (push to a queue or service or etc.)
            # ...but just printing it for this example
            print(reading)

            previous_reading = reading

        if reading['is_stable']:
            time.sleep(SLEEP_STABLE)
        else:
            time.sleep(SLEEP_UNSTABLE)