monster/yourovo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
88f18d98a1
yourovo/mac/daemon.py
type-two 3a741079d6 fix(daemon): move to 7791 so the label service and the gun daemon coexist 
Four devices, two daemons, one front door. This service drives the Urovo D812R+
and the Dymo LabelWriter over raw USB; the Node rfid-daemon on 7790 drives the
Chafon H102 gun and the Dymo M25 scale, and is what the PriceGod extension is
pointed at. Both were binding 7790.

The port was not the whole problem. /status, /read-tag, /write-tag and /recover
exist on BOTH services and mean different things — the gun in your hand versus
the antenna inside the printer. Merging them into one flat namespace would let a
"read-tag" land on whichever daemon answered, which is how you get a tag written
by the wrong device. So this keeps its own names and the Node side proxies
/printer/* here, leaving the extension with a single endpoint on 7790.

They stay separate processes on purpose: this side needs libusb for a real
bidirectional pipe (UHF READ replies as raw bytes on the bulk-IN endpoint, which
is why CUPS 'lp -o raw' cannot drive it) and the other needs node-serialport and
node-hid. Merging would mean porting one language's device stack to the other
for no gain.

--port still wins, UROVO_PORT overrides the default, and neither the launchd
plist nor start-daemon.command pins a port, so install.sh needed no change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-04 10:32:42 +10:00

383 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
import os
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():
# 7791, not 7790. The Node rfid-daemon owns 7790: it is what the PriceGod extension talks to,
# it drives the Chafon gun and the Dymo scale, and it has the larger route surface. Both
# services binding one port is not the only clash either — /status, /read-tag, /write-tag and
# /recover exist on BOTH and mean different things (the gun versus this printer), so they can
# never be merged into one flat namespace safely.
#
# The Node daemon proxies /printer/* here instead, which keeps one front door on 7790 for the
# extension and leaves these route names unambiguous. Override with --port or UROVO_PORT.
port = int(os.environ.get('UROVO_PORT', 7791))
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