yourovo/mac/urovo.py

#!/usr/bin/env python3
"""
Urovo D812R+ (Gainscha GTSPL) driver for macOS -- raw TSPL over USB, no vendor SDK.

WHY THIS WORKS WITHOUT THE WINDOWS DLL
--------------------------------------
ASSESSMENT.md sec.7 assumed the RFID encode was locked inside GTSPL_SDK.dll and would need
a USB sniff or the Java jar to recover. It doesn't. The DLL is a managed .NET assembly and
decompiling its IL shows every RFID function is a one-line String.Concat -> ASCII ->
WritePrinter. The exact wire format, recovered from ldstr order + argument order:

    writeUHF(fmt, start, len, bank, data) -> 'UHF WRITE <fmt>,<start>,<len>,<bank>,"<data>"'
    readUHF(fmt, start, len, bank)        -> 'UHF READ <fmt>,<start>,<len>,<bank>'
    query_UHF(fmt, pcStatus, crcStatus)   -> 'UHF QUERY <fmt>,<pc>,<crc>'
    EPCPWD_Action(action, pw)             -> 'UHF GEN2 EPC <action>,"<pw>"'
    Set_RFIDPorcedure(tagType, rw_pos, void_printout, tryEncode, errHandle, speed, retry)
                                          -> 'SET RFID <..7 fields..>'   (positions in DOTS)
    rfidSetupDefault()                    -> '&DEFAULT,i'
    RFIDAutoCalibration()                 -> '&CALIBRATE,A,R'
    printlabel(set, copy)                 -> 'PRINT <set>, <copy>'       (note space after comma)

Everything is plain ASCII, one command per line, CRLF-terminated. So the whole print+encode
recipe is portable to any platform that can push bytes at the printer's bulk-OUT endpoint.

READ-BACK SEMANTICS (matters -- this is why CUPS 'lp -o raw' is not enough)
UHF READ / UHF QUERY do NOT reply with text. The printer returns raw bytes on the bulk-IN
endpoint and the SDK hex-formats them itself. A reply whose second byte is 0 is the SDK's
"no tag" case (it substitutes 22 zeros). We need a real bidirectional pipe, hence libusb.

Requires: pyusb + libusb  ->  brew install libusb && pip install pyusb
"""
import ctypes.util
import os
import time

import usb.backend.libusb1
import usb.core
import usb.util

VID, PID = 0x0471, 0x8E92          # Urovo D812R+ ("PRINTER"/"PRINTER", printer class, proto 2)

# Homebrew installs libusb where the dynamic loader won't look by default.
_LIBUSB_HINTS = [
    '/opt/homebrew/lib/libusb-1.0.dylib',
    '/usr/local/lib/libusb-1.0.dylib',
]


def _backend():
    for path in _LIBUSB_HINTS:
        if os.path.exists(path):
            b = usb.backend.libusb1.get_backend(find_library=lambda _p=path: _p)
            if b is not None:
                return b
    if ctypes.util.find_library('usb-1.0'):
        return usb.backend.libusb1.get_backend()
    raise RuntimeError('libusb not found -- run: brew install libusb')


# Status byte -> meaning. 0x08 ("out of ribbon") is a LATCH: SET RIBBON OFF alone will not
# clear it, it needs a FORMFEED as well. Verified on this printer from macOS.
STATUS_TEXT = {
    0x00: 'Ready',
    0x01: 'Head opened',
    0x02: 'Paper jam',
    0x03: 'Paper jam and head opened',
    0x04: 'Out of paper',
    0x05: 'Out of paper and head opened',
    0x08: 'Out of ribbon (thermal-transfer mode latched -- needs SET RIBBON OFF + FORMFEED)',
    0x09: 'Out of ribbon and head opened',
    0x10: 'Pause',
    0x20: 'Printing',
    0x80: 'Other error',
}


def status_text(code):
    if code is None:
        return 'no response'
    return STATUS_TEXT.get(code, 'busy/unknown (0x%02X)' % code)


class UrovoError(RuntimeError):
    pass


class Urovo(object):
    def __init__(self):
        self.dev = usb.core.find(idVendor=VID, idProduct=PID, backend=_backend())
        if self.dev is None:
            raise UrovoError('Urovo D812R+ (%04x:%04x) not found on USB' % (VID, PID))
        try:
            self.dev.set_configuration()
        except usb.core.USBError:
            pass                                    # already configured is fine
        intf = next(i for i in self.dev.get_active_configuration()
                    if i.bInterfaceClass == 7)      # USB printer class
        self.out = next(e for e in intf if usb.util.endpoint_direction(
            e.bEndpointAddress) == usb.util.ENDPOINT_OUT)
        self.inp = next((e for e in intf if usb.util.endpoint_direction(
            e.bEndpointAddress) == usb.util.ENDPOINT_IN), None)

    def close(self):
        try:
            usb.util.dispose_resources(self.dev)
        except Exception:
            pass

    def __enter__(self):
        return self

    def __exit__(self, *a):
        self.close()

    # ---- raw pipe --------------------------------------------------------
    def write(self, data):
        if isinstance(data, str):
            data = data.encode('latin-1', 'replace')
        self.out.write(data, timeout=5000)

    def read(self, wait=1.0, size=4096):
        if self.inp is None:
            return b''
        out, end = b'', time.time() + wait
        while time.time() < end:
            try:
                chunk = self.inp.read(size, timeout=300)
            except usb.core.USBTimeoutError:
                continue
            except usb.core.USBError:
                break
            if chunk:
                out += bytes(chunk)
                end = time.time() + 0.3
        return out

    def flush_in(self):
        self.read(0.25)

    def cmd(self, line, wait=0.0):
        """Send one TSPL line (CRLF appended); optionally collect a reply."""
        self.write(line + '\r\n')
        return self.read(wait) if wait else b''

    # ---- status ----------------------------------------------------------
    def status(self, wait=0.8):
        """Real-time status: <ESC>!? -> one byte. 0x00 = Ready."""
        self.flush_in()
        self.write(b'\x1b!?')
        r = self.read(wait)
        return r[-1] if r else None

    def model(self):
        self.flush_in()
        return self.cmd('~!T', 1.5).decode('latin-1', 'replace').strip()

    def wait_ready(self, timeout=10.0):
        """Poll until Ready. The status byte is unreliable for 1-3s after a print
        (returns transient garbage like 0x20/0x31/0x45), so never judge a print by
        the immediate post-print byte -- wait for it to settle to 0x00."""
        end, last = time.time() + timeout, None
        while time.time() < end:
            last = self.status(0.5)
            if last == 0x00:
                return 0x00
            time.sleep(0.25)
        return last

    def recover(self):
        """Clear the latched 'out of ribbon' fault. SET RIBBON OFF on its own does not
        do it -- it takes a FORMFEED too (costs one blank label)."""
        self.cmd('SET RIBBON OFF')
        time.sleep(0.4)
        if self.wait_ready(3.0) == 0x00:
            return 0x00
        self.cmd('FORMFEED')
        time.sleep(1.5)
        self.cmd('SET RIBBON OFF')
        time.sleep(0.4)
        return self.wait_ready(5.0)

    # ---- geometry / drawing (exact GTSPL wire format) --------------------
    def cls(self):                    self.cmd('CLS')
    def formfeed(self):               self.cmd('FORMFEED')
    def ribbon_off(self):             self.cmd('SET RIBBON OFF')
    def tear_on(self):                self.cmd('SET TEAR ON')
    def size(self, w, h):             self.cmd('SIZE %s mm,%s mm' % (w, h))
    def gap(self, g, off=0):          self.cmd('GAP %s mm,%s mm' % (g, off))
    def density(self, d):             self.cmd('DENSITY %s' % d)
    def direction(self, d):           self.cmd('DIRECTION %s' % d)
    def speed(self, s):               self.cmd('SPEED %s' % s)

    def text(self, x, y, font, rot, xmul, ymul, s):
        self.cmd('TEXT %s,%s,"%s",%s,%s,%s,"%s"' % (x, y, font, rot, xmul, ymul, _esc(s)))

    def qrcode(self, x, y, ecc, cell, mode, rot, s):
        self.cmd('QRCODE %s,%s,%s,%s,%s,%s,"%s"' % (x, y, ecc, cell, mode, rot, _esc(s)))

    def bar(self, x, y, w, h):
        self.cmd('BAR %s,%s,%s,%s' % (x, y, w, h))

    def printlabel(self, sets=1, copies=1):
        self.cmd('PRINT %s, %s' % (sets, copies))

    # ---- RFID ------------------------------------------------------------
    def rfid_setup_default(self):     self.cmd('&DEFAULT,i')
    def rfid_autocalibrate(self):     self.cmd('&CALIBRATE,A,R')

    def set_rfid(self, tag_type=0, rw_position=0, void_printout=0,
                 try_encode=3, error_handle=0, speed=1, retry=3):
        """SET RFID -- positions are in DOTS (203dpi = 8 dots/mm), never mm."""
        self.cmd('SET RFID %s,%s,%s,%s,%s,%s,%s' % (
            tag_type, rw_position, void_printout, try_encode, error_handle, speed, retry))

    def write_uhf(self, hexdata, fmt='H', start=2, length=12, bank='E'):
        """Stage an RFID encode. Fires together with the next PRINT, in one pass."""
        self.cmd('UHF WRITE %s,%s,%s,%s,"%s"' % (fmt, start, length, bank, hexdata))

    def read_uhf(self, fmt='H', start=2, length=12, bank='E', wait=2.5):
        self.flush_in()
        return _tag_hex(self.cmd('UHF READ %s,%s,%s,%s' % (fmt, start, length, bank), wait))

    def query_uhf(self, fmt='H', pc=0, crc=0, wait=2.5):
        self.flush_in()
        return _tag_hex(self.cmd('UHF QUERY %s,%s,%s' % (fmt, pc, crc), wait))


def _esc(s):
    """TSPL quotes strings with " and escapes with \\."""
    return str(s).replace('\\', '\\\\').replace('"', '\\"')


def _tag_hex(raw):
    """Mirror the SDK's response handling: raw bytes -> uppercase hex, and treat an
    all-zero / empty reply as 'no tag at the antenna' (the SDK's own sentinel)."""
    if not raw or not any(raw):
        return None
    return ''.join('%02X' % b for b in raw)


# ---- SKU <-> EPC (v1 scheme, ported byte-for-byte from daemon.cs Epc) -----
# 96-bit EPC = 6 bytes of the 14-digit timestamp SKU + 4 bytes release_id + 0xEC01 marker.
# UNVERIFIED against whatever the Chafon-written tags use. If it must change, change ONLY
# these two functions -- nothing else depends on the byte layout.
EPC_MARKER = b'\xec\x01'


def epc_encode(sku14, release_id):
    sku14 = (sku14 or '').strip()
    if len(sku14) != 14 or not sku14.isdigit():
        raise ValueError('sku must be 14 digits')
    rel = int(str(release_id or '0').strip() or '0')
    if not 0 <= rel <= 0xFFFFFFFF:
        raise ValueError('releaseId must fit in uint32')
    b = int(sku14).to_bytes(6, 'big') + rel.to_bytes(4, 'big') + EPC_MARKER
    return b.hex().upper()


def epc_decode(hexstr):
    hexstr = (hexstr or '').strip()
    if len(hexstr) != 24:
        return None
    try:
        b = bytes.fromhex(hexstr)
    except ValueError:
        return None
    if b[10:12] != EPC_MARKER:
        return None                                  # not our scheme
    return {'sku': str(int.from_bytes(b[:6], 'big')).zfill(14),
            'releaseId': str(int.from_bytes(b[6:10], 'big'))}


if __name__ == '__main__':
    with Urovo() as p:
        print('model  :', p.model())
        st = p.status()
        print('status : 0x%02X  %s' % (st, status_text(st)) if st is not None else 'status : none')
        print('tag    :', p.read_uhf() or '(no tag at antenna)')