Adding set_rfid(tag_type=1) ahead of the read-back seemed like cheap insurance
and was the opposite. On this firmware a SET RFID sent while the UHF module is
already active left it answering NOTHING -- READ and QUERY returned no bytes at
all under every tag type, while model()/status() kept working -- and that state
survived every software command tried. It presents exactly like a media
alignment fault ("before: null", chip apparently never in the window) on a
roll whose inlay fills the label. The printer arms RFID itself at power-up.
Leave it. set_rfid()'s default is corrected to 1 (0 is OFF) for any caller that
does use it deliberately.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
329 lines
13 KiB
Python
329 lines
13 KiB
Python
#!/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):
|
|
"""Drain stale replies. Without this a slow answer from the previous command
|
|
gets attributed to the next one, which silently corrupts tag reads."""
|
|
while self.read(0.25):
|
|
pass
|
|
|
|
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. 0x00 = Ready.
|
|
|
|
Asks the one-byte <ESC>!? first. After &CALIBRATE,A,R this firmware stops answering
|
|
that form entirely -- the printer sat "no response" for minutes while model() kept
|
|
working over the same pipe -- and every caller that gates on status() (wait_ready,
|
|
recover, do_print) then refuses to print a printer that is in fact fine. So fall back to
|
|
the extended <ESC>!S, which it does keep answering: STX + 4 status bytes + ETX CR LF,
|
|
first byte 0x40 + code. That offset maps one-to-one onto STATUS_TEXT (0x41 head open,
|
|
0x48 ribbon latched, 0x40 ready), so the fallback returns the same codes as the primary.
|
|
Seen live: <ESC>!? silent, <ESC>!S -> 02 40 40 40 40 03 0D 0A = Ready."""
|
|
self.flush_in()
|
|
self.write(b'\x1b!?')
|
|
r = self.read(wait)
|
|
if r:
|
|
return r[-1]
|
|
self.flush_in()
|
|
self.write(b'\x1b!S')
|
|
r = self.read(wait)
|
|
i = r.find(b'\x02') if r else -1
|
|
if i >= 0 and len(r) > i + 1:
|
|
return (r[i + 1] - 0x40) & 0xFF
|
|
return 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=1, 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.
|
|
|
|
tag_type=1 is UHF Gen2. tag_type=0 is RFID OFF -- and it used to be the default here.
|
|
Seen live: after &DEFAULT,i the module went completely silent (UHF QUERY/READ returned
|
|
no bytes at all while model()/status() answered), reads failed at every media position,
|
|
and a 40 mm sweep found "no chip" on a roll whose inlay fills the label. SET RFID 1,...
|
|
brought it back on the first try. Never default to 0."""
|
|
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):
|
|
"""Parse a UHF READ/QUERY reply.
|
|
|
|
With dataFormat 'H' the printer answers in ASCII hex *text* -- e.g. it literally
|
|
sends b'E28069950000600625F600D0'. (Do NOT hex-encode it again; the SDK's own
|
|
byte.ToString("X2") loop is formatting a buffer it has already converted.)
|
|
An empty reply, all zeros, or a lone status byte such as 0xFC means no readable
|
|
tag at the antenna.
|
|
"""
|
|
if not raw or not any(raw):
|
|
return None
|
|
s = raw.decode('latin-1', 'replace').strip().strip('\x00').strip()
|
|
if len(s) < 4 or any(c not in '0123456789abcdefABCDEF' for c in s):
|
|
return None
|
|
return s.upper()
|
|
|
|
|
|
# ---- SKU <-> EPC ---------------------------------------------------------
|
|
# THE SHOP'S REAL SCHEME, matching the Chafon rfid-daemon (pliceclogs/rfid-daemon
|
|
# index.js buildEpcPayload/parseEpcData). The 96-bit EPC is ONE big-endian integer:
|
|
#
|
|
# value = sku(14 digits) * 10^9 + releaseId
|
|
#
|
|
# In decimal that reads as [digits 1-14 = SKU][digits 15-23 = releaseId], so any
|
|
# scanner that shows the EPC as hex can be converted to decimal and split at digit 14.
|
|
#
|
|
# NOTE: daemon.cs used a DIFFERENT layout (6 bytes SKU + 4 bytes releaseId + 0xEC01
|
|
# marker). That layout was never verified against a real tag and is NOT what the shop
|
|
# uses -- parseEpcData rejects it outright, so tags written that way are invisible to
|
|
# the extension. This module follows the Chafon scheme; the Windows daemon is the one
|
|
# that needs changing, not this.
|
|
RELID_FACTOR = 10 ** 9 # 9 digits: enough for any Discogs release id
|
|
_MAX_SKU = 10 ** 14
|
|
|
|
|
|
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 < RELID_FACTOR:
|
|
raise ValueError('releaseId must be under 9 digits')
|
|
return (int(sku14) * RELID_FACTOR + rel).to_bytes(12, 'big').hex().upper()
|
|
|
|
|
|
def epc_decode(hexstr):
|
|
"""Returns None for anything that is not one of ours -- a malformed read, or a
|
|
foreign/factory EPC such as E28069... whose value/10^9 exceeds 14 digits. That
|
|
guard is what stops a stray tag being reported to the operator as a real SKU."""
|
|
hexstr = (hexstr or '').strip()
|
|
if len(hexstr) != 24:
|
|
return None
|
|
try:
|
|
value = int.from_bytes(bytes.fromhex(hexstr), 'big')
|
|
except ValueError:
|
|
return None
|
|
sku_num = value // RELID_FACTOR
|
|
if sku_num >= _MAX_SKU:
|
|
return None # not a PriceGod SKU
|
|
rel = value % RELID_FACTOR
|
|
return {'sku': str(sku_num).zfill(14),
|
|
'releaseId': str(rel) if rel else None}
|
|
|
|
|
|
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)')
|