325 lines
10 KiB
Python
325 lines
10 KiB
Python
"""
|
|
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 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."""
|
|
|
|
|
|
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'] == 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',
|
|
}
|