monster/yourovo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
d4b16c075b
yourovo/mac/daemon.py
type-two d254d16278 Fix Dymo label geometry: rotation and 300x600 dpi halving 
Two bugs found on the bench, both from wrong assumptions about the LW450:

1. Orientation. I assumed the 57mm head meant the label feeds landscape. It
   doesn't -- the Dymo takes the SAME 24mm narrow-edge-first stock as the Urovo,
   so the landscape design needs the identical 90-degree rotation. Rendering
   656 dots onto a ~288-dot label also silently clipped the price and QR off the
   right-hand edge. Now rotates by default, overridable per request via "rotate",
   and the daemon rejects an image wider than the head instead of clipping it.

2. Half-length labels. print_image() sent ESC i (300x600 Barcode/Graphics mode),
   where the head stays 300 dpi across but each raster line becomes 1/600" in the
   travel direction -- so a 650-line label printed at 27.5mm instead of 55mm, with
   the across-head dimension correct. Default is now ESC h (300x300), matching our
   square 300 dpi bitmap. graphics_mode=True remains available and now stretches
   the image 2x vertically so physical size stays correct.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-22 16:39:58 +10:00

374 lines
15 KiB
Python
Raw Blame History

#!/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,
web_mm=body.get('w'), pitch_mm=body.get('h'),
margin_mm=body.get('margin'))
# The Dymo takes the SAME 24 mm stock as the Urovo, fed narrow-edge-first, so the
# landscape design has to be rotated 90 degrees here too. (The 57 mm head is wider
# than the label; that does NOT mean the label feeds landscape.) Overridable per
# request via "rotate" in case a wider Dymo roll is loaded.
rotate = body.get('rotate', True)
img = label.render(body, prof, rotate=rotate)
if img.width > dymo_mod.HEAD_BYTES * 8:
return {'ok': False, 'error': 'label is %d dots wide; Dymo head is %d'
% (img.width, dymo_mod.HEAD_BYTES * 8)}
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()
Reference in New Issue View Git Blame Copy Permalink