monster/yourovo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
ce5b2b5db5
yourovo/mac/daemon.py
type-two ce5b2b5db5 fix(urovo): SET RFID OFF after our encode so the firmware stops voiding the label 
We encode and verify the chip ourselves before PRINT, so the print job carries no
RFID data. With the firmware's own encoding armed it treats that as a failed
encode and VOIDs the label -- 'VOID VOID VOID' overprinted across the bottom
third, on top of the info line and the QR. Seen on the bench.

SET RFID OFF between our verify and the print stops it. Proven live that it is
free: UHF READ and UHF WRITE both keep working with the firmware pass off, and
the module read the next blank straight after the print. So it is never
re-armed -- no SET RFID 1, no &DEFAULT, no &CALIBRATE, all of which have wedged
the module today.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-06 14:33:06 +10:00

408 lines
18 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:
# ENCODE BEFORE GEOMETRY. Order matters on this firmware, and it cost most of a day:
# the SIZE/GAP/TEAR/DIRECTION/CLS burst that sets up a print silences the UHF module
# -- every UHF READ/QUERY afterwards returns no bytes at all until a power-cycle --
# so an encode placed after it always failed its own read-back with "before: null",
# which looks exactly like a chip that never reached the antenna. It had. The module
# was fine right up until the geometry went out (a standalone /read-tag a second
# earlier read the chip every time; SET RIBBON OFF + FORMFEED provably do not kill
# it). UHF WRITE commits immediately and needs no geometry, and printing works fine
# AFTER geometry, so: read, write, verify on the untouched module first; only then
# set up the print. Nothing feeds in between, so the label at the antenna is the one
# that prints.
before = verified = None
if encode:
# Deliberately NO `SET RFID` here. It was tried as an "arm the radio first" step and
# made things worse: on this firmware a SET RFID sent while the module is already
# active silenced every UHF command that followed (READ/QUERY returned no bytes at
# all, all tag types, until a power-cycle) -- and that looks identical to a media
# alignment problem. The printer arms RFID itself on power-up; leave that alone.
# Likewise never send &DEFAULT,i from here (README warns it re-arms/resets RFID).
# 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'}
# DISARM the firmware's own RFID pass before printing. We have already encoded and
# verified the chip ourselves (above), so the PRINT job carries no RFID data -- and with
# encoding armed the firmware treats that as a failed encode and VOIDs the label,
# overprinting "VOID VOID VOID" across the bottom third, on top of the info line and the
# QR. Seen on the bench. SET RFID OFF stops it, and it costs nothing: proven live that
# both UHF READ and UHF WRITE keep working with the firmware pass off, and the module
# stays alive for the next label. So it is simply never re-armed.
if encode:
p.cmd('SET RFID OFF')
time.sleep(0.4)
# 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 -- AFTER the encode, see above
geometry(p, prof)
p.cls()
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