diff --git a/app/services/ipp_client.py b/app/services/ipp_client.py new file mode 100644 index 0000000..163ae8a --- /dev/null +++ b/app/services/ipp_client.py @@ -0,0 +1,324 @@ +""" +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', +}