2727545ac7
Answers "can we direct print to the Dymo via CUPS, or is it the cmd-P dance?":
neither. The LW450 (0922:0020) is a bidirectional USB printer-class device just
like the Urovo, so its raster protocol goes straight to the bulk endpoint -- no
DYMO driver, no CUPS queue, no print dialog. Command set taken from DYMO's own
LabelWriter 450 Series Technical Reference Manual and documented in dymo.py.
Verified on the bench: revision 1750111r53, status 0x03 Ready, 283 raster lines
at 82 bytes/line accepted.
Also corrects an assumption inherited from daemon.cs: UHF WRITE commits
immediately and does NOT wait for PRINT. Confirmed by writing a chip and reading
it back with no PRINT sent. Two consequences:
- the Urovo can program a chip without printing or consuming a label, which is
what makes the Urovo-encodes/Dymo-prints split work at all;
- an encode can be verified BEFORE printing, so a failed write never costs a
label. /print-encode and /encode both do this now.
New routes: /devices (all three devices at a glance), /encode (chip only),
/print-paper (Dymo), /tag-and-print (Mode A in one call).
label.py is now resolution-aware (203 dpi Urovo / 300 dpi Dymo via profile(dpmm=))
and orientation-aware: the Urovo's 24mm web feeds narrow-edge-first so the design
rotates 90 degrees, while the Dymo's 57mm head is wider than the label so the same
design feeds landscape unrotated.
Also fixes tag read parsing: with dataFormat 'H' the printer replies in ASCII hex
TEXT, not raw bytes -- the old code hex-encoded it a second time. flush_in() now
drains fully, since a late reply to one command was being attributed to the next
and silently corrupting reads.
KNOWN ISSUE: the Node rfid-daemon (Chafon + scale + inventory) also binds :7790.
Both cannot run at once and this service does not implement its routes. Unresolved.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
365 lines
14 KiB
Python
365 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
pricegod-urovo-daemon (macOS) -- localhost:7790 HTTP bridge to the Urovo D812R+.
|
|
|
|
The macOS counterpart to daemon.cs. Same port, same routes, same JSON shape, CORS *,
|
|
so it is a drop-in for the PriceGod extension's existing daemon.js client. The Windows
|
|
build needs GTSPL_SDK.dll; this one needs nothing but libusb, because the SDK turned out
|
|
to be a thin string formatter over a raw byte pipe (see urovo.py for the recovered
|
|
wire format).
|
|
|
|
Routes
|
|
GET /status detect + status -> {ok,ready,status,statusText,printer}
|
|
POST /print-encode print a label AND encode the chip in one pass
|
|
POST /write-tag alias of /print-encode (the daemon.js name)
|
|
POST /print-test print only, no RFID (safe on plain thermal stock)
|
|
GET /read-tag read the tag at the antenna -> {ok,epc,sku,releaseId}
|
|
GET /read-raw dump every memory bank
|
|
GET /preview?... PNG of the rendered label -- costs no labels
|
|
GET /profiles the media profiles this daemon knows
|
|
POST /calibrate RFID auto-calibration
|
|
POST /recover clear a latched fault
|
|
|
|
Run: ./venv/bin/python daemon.py (or ./start-daemon.command)
|
|
"""
|
|
import glob
|
|
import io
|
|
import json
|
|
import sys
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlparse, parse_qs
|
|
|
|
import label
|
|
import dymo as dymo_mod
|
|
from urovo import Urovo, UrovoError, epc_encode, epc_decode, status_text
|
|
|
|
VERSION = 'pricegod-label-daemon-mac/1.1'
|
|
DENSITY = 15 # max darkness -- anything lower barely marks
|
|
LOCK = threading.Lock() # one USB handle each; serialise all hardware access
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
def geometry(p, prof):
|
|
"""Apply media geometry. gap>0 = die-cut via the optical gap sensor;
|
|
gap=0 = continuous, for RFID rolls whose inlay blinds the gap sensor."""
|
|
p.tear_on()
|
|
p.gap(prof['gap_mm'], 0)
|
|
p.size(prof['web_mm'], prof['pitch_mm'])
|
|
p.density(DENSITY)
|
|
p.direction(1)
|
|
|
|
|
|
def do_print(body, encode):
|
|
prof = label.profile(body.get('size'),
|
|
web_mm=body.get('w'), pitch_mm=body.get('h'),
|
|
gap_mm=body.get('gap'))
|
|
sku = str(body.get('sku', '') or '').strip()
|
|
release_id = str(body.get('releaseId', '') or '').strip()
|
|
|
|
epc = None
|
|
if encode:
|
|
try:
|
|
epc = epc_encode(sku, release_id)
|
|
except ValueError as e:
|
|
return {'ok': False, 'error': str(e)}
|
|
|
|
with LOCK, Urovo() as p:
|
|
# 1) direct thermal. Wait out transient busy states before feeding to clear,
|
|
# so we don't waste a label on a momentary post-print code.
|
|
p.ribbon_off()
|
|
time.sleep(0.4)
|
|
if p.wait_ready(3.0) != 0x00:
|
|
st = p.recover()
|
|
if st != 0x00:
|
|
return {'ok': False, 'error': 'printer not ready',
|
|
'status': st, 'statusText': status_text(st)}
|
|
|
|
# 2) geometry, then the artwork
|
|
geometry(p, prof)
|
|
p.cls()
|
|
|
|
before = verified = None
|
|
if encode:
|
|
# UHF WRITE commits immediately -- it does NOT wait for PRINT (daemon.cs
|
|
# assumed it did). That lets us read the chip back and confirm the encode
|
|
# BEFORE spending a label on it.
|
|
before = p.read_uhf()
|
|
p.write_uhf(epc)
|
|
time.sleep(1.0)
|
|
verified = p.read_uhf() == epc
|
|
if not verified:
|
|
return {'ok': False, 'error': 'RFID encode failed verification',
|
|
'epc': epc, 'epcBefore': before, 'readBack': p.read_uhf(),
|
|
'note': 'nothing printed -- no label wasted'}
|
|
p.write(label.to_tspl(label.render(body, prof)))
|
|
|
|
# 4) re-assert ribbon-off immediately before firing
|
|
p.ribbon_off()
|
|
time.sleep(0.4)
|
|
p.printlabel(1, 1)
|
|
|
|
# 5) the status byte returns garbage for 1-3 s during the RFID write + feed,
|
|
# so poll until it settles rather than trusting the first read.
|
|
st = p.wait_ready(12.0)
|
|
after = p.read_uhf() if encode else None
|
|
p.ribbon_off()
|
|
|
|
ok = st == 0x00
|
|
return {
|
|
'ok': ok, 'sku': sku, 'releaseId': release_id, 'epc': epc,
|
|
'epcBefore': before, 'verified': verified, 'nextTagAtAntenna': after,
|
|
'status': st, 'statusText': status_text(st),
|
|
'profile': prof['name'], 'size': [prof['web_mm'], prof['pitch_mm']],
|
|
'mode': 'encode+print' if encode else 'thermal-test',
|
|
'note': ('printer returned to Ready (no VOID) -- verify the chip with a scan'
|
|
if ok else 'printer did NOT return to Ready -- check for VOID / jam / tag position'),
|
|
}
|
|
|
|
|
|
def do_encode(body):
|
|
"""Program the chip at the antenna and read it straight back. Prints nothing.
|
|
|
|
This is Mode A ('Separate'): the Urovo programs the chip while the paper label is
|
|
printed elsewhere (the Dymo). Possible because UHF WRITE commits on its own rather
|
|
than waiting for PRINT, so no label is consumed and the write is verifiable.
|
|
"""
|
|
sku = str(body.get('sku', '') or '').strip()
|
|
release_id = str(body.get('releaseId', '') or '').strip()
|
|
try:
|
|
epc = epc_encode(sku, release_id)
|
|
except ValueError as e:
|
|
return {'ok': False, 'error': str(e)}
|
|
|
|
with LOCK, Urovo() as p:
|
|
before = p.read_uhf()
|
|
if before is None:
|
|
return {'ok': False, 'error': 'no tag at the antenna',
|
|
'note': 'seat an RFID label at the print head and retry'}
|
|
p.write_uhf(epc)
|
|
time.sleep(1.0)
|
|
after = p.read_uhf()
|
|
|
|
ok = after == epc
|
|
return {'ok': ok, 'sku': sku, 'releaseId': release_id, 'epc': epc,
|
|
'epcBefore': before, 'epcAfter': after, 'verified': ok,
|
|
'decoded': epc_decode(after or ''),
|
|
'note': ('chip programmed and read back -- scan it to confirm' if ok
|
|
else 'read-back did NOT match; tag may be locked or out of position')}
|
|
|
|
|
|
def do_print_paper(body):
|
|
"""Print the paper label on the Dymo LabelWriter 450 (raw USB, no CUPS driver)."""
|
|
prof = label.profile(body.get('size'), dpmm=label.DPMM_DYMO,
|
|
art_mm=body.get('art_mm'), margin_mm=body.get('margin'))
|
|
img = label.render(body, prof, rotate=False) # 57 mm head: no rotation needed
|
|
with LOCK, dymo_mod.Dymo() as d:
|
|
before = d.status()
|
|
if before is not None and not before & 0x01:
|
|
return {'ok': False, 'status': before,
|
|
'statusText': dymo_mod.status_text(before),
|
|
'error': 'Dymo not ready'}
|
|
lines = d.print_image(img, density=body.get('density', 'normal'))
|
|
time.sleep(1.5)
|
|
st = d.status()
|
|
return {'ok': st is None or bool(st & 0x01), 'printer': 'dymo',
|
|
'status': st, 'statusText': dymo_mod.status_text(st),
|
|
'profile': prof['name'], 'lines': lines, 'bytesPerLine': img.width // 8}
|
|
|
|
|
|
def do_tag_and_print(body):
|
|
"""Mode A, one call: the Urovo programs the chip, the Dymo prints the paper.
|
|
|
|
Encode first and verify it, so a failed write never costs a printed label.
|
|
"""
|
|
out = {'ok': False}
|
|
enc = do_encode(body)
|
|
out['encode'] = enc
|
|
if not enc.get('ok'):
|
|
out['error'] = 'chip not programmed -- nothing printed'
|
|
return out
|
|
pr = do_print_paper(body)
|
|
out['print'] = pr
|
|
out['ok'] = bool(pr.get('ok'))
|
|
out['epc'] = enc.get('epc')
|
|
out['note'] = ('chip programmed and paper printed -- stick the paper over the chip'
|
|
if out['ok'] else 'chip programmed but the paper print failed')
|
|
return out
|
|
|
|
|
|
def do_devices():
|
|
"""One glance at all three pieces of hardware."""
|
|
out = {'ok': True, 'service': VERSION}
|
|
try:
|
|
with LOCK, Urovo() as p:
|
|
st = p.status()
|
|
out['urovo'] = {'present': True, 'model': p.model(), 'status': st,
|
|
'statusText': status_text(st), 'ready': st == 0x00}
|
|
except Exception as e:
|
|
out['urovo'] = {'present': False, 'error': str(e)}
|
|
try:
|
|
with LOCK, dymo_mod.Dymo() as d:
|
|
st = d.status()
|
|
out['dymo'] = {'present': True, 'revision': d.revision(), 'status': st,
|
|
'statusText': dymo_mod.status_text(st),
|
|
'ready': bool(st and st & 0x01)}
|
|
except Exception as e:
|
|
out['dymo'] = {'present': False, 'error': str(e)}
|
|
# The Chafon gun speaks its own serial protocol on a CH340 bridge. This daemon only
|
|
# reports the port; the Node rfid-daemon owns that protocol (see README).
|
|
ports = sorted(glob.glob('/dev/cu.usbserial*'))
|
|
out['chafon'] = {'present': bool(ports), 'ports': ports,
|
|
'handledBy': 'node rfid-daemon (not this service)'}
|
|
return out
|
|
|
|
|
|
def do_status():
|
|
with LOCK, Urovo() as p:
|
|
st = p.status()
|
|
return {'ok': True, 'ready': st == 0x00, 'status': st,
|
|
'statusText': status_text(st), 'printer': p.model(),
|
|
'service': VERSION}
|
|
|
|
|
|
def do_read_tag():
|
|
with LOCK, Urovo() as p:
|
|
epc = p.read_uhf()
|
|
out = {'ok': epc is not None, 'epc': epc}
|
|
if epc:
|
|
out.update(epc_decode(epc) or {})
|
|
out['recognized'] = epc_decode(epc) is not None
|
|
else:
|
|
out['error'] = 'no tag at the antenna'
|
|
return out
|
|
|
|
|
|
def do_read_raw():
|
|
with LOCK, Urovo() as p:
|
|
banks = {'query': p.query_uhf()}
|
|
for code, name in (('E', 'epc'), ('T', 'tid'), ('U', 'user')):
|
|
banks[name] = p.read_uhf(bank=code, start=0, length=12)
|
|
return {'ok': any(banks.values()), 'banks': banks}
|
|
|
|
|
|
def do_calibrate():
|
|
with LOCK, Urovo() as p:
|
|
p.rfid_autocalibrate()
|
|
st = p.wait_ready(20.0)
|
|
return {'ok': st == 0x00, 'status': st, 'statusText': status_text(st)}
|
|
|
|
|
|
def do_recover():
|
|
with LOCK, Urovo() as p:
|
|
st = p.recover()
|
|
return {'ok': st == 0x00, 'status': st, 'statusText': status_text(st)}
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
class Handler(BaseHTTPRequestHandler):
|
|
protocol_version = 'HTTP/1.1'
|
|
|
|
def log_message(self, fmt, *args):
|
|
sys.stdout.write('%s %s\n' % (time.strftime('%H:%M:%S'), fmt % args))
|
|
sys.stdout.flush()
|
|
|
|
def _cors(self):
|
|
self.send_header('Access-Control-Allow-Origin', '*')
|
|
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
|
|
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
|
|
|
|
def _json(self, obj, code=200):
|
|
raw = json.dumps(obj, default=str).encode()
|
|
self.send_response(code)
|
|
self._cors()
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.send_header('Content-Length', str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
|
|
def _png(self, img):
|
|
buf = io.BytesIO()
|
|
img.convert('L').save(buf, 'PNG')
|
|
raw = buf.getvalue()
|
|
self.send_response(200)
|
|
self._cors()
|
|
self.send_header('Content-Type', 'image/png')
|
|
self.send_header('Content-Length', str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
|
|
def do_OPTIONS(self):
|
|
self.send_response(204)
|
|
self._cors()
|
|
self.send_header('Content-Length', '0')
|
|
self.end_headers()
|
|
|
|
def do_GET(self):
|
|
u = urlparse(self.path)
|
|
path = u.path.rstrip('/') or '/'
|
|
q = {k: v[0] for k, v in parse_qs(u.query).items()}
|
|
try:
|
|
if path == '/preview':
|
|
prof = label.profile(q.get('size'))
|
|
return self._png(label.render_art(q or label.DEMO, prof))
|
|
if path == '/profiles':
|
|
return self._json({'ok': True, 'default': label.DEFAULT_PROFILE,
|
|
'profiles': label.PROFILES})
|
|
handler = {'/status': do_status, '/read-tag': do_read_tag,
|
|
'/read-raw': do_read_raw, '/devices': do_devices}.get(path)
|
|
if handler:
|
|
return self._json(handler())
|
|
if path == '/':
|
|
return self._json({'ok': True, 'service': VERSION})
|
|
self._json({'ok': False, 'error': 'unknown route ' + path}, 404)
|
|
except UrovoError as e:
|
|
self._json({'ok': False, 'error': str(e)}, 503)
|
|
except Exception as e:
|
|
self._json({'ok': False, 'error': '%s: %s' % (type(e).__name__, e)}, 500)
|
|
|
|
def do_POST(self):
|
|
path = urlparse(self.path).path.rstrip('/') or '/'
|
|
n = int(self.headers.get('Content-Length') or 0)
|
|
raw = self.rfile.read(n).decode('utf-8', 'replace') if n else ''
|
|
try:
|
|
body = json.loads(raw) if raw.strip() else {}
|
|
except ValueError:
|
|
body = {}
|
|
try:
|
|
if path in ('/print-encode', '/write-tag'):
|
|
return self._json(do_print(body, True))
|
|
if path == '/print-test':
|
|
return self._json(do_print(body, False))
|
|
if path == '/encode':
|
|
return self._json(do_encode(body))
|
|
if path == '/print-paper':
|
|
return self._json(do_print_paper(body))
|
|
if path == '/tag-and-print':
|
|
return self._json(do_tag_and_print(body))
|
|
handler = {'/calibrate': do_calibrate, '/recover': do_recover}.get(path)
|
|
if handler:
|
|
return self._json(handler())
|
|
self._json({'ok': False, 'error': 'unknown route ' + path}, 404)
|
|
except UrovoError as e:
|
|
self._json({'ok': False, 'error': str(e)}, 503)
|
|
except Exception as e:
|
|
self._json({'ok': False, 'error': '%s: %s' % (type(e).__name__, e)}, 500)
|
|
|
|
|
|
def main():
|
|
port = 7790
|
|
if '--port' in sys.argv:
|
|
port = int(sys.argv[sys.argv.index('--port') + 1])
|
|
srv = ThreadingHTTPServer(('127.0.0.1', port), Handler)
|
|
print('%s listening on http://localhost:%d/ (Ctrl+C to stop)' % (VERSION, port))
|
|
print('routes: /status /print-encode /write-tag /print-test /read-tag /read-raw '
|
|
'/preview /profiles /calibrate /recover')
|
|
try:
|
|
srv.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print('\nbye')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|