jessikitty/lego-instructions-manager
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
lego-instructions-manager/app/services/ipp_client.py
T
jessikittyandClaude Opus 5 deb009b80e Treat "busy" as retryable and pace jobs behind the printer 
The VC-500W accepts one job at a time; anything sent while the tape is
moving comes back server-error-busy. Add IPPBusy, wait_until_idle and
print_job_paced so a batch goes through instead of failing from the
second label on.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Vunz4VLxAYDBnU5tLciLDC
2026-10-13 10:27:14 +11:00

401 lines
13 KiB
Python
Raw Permalink Blame History

"""
A small IPP client, enough to drive an AirPrint label printer.
The Brother VC-500W is AirPrint-enabled, which means it speaks IPP over port
631. That is a documented, open protocol, so LIM can send a job to the
printer directly over the network - no Brother driver, no P-touch Editor, no
"save a PDF and print it by hand".
Brother publish no raw command protocol for the VC-500W, so this does NOT
try to reverse-engineer one. It speaks standard IPP:
Get-Printer-Attributes (0x000B) - what are you, what can you accept
Print-Job (0x0002) - here is a document, print it
What a given printer will accept as a document format varies, so always
probe first (the Test connection button) and send a format the printer
actually lists in document-format-supported.
Reference: RFC 8010 (encoding) and RFC 8011 (semantics).
Created by: Jess Rogerson (yelling commands at Claude.AI)
"""
import logging
import random
import struct
import time
import requests
logger = logging.getLogger(__name__)
DEFAULT_PORT = 631
TIMEOUT = 25
# Operation ids
OP_PRINT_JOB = 0x0002
OP_GET_PRINTER_ATTRIBUTES = 0x000B
# Delimiter tags
TAG_OPERATION_ATTRIBUTES = 0x01
TAG_JOB_ATTRIBUTES = 0x02
TAG_END_OF_ATTRIBUTES = 0x03
TAG_PRINTER_ATTRIBUTES = 0x04
TAG_UNSUPPORTED_ATTRIBUTES = 0x05
# Value tags
TAG_INTEGER = 0x21
TAG_BOOLEAN = 0x22
TAG_ENUM = 0x23
TAG_OCTET_STRING = 0x30
TAG_RESOLUTION = 0x32
TAG_RANGE_OF_INTEGER = 0x33
TAG_TEXT = 0x41
TAG_NAME = 0x42
TAG_KEYWORD = 0x44
TAG_URI = 0x45
TAG_CHARSET = 0x47
TAG_NATURAL_LANGUAGE = 0x48
TAG_MIME_MEDIA_TYPE = 0x49
TEXTUAL_TAGS = {TAG_TEXT, TAG_NAME, TAG_KEYWORD, TAG_URI, TAG_CHARSET,
TAG_NATURAL_LANGUAGE, TAG_MIME_MEDIA_TYPE}
INTEGER_TAGS = {TAG_INTEGER, TAG_ENUM}
# Document formats we can actually produce, best first. The printer's own
# document-format-supported list decides which of these gets used.
PREFERRED_FORMATS = ['image/jpeg', 'image/png', 'application/pdf']
class IPPError(Exception):
"""Anything that stopped us talking to the printer."""
class IPPBusy(IPPError):
"""
The printer refused the job because it is still working.
Worth separating from a real failure: a ZINK tape printer runs at
about 7.5mm/s, so a 15mm label takes a couple of seconds plus feed and
cut. Fire a batch at it with no pacing and everything after the first
job comes back busy. The cure is to wait and try again, not to give up.
"""
# Statuses that mean "not now" rather than "no".
RETRYABLE_STATUSES = {
0x0503, # server-error-service-unavailable
0x0507, # server-error-busy
}
def _encode_attribute(tag, name, value):
"""One IPP attribute: tag, name, value. Empty name = extra set member."""
if isinstance(value, bool):
payload = b'\x01' if value else b'\x00'
elif isinstance(value, int):
payload = struct.pack('>i', value)
else:
payload = str(value).encode('utf-8')
name_bytes = name.encode('utf-8') if name else b''
return (struct.pack('>B', tag)
+ struct.pack('>H', len(name_bytes)) + name_bytes
+ struct.pack('>H', len(payload)) + payload)
def _printer_uri(host, port=DEFAULT_PORT, path='/ipp/print'):
if not path.startswith('/'):
path = '/' + path
suffix = '' if port == DEFAULT_PORT else f':{port}'
return f'ipp://{host}{suffix}{path}'
def _http_url(host, port=DEFAULT_PORT, path='/ipp/print'):
if not path.startswith('/'):
path = '/' + path
return f'http://{host}:{port}{path}'
def _build_request(operation, host, port, path, user='lim',
extra_operation=(), job_attributes=(), document=b''):
"""Assemble a complete IPP request."""
request_id = random.randint(1, 0x7FFFFFFF)
out = bytearray()
out += struct.pack('>HHI', 0x0200, operation, request_id) # IPP 2.0
out += struct.pack('>B', TAG_OPERATION_ATTRIBUTES)
out += _encode_attribute(TAG_CHARSET, 'attributes-charset', 'utf-8')
out += _encode_attribute(TAG_NATURAL_LANGUAGE,
'attributes-natural-language', 'en')
out += _encode_attribute(TAG_URI, 'printer-uri',
_printer_uri(host, port, path))
out += _encode_attribute(TAG_NAME, 'requesting-user-name', user)
for tag, name, value in extra_operation:
out += _encode_attribute(tag, name, value)
if job_attributes:
out += struct.pack('>B', TAG_JOB_ATTRIBUTES)
for tag, name, value in job_attributes:
out += _encode_attribute(tag, name, value)
out += struct.pack('>B', TAG_END_OF_ATTRIBUTES)
out += document
return bytes(out), request_id
def parse_response(data):
"""
Decode an IPP response into {status, status_name, attributes}.
attributes maps name -> list of values. Unknown value types come back as
raw bytes rather than being dropped, so a surprising printer still
produces something readable in the log.
"""
if len(data) < 8:
raise IPPError('Printer sent a truncated IPP response.')
_version, status, _request_id = struct.unpack('>HHI', data[:8])
attributes = {}
offset = 8
current = None
while offset < len(data):
tag = data[offset]
offset += 1
if tag == TAG_END_OF_ATTRIBUTES:
break
if tag <= 0x05: # a delimiter: start of a new attribute group
current = None
continue
if offset + 2 > len(data):
break
(name_len,) = struct.unpack('>H', data[offset:offset + 2])
offset += 2
name = data[offset:offset + name_len].decode('utf-8', 'replace')
offset += name_len
if offset + 2 > len(data):
break
(value_len,) = struct.unpack('>H', data[offset:offset + 2])
offset += 2
raw = data[offset:offset + value_len]
offset += value_len
if tag in TEXTUAL_TAGS:
value = raw.decode('utf-8', 'replace')
elif tag in INTEGER_TAGS and len(raw) == 4:
value = struct.unpack('>i', raw)[0]
elif tag == TAG_BOOLEAN and len(raw) == 1:
value = bool(raw[0])
elif tag == TAG_RESOLUTION and len(raw) == 9:
cross, feed, units = struct.unpack('>iiB', raw)
value = f'{cross}x{feed}' + ('dpi' if units == 3 else 'dpcm')
else:
value = raw
if name:
current = name
attributes.setdefault(name, []).append(value)
elif current:
attributes[current].append(value)
return {
'status': status,
'ok': status < 0x0100,
'status_name': STATUS_NAMES.get(status, f'0x{status:04x}'),
'attributes': attributes,
}
STATUS_NAMES = {
0x0000: 'successful-ok',
0x0001: 'successful-ok-ignored-or-substituted-attributes',
0x0002: 'successful-ok-conflicting-attributes',
0x0400: 'client-error-bad-request',
0x0401: 'client-error-forbidden',
0x0402: 'client-error-not-authenticated',
0x0403: 'client-error-not-authorized',
0x0405: 'client-error-not-found',
0x040A: 'client-error-document-format-not-supported',
0x040B: 'client-error-attributes-or-values-not-supported',
0x0500: 'server-error-internal-error',
0x0501: 'server-error-operation-not-supported',
0x0503: 'server-error-service-unavailable',
0x0507: 'server-error-busy',
0x0508: 'server-error-job-canceled',
}
def _post(host, port, path, body):
url = _http_url(host, port, path)
try:
response = requests.post(
url, data=body, timeout=TIMEOUT,
headers={'Content-Type': 'application/ipp'})
except requests.RequestException as exc:
raise IPPError(f'Could not reach the printer at {host}:{port} - {exc}')
if response.status_code != 200:
raise IPPError(f'Printer returned HTTP {response.status_code} '
f'from {url}.')
return parse_response(response.content)
def get_printer_attributes(host, port=DEFAULT_PORT, path='/ipp/print'):
"""
Ask the printer what it is and what it accepts.
Returns a tidy summary dict. Raises IPPError if it cannot be reached.
"""
body, _ = _build_request(
OP_GET_PRINTER_ATTRIBUTES, host, port, path,
extra_operation=[
(TAG_KEYWORD, 'requested-attributes', 'all'),
])
result = _post(host, port, path, body)
attributes = result['attributes']
def first(name, default=None):
values = attributes.get(name)
return values[0] if values else default
def all_of(name):
return [v for v in attributes.get(name, []) if isinstance(v, str)]
formats = all_of('document-format-supported')
return {
'ok': result['ok'],
'status': result['status_name'],
'name': first('printer-name'),
'model': first('printer-make-and-model'),
'state': PRINTER_STATES.get(first('printer-state'),
first('printer-state')),
'state_message': first('printer-state-message'),
'reasons': all_of('printer-state-reasons'),
'formats': formats,
'resolutions': all_of('printer-resolution-supported'),
'media': all_of('media-supported')[:40],
'usable_format': pick_format(formats),
'uri': _printer_uri(host, port, path),
}
PRINTER_STATES = {3: 'idle', 4: 'processing', 5: 'stopped'}
def pick_format(supported):
"""Best document format we can produce that the printer also accepts."""
if not supported:
return None
for candidate in PREFERRED_FORMATS:
if candidate in supported:
return candidate
return None
def print_job(host, document, mime_type, job_name='LEGO part labels',
port=DEFAULT_PORT, path='/ipp/print', copies=1, user='lim'):
"""
Send one document to the printer. Returns {job_id, state, status_name}.
Raises IPPError on anything that stopped the job being accepted,
including the printer rejecting the document format.
"""
operation_extras = [
(TAG_MIME_MEDIA_TYPE, 'document-format', mime_type),
(TAG_NAME, 'job-name', job_name[:120]),
]
job_attributes = []
if copies and copies > 1:
job_attributes.append((TAG_INTEGER, 'copies', int(copies)))
body, _ = _build_request(
OP_PRINT_JOB, host, port, path, user=user,
extra_operation=operation_extras,
job_attributes=job_attributes,
document=document)
result = _post(host, port, path, body)
attributes = result['attributes']
if not result['ok']:
detail = result['status_name']
if result['status'] in RETRYABLE_STATUSES:
raise IPPBusy(f'Printer is busy ({detail}).')
if result['status'] == 0x040A:
detail += (f' - the printer will not accept {mime_type}. '
'Run Test connection to see what it does accept.')
raise IPPError(f'Printer rejected the job: {detail}')
job_id = attributes.get('job-id', [None])[0]
job_state = attributes.get('job-state', [None])[0]
return {
'job_id': job_id,
'state': JOB_STATES.get(job_state, job_state),
'status_name': result['status_name'],
}
JOB_STATES = {
3: 'pending', 4: 'pending-held', 5: 'processing', 6: 'processing-stopped',
7: 'canceled', 8: 'aborted', 9: 'completed',
}
# --------------------------------------------------------------- pacing
IDLE_STATES = ('idle',)
BUSY_STATES = ('processing', 'stopped')
def wait_until_idle(host, port=DEFAULT_PORT, path='/ipp/print',
timeout=120.0, poll=1.0):
"""
Block until the printer reports idle again, or the timeout runs out.
Returns True if it went idle. A probe failure returns False rather than
raising: the caller should still try the job and let that report the
real problem. A state we don't recognise also returns straight away -
waiting on a reading we can't interpret would just stall the batch for
the whole timeout on every single label.
"""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
info = get_printer_attributes(host, port, path)
except IPPError:
return False
state = info.get('state')
if state in IDLE_STATES:
return True
if state not in BUSY_STATES:
return False
time.sleep(poll)
return False
def print_job_paced(host, document, mime_type, job_name='LEGO part labels',
port=DEFAULT_PORT, path='/ipp/print', user='lim',
attempts=6, settle=1.5, wait_timeout=120.0):
"""
Send one job, waiting for the printer and retrying while it is busy.
This is what a batch should use. print_job on its own fires and hopes,
which is fine for one label and reliably fails from the second one on.
"""
last = None
for attempt in range(attempts):
wait_until_idle(host, port, path, timeout=wait_timeout)
try:
return print_job(host, document, mime_type, job_name=job_name,
port=port, path=path, user=user)
except IPPBusy as exc:
last = exc
# Back off a little further each time round.
time.sleep(settle * (attempt + 1))
raise IPPError(f'Printer stayed busy after {attempts} attempts'
+ (f' ({last})' if last else '') + '.')
Reference in new issue View Git Blame Copy Permalink