#!/usr/bin/env python3 """ zorros_print_agent.py — Local Kitchen Print Bridge Run this on ANY computer at Zorro's on the same WiFi as the printer. Polls Railway every 10 seconds for new orders → prints ESC/POS tickets directly to 192.168.20.68:9100 Usage: pip install requests export PRINT_BRIDGE_TOKEN=your-token export PRINTER_IP=192.168.20.68 # optional; this is the default export BACKUP_PRINTER_IP=192.168.213.110 # optional 2nd printer; used if primary is down export RECEIPT_PRINTER_IP=192.168.20.60 # optional receipt station; customer/store/driver copies print there python3 zorros_print_agent.py --status # no-token readiness check python3 zorros_print_agent.py --test-print # local printer test python3 zorros_print_agent.py --test-bridge-print # Railway + token + printer test python3 zorros_print_agent.py --doctor # no-paper readiness check python3 zorros_print_agent.py --probe-completion # certify GS ( H support (prints 1 slip) python3 zorros_print_agent.py # start the live spooler """ import time, socket, datetime, json, os, re, sys, hashlib, requests try: import fcntl except ImportError: # pragma: no cover - store agent runs on macOS/Linux fcntl = None try: from dotenv import load_dotenv load_dotenv() except ImportError: pass def _load_agent_env(): """Fill missing config from print-agent.env, the file the wrapper sources. Owner report 2026-08-06: hand-run prints (importing this module directly to push a single ticket) came out all red and routed every duplicate to the kitchen printer. Cause: only the launchd wrapper sources print-agent.env, so an ad-hoc interpreter had no KITCHEN_DARKNESS_BOOST (under-driven head => red-only on two-color paper) and no RECEIPT_PRINTER_IP (copies fell back to the kitchen rail). Reading the file here makes every entry point — wrapper, stale launchd plist, or a bare `import` — configure itself identically. Real environment always wins; this only supplies what is absent. """ path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "print-agent.env") try: with open(path, "r", encoding="utf-8") as fh: lines = fh.readlines() except OSError: return for line in lines: line = line.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[7:] key, sep, val = line.partition("=") if not sep: continue key = key.strip() val = val.strip().strip("'\"") if key and not os.environ.get(key): os.environ[key] = val _load_agent_env() RAILWAY_URL = os.getenv("RAILWAY_URL", "https://zorro-phone-agent-production.up.railway.app").rstrip("/") PRINTER_IP = os.getenv("PRINTER_IP", "192.168.20.68").strip() PRINTER_PORT = int(os.getenv("PRINTER_PORT", "9100")) # Optional second printer on the same LAN. When set, a ticket that fails to # reach the primary is retried on the backup before the order is left for the # next poll — so a single jammed/offline printer doesn't stall the whole line. BACKUP_PRINTER_IP = os.getenv("BACKUP_PRINTER_IP", "").strip() BACKUP_PRINTER_PORT = int(os.getenv("BACKUP_PRINTER_PORT", str(PRINTER_PORT))) # Optional receipt-station printer (owner station split 2026-08-02: kitchen # ticket stays on the primary; CUSTOMER/STORE/DRIVER copies go here). When # unset, or when this printer is down, copies print on the primary as before — # a dead receipt printer must never cost the cashier their PAID/COLLECT copy. RECEIPT_PRINTER_IP = os.getenv("RECEIPT_PRINTER_IP", "").strip() RECEIPT_PRINTER_PORT = int(os.getenv("RECEIPT_PRINTER_PORT", str(PRINTER_PORT))) POLL_INTERVAL = int(os.getenv("PRINT_POLL_INTERVAL", "10")) PRINT_BRIDGE_TOKEN = os.getenv("PRINT_BRIDGE_TOKEN", "").strip() # Real-time printer status gate. TCP-send success does NOT mean a ticket printed: # a powered, networked printer that is out of paper or has its cover open accepts # the bytes and silently drops the job. When enabled (default), we ask the printer # whether it can actually print (ESC/POS DLE EOT) before marking an order printed. # Set PRINTER_STATUS_CHECK=0 to disable if a printer misbehaves in the store. STATUS_CHECK_ENABLED = os.getenv("PRINTER_STATUS_CHECK", "1").strip().lower() not in ("0", "false", "no", "off") # Post-send STATUS SAMPLE. socket.sendall() only proves the OS accepted the # bytes into the LOCAL send buffer — over the store VPN the tunnel can drop # right after sendall returns and the ticket never reaches the printer while # the order still gets marked printed (order #59, 2026-07-26 18:25Z). The # pre-send status gate above cannot catch this: the connection was healthy # when it ran. So after the ticket bytes we request real-time status ON THE # SAME connection. IMPORTANT LIMITS (Epson data-processing reference): DLE EOT # is a REAL-TIME command — the printer answers it as soon as it is received, # while the ticket bytes may still sit unprocessed in its receive/print # buffers. A reply is therefore evidence the link survived past the send plus # a sample of current printer status — NOT proof the ticket was parsed or # physically printed. Silence from a printer that is known to answer DLE EOT # means the link likely died mid-stream: the ticket enters a bounded # sent-unconfirmed retry (below) instead of being marked printed. # Set PRINT_SEND_STATUS_CHECK=0 to disable if a printer misbehaves. SEND_STATUS_CHECK_ENABLED = os.getenv("PRINT_SEND_STATUS_CHECK", "1").strip().lower() not in ("0", "false", "no", "off") SEND_STATUS_TIMEOUT = float(os.getenv("PRINT_SEND_STATUS_TIMEOUT", "5")) # A sent-but-unconfirmed ticket retries at most this many times, then is HELD # (not marked printed, staff paged) — bounded so a printer that prints fine # but whose replies get lost cannot spew duplicate tickets every poll forever. MAX_UNCONFIRMED_SENDS = int(os.getenv("PRINT_MAX_UNCONFIRMED_SENDS", "3")) # TRUE job-completion barrier (ESC/POS GS ( H fn=48 "process ID response"): # unlike real-time DLE EOT, GS ( H is an ordinary buffered command — the # printer transmits the echoed process ID only AFTER all receive/print data # before it has been processed, so its reply DOES confirm the ticket was # consumed. OFF by default: BTP-M300A support is UNCERTIFIED, and a printer # that does not know the command may print the trailing ID bytes as garbage # on a live kitchen ticket. Certify with --probe-completion (prints one small # slip; run owner-approved, in a quiet window) before setting # PRINT_COMPLETION_CHECK=1. COMPLETION_CHECK_ENABLED = os.getenv("PRINT_COMPLETION_CHECK", "0").strip().lower() in ("1", "true", "yes", "on") COMPLETION_TIMEOUT = float(os.getenv("PRINT_COMPLETION_TIMEOUT", "15")) # Cash-drawer kick (owner spec 2026-07-24, drawer control): the drawer plugs # into the receipt printer's DK port; an ESC p pulse pops it. OFF by default — # enable per store with DRAWER_KICK_ENABLED=1 once a drawer is physically # attached, else the pulse harmlessly no-ops on drawerless printers. DRAWER_KICK_ENABLED = os.getenv("DRAWER_KICK_ENABLED", "").strip().lower() in ("1", "true", "yes", "on") SPOOLER_VERSION = "2026-08-09-store-sync-v6" SPOOLER_LOCK_PATH = os.getenv( "PRINT_SPOOLER_LOCK_PATH", "/tmp/zorros_print_agent.lock" ).strip() ESC=b'\x1b'; GS=b'\x1d'; DLE=b'\x10'; EOT=b'\x04' INIT=ESC+b'@'; CUT=GS+b'V\x41\x05' # Same pin-2 pulse backend/printer_client.py uses for its owner-gated kick. OPEN_DRAWER=ESC+b'p\x00\x19\xfa' BOLD_ON=ESC+b'E\x01'; BOLD_OFF=ESC+b'E\x00' ALIGN_CTR=ESC+b'a\x01'; ALIGN_LEFT=ESC+b'a\x00' DOUBLE_ON=GS+b'!\x11'; DOUBLE_OFF=GS+b'!\x00' # Double HEIGHT only (GS ! 0x01). DOUBLE_ON is 0x11 — double width AND height — # which halves the line to 16 columns and destroys the right-aligned price # column. Tall-only keeps all 32 columns, so an item line reads twice as big # across the kitchen while its price still lands flush right. TALL_ON=GS+b'!\x01'; TALL_OFF=GS+b'!\x00' INVERSE_ON=GS+b'B\x01'; INVERSE_OFF=GS+b'B\x00' # Second print color (ESC r n). The BTP-M300A honors it on two-color thermal # paper (black + red OR black + blue — the accent color is a property of the # paper roll, not selectable). Ignored harmlessly on plain paper. COLOR2_ON=ESC+b'r\x01'; COLOR2_OFF=ESC+b'r\x00' LF=b'\n' # Kitchen darkness boost (owner report 2026-08-02: BTP printed EVERYTHING red # — on two-color paper, black only develops at high head energy; red develops # at low energy, so an under-driven head turns the whole ticket red). When # KITCHEN_DARKNESS_BOOST=1, kitchen-path tickets get GS ( K speed=2 + # density=8 injected right after ESC @ (INIT resets these, so they must # follow it). Non-persistent per ESC/POS — a power-cycle or ESC @ clears # them — and printers that don't implement GS ( K ignore the bytes. DARKNESS_BOOST_ENABLED = os.getenv("KITCHEN_DARKNESS_BOOST", "").strip().lower() in ("1", "true", "yes", "on") DARKNESS_BOOST = GS+b'(K\x02\x00\x32\x02' + GS+b'(K\x02\x00\x31\x08' def _apply_darkness_boost(data): """Inject the boost after the leading ESC @ so INIT can't wipe it.""" if not DARKNESS_BOOST_ENABLED: return data if data.startswith(INIT): return INIT + DARKNESS_BOOST + data[len(INIT):] return DARKNESS_BOOST + data # ESC/POS real-time status transmission (DLE EOT n): the printer answers with one # status byte even while offline. n=2 = offline cause (cover open / paper end), # n=4 = roll-paper sensor (paper end/near-end). STATUS_OFFLINE = DLE + EOT + b'\x02' STATUS_PAPER = DLE + EOT + b'\x04' # GS ( H fn=48: request a "process ID response" — the printer echoes the 4-byte # ID (framed 0x37 ... 0x00) only after everything received before the command # has been processed. Ordinary buffered command, NOT real-time: that is what # makes it a completion barrier. Format: GS ( H pL pH fn m d1..d4. COMPLETION_REQ_PREFIX = GS + b'(H\x06\x00\x30\x30' COMPLETION_RSP_HEADER = b'\x37' # ── Printed-order memory SURVIVES restarts (owner 2026-09-05: "please dont # send milion prints for 1 single order"). This set used to be in-memory # only, so every restart forgot everything and re-printed whatever the # server still listed as pending — T-4 Kelly got six tickets. Backed by a # small JSON file now: load at boot, append on every print, keep the last # 2000 ids. A restart can no longer cause a duplicate ticket. # ZORROS_PRINT_STATE_DIR lets tests (and a relocated agent) keep the registry # files somewhere other than beside this script. Default: beside the script, # exactly where the store's printed_ids.json has always lived. _STATE_DIR = (os.getenv("ZORROS_PRINT_STATE_DIR") or "").strip() or \ os.path.dirname(os.path.abspath(__file__)) _PRINTED_STORE = os.path.join(_STATE_DIR, "printed_ids.json") def _load_printed(): try: with open(_PRINTED_STORE, "r", encoding="utf-8") as fh: ids = json.load(fh) return set(ids if isinstance(ids, list) else []) except Exception: return set() def _persist_printed(): try: ids = sorted(_printed, key=lambda x: str(x))[-2000:] tmp = _PRINTED_STORE + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(ids, fh) os.replace(tmp, _PRINTED_STORE) except Exception as e: log(f"[PRINTED STORE WARN] could not persist printed ids: {e}") def note_manually_printed(order_id, order=None): """Record an order printed by hand (operator script) so the spooler never prints it again — even after a restart. Pass the order payload too when you have it: the requeue detector below then knows WHAT was printed, so a later counter correction or payment on that order still gets its one update ticket instead of a silent heal.""" _printed.add(order_id) _persist_printed() if isinstance(order, dict): try: _record_print_state(order_id, order, marked=False) except Exception as e: log(f"[PRINT STATE WARN] could not record hand-print state for #{order_id}: {e}") _printed = _load_printed() # Ids whose server "printed" flag this process has already tried to heal # (printed locally, server still said pending) — once per id per process. _healed_marks = set() # ── Deliberate REQUEUE vs lost mark (2026-09-16) ────────────────────────────── # The registry above answers "did this Mac ever print #N?" — and since 9/6 a # "yes" turned every re-served id into a HEAL: repair the server's flag, print # nothing. Right for T-24 (the mark-printed POST was lost on a blip), wrong # for every DELIBERATE requeue: a counter correction (order_mods.replace_items # → kitchen_printed=0 + reprint_marker, served with is_update_reprint and an # '*** UPDATED ORDER — NOT A NEW ORDER ***' item), a payment landing on a # ticket already on the rail (webhook → kitchen_printed=0 for the PAID # reprint), a scheduled order released at prep time. 75 heals by 9/15 — every # one followed a CLEAN mark, i.e. the server had acked the print and a writer # then cleared the flag on purpose. T-43/T-44/D-51/T-62/T-64 on 9/15 were # payments settling minutes after the print; their PAID tickets never reached # paper, and no correction has printed since 9/6. # # printed_state.json remembers, per id, WHAT was printed (a fingerprint of the # ticket-relevant payload), whether it was PAID, and whether the server ACKED # the mark. A re-served id is then: # • the very ticket we last printed and the mark never acked → a lost mark: # heal the flag, print nothing (the 9/6 behaviour); # • bridge marker present / payload changed / mark previously acked → a # deliberate requeue: print ONE kitchen ticket bannered NOT A NEW ORDER, # mark again. No PACK / DRIVER / PAID copies — the owner's 9/6 maximum # (2 kitchen + 1 driver + 1 paid) was already spent on the first print. # MAX_REQUEUE_PRINTS caps update tickets per id so a server that keeps # re-clearing a flag can never become a ticket storm. _STATE_STORE = os.path.join(_STATE_DIR, "printed_state.json") MAX_REQUEUE_PRINTS = int(os.getenv("PRINT_MAX_REQUEUE_PRINTS", "3")) _PAID_WORDS = ("paid", "captured", "settled") _UPDATE_MARKER_WORDS = ("updated order", "not a new order") def _load_print_state(): try: with open(_STATE_STORE, "r", encoding="utf-8") as fh: data = json.load(fh) return {str(k): v for k, v in data.items() if isinstance(v, dict)} \ if isinstance(data, dict) else {} except Exception: return {} def _persist_print_state(): try: keys = sorted(_print_state, key=lambda k: str(_print_state[k].get("ts", "")))[-2000:] tmp = _STATE_STORE + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump({k: _print_state[k] for k in keys}, fh) os.replace(tmp, _STATE_STORE) except Exception as e: log(f"[PRINT STATE WARN] could not persist print state: {e}") def _merge_print_state_from_disk(): """Another process (a hand-print script) may have recorded state since we loaded — newest timestamp wins, same idea as re-reading printed_ids.json.""" for k, v in _load_print_state().items(): cur = _print_state.get(k) if cur is None or str(v.get("ts", "")) > str(cur.get("ts", "")): _print_state[k] = v def _is_paid_payload(o): return str(o.get("payment_status") or "").strip().lower() in _PAID_WORDS def _is_bridge_banner_item(it): return "***" in str(it.get("name") or it.get("display_name") or "") def _requeue_marker_text(o): """The bridge's UPDATED ORDER marker when this payload carries one, else ''. print_bridge_api prepends orders.reprint_marker as the first item and sets is_update_reprint=True. build_ticket drops every '***' pseudo-item, so the marker is rendered here as the requeue banner instead.""" try: for it in _coerce_items(o.get("items") or o.get("order_items") or []): if not isinstance(it, dict) or not _is_bridge_banner_item(it): continue name = str(it.get("name") or it.get("display_name") or "") if any(w in name.lower() for w in _UPDATE_MARKER_WORDS): return name.strip("* ").strip() except Exception: pass if o.get("is_update_reprint") in (True, 1, "1", "true", "True"): return "UPDATED ORDER" return "" def _ticket_fingerprint(o): """Hash of everything on this payload that changes what the ticket SAYS. Deliberately a curated subset, not the whole payload: money display fields and server-derived extras may be recomputed between polls, and a spurious difference here would print a duplicate ticket.""" try: items = [it for it in _coerce_items(o.get("items") or o.get("order_items") or []) if isinstance(it, dict) and not _is_bridge_banner_item(it)] except Exception: items = [] basis = { "items": items, "paid": _is_paid_payload(o), "type": str(o.get("order_type") or "").strip().lower(), "sched": str(o.get("scheduled_for") or o.get("scheduled_time") or "").strip(), "notes": str(o.get("special_instructions") or "").strip(), "name": str(o.get("customer_name") or "").strip(), "addr": str(o.get("address") or o.get("delivery_address") or "").strip(), "marker": _requeue_marker_text(o), } raw = json.dumps(basis, sort_keys=True, default=str, ensure_ascii=True) return hashlib.sha1(raw.encode("utf-8")).hexdigest() def _record_print_state(order_id, o, marked, requeue=False): key = str(order_id) prev = _print_state.get(key) or {} _print_state[key] = { "fp": _ticket_fingerprint(o), "paid": _is_paid_payload(o), "marked": bool(marked), "reprints": int(prev.get("reprints") or 0) + (1 if requeue else 0), "ts": _log_ts(), } _persist_print_state() def _note_state_marked(order_id): st = _print_state.get(str(order_id)) if st is not None: st["marked"] = True st["ts"] = _log_ts() _persist_print_state() def _classify_relisted(order_id, o): """Decide what a re-served id that is already in the registry means. Returns ("requeue", ) — print ONE update ticket — or ("heal", ) — repair the server flag, print nothing.""" st = _print_state.get(str(order_id)) marker = _requeue_marker_text(o) if st is None: # Printed before this build, or by hand without a payload: the only # proof of a deliberate requeue is the bridge's own marker. if marker: return "requeue", "UPDATED ORDER" return "heal", "no local print state and no update marker" if int(st.get("reprints") or 0) >= MAX_REQUEUE_PRINTS: return "heal", f"requeue cap of {MAX_REQUEUE_PRINTS} update tickets reached" fp = _ticket_fingerprint(o) if fp == st.get("fp") and not st.get("marked"): # The exact ticket we last printed, and the server never acked the # mark: the T-24 lost-POST case. Not a requeue. return "heal", "same ticket, mark never acked" if marker: return "requeue", "UPDATED ORDER" if fp != st.get("fp"): if _is_paid_payload(o) and not st.get("paid"): return "requeue", "NOW PAID" return "requeue", "UPDATED ORDER" # Same content, and the server HAD acked our mark: only a writer clearing # kitchen_printed on purpose puts an acked ticket back in the feed. return "requeue", "REPRINT" def _requeue_banner(head): """Two big centred lines above an update ticket: the reason, then NOT A NEW ORDER — so the line never makes the food twice.""" b = bytearray(INIT + ALIGN_CTR + BOLD_ON + DOUBLE_ON) b += enc(str(head)[:16]) + LF b += enc("NOT A NEW ORDER") + LF b += DOUBLE_OFF + BOLD_OFF b += enc("=" * 32) + LF + ALIGN_LEFT return bytes(b) _print_state = _load_print_state() _last_printer_error = "" _last_pending_orders_seen = 0 _last_print_failure = "" # True when the last print_ticket failure happened AFTER ticket bytes were # sent (silence/reset during the post-send sample) — i.e. the ticket may or # may not have printed. Distinguishes the bounded sent-unconfirmed path from # plain connect/send errors, which never delivered bytes and retry freely. _last_send_unconfirmed = False # True when the last successful print_ticket was confirmed by the GS ( H # completion barrier (not just status-sampled). _last_print_completed = False # Per-order count of sent-but-unconfirmed attempts, and orders held once the # count hits MAX_UNCONFIRMED_SENDS (dedup: no more automatic reprints). _unconfirmed_sends = {} _held_unconfirmed = set() _completion_seq = 0 # Printers seen answering an identifying DLE EOT byte this process lifetime. # Only for these does post-send silence mean "connection died"; a printer that # has never spoken (a mute Star model, an lpr relay) keeps legacy behavior so # the status sample cannot put it into an infinite reprint loop. _status_capable = set() # Set when the bridge rejects our print token (401/403). A rejected agent used # to be INVISIBLE: fetch_orders swallowed the 401, returned no orders, and the # next heartbeat reported a reachable printer — identical to a healthy store # with nothing to print. Tickets piled up in the cloud and nobody was paged. _auth_rejected = "" _spooler_lock_file = None def acquire_spooler_lock(): """Allow exactly one live polling process on this store computer. The LaunchAgent and a manually started copy previously could poll the same pending order at the same time. Both printed the full kitchen/customer/store packet before either one marked the order printed, producing six slips for one pickup. The OS releases this lock automatically if the process exits or crashes, so it cannot leave printing permanently wedged. """ global _spooler_lock_file if fcntl is None: log("[FATAL] Single-instance locking is unavailable on this computer") return False lock_path = SPOOLER_LOCK_PATH or "/tmp/zorros_print_agent.lock" lock_file = None try: lock_file = open(lock_path, "a+", encoding="ascii") fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) lock_file.seek(0) lock_file.truncate() lock_file.write(str(os.getpid())) lock_file.flush() _spooler_lock_file = lock_file return True except (OSError, IOError): try: if lock_file is not None: lock_file.close() except Exception: pass log( "[FATAL] Another Zorro's print agent is already running. " "This copy will exit to prevent duplicate tickets." ) return False def _log_ts(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def log(msg): """Timestamped stdout for the unattended polling loop. This agent runs headless behind a VPN that drops (7/25: two flapping episodes inside store hours). Undated lines made it impossible to tell a printer-unreachable burst from this minute apart from one from this morning — the log had to be dated by cross-referencing teleport-watchdog.log. Interactive commands keep plain print(). """ print(f"{_log_ts()} {msg}") _TRANSLIT = str.maketrans({ "—": "-", "–": "-", "‘": "'", "’": "'", "“": '"', "”": '"', "…": "...", " ": " ", # Web item names carry typographic marks the phone path never sent: # "Grandma Pizza — Sicilian 16×16" printed its × as "?". "×": "x", "″": '"', "′": "'", "•": "-", "½": "1/2", "¼": "1/4", "¾": "3/4", }) def enc(t): # Transliterate common punctuation BEFORE ascii-replace: em-dashes were # printing as '?' on live tickets (owner photos 2026-08-03, T-31). return str(t).translate(_TRANSLIT).encode("ascii", "replace") # HOT/COLD item marks (owner 2026-07-24). Self-contained keyword classifier — # the on-prem agent has no menu brain; COLD keywords win over HOT ones so a # grilled-chicken SALAD still reads cold. Default HOT (kitchen looks at it). COLD_WORDS=("take-and-bake","take and bake","ready to bake","iced","cold", "bottled","chilled","salad","juice","lemonade","water","matcha", "yogurt","fruit","cookie","muffin","cannoli") HOT_WORDS=("hot","grilled","toasted","baked","fries","soup","wings","pizza", "calzone","stromboli","burger","cheesesteak","parm","nashville", "quesadilla","meatball","nuggets","tenders","cutlet","breakfast", "egg","bacon","sausage") def item_temperature(item): name=str(item.get("name") or "").lower() for w in COLD_WORDS: if w in name: return "COLD" for w in HOT_WORDS: if w in name: return "HOT" cat=str(item.get("category") or item.get("category_id") or "").lower() if cat in ("drinks","salad","cold-subs","cookies","muffins", "sweet-croissants","bread","dough-ready-to-bake","fresh-fruit-cups"): return "COLD" return "HOT" def temperature_tag(temp): # ESC r (second colour) is a LINE attribute on this printer class, not a # character one: switching it on anywhere in a line paints the WHOLE line, # and switching it back off mid-line does not undo that. The HOT tag used # COLOR2 and leads the item line, so every item name printed red too — # owner report 2026-08-06, "all modifiers and the item name print in red". # The item name is the line the kitchen reads from across the room and it # is meant to be big and BLACK, so the tag is emphasised without colour. # Red now survives only on lines that are red end-to-end: removals, sides, # extras and the payment banner (owner ruling 2026-07-28). if temp=="COLD": return INVERSE_ON+enc(" COLD ")+INVERSE_OFF+enc(" ") return INVERSE_ON+enc(" HOT ")+INVERSE_OFF+enc(" ") # ── Ticket width (standard 32-column ESC/POS at normal size) ──────────────── TICKET_WIDTH = 32 # Modifier lines print DOUBLE width + height to match the POS ticket the owner # holds as the standard (photos IMG_0591/IMG_0592, 2026-08-06) — that size is # what the make station reads across the room. Double width halves the roll to # 16 columns, which is why the POS ticket truncates ("fresh apple slic", # "ITALIAN SOFT ROL"). We wrap at 16 instead of truncating, so the last word # of a build never goes missing. MOD_WIDTH = 16 # ── Half-and-half modifier grouping (mirrors backend/printer_client.py) ───── # The server flattens pizza builds into modifier strings like # "left half: pepperoni" / "1st half no garlic". Those strings are a contract # (the KDS board stores them verbatim), so grouping into a LEFT:/RIGHT: block # happens here at print time only. HALF_LINE_PREFIXES = ( # (prefix, side label, render topping as a removal) ("left half: ", "LEFT", False), ("right half: ", "RIGHT", False), ("1st half no ", "1ST HALF", True), ("2nd half no ", "2ND HALF", True), ("1st half: ", "1ST HALF", False), ("2nd half: ", "2ND HALF", False), ("half: ", "HALF", False), ) def parse_half_modifier(mod_str): """Return (side_label, TOPPING_TEXT) for a half-scoped modifier, else None.""" mod_lower = mod_str.lower() for prefix, label, is_removal in HALF_LINE_PREFIXES: if mod_lower.startswith(prefix): topping = mod_str[len(prefix):].strip().upper() if not topping: return None return label, ("NO " + topping) if is_removal else topping return None def group_half_modifiers(mods): """Split modifier strings into ({side: [toppings…]}, [non-half mods]).""" half_groups, other_mods = {}, [] for mod in mods: if isinstance(mod, dict): mod = mod.get("name", str(mod)) mod_str = str(mod).strip() # The bridge's pending-print payload stringifies web-order modifier # dicts, so "{'name': 'Well Done', 'price': 0}" arrives as TEXT and # printed verbatim (owner photos 2026-08-07, D-17: nobody in the # kitchen could read the tickets). Parse the repr back to its name; # a paid add-on keeps its price visible. if mod_str.startswith("{") and "name" in mod_str: try: import ast parsed = ast.literal_eval(mod_str) if isinstance(parsed, dict): nm = str(parsed.get("name") or parsed.get("NAME") or "").strip() try: pr = float(parsed.get("price") or parsed.get("PRICE") or 0) except Exception: pr = 0 if nm: mod_str = f"{nm} (+${pr:.0f})" if pr else nm except Exception: pass if not mod_str: continue parsed = parse_half_modifier(mod_str) if parsed: half_groups.setdefault(parsed[0], []).append(parsed[1]) else: other_mods.append(mod_str) return half_groups, other_mods def wrap_lines(text, avail): """Word-wrap text to at most `avail` chars per line, hard-breaking any single word longer than `avail` (wrap, never truncate).""" avail = max(1, int(avail)) lines, cur = [], '' for word in text.split(): while len(word) > avail: if cur: lines.append(cur); cur = '' lines.append(word[:avail]); word = word[avail:] if not word: continue if len(cur) + len(word) + (1 if cur else 0) <= avail: cur = (cur + ' ' + word).strip() else: lines.append(cur); cur = word if cur: lines.append(cur) return lines or [''] def mod_block(prefix, text, width=TICKET_WIDTH): """Render one modifier line wrapped to the ticket width; continuation lines indent to align under the text.""" indent = len(prefix) parts = wrap_lines(text, width - indent) out = enc(prefix + parts[0]) + LF for cont in parts[1:]: out += enc(' ' * indent + cont) + LF return out # Words that carry no build meaning — the marker glyph already says "side", # and articles/politeness never change what the kitchen does. _NOTE_FILLER = {"a", "an", "the", "of", "on", "in", "it", "and", "to", "with", "my", "our", "please", "pls", "thanks", "thank", "you", "side", "sides", "order", "just"} def _note_words(text): return {w for w in re.findall(r"[a-z]+", str(text).lower()) if w not in _NOTE_FILLER} def _note_already_said(special, mods): """True when a modifier already carries the note's whole meaning. Owner 2026-08-06, salad sample: "side of caesar dressing" printed as a side AND "dressing on the side" printed again as a note — the same instruction twice, on the one line that has to be unambiguous. The old check was exact string equality, so any rewording slipped through. Compare significant words instead: suppress only when every meaningful word of the note also appears in one modifier, so a note that adds anything new ("no croutons" against "extra croutons") still prints. """ note = _note_words(special) if not note: return True return any(note <= _note_words(m) for m in (mods or [])) _BUILDS_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "menu_builds.json") _builds_cache = {"mtime": None, "index": {}} def _load_menu_builds(): """Read menu_builds.json, re-reading it whenever the file changes. Regenerated by build_menu_index.py from the menu descriptions. A missing or unreadable file is not an error — tickets simply print without build lists, exactly as they did before. """ try: mtime = os.path.getmtime(_BUILDS_PATH) except OSError: _builds_cache["mtime"], _builds_cache["index"] = None, {} return _builds_cache["index"] if _builds_cache["mtime"] != mtime: try: with open(_BUILDS_PATH, "r", encoding="utf-8") as fh: _builds_cache["index"] = json.load(fh) or {} except Exception as e: log(f"[BUILD INDEX] unreadable, printing without builds — {e}") _builds_cache["index"] = {} _builds_cache["mtime"] = mtime return _builds_cache["index"] def _normalize_item_name(name): """Strip size and shape noise so an ordered item matches its menu row.""" n = str(name or "").lower() n = re.sub(r'\d+\s*(?:"|inch|in\b)', " ", n) n = re.sub(r"\b(?:small|medium|large|lg|sm|md|personal|sicilian|round|" r"half|whole|slice|x\d+|\d+)\b", " ", n) n = re.sub(r"[^a-z ]", " ", n) return re.sub(r"\s+", " ", n).strip() def _removed_words(mods): """Significant words of every "no X" / "without X" modifier on an item.""" out = [] for mod in mods or []: if isinstance(mod, dict): mod = mod.get("name", "") text = str(mod).strip().lower() if not text.startswith(("no ", "without ")): continue words = {w for w in re.findall(r"[a-z]+", text)} - {"no", "without"} if words: out.append(words) return out def menu_build_for(name): """Component list for an ordered item, or [] when we have no recipe.""" index = _load_menu_builds() if not index: return [] entry = index.get(_normalize_item_name(name)) return list(entry.get("components") or []) if entry else [] def _coerce_items(raw): """Accept an items list OR the raw JSON string the orders table stores. The print bridge hands back parsed lists, so build_ticket only ever saw lists — until a ticket was pushed straight from a database row, where `items` is a JSON *string*. list("[{...}]") then yields single characters, every one of which fails the isinstance(dict) filter, and the ticket printed with NO ITEMS AT ALL between the rules (owner report 2026-08-06, "the print from web orders is weird"). Silent, total loss of the only lines that matter, so parse defensively here instead of trusting callers. """ if isinstance(raw, str): try: raw = json.loads(raw) except Exception: return [] if isinstance(raw, dict): raw = [raw] return [it for it in list(raw or []) if isinstance(it, dict)] def money(v): try: return f"${float(v):.2f}" except Exception: return "" def is_positive(v): try: return float(v) > 0 except Exception: return False def auth_headers(): if not PRINT_BRIDGE_TOKEN: raise RuntimeError("PRINT_BRIDGE_TOKEN is required for the print bridge") return {"X-Print-Bridge-Token": PRINT_BRIDGE_TOKEN} def post_heartbeat(printer_reachable=False, pending_orders_seen=None, mode="polling"): """Tell Railway the in-store print spooler is alive without sending PII.""" payload = { "printer_ip": PRINTER_IP, "printer_port": PRINTER_PORT, "printer_reachable": bool(printer_reachable), "mode": mode, "version": SPOOLER_VERSION, "last_error": _last_printer_error or None, "pending_orders_seen": _last_pending_orders_seen if pending_orders_seen is None else pending_orders_seen, } try: r = requests.post( f"{RAILWAY_URL}/api/print/heartbeat", json=payload, headers=auth_headers(), timeout=5, ) if r.status_code != 200: log(f"[HEARTBEAT WARN] print bridge returned HTTP {r.status_code}") return False return True except Exception as e: log(f"[HEARTBEAT WARN] {e}") return False def _item_line_total(item): """Item line price for the ticket: line_total (qty-aware, includes paid modifiers), falling back to unit_total*qty then price*qty. 0 = unpriced.""" for key, per_unit in (("line_total", False), ("unit_total", True), ("price", True)): try: v = float(item.get(key)) except (TypeError, ValueError): continue if v > 0: if per_unit: try: v *= float(item.get("quantity") or item.get("qty") or 1) except (TypeError, ValueError): pass return v return 0.0 def _derive_ticket_money(o): """Classify total_price and return (food, fee_display, grand, shape). total_price is the GRAND total (food+tax+fee) for web/POS orders but food-only for phone orders; assuming food-only for everyone printed COLLECT $159.68 on the $150.34 web order #78 (2026-07-26). Local copy of backend/ticket_money.py's derive_ticket_money — keep the two in lockstep (backend/test_kitchen_ticket_money_shapes.py asserts parity). Only used when the server payload has no explicit food_subtotal/grand_total. Ambiguous payloads (unpriced items) keep the historical food-only math: a too-low COLLECT loses money silently. grand INCLUDES the tip. """ def _f(key): try: return float(o.get(key) or 0) except Exception: return 0.0 total, tax, fee, tip = _f("total_price"), _f("tax"), _f("delivery_fee"), _f("tip") items = o.get("items") or o.get("order_items") or [] items_sum = round(sum(_item_line_total(i) for i in items if isinstance(i, dict)), 2) is_delivery = "deliv" in str(o.get("order_type") or "").lower() shape, food, fee_display = "food_only", total, fee grand = round(total + tax + fee + tip, 2) if items_sum > 0.005 and total > items_sum + 0.02: if abs(total - (items_sum + tax + fee + tip)) <= 0.02: shape, food, grand = "grand_tip_included", round(total - tax - fee - tip, 2), total elif abs(total - (items_sum + tax + fee)) <= 0.02: shape, food = "grand_tip_separate", round(total - tax - fee, 2) grand = round(total + tip, 2) elif fee <= 0.005 and is_delivery: # Fee folded into the total while the delivery_fee column is 0 # (the order #78 family) — recover it so lines sum to COLLECT. hidden_tip_in = round(total - items_sum - tax - tip, 2) hidden_tip_out = round(total - items_sum - tax, 2) if tip > 0.005 and 0.02 < hidden_tip_in <= 25.0: shape = "grand_hidden_fee_tip_included" food, fee_display, grand = items_sum, hidden_tip_in, total elif 0.02 < hidden_tip_out <= 25.0: shape, food, fee_display = "grand_hidden_fee", items_sum, hidden_tip_out grand = round(total + tip, 2) if food <= 0: shape, food, fee_display = "food_only", total, fee grand = round(total + tax + fee + tip, 2) return round(food, 2), round(fee_display, 2), round(grand, 2), shape _SCHED_TAG_RE = re.compile(r"\[SCHEDULED for ([^\]]+)\]", re.IGNORECASE) _SCHED_FREE_RE = re.compile( r"SCHEDULED[^\d\n]{0,40}?(\d{1,2}:\d{2}\s*[AP]\.?M\.?)", re.IGNORECASE) def _scheduled_banner_text(o, otype=""): """Local-time text for a scheduled order's banner, or '' when not scheduled. Checks the three ways a schedule reaches a ticket (see build_ticket). Never raises — a malformed value just means no banner, never a lost ticket. """ try: # 1. Explicit field (tracker's naive-UTC ISO, e.g. 2026-09-05T22:00:00). raw = str(o.get("scheduled_for") or o.get("scheduled_time") or "").strip() if raw and re.match(r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}", raw): dt = datetime.datetime.strptime(raw[:16].replace("T", " "), "%Y-%m-%d %H:%M") dt = dt.replace(tzinfo=datetime.timezone.utc).astimezone() today = datetime.datetime.now(datetime.timezone.utc).astimezone().date() t = dt.strftime("%I:%M %p").lstrip("0") return t if dt.date() == today else f"{dt.strftime('%a %b %d')} {t}" si = str(o.get("special_instructions") or o.get("delivery_instructions") or "") # 2. record_order's stamp. m = _SCHED_TAG_RE.search(si) if m: return m.group(1).strip() # 3. Free-text note with a clock time. m = _SCHED_FREE_RE.search(si) if m: return m.group(1).upper().replace(".", "").replace(" ", " ").strip() except Exception: pass return "" def build_ticket(o, include_specs=True): """Wawa-style kitchen ticket: big black item names, every modifier on its own marked line, accent colour only on lines the kitchen acts on (owner ruling 2026-07-28), ***FULFILMENT*** banner, instructions impossible to miss. Mirrors the layout language of backend/printer_client.py's Wawa ticket, scaled to 32 columns. include_specs=False drops the prep recipe lines (owner 2026-08-02: 'MEASUREMENT PENDING' guidance on a customer's receipt looks bad) — used for the duplicate CUSTOMER/STORE/DRIVER copies; the kitchen ticket always keeps them. Kitchen recipe components use the same large, one-line visual hierarchy as the reference ticket in IMG_0763 (owner 2026-08-30); driver and customer copies remain unchanged.""" b = bytearray() b += INIT + ALIGN_CTR + BOLD_ON + DOUBLE_ON b += enc(" ZORRO'S ") + LF + DOUBLE_OFF + BOLD_OFF ticket = o.get("ticket_number") or o.get("id","??") src = (o.get("source","ONLINE")).upper() otype = (o.get("order_type") or "PICKUP").upper() # Ticket number BIG (owner 2026-07-26, order #62): the small one-line # "ORDER #D-12" was missed at the counter. b += BOLD_ON + DOUBLE_ON + enc(f"TICKET #{ticket}") + LF + DOUBLE_OFF + BOLD_OFF b += BOLD_ON + enc(f"[{src}]") + LF + BOLD_OFF b += enc(datetime.datetime.now().strftime("%m/%d %I:%M %p")) + LF # Owner 2026-08-03: precise RECEIVED (order intake, server UTC → local) # and PRINTED times on every ticket, so staff can time the make. recv_txt = "" try: raw_created = str(o.get("created_at") or "").strip() if raw_created: cdt = datetime.datetime.strptime(raw_created[:19], "%Y-%m-%d %H:%M:%S") cdt = cdt.replace(tzinfo=datetime.timezone.utc).astimezone() recv_txt = cdt.strftime("%I:%M %p").lstrip("0") except Exception: recv_txt = "" printed_txt = (datetime.datetime.now(datetime.timezone.utc) .astimezone().strftime("%I:%M %p").lstrip("0")) if recv_txt: # "RECEIVED ... PRINTED ..." was 33 columns and wrapped a stray # fragment onto its own line on a 32-column roll. b += BOLD_ON + enc(f"RECV {recv_txt} PRINT {printed_txt}") + LF + BOLD_OFF else: b += BOLD_ON + enc(f"PRINTED {printed_txt}") + LF + BOLD_OFF b += enc("="*32) + LF + ALIGN_LEFT # ── SCHEDULED banner (owner 2026-09-05 "fix this now") ───────────────── # A scheduled order must announce its time at the top of the ticket, or # the kitchen fires a 6 PM order at noon. D-6 (9/4), T-4 and D-5 (9/5) # all needed a hand-printed banner because nothing rendered the time. # Three ways a schedule reaches a ticket — check them all: # 1. a scheduled_for / scheduled_time field (naive-UTC ISO from the # tracker → shown in local time); # 2. record_order's "[SCHEDULED for Fri Sep 5, 6:00 PM]" stamp; # 3. a free-text "SCHEDULED PICKUP 6:00 PM" note. sched_txt = _scheduled_banner_text(o, otype) if sched_txt: # Owner 2026-09-06: "enlarge and bold the line on the printed ticket # that shows the scheduled pickup or delivery time." Every line of # the block prints DOUBLE width+height, bold, wrapped at the 16 # double-width columns — the fulfilment word ("PICKUP FOR" / # "DELIVERY FOR") and the time each get their own big line so the # kitchen can read them from the make station. b += ALIGN_CTR + BOLD_ON + DOUBLE_ON b += enc("** SCHEDULED **") + LF b += enc(f"{otype} FOR"[:16]) + LF # Day-form text ("Fri Sep 05, 7:00 PM") breaks at the comma, and the # clock time is one unbreakable token — never "7:00" / "PM" split. _pieces = [p.strip() for p in sched_txt.split(",") if p.strip()] for _piece in _pieces: _piece = re.sub(r"(\d{1,2}:\d{2})\s+([AP]M)", r"\1 \2", _piece) _words, _line = _piece.split(" "), "" for _w in _words: if _line and len(_line) + 1 + len(_w) > 16: b += enc(_line.replace(" ", " ")) + LF _line = _w else: _line = (_line + " " + _w).strip() if _line else _w if _line: b += enc(_line.replace(" ", " ")) + LF b += DOUBLE_OFF + BOLD_OFF b += enc("="*32) + LF + ALIGN_LEFT items = _coerce_items(o.get("items") or o.get("order_items") or []) # Server payloads sometimes inject payment-banner pseudo-items like # "*** PAID (PAYMENT_LINK) ***" / "*** NOT PAID — COLLECT $31.99 ***". # They printed as "*HOT* 1x *** PAID ***" item lines (owner photos 8/3, # D-7 + T-31). We render real PAID/NOT PAID banners ourselves — drop them. def _is_payment_pseudo_item(it): n = str(it.get("name") or it.get("display_name") or "") nl = n.lower() return ("***" in n) or nl.startswith(("paid", "not paid")) or "collect $" in nl items = [it for it in items if isinstance(it, dict) and not _is_payment_pseudo_item(it)] # Internal build-guidance strings stored as plain modifiers (not "spec:" # prefixed) leaked onto customer/driver copies. Treat them as spec-class. _INTERNAL_MOD_MARKS = ("measurement pending", "current kitchen standard", "build not verified", "staff confirm", "follow current kitchen") for idx, item in enumerate(items): name = item.get("name") or item.get("display_name","Item") qty = item.get("quantity") or item.get("qty",1) # HOT/COLD mark leads every item line (owner 2026-07-24): HOT in the # accent color on two-color paper, COLD inverse. temp = item_temperature(item) b += TALL_ON + BOLD_ON + temperature_tag(temp) # Per-item price on the right (owner 2026-07-26, order #62): the one # paper ticket must price every line like any other receipt. The # HOT/COLD tag occupies printable columns, so budget for it. # The WHAT-TO-MAKE line is the one the kitchen reads from a distance, # so it prints double-height and BLACK — the same hierarchy the POS # ticket uses (big black items, colour reserved for exceptions). # Everything around it stays normal size; if every line is the same # weight, none of them are. name_txt = f"{qty}x {name.upper()}" price_txt = money(_item_line_total(item)) if _item_line_total(item) > 0 else "" # Owner 2026-09-05 ("kitchen prints are not visible for kitchen # staff need bold bigger letters"): the WHAT-TO-MAKE line was only # double-HEIGHT while the modifiers under it were already double # width+height — the main item was the least readable big element. # It now prints DOUBLE width AND height, bold, wrapped at the 16 # double-width columns so a long name never truncates, and the # price drops to its own normal-size line so it can't crowd it. b += LF + DOUBLE_ON + BOLD_ON _half = TICKET_WIDTH // 2 _words, _line = name_txt.split(" "), "" for _w in _words: if _line and len(_line) + 1 + len(_w) > _half: b += enc(_line) + LF _line = _w else: _line = (_line + " " + _w).strip() if _line else _w if _line: b += enc(_line) + LF b += DOUBLE_OFF + BOLD_OFF + TALL_OFF if price_txt: b += enc(price_txt.rjust(TICKET_WIDTH)) + LF # Kitchen-only standard build (owner 2026-08-30, reference IMG_0763): # one ingredient per line, large enough to read from the make station. # Customer/store/driver copies pass include_specs=False and preserve # their existing clean layout. Customer removals are omitted from the # base build because the explicit large red NO line below is the only # instruction the cook should see for that ingredient. if include_specs: build = menu_build_for(name) if build: removed = _removed_words(item.get("modifiers") or []) b += BOLD_ON + enc("MAKE:") + LF + BOLD_OFF for comp in build: words = {w for w in re.findall(r"[a-z]+", comp.lower())} if removed and any(words & r for r in removed): continue b += (DOUBLE_ON + BOLD_ON + mod_block("", comp.upper(), MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF) # Half-and-half sides print as one grouped, bold block (LEFT:/RIGHT:) # so each side reads as a complete build. Removals print inverse, # add-ons in the accent color, "side of" lines get a pack checkbox # (owner 2026-07-24: highlighted for visibility). mods = item.get("modifiers") or [] half_groups, other_mods = group_half_modifiers(mods) # Wawa discipline (owner 2026-08-07, T-23 photo): ONE ingredient per # line, no sentences. A compound modifier like "NO veggies — steak & # mozzarella ONLY" wrapped over four double-width lines and the make # station couldn't read it. Split on the dash connectors, drop a # clause whose words are already covered by the other modifier lines, # and strip trailing filler ("only", "please"). _expanded = [] _seen_words = set() for _m in other_mods: _seen_words.update(re.findall(r"[a-z]+", str(_m).lower())) for _m in other_mods: for _clause in re.split(r"\s+[—–-]{1,2}\s+", str(_m)): _clause = re.sub(r"\s+(?:only|please|pls)\s*$", "", _clause.strip(), flags=re.I).strip(" ,;") if not _clause: continue _cw = set(re.findall(r"[a-z]+", _clause.lower())) _others = _seen_words - set(re.findall(r"[a-z]+", str(_m).lower())) \ | {w for x in _expanded for w in re.findall(r"[a-z]+", x.lower())} if _clause is not _m and _cw and _cw <= _others: continue if _clause not in _expanded: _expanded.append(_clause) other_mods = _expanded for side_label, toppings in half_groups.items(): b += TALL_ON + BOLD_ON + mod_block(" ", side_label + ": " + " + ".join(toppings)) + BOLD_OFF + DOUBLE_OFF for mod in other_mods: m = str(mod); ml = m.lower() if (ml.startswith(("spec:", "build:")) or ml == "build not verified - staff confirm" or any(k in ml for k in _INTERNAL_MOD_MARKS)): # Defensive compatibility with orders saved by older servers: # internal recipe/spec diagnostics never belong on a live # human ticket. The _INTERNAL_MOD_MARKS net also catches # un-prefixed guidance strings ("MEASUREMENT PENDING") that # leaked onto customer/driver copies (owner photos 8/3). continue # Owner ruling 2026-07-28: the accent colour marks ONLY the lines # the kitchen has to act on differently — removals, sides # (dressing/sauce) and extras. Everything else stays black so the # marked lines are the ones that catch the eye. # Owner 2026-08-06 (POS ticket photo IMG_0591): modifiers on the # Shift4 ticket print as large as the item lines — the make station # reads them from the same distance. Ours printed at normal size # and got lost. Double width AND height (DOUBLE_ON) at MOD_WIDTH # so nothing truncates ("ITALIAN SOFT ROL" on that very photo). if ml.startswith("no ") or ml.startswith("without "): b += (DOUBLE_ON + BOLD_ON + COLOR2_ON + mod_block("! ", m.upper(), MOD_WIDTH) + COLOR2_OFF + BOLD_OFF + DOUBLE_OFF) elif ml.startswith("side of ") or ml.startswith("side "): # The [] glyph already says "side" — repeating it burns 8 of # the 16 double-width columns and forces a needless wrap. txt = m.upper() for lead in ("SIDE OF ", "SIDE "): if txt.startswith(lead): txt = txt[len(lead):] break b += DOUBLE_ON + BOLD_ON + mod_block("[]", txt, MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF elif ml.startswith("extra ") or ml.startswith("add "): # Same for "ADD" against the + glyph. "EXTRA" stays: it means # a double portion of something already on the build, which is # not the same instruction as adding it. txt = m.upper() if txt.startswith("ADD "): txt = txt[4:] b += DOUBLE_ON + BOLD_ON + mod_block("+ ", txt, MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF elif ml.startswith("light ") or ml.startswith("easy "): b += DOUBLE_ON + BOLD_ON + mod_block("~ ", m.upper(), MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF elif ml.startswith("sub ") or ml.startswith("swap "): b += DOUBLE_ON + BOLD_ON + mod_block("> ", m.upper(), MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF else: b += DOUBLE_ON + BOLD_ON + mod_block("+ ", m.upper(), MOD_WIDTH) + BOLD_OFF + DOUBLE_OFF # Per-item special instructions (owner order #78: "SUB SPINACH for # lettuce" was stored on the item but the kitchen never saw it — # only order-level notes printed). Same key precedence as the POS # path (pos_submission._pos_item_for). Skip if a modifier already # says the same thing. special = str(item.get("special_instructions") or item.get("notes") or "").strip() if special and not _note_already_said(special, other_mods): b += DOUBLE_ON + BOLD_ON + COLOR2_ON + mod_block(">>", special.upper(), MOD_WIDTH) b += COLOR2_OFF + BOLD_OFF + DOUBLE_OFF if idx < len(items) - 1: b += enc("- " * (TICKET_WIDTH // 2)) + LF b += enc("="*32) + LF # Fulfilment banner, styled like the POS ticket's ***TO GO*** — accent # colour, double height, starred so it is unmissable at a glance. Whether # this goes out the door or waits at the counter is the single fact staff # most need off the paper. # Owner 2026-09-11: web orders say WEB PICKUP / WEB DELIVERY in the # banner itself, same tall bold accent style — the small [WEB] tag by the # ticket number was not enough for the cooks to tell the channel apart. banner = (f"*** WEB {otype} ***" if src in ("WEB", "ONLINE", "WEBSITE") else f"***{otype}***") b += (ALIGN_CTR + BOLD_ON + COLOR2_ON + TALL_ON + enc(banner) + LF + TALL_OFF + COLOR2_OFF + BOLD_OFF + ALIGN_LEFT) # The name is how the counter matches ticket to customer at handover, so # it carries the accent colour too (owner ruling 2026-07-28). if o.get("customer_name"): b += (BOLD_ON + COLOR2_ON + enc(f"Name: {o['customer_name']}") + COLOR2_OFF + BOLD_OFF + LF) if o.get("customer_phone"): b += enc(f"Phone: {o['customer_phone']}") + LF if o.get("address") and "DELIVERY" in otype: # Delivery address tall + bold, wrapped with continuation indent so a # long street name never hard-wraps mid-word at the printer. b += TALL_ON + BOLD_ON + mod_block("ADDR: ", str(o["address"])) + BOLD_OFF + DOUBLE_OFF if o.get("payment_method"): pay = str(o.get("payment_method") or "").upper() status = str(o.get("payment_status") or "").upper() b += enc(f"Pay: {pay}{(' / ' + status) if status else ''}") + LF def _f(key): try: return float(o.get(key) or 0) except Exception: return 0.0 tax_v, tip_v = _f("tax"), _f("tip") # Prefer the server's explicit money fields (grand_total INCLUDES tip); # older servers only send raw columns, where total_price means "grand" # for web/POS orders but "food" for phone orders — derive locally then. # Naively summing the columns double-counted tax+fee on web orders # (order #78 printed COLLECT $159.68 for a $150.34 order). food, collect = _f("food_subtotal"), _f("grand_total") if collect > 0: fee_v = _f("delivery_fee_display") or _f("delivery_fee") else: food, fee_v, collect, _shape = _derive_ticket_money(o) if food > 0: b += enc(f"Food: ${food:.2f}") + LF if tax_v > 0: b += enc(f"Tax: ${tax_v:.2f}") + LF if fee_v > 0: b += enc(f"Deliv: ${fee_v:.2f}") + LF if tip_v > 0: b += BOLD_ON + enc(f"TIP: ${tip_v:.2f}") + LF + BOLD_OFF # PAID / NOT PAID banner (owner 2026-07-26, order #62): the single paper # ticket must say unmistakably whether money is still owed. An order with # no payment_status at all (older payloads) keeps the legacy COLLECT line. status_l = str(o.get("payment_status") or "").strip().lower() if status_l == "paid": # Owner 2026-08-03: when paid by card, the ticket must state the exact # amount charged and the tip on its own loud line — drivers/cashiers # were calling to ask what was collected and whether the tip is theirs. paid_total = collect if collect > 0 else round(food + tax_v + fee_v + tip_v, 2) b += ALIGN_CTR + BOLD_ON + DOUBLE_ON + enc(" ** PAID ** ") + LF b += DOUBLE_OFF + BOLD_OFF if paid_total > 0: b += BOLD_ON + enc(f"CARD TOTAL ${paid_total:.2f}") + LF + BOLD_OFF if tip_v > 0: b += BOLD_ON + COLOR2_ON + TALL_ON b += enc(f"TIP INCLUDED ${tip_v:.2f}") + LF b += TALL_OFF + COLOR2_OFF + BOLD_OFF b += enc("COLLECT NOTHING") + LF + ALIGN_LEFT elif status_l and collect > 0: b += ALIGN_CTR + INVERSE_ON + BOLD_ON + DOUBLE_ON b += enc(" NOT PAID ") + LF b += enc(f"COLLECT ${collect:.2f}") + LF b += DOUBLE_OFF + BOLD_OFF + INVERSE_OFF + ALIGN_LEFT elif collect > 0: b += BOLD_ON + DOUBLE_ON + enc(f"COLLECT ${collect:.2f}") + LF + DOUBLE_OFF + BOLD_OFF # Owner 2026-08-03: timer targets on the kitchen ticket. Until the owner # sets standard quote minutes we print write-in boxes — never invent an # ETA the customer wasn't promised. Kitchen tickets only (include_specs); # customer/driver copies skip the internal timer boxes. if include_specs: b += enc("-"*32) + LF if "DELIVERY" in otype: b += TALL_ON + BOLD_ON b += enc("OUT THE DOOR BY: ______") + LF b += enc("AT CUSTOMER BY: ______") + LF b += BOLD_OFF + TALL_OFF else: b += TALL_ON + BOLD_ON b += enc("READY BY: ______") + LF b += BOLD_OFF + TALL_OFF b += enc("[ ] waiting at counter") + LF # Order-level notes get a full SPECIAL INSTRUCTIONS section (Wawa-style) # instead of one small NOTE line that hard-wrapped past 32 columns. # Internal staff annotations (review-hold reasons etc.) must NEVER reach # a customer's hands: Anna's CUSTOMER COPY printed "UNCONFIRMED - CALL # ENDED EARLY - VERIFY WITH CUSTOMER" (owner photo 8/3). Kitchen tickets # keep them; customer/store/driver copies drop them. _notes_l = str(o.get("notes") or "").lower() _internal_note = any(k in _notes_l for k in ( "unconfirmed", "verify with customer", "staff review", "staff must", "superseded by", "review", "owner ruling", "pay-link failed")) if _internal_note and not include_specs: o = dict(o); o["notes"] = "" if o.get("notes"): b += enc("-"*32) + LF b += ALIGN_CTR + INVERSE_ON + BOLD_ON + enc(" SPECIAL INSTRUCTIONS ") + LF b += BOLD_OFF + INVERSE_OFF + ALIGN_LEFT b += TALL_ON + BOLD_ON for line in wrap_lines(str(o["notes"]).upper(), TICKET_WIDTH - 1): b += enc(" " + line) + LF b += BOLD_OFF + DOUBLE_OFF b += enc(" "*32) + LF + CUT return bytes(b) def build_test_order(): return { "id": "TEST", "ticket_number": "TEST", "source": "SETUP", "order_type": "pickup", "customer_name": "Printer Test", "customer_phone": "", "notes": "If you can read this, the kitchen printer is connected.", "items": [ { "name": "TEST PRINT - ZORRO'S KITCHEN", "quantity": 1, "modifiers": [ "Local printer path works", "Phone orders will print here", "Keep this computer running", ], } ], } def _send_raw(ip, port, data, timeout=5): with socket.create_connection((ip, port), timeout=timeout) as s: s.sendall(data) def _note_status_capable(ip, port, *replies): if any(_status_identifies(b) for b in replies): _status_capable.add((ip, port)) def _post_send_status_sample(sock, ip, port): """Sample printer status on the SAME socket, after the ticket bytes. This is a STATUS SAMPLE, not a completion barrier: DLE EOT is a real-time command the printer answers on receipt, while the ticket bytes may still sit unprocessed in its buffers. So a reply proves only that the connection survived past the send and reports the printer's current condition; it does NOT prove the ticket was parsed or physically printed. What silence from a known DLE-EOT-capable printer DOES tell us is that the link likely died mid-stream — the order-#59 hole — so that case fails the print into the bounded sent-unconfirmed retry instead of being marked printed. Returns None to proceed with mark-printed (sampled OK, or unprovable on a printer that never answers status), else a short failure reason. """ global _last_send_unconfirmed offline = _query_status_byte(sock, STATUS_OFFLINE, timeout=SEND_STATUS_TIMEOUT) if offline is None: if (ip, port) in _status_capable: _last_send_unconfirmed = True return "no status reply after send — link lost, delivery unconfirmed" # Never seen this printer speak DLE EOT: silence is ambiguous (mute # printer vs dead link). Keep legacy send-and-mark rather than reprint # forever, but say so once per ticket. log(f"[SEND CHECK WARN] {ip}:{port} gave no post-send status reply; delivery unverified") return None paper = _query_status_byte(sock, STATUS_PAPER) _note_status_capable(ip, port, offline, paper) fault = _interpret_status(offline, paper) if fault: return f"printer reported '{fault}' right after send — ticket may not have printed" return None def _next_completion_id(): global _completion_seq _completion_seq = (_completion_seq + 1) % 10000 return "{:04d}".format(_completion_seq).encode("ascii") def _confirm_completion(sock, ident, timeout=None): """Wait for the GS ( H process-ID echo: a TRUE job-completion signal. The request must already have been sent after the ticket bytes. Because GS ( H is an ordinary buffered command, the printer replies only after all preceding data has been processed — so seeing our ID back means the ticket was consumed. Scans a small window for 0x37 0x00 to tolerate an interleaved ASB/status byte. Returns True on confirmation. """ if timeout is None: timeout = COMPLETION_TIMEOUT expected = COMPLETION_RSP_HEADER + ident buf = b"" try: sock.settimeout(timeout) while len(buf) < 64: chunk = sock.recv(16) if not chunk: return False buf += chunk if expected in buf: return True except (OSError, socket.timeout): return False return False def _send_ticket(ip, port, data, timeout=5): """Send a ticket on one connection, then check on it as well as we can. With PRINT_COMPLETION_CHECK=1 (requires --probe-completion certification): appends a GS ( H process-ID request and waits for the echo — genuine confirmation the ticket was processed. Otherwise falls back to the DLE EOT status sample (link survival + status, not proof of printing). Returns None on success, else a failure-reason string. Raises on connect/send errors like _send_raw did. """ global _last_send_unconfirmed with socket.create_connection((ip, port), timeout=timeout) as s: if COMPLETION_CHECK_ENABLED: global _last_print_completed ident = _next_completion_id() s.sendall(data + COMPLETION_REQ_PREFIX + ident) if _confirm_completion(s, ident): _last_print_completed = True return None _last_send_unconfirmed = True return "no completion response — printing unconfirmed" s.sendall(data) if not SEND_STATUS_CHECK_ENABLED: return None return _post_send_status_sample(s, ip, port) def print_ticket(data): global _last_print_failure, _last_send_unconfirmed, _last_print_completed _last_print_failure = "" _last_send_unconfirmed = False _last_print_completed = False # Kitchen path only — the receipt-station Epson prints black already. data = _apply_darkness_boost(data) try: why = _send_ticket(PRINTER_IP, PRINTER_PORT, data) if why is None: return True log(f"[PRINTER ERROR] {why}") except Exception as e: why = str(e) log(f"[PRINTER ERROR] {e}") _last_print_failure = why # Primary unreachable — fall back to the backup printer if one is configured. # Only ONE of the two ever prints a given ticket in a poll: we return True as # soon as either succeeds, so mark-printed acks it and it won't reprint. if BACKUP_PRINTER_IP: try: why = _send_ticket(BACKUP_PRINTER_IP, BACKUP_PRINTER_PORT, data) if why is None: _last_print_failure = "" log(f"[BACKUP OK] Printed on backup {BACKUP_PRINTER_IP}:{BACKUP_PRINTER_PORT} (primary down)") return True log(f"[BACKUP PRINTER ERROR] {why}") except Exception as e: why = str(e) log(f"[BACKUP PRINTER ERROR] {e}") _last_print_failure = f"{_last_print_failure}; backup: {why}" return False def print_receipt_copy(data, label=""): """Print a duplicate receipt on the receipt-station printer when configured. Falls back to the normal kitchen path (print_ticket, incl. its backup chain) if no receipt printer is set or it fails — copies are best-effort everywhere, but the fallback keeps them flowing when the Epson is down. Every slip logs its destination so a pile of unexpected paper at either station can be traced to the order and path that produced it. """ if RECEIPT_PRINTER_IP: # OWNER REPORT 2026-08-03: every order came out ~4x on the receipt # printer AND ~4x in the kitchen. The bytes had already reached the # Epson and PRINTED — only the post-send status check came back # unanswered (the Epson does not answer like the BTP does) — and this # fallback then reprinted the very same slip in the kitchen. # # So: once the socket send SUCCEEDS the paper is already out, and the # copy must NEVER be re-sent to another printer. Only a genuine # delivery failure (connect/send raised — nothing left the Mac) may # fall back. Duplicate paper at two stations is far worse than one # unverified customer copy, which staff can reprint on demand. try: why = _send_ticket(RECEIPT_PRINTER_IP, RECEIPT_PRINTER_PORT, data) if why is None: log(f"[COPY] {label or 'copy'} -> receipt {RECEIPT_PRINTER_IP}") else: log(f"[COPY] {label or 'copy'} -> receipt {RECEIPT_PRINTER_IP} " f"(delivered, unverified: {why}) — NOT reprinting elsewhere") return True except Exception as e: log(f"[RECEIPT PRINTER ERROR] {e} — nothing delivered; " f"falling back to kitchen printer") ok = print_ticket(data) if ok: log(f"[COPY] {label or 'copy'} -> kitchen {PRINTER_IP}") return ok def _status_identifies(byte_val): """True if byte_val looks like an ESC/POS real-time status byte. The ESC/POS spec fixes bit0=0, bit1=1, bit4=1, bit7=0 in every DLE EOT reply — the store BTP-M300A answers exactly that (certified 7/25: 0x16, 0x12, 0x12, 0x12). This function previously required bit4=0, which no spec-conforming reply has — so the REAL printer's bytes never identified and a genuine paper-out (0x32) fell through as "unknown printer". Both patterns are accepted now: the spec one, plus the legacy bit4=0 shape for any emulation that sends it. A printer that does not speak DLE EOT (some Star models, a CUPS/lpr relay, a USB bridge) returns nothing or noise; those fail this check and are treated as "status unknown" so we never wedge the line on an un-queryable printer. """ if byte_val is None: return False return (byte_val & 0b10010011) == 0b00010010 or (byte_val & 0b10010011) == 0b00000010 def _query_status_byte(sock, command, timeout=1.5): """Send one DLE EOT command and read its single-byte reply (None on silence).""" try: sock.sendall(command) sock.settimeout(timeout) chunk = sock.recv(1) except (OSError, socket.timeout): return None return chunk[0] if chunk else None def _interpret_status(offline, paper): """Map DLE EOT status bytes to a human fault reason, or None if healthy/unknown. Returns None when neither byte identifies as ESC/POS status (unknown printer) so the caller falls back to its prior send-and-mark behavior. """ if not _status_identifies(offline) and not _status_identifies(paper): return None reasons = [] if _status_identifies(offline): if offline & 0x04: # bit2: cover open reasons.append("cover open") if offline & 0x20: # bit5: printing stopped, paper end reasons.append("out of paper") if offline & 0x40: # bit6: error occurred reasons.append("printer error") if _status_identifies(paper) and (paper & 0x60) == 0x60: # bits5,6: roll paper end if "out of paper" not in reasons: reasons.append("out of paper") return ", ".join(reasons) if reasons else None def check_printer_fault(ip=None, port=None, timeout=3): """Ask the printer whether it can actually print RIGHT NOW. Returns a short human reason string ("out of paper", "cover open", ...) when the printer reports a definite fault, else None. Pure: no global mutation. A None result means "healthy OR can't tell" — including a printer that is unreachable (that failure is already handled by print_ticket's retry) or one that does not answer DLE EOT. The caller only blocks marking-printed on a DEFINITE fault, so an un-queryable printer keeps its prior behavior. """ if not STATUS_CHECK_ENABLED: return None ip = ip or PRINTER_IP port = port or PRINTER_PORT try: with socket.create_connection((ip, port), timeout=timeout) as s: offline = _query_status_byte(s, STATUS_OFFLINE) paper = _query_status_byte(s, STATUS_PAPER) except Exception: return None # Remember that this printer answers DLE EOT: from then on, post-send # silence in _confirm_drained means a dead link, not a mute printer. _note_status_capable(ip, port, offline, paper) return _interpret_status(offline, paper) def _note_bridge_status(status_code, what): """Remember a print-token rejection so the agent stops pretending it is well. 401/403 means this spooler is locked out of the cloud: it will never see a ticket and every report it sends is discarded. That is an outage, not an idle store, so it must be loud locally and must not be reported as a healthy printer. Returns True when the call was rejected. """ global _auth_rejected if status_code not in (401, 403): return False reason = (f"print bridge rejected our PRINT_BRIDGE_TOKEN (HTTP {status_code} " f"on {what}) — this spooler cannot see or acknowledge tickets") if _auth_rejected != reason: log(f"[AUTH ERROR] {reason}. Fix PRINT_BRIDGE_TOKEN on this computer.") _auth_rejected = reason return True def _flag_printer_fault(reason): """Record a printer fault and page staff immediately. Reports the printer as NOT reachable in an out-of-band heartbeat so the Railway spooler watchdog (ops_alerting.check_print_spooler) pages the owner with the printer power/Wi-Fi/paper runbook — instead of waiting for the next polling heartbeat, which bare TCP reachability would report as healthy. """ global _last_printer_error _last_printer_error = reason post_heartbeat(printer_reachable=False, mode="fault") def print_test_ticket(): ok = print_ticket(build_ticket(build_test_order())) if ok: print("[OK] Test kitchen ticket sent. Printer spooler path is ready.") else: print("[WARN] Test kitchen ticket failed. Check printer IP, port, WiFi, and power.") return ok def probe_completion(ip=None, port=None): """Certify GS ( H fn=48 process-ID support on the printer (owner-approved). PRINTS ONE SMALL SLIP. If the printer does not implement GS ( H it may render the trailing request bytes as stray characters on that slip — which is exactly the evidence needed, and why this must never run unattended or against a printer mid-service. Only after this probe passes is it safe to set PRINT_COMPLETION_CHECK=1. Interactive command: plain print(). """ ip = ip or PRINTER_IP port = port or PRINTER_PORT ident = _next_completion_id() slip = (INIT + ALIGN_CTR + BOLD_ON + enc("COMPLETION PROBE (GS ( H)") + LF + BOLD_OFF + enc(_log_ts()) + LF + enc("ID " + ident.decode("ascii")) + LF + CUT) print(f"Probing {ip}:{port} for GS ( H process-ID completion support...") try: with socket.create_connection((ip, port), timeout=5) as s: t0 = time.time() s.sendall(slip + COMPLETION_REQ_PREFIX + ident) ok = _confirm_completion(s, ident, timeout=10) elapsed = time.time() - t0 except Exception as e: print(f"[FAIL] Probe could not run: {e}") return False if ok: print(f"[OK] Process-ID echo received {elapsed:.2f}s after send — and GS ( H replies only") print(" after all preceding data is processed, so this printer supports the true") print(" completion barrier. PRINT_COMPLETION_CHECK=1 is safe to enable.") return True print(f"[NO] No process-ID echo within 10s (waited {elapsed:.2f}s).") print(" Inspect the probe slip: stray characters after the ID line mean the printer") print(" does NOT implement GS ( H. Leave PRINT_COMPLETION_CHECK unset/0.") return False def fetch_bridge_test_order(): try: r = requests.get( f"{RAILWAY_URL}/api/print/test", headers=auth_headers(), timeout=8, ) if r.status_code == 200: return r.json().get("order") or build_test_order() print(f"[TEST FETCH ERROR] print bridge returned HTTP {r.status_code}") except Exception as e: print(f"[TEST FETCH ERROR] {e}") return None def check_railway_health(): try: r = requests.get(f"{RAILWAY_URL}/api/print/health", timeout=5) if r.status_code == 200: return True, "Railway print bridge reachable." return False, f"Railway print bridge returned HTTP {r.status_code}." except Exception as e: return False, f"Railway print bridge unreachable: {e}" def check_pending_print_auth(): try: r = requests.get( f"{RAILWAY_URL}/api/orders/pending-print", headers=auth_headers(), timeout=8, ) if r.status_code == 200: orders = r.json().get("orders", []) return True, f"Pending-print API accepted token; {len(orders)} order(s) waiting." if r.status_code in {401, 403}: return False, "Pending-print API rejected PRINT_BRIDGE_TOKEN." return False, f"Pending-print API returned HTTP {r.status_code}." except Exception as e: return False, f"Pending-print API check failed: {e}" def check_phone_order_readiness(): """Read the public, non-PII phone-order readiness report from Railway.""" try: r = requests.get(f"{RAILWAY_URL}/api/phone-order/readiness", timeout=8) if r.status_code == 200: return True, r.json() return False, {"error": f"Phone-order readiness returned HTTP {r.status_code}."} except Exception as e: return False, {"error": f"Phone-order readiness check failed: {e}"} def _yes_no(value): return "yes" if value else "no" def print_public_status(): """Print a no-token, no-paper status summary for the store computer.""" print("Zorro's Print Agent Status") print(f"Railway: {RAILWAY_URL}") print(f"Printer target: {PRINTER_IP}:{PRINTER_PORT}") print("") ok, report = check_phone_order_readiness() if not ok: print(f"[FAIL] Railway readiness: {report.get('error', 'unavailable')}") print("Next: check internet connection, then run --status again.") return False checks = report.get("checks") or {} print(f"Status: {report.get('status', 'unknown')}") print(f"Readiness: {report.get('readiness_pct', '?')}%") print(f"Menu items: {report.get('menu_item_count', '?')}") print("") for key, label in ( ("menu_brain_loaded", "Menu brain loaded"), ("live_order_enabled", "Live orders enabled"), ("phone_auto_activation_enabled", "Phone auto POS/KDS activation"), ("orders_schema_ready", "Order schema ready"), ("print_bridge_token_configured", "Railway print token configured"), ("store_spooler_auto_print_ready", "Store spooler + printer ready"), ): print(f"[{_yes_no(checks.get(key)).upper()}] {label}") blockers = report.get("blockers") or [] if blockers: print("") print("Blockers:") for blocker in blockers: print(f" - {blocker}") print("") if report.get("ready_for_phone_orders"): print("[READY] Phone orders can enter POS/KDS and print automatically.") elif report.get("software_ready_without_physical_printer"): print("[WAITING] Software is ready. Start this spooler on the store Wi-Fi.") else: print("[NOT READY] Fix the failed software checks before taking live phone orders.") print(f"Next: {report.get('next_action', 'Run --doctor on the store computer.')}") return True def print_doctor(): """Run no-paper setup diagnostics for the store computer.""" print("Zorro's Print Agent Doctor") print(f"Railway: {RAILWAY_URL}") print(f"Printer: {PRINTER_IP}:{PRINTER_PORT}") print("") checks = [] token_ready = bool(PRINT_BRIDGE_TOKEN) checks.append(("PRINT_BRIDGE_TOKEN", token_ready, "Set" if token_ready else "Missing")) if not token_ready: for label, ok, detail in checks: print(f"[{'OK' if ok else 'FAIL'}] {label}: {detail}") print("") print("Next: set PRINT_BRIDGE_TOKEN, then run --doctor again.") return False railway_ok, railway_detail = check_railway_health() checks.append(("Railway bridge", railway_ok, railway_detail)) pending_ok, pending_detail = check_pending_print_auth() checks.append(("Pending-print auth", pending_ok, pending_detail)) printer_ok = ping_printer(timeout=3) printer_detail = "Printer TCP port reachable." if printer_ok else (_last_printer_error or "Printer not reachable.") checks.append(("Printer network", printer_ok, printer_detail)) # Reachable over TCP isn't enough — a powered printer can be out of paper or # cover-open. Ask its real status so setup catches a no-paper printer here. paper_fault = check_printer_fault() if printer_ok else None if printer_ok: checks.append(("Printer status", paper_fault is None, f"Printer reports: {paper_fault}." if paper_fault else "Printer reports ready (or does not support status).")) printer_ok = printer_ok and not paper_fault heartbeat_ok = post_heartbeat(printer_reachable=printer_ok, mode="doctor") checks.append(("Heartbeat", heartbeat_ok, "Railway received heartbeat." if heartbeat_ok else "Heartbeat failed.")) for label, ok, detail in checks: print(f"[{'OK' if ok else 'FAIL'}] {label}: {detail}") ready = all(ok for _, ok, _ in checks) print("") if ready: print("[READY] Spooler prerequisites are ready. Start the live spooler or run --test-bridge-print.") else: print("[NOT READY] Fix the FAIL item(s), then run --doctor again.") return ready def print_bridge_test_ticket(): order = fetch_bridge_test_order() if not order: print("[WARN] Bridge test ticket failed. Check RAILWAY_URL and PRINT_BRIDGE_TOKEN.") return False ok = print_ticket(build_ticket(order)) if ok: print("[OK] Bridge test kitchen ticket sent. Railway, token, and printer path are ready.") else: print("[WARN] Bridge test print failed. Check printer IP, port, WiFi, and power.") return ok def ping_printer(timeout=8, attempts=3): """Is the kitchen printer answering? Retries before declaring it down. Owner report 2026-08-07: a stream of printer-offline SMS alerts while the printer was in fact printing. Measured LAN round-trip to the BTP that morning was 0.7-1.3 SECONDS (a degraded wireless link, not a dead printer), and the polling loop pinged with timeout=1 — so the check could not succeed even when the printer was healthy. One slow answer must not page anyone; only a printer that misses every attempt is treated as down. """ global _last_printer_error last = "" for attempt in range(max(1, attempts)): try: with socket.create_connection((PRINTER_IP, PRINTER_PORT), timeout=timeout) as s: s.close() _last_printer_error = "" if attempt: log(f"[PRINTER CHECK] Reachable on attempt {attempt + 1} — slow link, not down.") return True except Exception as e: last = str(e) if attempt + 1 < max(1, attempts): time.sleep(1.0) _last_printer_error = last log(f"[PRINTER CHECK] Not reachable at {PRINTER_IP}:{PRINTER_PORT} " f"after {max(1, attempts)} tries — {last}") return False def fetch_orders(): global _last_pending_orders_seen try: r = requests.get( f"{RAILWAY_URL}/api/orders/pending-print", headers=auth_headers(), timeout=8, ) if r.status_code == 200: orders = r.json().get("orders",[]) _last_pending_orders_seen = len(orders) return orders if not _note_bridge_status(r.status_code, "pending-print"): log(f"[FETCH ERROR] print bridge returned HTTP {r.status_code}") except Exception as e: log(f"[FETCH ERROR] {e}") _last_pending_orders_seen = 0 return [] def mark_printed(order_id): return requests.post( f"{RAILWAY_URL}/api/orders/{order_id}/mark-printed", headers=auth_headers(), timeout=5, ) def mark_printed_or_warn(order_id): try: response = mark_printed(order_id) if response.status_code == 200: return True if not _note_bridge_status(response.status_code, "mark-printed"): log(f"[MARK ERROR] order #{order_id} returned HTTP {response.status_code}") except Exception as e: log(f"[MARK ERROR] order #{order_id}: {e}") return False def fetch_print_jobs(): """Generic print jobs (Digital Fridge labels): opaque ESC/POS payloads the cloud composed — push verbatim, never rendered here.""" try: r = requests.get( f"{RAILWAY_URL}/api/print/pending-jobs", headers=auth_headers(), timeout=8, ) if r.status_code == 200: return r.json().get("jobs", []) if not _note_bridge_status(r.status_code, "pending-jobs"): print(f"[JOB FETCH ERROR] print bridge returned HTTP {r.status_code}") except Exception as e: print(f"[JOB FETCH ERROR] {e}") return [] #: Bridge job types that print at the COUNTER'S RECEIPT STATION instead of the #: kitchen BTP. Everything not listed here (fridge labels, void slips, requeued #: kitchen tickets) keeps the kitchen path. RECEIPT_STATION_JOB_TYPES = ("receipt_copy", "recipe_card") def process_pending_jobs_once(): import base64 printed_count = 0 for job in fetch_print_jobs(): jid = job.get("id") if not jid or jid in _printed_jobs: continue # 2026-09-09: the register's "reprint receipt" button queues a # receipt_copy job. That is customer paper and belongs on the # RECEIPT station (RECEIPT_PRINTER_IP), never the kitchen BTP — # print_receipt_copy() sends there and only falls back to the # kitchen when nothing left the Mac. Every other job type (fridge # labels, void slips, requeued tickets) keeps the kitchen path. # # 2026-09-21: the POS RECIPES button queues a recipe_card job and # rides the same lane. A recipe card is paper a cook reads standing # at the bench, so it belongs at the counter — and it must never # shoulder into the kitchen BTP's ticket stream mid-rush, nor be # held back by a kitchen-printer fault that has nothing to do with # it. job_type = str(job.get("job_type") or "").strip().lower() to_receipt_station = job_type in RECEIPT_STATION_JOB_TYPES if not to_receipt_station: # The fault probe is the KITCHEN printer's status; a receipt # copy never touches it, so a kitchen fault must not hold one. fault = check_printer_fault() if fault: _flag_printer_fault(fault) print(f"[PRINTER FAULT] {fault} — holding job #{jid}; will retry") break try: payload = base64.b64decode(job.get("payload_b64") or "") except Exception as e: requests.post( f"{RAILWAY_URL}/api/print/jobs/{jid}/mark-failed", headers=auth_headers(), json={"error": f"bad payload: {e}"}, timeout=5) continue if to_receipt_station: sent = bool(payload) and print_receipt_copy( payload, label=f"{job_type} job #{jid} {job.get('title', '')}".strip()) else: sent = bool(payload) and print_ticket(payload) if sent: try: requests.post( f"{RAILWAY_URL}/api/print/jobs/{jid}/mark-printed", headers=auth_headers(), timeout=5) _printed_jobs.add(jid) printed_count += 1 print(f"Printed job #{jid} — {job.get('title', '')}") except Exception as e: print(f"[JOB MARK ERROR] #{jid}: {e} — may reprint") else: print(f"Failed job #{jid} — will retry") return printed_count _printed_jobs = set() def maybe_kick_drawer(order): """Pop the cash drawer for a COMPLETED cash sale (best-effort). Only a cash-method ticket already marked paid kicks — an unpaid pay-at-counter order must not open the drawer before money changes hands, and card sales never do. Fires after the ticket printed and was acked, so a print failure never leaves the drawer open with no ticket. """ if not DRAWER_KICK_ENABLED: return False method = str(order.get("payment_method") or "").strip().lower() status = str(order.get("payment_status") or "").strip().lower() if method != "cash" or status != "paid": return False try: _send_raw(PRINTER_IP, PRINTER_PORT, OPEN_DRAWER) log(f"[DRAWER] kicked for cash order #{order.get('id')}") return True except Exception as e: log(f"[DRAWER ERROR] {e}") return False def report_print_failed(order_id, error): """Tell the bridge this ticket could NOT print (W5 print-job ledger). Best-effort: a failure to report must never affect the retry loop — the ticket stays pending and retries either way. Skipped once the bridge has rejected our token: re-POSTing a report it will 401 every poll forever just floods the request log with a symptom while hiding the cause (seen in prod 2026-08-03 — the same order reported every cycle for hours, all 401).""" if _auth_rejected: return try: r = requests.post( f"{RAILWAY_URL}/api/orders/{order_id}/print-failed", headers=auth_headers(), json={"error": str(error)[:300]}, timeout=5, ) _note_bridge_status(r.status_code, "print-failed") except Exception as e: log(f"[REPORT ERROR] order #{order_id}: {e}") def _copy_banner(label): """Centered, emphasized copy label prepended to duplicate receipts.""" return (b"\x1b@" + b"\x1ba\x01" + b"\x1d!\x11" + ("*** %s ***" % label).encode("ascii", "replace") + b"\n" + b"\x1d!\x00" + b"\x1ba\x00") def _print_extra_copies(o): """Owner directive 2026-07-31: three receipts per order, four on delivery. Copy 1 (kitchen) is the unlabeled ticket printed on the critical path above — its success is what marks the order printed. Copies 2..N print here BEST-EFFORT with a banner naming who each is for; a failed extra copy logs loudly but never fails the order or triggers reprint holds (a duplicate-confused kitchen is worse than a missing customer copy). """ # Owner directive 2026-09-04: "just 2 kitchen 1 drivers copy" — the pile # of CUSTOMER + STORE + DRIVER receipts was confusing the line. Now: # • the kitchen gets exactly TWO — the main ticket printed on the # critical path above (prep), plus ONE "PACK COPY" here, both on the # kitchen printer; # • a DELIVERY adds ONE driver copy on the receipt/delivery printer; # • no customer or store copies auto-print — the POS reprints on demand. oid = o.get("id") or o.get("order_id") is_delivery = "deliver" in str(o.get("order_type", "")).lower() full = build_ticket(o) # 2nd kitchen copy (PACK) — same kitchen printer as the prep ticket. try: why = _send_ticket(PRINTER_IP, PRINTER_PORT, _copy_banner("PACK COPY") + full) if why: log("[COPY ERROR] PACK COPY #%s did not print (%s)" % (oid, why)) else: log("[COPY] PACK COPY #%s -> kitchen %s" % (oid, PRINTER_IP)) except Exception as e: log("[COPY ERROR] PACK COPY #%s: %s" % (oid, e)) # 1 driver copy, delivery only, on the delivery/receipt printer. if is_delivery: try: if not print_receipt_copy(_copy_banner("DRIVER COPY") + full, label=f"DRIVER COPY #{oid}"): log("[COPY ERROR] DRIVER COPY #%s did not print (%s)" % ( oid, _last_print_failure)) except Exception as e: log("[COPY ERROR] DRIVER COPY #%s: %s" % (oid, e)) # Owner 2026-09-06: "if it's paid, just one, to confirm a payment. So # maximum four in total." ONE PAID-confirmation copy on the receipt # printer when the order is already paid at print time — never more. if str(o.get("payment_status", "")).lower() == "paid": try: if not print_receipt_copy(_copy_banner("PAID - CONFIRMATION") + full, label=f"PAID COPY #{oid}"): log("[COPY ERROR] PAID COPY #%s did not print (%s)" % ( oid, _last_print_failure)) except Exception as e: log("[COPY ERROR] PAID COPY #%s: %s" % (oid, e)) def process_pending_orders_once(): """Fetch, print, and mark each currently pending order once. Extracted from the infinite loop so setup tests can prove the store-computer spooler path without blocking forever. """ printed_count = 0 # Owner 2026-09-06 ("stop printing million tickets for one order"): an # operator hand-print registers the id in printed_ids.json from ANOTHER # process, but this running spooler only loaded that file at boot — so # in the seconds between a hand-print and the server's kitchen_printed # mark, this poll printed the same order again. Re-read the registry # every poll (it is tiny) so a hand-printed order is never re-printed. try: _printed.update(_load_printed()) _merge_print_state_from_disk() except Exception: pass seen_this_poll = set() for o in fetch_orders(): oid = o.get("id") or o.get("order_id") if not oid: log("[ORDER ERROR] Pending order missing id/order_id; skipping") continue if oid in seen_this_poll: # The same id twice in ONE response is a duplicate row, not a # requeue — nothing can have cleared the flag between them. continue seen_this_poll.add(oid) if oid in _printed: # Already on paper once. Two very different reasons the server # can serve it again — see the REQUEUE block near the top. decision, why = _classify_relisted(oid, o) if decision == "requeue": if oid in _held_unconfirmed: continue fault = check_printer_fault() if fault: _flag_printer_fault(fault) report_print_failed(oid, fault) log(f"[PRINTER FAULT] {fault} — holding update ticket for #{oid} and the rest of this batch; will retry") break ticket = o.get("ticket_number") or oid if print_ticket(_requeue_banner(why) + build_ticket(o)): _unconfirmed_sends.pop(oid, None) ok = mark_printed_or_warn(oid) # ONE kitchen ticket. No PACK / DRIVER / PAID copies: the # owner's 9/6 maximum was spent on the first print, and a # correction the line cannot tell from a second order is # exactly what the banner exists to prevent. _record_print_state(oid, o, marked=ok, requeue=True) _healed_marks.discard(oid) # a failed re-mark may heal once more _persist_printed() printed_count += 1 tail = "" if ok else "; Railway did not mark it — server flag heals next poll" log(f"[REQUEUE] #{ticket} {why} — update ticket printed once for {o.get('customer_name', '')}{tail}") elif _last_send_unconfirmed: # Bytes left the Mac; the update MAY be on paper. Never # spray a correction: count it printed, let the heal path # repair the flag, and say so loudly. _record_print_state(oid, o, marked=False, requeue=True) _healed_marks.discard(oid) _unconfirmed_sends.pop(oid, None) log(f"[REQUEUE] #{ticket} {why} — sent without confirmation, counted as printed; check the rail ({_last_print_failure})") else: report_print_failed(oid, _last_print_failure or "print_ticket failed (send error)") log(f"[REQUEUE] Failed update ticket for #{ticket} — will retry") continue # Owner 2026-09-06 (T-24 #999893): this spooler printed the ticket # at 15:49, but the mark-printed POST to the server failed on a # network blip, so the server kept serving it as unprinted — the # watchdog paged "NOT PRINTED" and an operator hand-printed a # DUPLICATE. The local registry is the truth here: heal the # server flag (idempotent) instead of leaving it to drift. Once # per id per process — a dead server must not turn into a storm. if oid not in _healed_marks: _healed_marks.add(oid) try: if mark_printed_or_warn(oid): if str(oid) in _print_state: _note_state_marked(oid) else: # Printed before this build: remember what the # server is serving now, so the NEXT change to # this order is recognised as a requeue. _record_print_state(oid, o, marked=True) log(f"[HEAL] server flag for #{oid} repaired ({why})") except Exception as e: log(f"[HEAL WARN] could not repair mark for #{oid}: {e}") continue # Belt-and-suspenders: if the server itself says this order is # already printed (an operator hand-printed and marked it), never # print it again regardless of what the pending list returned. if o.get("kitchen_printed") in (1, True, "1", "true"): _printed.add(oid) continue # Held after MAX_UNCONFIRMED_SENDS sent-but-unconfirmed attempts: do # NOT keep reprinting (each attempt may have physically printed) and do # NOT mark printed (it may never have). Staff were paged at hold time; # the order stays visibly pending for a human to resolve. if oid in _held_unconfirmed: continue # Confirm the printer can actually print BEFORE we send + mark-printed. # A powered printer that is out of paper / cover-open accepts the bytes # and silently drops the ticket; marking it printed would lose the order. # One fault holds this whole batch (all orders share the printer) and # pages staff; the held orders stay pending and retry next poll. fault = check_printer_fault() if fault: _flag_printer_fault(fault) report_print_failed(oid, fault) log(f"[PRINTER FAULT] {fault} — holding #{oid} and the rest of this batch; will retry") break if print_ticket(build_ticket(o)): _unconfirmed_sends.pop(oid, None) if mark_printed_or_warn(oid): _printed.add(oid) _persist_printed() # survive a restart — never re-print _record_print_state(oid, o, marked=True) printed_count += 1 # Extra copies come AFTER the order is marked printed. Before # this, a failed mark-printed left the order retryable and # every retry sprayed a fresh set of customer/store copies at # the counter. _print_extra_copies(o) maybe_kick_drawer(o) # "Printed" = dispatched to the printer and acked by Railway # (the mark-printed state). Only the completion barrier proves # the paper actually moved; say so when it did. suffix = " (completion-confirmed)" if _last_print_completed else "" log(f"Printed #{o.get('ticket_number','?')} — {o.get('customer_name','')}{suffix}") else: # The PAPER already printed. Owner 2026-09-08 ("no more stuck # orders"): T-22 Dan printed at 00:00:27, the mark-printed # POST timed out on a Railway blip, the id was NOT put in the # local registry, and the next poll printed the same ticket # AGAIN at 00:03:13. The local registry is the truth # (T-24 lesson): register it now so no poll re-prints it, # and let the [HEAL] path repair the server flag on the next # poll. Extra copies wait for the heal, so a duplicate never # sprays the counter either. _printed.add(oid) _persist_printed() _record_print_state(oid, o, marked=False) _unconfirmed_sends.pop(oid, None) printed_count += 1 log(f"Printed #{oid}, but Railway did not mark it printed — registered locally; server flag will be healed next poll") else: if _last_send_unconfirmed: # Ticket bytes went out but nothing confirmed them: the ticket # MAY have printed. Retry a bounded number of times (a reprint # beats a silently lost order), then hold and page a human — # unbounded retries could spew duplicates every poll. n = _unconfirmed_sends.get(oid, 0) + 1 _unconfirmed_sends[oid] = n if n >= MAX_UNCONFIRMED_SENDS: _held_unconfirmed.add(oid) reason = (f"held after {n} unconfirmed sends — may have printed " f"{n}x or 0x; needs human check ({_last_print_failure})") report_print_failed(oid, reason) _flag_printer_fault(f"order #{oid} {reason}") log(f"[HOLD] #{oid} sent {n}x without confirmation — no more auto-reprints; staff paged") continue report_print_failed(oid, _last_print_failure or "print_ticket failed (send error)") log(f"Failed #{oid} — will retry") return printed_count def run(): global _last_printer_error if not acquire_spooler_lock(): return False backup = f" | Backup: {BACKUP_PRINTER_IP}:{BACKUP_PRINTER_PORT}" if BACKUP_PRINTER_IP else "" receipt = f" | Receipts: {RECEIPT_PRINTER_IP}:{RECEIPT_PRINTER_PORT}" if RECEIPT_PRINTER_IP else "" log(f"Zorro's Print Agent | Printer: {PRINTER_IP}:{PRINTER_PORT}{backup}{receipt} | Railway: {RAILWAY_URL}") auth_headers() printer_ready = ping_printer() post_heartbeat(printer_reachable=printer_ready, mode="startup") if printer_ready: log("[OK] Printer reachable. Spooler is ready.") else: log("[WARN] Printer is not reachable yet. Orders will retry until the printer is online.") log("Press Ctrl+C to stop\n") while True: try: process_pending_orders_once() process_pending_jobs_once() # A printer that is reachable over TCP but out of paper / cover-open is # NOT ready — report it as not-reachable so the watchdog keeps paging. reachable = ping_printer() fault = check_printer_fault() if reachable else None if fault: _last_printer_error = fault # A locked-out spooler is an outage even when the printer itself is # fine — it will never receive a ticket. Never report it as healthy. if _auth_rejected: _last_printer_error = _auth_rejected post_heartbeat( printer_reachable=(reachable and not fault and not _auth_rejected), mode="polling") except Exception as e: log(f"[ERROR] {e}") time.sleep(POLL_INTERVAL) return True def main(argv=None): args = list(sys.argv[1:] if argv is None else argv) if "--status" in args or "status" in args: return 0 if print_public_status() else 1 if "--test-print" in args or "test-print" in args: return 0 if print_test_ticket() else 1 if "--test-bridge-print" in args or "test-bridge-print" in args: auth_headers() return 0 if print_bridge_test_ticket() else 1 if "--doctor" in args or "doctor" in args: return 0 if print_doctor() else 1 if "--probe-completion" in args or "probe-completion" in args: return 0 if probe_completion() else 1 if "--check" in args or "check" in args: auth_headers() printer_ready = ping_printer() post_heartbeat(printer_reachable=printer_ready, mode="check") return 0 if printer_ready else 1 if "--once" in args or "once" in args: auth_headers() processed = process_pending_orders_once() post_heartbeat(printer_reachable=ping_printer(), mode="once") return 0 if processed >= 0 else 1 return 0 if run() else 1 if __name__ == "__main__": raise SystemExit(main())