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
This commit is contained in:
jessikittyandClaude Opus 5 committed 2026-10-13 10:27:14 +11:00
1 parent 041eb60485
commit deb009b80e
1 file changed
+76
+76
View File
@@ -24,6 +24,7 @@ Created by: Jess Rogerson (yelling commands at Claude.AI)
import logging import logging
import random import random
import struct import struct
import time
import requests import requests
@@ -71,6 +72,24 @@ class IPPError(Exception):
"""Anything that stopped us talking to the printer.""" """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): def _encode_attribute(tag, name, value):
"""One IPP attribute: tag, name, value. Empty name = extra set member.""" """One IPP attribute: tag, name, value. Empty name = extra set member."""
if isinstance(value, bool): if isinstance(value, bool):
@@ -304,6 +323,8 @@ def print_job(host, document, mime_type, job_name='LEGO part labels',
if not result['ok']: if not result['ok']:
detail = result['status_name'] detail = result['status_name']
if result['status'] in RETRYABLE_STATUSES:
raise IPPBusy(f'Printer is busy ({detail}).')
if result['status'] == 0x040A: if result['status'] == 0x040A:
detail += (f' - the printer will not accept {mime_type}. ' detail += (f' - the printer will not accept {mime_type}. '
'Run Test connection to see what it does accept.') 'Run Test connection to see what it does accept.')
@@ -322,3 +343,58 @@ JOB_STATES = {
3: 'pending', 4: 'pending-held', 5: 'processing', 6: 'processing-stopped', 3: 'pending', 4: 'pending-held', 5: 'processing', 6: 'processing-stopped',
7: 'canceled', 8: 'aborted', 9: 'completed', 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 '') + '.')