Autoparts-DB/pos/services/thermal_printer.py

"""ESC/POS thermal printer commands for 58mm and 80mm printers.

Generates raw ESC/POS byte commands that can be sent to a thermal printer
via USB, serial, or network connection.
"""

# ESC/POS command constants
ESC = b'\x1b'
GS = b'\x1d'
INIT = ESC + b'@'           # Initialize printer
CUT = GS + b'V' + b'\x00'  # Full cut
PARTIAL_CUT = GS + b'V' + b'\x01'
FEED = ESC + b'd'           # Feed N lines
ALIGN_LEFT = ESC + b'a' + b'\x00'
ALIGN_CENTER = ESC + b'a' + b'\x01'
ALIGN_RIGHT = ESC + b'a' + b'\x02'
BOLD_ON = ESC + b'E' + b'\x01'
BOLD_OFF = ESC + b'E' + b'\x00'
DOUBLE_HEIGHT = ESC + b'!' + b'\x10'
NORMAL_SIZE = ESC + b'!' + b'\x00'
LARGE_SIZE = ESC + b'!' + b'\x30'  # Double width + double height


def generate_ticket(sale_data, business_info, width=80):
    """Generate ESC/POS bytes for a sale ticket.

    Args:
        sale_data: dict with sale info (items, totals, payment, folio)
        business_info: dict with business name, RFC, address
        width: 58 or 80 (mm)

    Returns: bytes ready to send to printer
    """
    chars = 32 if width == 58 else 48  # characters per line
    buf = bytearray()
    buf += INIT

    # Header: business name (centered, bold, large)
    buf += ALIGN_CENTER
    buf += LARGE_SIZE
    buf += (business_info.get('name', 'NEXUS POS') + '\n').encode('cp437', errors='replace')
    buf += NORMAL_SIZE
    buf += (business_info.get('rfc', '') + '\n').encode('cp437', errors='replace')
    buf += (business_info.get('address', '') + '\n').encode('cp437', errors='replace')
    buf += b'\n'

    # Folio + date
    buf += ALIGN_LEFT
    buf += BOLD_ON
    folio = sale_data.get('folio', 'N/A')
    date = sale_data.get('date', '')
    buf += f'Folio: {folio}\n'.encode('cp437', errors='replace')
    buf += BOLD_OFF
    buf += f'Fecha: {date}\n'.encode('cp437', errors='replace')
    buf += f'Cajero: {sale_data.get("employee", "")}\n'.encode('cp437', errors='replace')
    if sale_data.get('customer'):
        buf += f'Cliente: {sale_data["customer"]}\n'.encode('cp437', errors='replace')
    buf += ('-' * chars + '\n').encode()

    # Column header
    buf += BOLD_ON
    hdr = _format_line('Cant  Descripcion', 'Importe', chars)
    buf += (hdr + '\n').encode('cp437', errors='replace')
    buf += BOLD_OFF
    buf += ('-' * chars + '\n').encode()

    # Items
    for item in sale_data.get('items', []):
        name = item.get('name', '')[:chars - 10]
        qty = item.get('quantity', 1)
        subtotal = item.get('subtotal', 0)
        buf += f'{qty}x {name}\n'.encode('cp437', errors='replace')
        buf += ALIGN_RIGHT
        buf += f'${subtotal:,.2f}\n'.encode('cp437', errors='replace')
        buf += ALIGN_LEFT

    buf += ('-' * chars + '\n').encode()

    # Totals
    buf += ALIGN_RIGHT
    buf += _total_line('Subtotal:', sale_data.get('subtotal', 0), chars).encode('cp437', errors='replace')
    if sale_data.get('discount_total', 0) > 0:
        buf += _total_line('Descuento:', -sale_data['discount_total'], chars).encode('cp437', errors='replace')
    buf += _total_line('IVA 16%:', sale_data.get('tax_total', 0), chars).encode('cp437', errors='replace')
    buf += BOLD_ON + DOUBLE_HEIGHT
    buf += _total_line('TOTAL:', sale_data.get('total', 0), chars).encode('cp437', errors='replace')
    buf += NORMAL_SIZE + BOLD_OFF

    # Payment
    buf += ALIGN_LEFT
    buf += f'\nPago: {sale_data.get("payment_method", "Efectivo")}\n'.encode('cp437', errors='replace')
    if sale_data.get('amount_paid'):
        buf += f'Recibido: ${sale_data["amount_paid"]:,.2f}\n'.encode('cp437', errors='replace')
    if sale_data.get('change_given'):
        buf += f'Cambio: ${sale_data["change_given"]:,.2f}\n'.encode('cp437', errors='replace')

    # Footer
    buf += b'\n'
    buf += ALIGN_CENTER
    buf += 'Gracias por su compra!\n'.encode('cp437', errors='replace')
    buf += 'Nexus Autoparts POS\n'.encode('cp437', errors='replace')
    buf += b'\n\n\n'
    buf += PARTIAL_CUT

    return bytes(buf)


def generate_quotation_ticket(quote_data, business_info, width=80):
    """Generate ESC/POS bytes for a quotation ticket.

    Args:
        quote_data: dict with keys: id, items[{name, part_number, quantity, unit_price, subtotal}],
                    subtotal, tax_total, total, valid_until, customer_name, wa_phone, created_at
        business_info: dict with name, rfc, address
        width: 58 or 80 (mm)

    Returns: bytes ready to send to printer
    """
    chars = 32 if width == 58 else 48
    buf = bytearray()
    buf += INIT

    # Header
    buf += ALIGN_CENTER
    buf += LARGE_SIZE
    buf += (business_info.get('name', 'NEXUS POS') + '\n').encode('cp437', errors='replace')
    buf += NORMAL_SIZE
    if business_info.get('rfc'):
        buf += (business_info['rfc'] + '\n').encode('cp437', errors='replace')
    if business_info.get('address'):
        buf += (business_info['address'] + '\n').encode('cp437', errors='replace')
    buf += b'\n'

    # Title
    buf += BOLD_ON + DOUBLE_HEIGHT
    buf += 'COTIZACION\n'.encode('cp437', errors='replace')
    buf += NORMAL_SIZE + BOLD_OFF
    buf += b'\n'

    # Folio + date
    buf += ALIGN_LEFT
    buf += BOLD_ON
    buf += f'Folio: COT-{quote_data.get("id", "")}\n'.encode('cp437', errors='replace')
    buf += BOLD_OFF
    buf += f'Fecha: {str(quote_data.get("created_at", ""))[:10]}\n'.encode('cp437', errors='replace')
    buf += f'Vigencia: {quote_data.get("valid_until", "7 dias")}\n'.encode('cp437', errors='replace')
    if quote_data.get('customer_name'):
        buf += f'Cliente: {quote_data["customer_name"]}\n'.encode('cp437', errors='replace')
    if quote_data.get('wa_phone'):
        buf += f'WhatsApp: {quote_data["wa_phone"]}\n'.encode('cp437', errors='replace')
    buf += ('-' * chars + '\n').encode()

    # Column header
    buf += BOLD_ON
    hdr = _format_line('Cant  Descripcion', 'Importe', chars)
    buf += (hdr + '\n').encode('cp437', errors='replace')
    buf += BOLD_OFF
    buf += ('-' * chars + '\n').encode()

    # Items
    for item in quote_data.get('items', []):
        name = item.get('name', '')[:chars - 10]
        part_no = item.get('part_number', '')
        qty = item.get('quantity', 1)
        subtotal = item.get('subtotal', 0)
        buf += f'{qty}x {name}\n'.encode('cp437', errors='replace')
        if part_no:
            buf += f'   #{part_no}\n'.encode('cp437', errors='replace')
        buf += ALIGN_RIGHT
        buf += f'${subtotal:,.2f}\n'.encode('cp437', errors='replace')
        buf += ALIGN_LEFT

    buf += ('-' * chars + '\n').encode()

    # Totals
    buf += ALIGN_RIGHT
    buf += _total_line('Subtotal:', quote_data.get('subtotal', 0), chars).encode('cp437', errors='replace')
    buf += _total_line('IVA:', quote_data.get('tax_total', 0), chars).encode('cp437', errors='replace')
    buf += BOLD_ON + DOUBLE_HEIGHT
    buf += _total_line('TOTAL:', quote_data.get('total', 0), chars).encode('cp437', errors='replace')
    buf += NORMAL_SIZE + BOLD_OFF

    # Footer
    buf += b'\n'
    buf += ALIGN_CENTER
    buf += 'Esta cotizacion no es comprobante fiscal\n'.encode('cp437', errors='replace')
    buf += 'Precios sujetos a disponibilidad\n'.encode('cp437', errors='replace')
    buf += 'Nexus Autoparts POS\n'.encode('cp437', errors='replace')
    buf += b'\n\n\n'
    buf += PARTIAL_CUT

    return bytes(buf)


def _format_line(left, right, width):
    """Pad a left-right line to fill the ticket width."""
    space = width - len(left) - len(right)
    if space < 1:
        space = 1
    return left + ' ' * space + right


def _total_line(label, amount, width):
    """Format a totals line like 'Subtotal:        $1,234.56'."""
    val = f'${abs(amount):,.2f}' if amount >= 0 else f'-${abs(amount):,.2f}'
    return _format_line(label, val, width) + '\n'


def generate_service_order_ticket(so_data, business_info, width=80):
    """Generate ESC/POS bytes for a workshop service order ticket.

    Args:
        so_data: dict with service order info:
            order_number, status, customer_name, vehicle_plate, vehicle_make,
            vehicle_model, mileage_in, fuel_level, reception_notes,
            employee_name, created_at, items[{name, part_number, quantity,
            unit_price}], labor[{description, hours, hourly_rate, total_cost}],
            estimated_cost, total
        business_info: dict with name, rfc, address
        width: 58 or 80 (mm)

    Returns: bytes ready to send to printer
    """
    chars = 32 if width == 58 else 48
    buf = bytearray()
    buf += INIT

    # Header
    buf += ALIGN_CENTER
    buf += LARGE_SIZE
    buf += (business_info.get("name", "NEXUS POS") + "\n").encode("cp437", errors="replace")
    buf += NORMAL_SIZE
    if business_info.get("rfc"):
        buf += (business_info["rfc"] + "\n").encode("cp437", errors="replace")
    if business_info.get("address"):
        buf += (business_info["address"] + "\n").encode("cp437", errors="replace")
    buf += b"\n"

    # Title
    buf += BOLD_ON + DOUBLE_HEIGHT
    buf += "ORDEN DE SERVICIO\n".encode("cp437", errors="replace")
    buf += NORMAL_SIZE + BOLD_OFF
    buf += b"\n"

    # Order info
    buf += ALIGN_LEFT
    buf += BOLD_ON
    buf += f"Folio: {so_data.get('order_number', 'N/A')}\n".encode("cp437", errors="replace")
    buf += BOLD_OFF
    buf += f"Estado: {so_data.get('status', '')}\n".encode("cp437", errors="replace")
    buf += f"Fecha: {str(so_data.get('created_at', ''))[:19]}\n".encode("cp437", errors="replace")
    if so_data.get("employee_name"):
        buf += f"Mecanico: {so_data['employee_name']}\n".encode("cp437", errors="replace")
    buf += ("-" * chars + "\n").encode()

    # Customer / vehicle
    if so_data.get("customer_name"):
        buf += BOLD_ON
        buf += f"Cliente: {so_data['customer_name']}\n".encode("cp437", errors="replace")
        buf += BOLD_OFF
    vehicle = " ".join(
        str(v) for v in [
            so_data.get("vehicle_plate", ""),
            so_data.get("vehicle_make", ""),
            so_data.get("vehicle_model", ""),
        ] if v
    ).strip()
    if vehicle:
        buf += f"Vehiculo: {vehicle}\n".encode("cp437", errors="replace")
    if so_data.get("mileage_in"):
        buf += f"Kilometraje: {so_data['mileage_in']}\n".encode("cp437", errors="replace")
    if so_data.get("fuel_level"):
        buf += f"Gasolina: {so_data['fuel_level']}\n".encode("cp437", errors="replace")
    buf += ("-" * chars + "\n").encode()

    # Reception notes
    if so_data.get("reception_notes"):
        buf += BOLD_ON
        buf += "Falla / Observaciones:\n".encode("cp437", errors="replace")
        buf += BOLD_OFF
        for line in str(so_data["reception_notes"]).splitlines():
            buf += (line[:chars] + "\n").encode("cp437", errors="replace")
        buf += ("-" * chars + "\n").encode()

    # Parts
    items = so_data.get("items", [])
    if items:
        buf += BOLD_ON
        buf += "REFACCIONES\n".encode("cp437", errors="replace")
        buf += BOLD_OFF
        for item in items:
            name = item.get("name", "")[:chars - 10]
            part_no = item.get("part_number", "")
            qty = item.get("quantity", 1)
            unit_price = item.get("unit_price", 0)
            line_total = qty * unit_price
            buf += f"{qty}x {name}\n".encode("cp437", errors="replace")
            if part_no:
                buf += f"   #{part_no}\n".encode("cp437", errors="replace")
            buf += ALIGN_RIGHT
            buf += f"${line_total:,.2f}\n".encode("cp437", errors="replace")
            buf += ALIGN_LEFT
        buf += ("-" * chars + "\n").encode()

    # Labor
    labor_items = so_data.get("labor", [])
    if labor_items:
        buf += BOLD_ON
        buf += "MANO DE OBRA\n".encode("cp437", errors="replace")
        buf += BOLD_OFF
        for labor in labor_items:
            desc = labor.get("description", "")[:chars - 10]
            hours = labor.get("hours", 0)
            rate = labor.get("hourly_rate", 0)
            total = labor.get("total_cost", hours * rate)
            buf += f"{desc}\n".encode("cp437", errors="replace")
            buf += f"   {hours} hrs x ${rate:,.2f}\n".encode("cp437", errors="replace")
            buf += ALIGN_RIGHT
            buf += f"${total:,.2f}\n".encode("cp437", errors="replace")
            buf += ALIGN_LEFT
        buf += ("-" * chars + "\n").encode()

    # Totals
    buf += ALIGN_RIGHT
    if items or labor_items:
        total = so_data.get("total") or sum(
            i.get("quantity", 1) * i.get("unit_price", 0) for i in items
        ) + sum(labor.get("total_cost", 0) for labor in labor_items)
        buf += BOLD_ON + DOUBLE_HEIGHT
        buf += _total_line("TOTAL ESTIMADO:", total, chars).encode("cp437", errors="replace")
        buf += NORMAL_SIZE + BOLD_OFF
    if so_data.get("estimated_cost"):
        buf += _total_line("Costo estimado:", so_data["estimated_cost"], chars).encode("cp437", errors="replace")

    # Footer
    buf += b"\n"
    buf += ALIGN_CENTER
    buf += "No es comprobante fiscal\n".encode("cp437", errors="replace")
    buf += "Nexus Autoparts POS\n".encode("cp437", errors="replace")
    buf += b"\n\n\n"
    buf += PARTIAL_CUT

    return bytes(buf)