lego-instructions-manager/app/routes/labels_tape.py

"""
Tape printing - admin only. Drives the Brother VC-500W over the network.

Routes (all under /admin/labels/tape):
    /                 choose labels, set tape options, see a live preview
    /preview.png      rendered preview of one physical label
    /print            send the job to the printer over IPP
    /download.pdf     the same labels as a PDF, for when you'd rather not
    /test             probe the printer: model, state, accepted formats

A "physical label" is not the same as a PartLabel row: a drawer holding
two part types prints as one 40 x 20mm label. tape_printer.plan_groups
works that out, and everything here follows its plan.

Created by: Jess Rogerson (yelling commands at Claude.AI)
"""

import logging

from flask import (Blueprint, abort, current_app, flash, jsonify, redirect,
                   render_template, request, send_file, url_for)
from flask_login import login_required

from app import db
from app.models.part_label import PartCategory, PartLabel
from app.models.site_setting import SiteSetting
from app.routes.admin import admin_required
from app.services import ipp_client, tape_printer

logger = logging.getLogger(__name__)

tape_bp = Blueprint('tape', __name__, url_prefix='/admin/labels/tape')


def printer_config():
    return {
        'host': (SiteSetting.get('tape_printer_host', '') or '').strip(),
        'port': int(SiteSetting.get('tape_printer_port', '631') or 631),
        'path': (SiteSetting.get('tape_printer_path', '/ipp/print')
                 or '/ipp/print'),
        'format': (SiteSetting.get('tape_printer_format', '') or '').strip(),
        'tape': SiteSetting.get('tape_default', tape_printer.DEFAULT_TAPE),
        'rotate': SiteSetting.get_bool('tape_rotate_default', True),
        'size_mode': SiteSetting.get('tape_size_mode',
                                     tape_printer.DEFAULT_SIZE_MODE),
    }


def _group_spec():
    """
    The manual grouping from the page: "1,2|3" means parts 1 and 2 share a
    label and part 3 is on its own. Empty means "work it out from drawers".
    """
    return (request.values.get('groups') or '').strip()


def _options_query(options, labels, group_spec=None):
    """
    The chosen options as query args, so a redirect back to the tape page
    comes back exactly as the user left it. Without this, printing bounced
    you to a fresh page and every choice - rotation especially - reverted
    to the default.
    """
    query = {
        'ids': ','.join(str(l.id) for l in labels),
        tape_printer.FORM_MARKER: '1',
        'tape': options.tape,
        'size_mode': options.size_mode,
        'copies': options.copies,
        'font_scale': options.font_scale,
        'image_percent': options.image_percent,
        'padding_mm': options.padding_mm,
        'align': options.align,
        'fields': options.fields,
    }
    if group_spec:
        query['groups'] = group_spec
    # Checkboxes: only present when on, which is what the parser expects.
    if options.rotate:
        query['rotate'] = '1'
    if options.group_drawers:
        query['group_drawers'] = '1'
    if options.outline:
        query['outline'] = '1'
    if options.fixed_length_mm:
        query['fixed_length_mm'] = options.fixed_length_mm
    return query


def _selected_labels():
    """Resolve the selection from ids= or category=, same as sheet printing."""
    ids = request.values.getlist('ids')
    if len(ids) == 1 and ',' in ids[0]:
        ids = ids[0].split(',')
    ids = [int(v) for v in ids if str(v).strip().isdigit()]

    if ids:
        found = {row.id: row for row in
                 PartLabel.query.filter(PartLabel.id.in_(ids)).all()}
        return [found[i] for i in ids if i in found]

    category_id = request.values.get('category', type=int)
    if category_id:
        category = PartCategory.query.get_or_404(category_id)
        return (PartLabel.query
                .filter(PartLabel.category_id.in_(category.descendant_ids()))
                .order_by(PartLabel.location, PartLabel.name).all())
    return []


@tape_bp.route('/', methods=['GET', 'POST'])
@login_required
@admin_required
def index():
    labels = _selected_labels()
    if not labels:
        flash('Select some labels to print first.', 'warning')
        return redirect(url_for('labels.index'))

    config = printer_config()
    source = request.form if request.method == 'POST' else request.args
    # The *_default values only apply when the form was not submitted, so an
    # explicit choice on the page always wins over the saved preference.
    options = tape_printer.options_from_form(
        source, rotate_default=config['rotate'],
        size_default=config['size_mode'], tape_default=config['tape'])

    group_spec = _group_spec()
    plan = tape_printer.plan_groups(labels, options, group_spec)
    tape = tape_printer.get_tape(options.tape)

    return render_template(
        'admin/labels/tape.html',
        labels=labels,
        plan=plan,
        group_spec=tape_printer.spec_from_groups(
            [item['labels'] for item in plan]),
        ids=','.join(str(l.id) for l in labels),
        options=options,
        tapes=tape_printer.TAPE_PRESETS,
        tape=tape,
        size_modes=tape_printer.SIZE_MODES,
        label_width_mm=tape_printer.LABEL_WIDTH_MM,
        tape_fits=tape_printer.tape_fits_label(options.tape),
        tape_used_mm=tape_printer.tape_used_mm(plan, options.copies),
        field_options=[
            ('image', 'Part image'), ('name', 'Part name'),
            ('dimensions', 'Dimensions'), ('part_number', 'Part number'),
            ('alt_numbers', 'Alternate numbers'), ('category', 'Category'),
            ('category_code', 'Category code'), ('location', 'Drawer'),
            ('colour', 'Colour name'), ('accent', 'Colour stripe'),
        ],
        printer=config,
    )


@tape_bp.route('/preview.png')
@login_required
@admin_required
def preview():
    """
    One physical label. `ids` is every part on it, because a drawer holding
    two part types prints as a single 40 x 20mm label, not two labels.
    """
    raw = request.args.get('ids') or request.args.get('id') or ''
    wanted = [int(v) for v in str(raw).split(',') if v.strip().isdigit()]
    if not wanted:
        abort(400, 'preview.png needs ?ids= of one or more labels.')

    found = {row.id: row for row in
             PartLabel.query.filter(PartLabel.id.in_(wanted)).all()}
    group = [found[i] for i in wanted if i in found]
    if not group:
        abort(404)

    config = printer_config()
    options = tape_printer.options_from_form(
        request.args, rotate_default=config['rotate'],
        size_default=config['size_mode'], tape_default=config['tape'])
    buffer = tape_printer.preview_png(
        group, options, current_app.config['UPLOAD_FOLDER'])
    return send_file(buffer, mimetype='image/png')


@tape_bp.route('/test', methods=['POST'])
@login_required
@admin_required
def test_printer():
    """Ask the printer what it is. Always answers; never 500s on a timeout."""
    host = (request.form.get('host')
            or printer_config()['host'] or '').strip()
    if not host:
        return jsonify({'ok': False,
                        'error': 'No printer address set.'})
    try:
        port = int(request.form.get('port') or printer_config()['port'])
    except ValueError:
        port = 631
    path = request.form.get('path') or printer_config()['path']

    try:
        info = ipp_client.get_printer_attributes(host, port, path)
    except ipp_client.IPPError as exc:
        return jsonify({'ok': False, 'error': str(exc)})
    except Exception as exc:
        logger.exception('Printer probe blew up')
        return jsonify({'ok': False,
                        'error': f'{type(exc).__name__}: {exc}'})

    if info.get('usable_format'):
        info['advice'] = (f'Ready. Jobs will be sent as '
                          f'{info["usable_format"]}.')
    else:
        info['advice'] = (
            'Reached the printer, but it did not list a format LIM can '
            'produce (JPEG, PNG or PDF). Use Download PDF and print from '
            'the Brother app instead.')
    return jsonify(info)


@tape_bp.route('/print', methods=['POST'])
@login_required
@admin_required
def print_tape():
    labels = _selected_labels()
    if not labels:
        flash('Nothing selected to print.', 'warning')
        return redirect(url_for('labels.index'))

    config = printer_config()
    options = tape_printer.options_from_form(
        request.form, rotate_default=config['rotate'],
        size_default=config['size_mode'], tape_default=config['tape'])
    host = config['host']
    if not host:
        flash('No tape printer address configured. Set it in Label '
              'Settings.', 'warning')
        return redirect(url_for('labels.settings'))

    # Work out what the printer will take. A stored choice wins; otherwise
    # probe, so a firmware change can't silently break printing.
    mime_type = config['format']
    if not mime_type:
        try:
            info = ipp_client.get_printer_attributes(
                host, config['port'], config['path'])
            mime_type = info.get('usable_format')
        except ipp_client.IPPError as exc:
            flash(f'Could not reach the printer: {exc}', 'danger')
            return redirect(url_for('tape.index',
                                    **_options_query(options, labels,
                                                     _group_spec())))
    if not mime_type:
        flash('The printer did not offer a format LIM can produce. Use '
              'Download PDF instead.', 'warning')
        return redirect(url_for('tape.index',
                                **_options_query(options, labels,
                                                 _group_spec())))

    upload_folder = current_app.config['UPLOAD_FOLDER']
    group_spec = _group_spec()
    plan = tape_printer.plan_groups(labels, options, group_spec)

    if options.is_fixed and not tape_printer.tape_fits_label(options.tape):
        flash(f'{tape_printer.LABEL_WIDTH_MM:.0f}mm labels need 50mm tape - '
              f'{tape_printer.get_tape(options.tape).name} is narrower, so '
              'they were scaled down to fit.', 'warning')

    sent = 0
    failures = []

    for item in plan:
        group = item['labels']
        name = ' + '.join(l.display_title for l in group)
        try:
            image = tape_printer.render_group(group, options, upload_folder)
            document = tape_printer.image_to_bytes(image, mime_type)
            # copies is handled by repeating the job rather than the IPP
            # copies attribute: tape printers vary wildly in honouring it.
            #
            # Paced rather than fired in a burst: the VC-500W takes one job
            # at a time and answers "busy" to anything arriving while the
            # tape is still moving, so each job waits for the printer to go
            # idle and retries while it is still working.
            for _ in range(options.copies):
                ipp_client.print_job_paced(
                    host, document, mime_type,
                    job_name=f'LIM: {name}'[:120],
                    port=config['port'], path=config['path'])
                sent += 1
        except ipp_client.IPPError as exc:
            failures.append(f'{name}: {exc}')
        except Exception as exc:
            logger.exception('Tape print failed for %s', name)
            failures.append(f'{name}: {exc}')

    if sent:
        flash(f'Sent {sent} label{"" if sent == 1 else "s"} to the printer.',
              'success')
    for problem in failures[:5]:
        flash(problem, 'danger')

    return redirect(url_for('tape.index',
                            **_options_query(options, labels, group_spec)))


@tape_bp.route('/download.pdf', methods=['GET', 'POST'])
@login_required
@admin_required
def download_pdf():
    """Every selected label as one PDF, one tape label per page."""
    labels = _selected_labels()
    if not labels:
        flash('Nothing selected.', 'warning')
        return redirect(url_for('labels.index'))

    config = printer_config()
    source = request.form if request.method == 'POST' else request.args
    options = tape_printer.options_from_form(
        source, rotate_default=config['rotate'],
        size_default=config['size_mode'], tape_default=config['tape'])
    buffer = tape_printer.render_batch_pdf(
        labels, options, current_app.config['UPLOAD_FOLDER'], _group_spec())
    if buffer is None:
        flash('Nothing to render.', 'warning')
        return redirect(url_for('labels.index'))

    return send_file(buffer, mimetype='application/pdf', as_attachment=True,
                     download_name='lego-tape-labels.pdf')