Bon fuer die Kasse auf einem USB-Bondrucker ausgeben
Der Kunde soll mit einem Zettel zur Kasse gehen koennen, auf dem steht, was er zahlen muss. Drei Wege fuehren jetzt zu einem Bon: der Knopf "Bon" auf der Timer-Karte druckt den Zwischenstand und laesst den Timer weiterlaufen, "Beenden" druckt den Endbeleg, und in der Historie laesst sich jeder Beleg nachdrucken. Der Drucker haengt per USB an einem Raspberry Pi und ist damit vom Timer-Server aus nicht direkt erreichbar. Auf dem Pi laeuft deshalb eine kleine Bruecke (deploy/bondrucker/, nur Python-Standardbibliothek), die fertige ESC/POS-Daten per HTTP annimmt. Das Bon-Layout bleibt bewusst hier im Repo, damit Aenderungen daran nicht auf dem Pi angefasst werden muessen. Beim Beenden wird erst gespeichert und dann gedruckt. Eine leere Papierrolle ist damit ein Papierproblem und kein Datenproblem - der Vorgang steht in der Historie, die Oberflaeche sagt ausdruecklich, dass der Bon nicht gedruckt wurde, und nennt den Grund. Ohne PRINTER_HOST in der .env ist die Druckfunktion aus und die App verhaelt sich unveraendert. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Kerim (portable/claudecode)
co-authored by
Claude Opus 5
Claude Opus 5
parent
5e83999950
commit
70756f3762
@@ -0,0 +1,197 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Bondrucker-Bruecke — laeuft auf dem Raspberry Pi am Kassendrucker.
|
||||
|
||||
Warum es diesen Dienst ueberhaupt gibt: der Bondrucker haengt per USB am Pi
|
||||
(/dev/usb/lp0) und ist damit vom Timer-Server aus nicht erreichbar. Dieser
|
||||
Dienst nimmt fertige ESC/POS-Daten per HTTP entgegen und schiebt sie in das
|
||||
Geraet. Die Timer-App baut den Bon, der Pi druckt ihn nur noch.
|
||||
|
||||
Bewusst nur Python-Standardbibliothek: auf dem Pi muss dafuer nichts
|
||||
nachinstalliert werden (kein pip, kein CUPS, kein node).
|
||||
|
||||
Endpunkte
|
||||
GET /status Zustand des Druckers als JSON (fuer Health-Checks)
|
||||
POST /print Rumpf = rohe ESC/POS-Bytes, wird 1:1 an den Drucker gereicht
|
||||
|
||||
Konfiguration ueber Umgebungsvariablen (siehe bondrucker.service):
|
||||
PRINT_DEVICE Geraetedatei, Standard /dev/usb/lp0
|
||||
PRINT_HOST Lausch-Adresse, Standard 0.0.0.0
|
||||
PRINT_PORT Lausch-Port, Standard 9100
|
||||
PRINT_TOKEN gemeinsames Geheimnis; leer = ohne Anmeldung (nur im LAN!)
|
||||
PRINT_MAX_BYTES Obergrenze je Auftrag, Standard 262144
|
||||
"""
|
||||
|
||||
import glob
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
import socketserver
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
VERSION = "1.0"
|
||||
|
||||
DEVICE = os.environ.get("PRINT_DEVICE", "/dev/usb/lp0")
|
||||
HOST = os.environ.get("PRINT_HOST", "0.0.0.0")
|
||||
PORT = int(os.environ.get("PRINT_PORT", "9100"))
|
||||
TOKEN = os.environ.get("PRINT_TOKEN", "").strip()
|
||||
MAX_BYTES = int(os.environ.get("PRINT_MAX_BYTES", str(256 * 1024)))
|
||||
|
||||
# Zwei Auftraege duerfen sich nicht ineinander schieben - sonst kommt Buchstaben-
|
||||
# salat aus dem Drucker. Ein Prozess, ein Schloss.
|
||||
_print_lock = threading.Lock()
|
||||
|
||||
_last_error = None
|
||||
_last_ok = None
|
||||
|
||||
|
||||
def log(msg):
|
||||
print(msg, flush=True)
|
||||
|
||||
|
||||
def find_device():
|
||||
"""Konfiguriertes Geraet, sonst das erste vorhandene /dev/usb/lp*.
|
||||
|
||||
Nach dem Aus- und Wiedereinstecken des USB-Kabels kann aus lp0 ein lp1
|
||||
werden. Ohne diesen Fallback druckt danach nichts mehr, bis jemand den
|
||||
Dienst anfasst.
|
||||
"""
|
||||
if os.path.exists(DEVICE):
|
||||
return DEVICE
|
||||
found = sorted(glob.glob("/dev/usb/lp*"))
|
||||
return found[0] if found else None
|
||||
|
||||
|
||||
def device_status():
|
||||
dev = find_device()
|
||||
status = {
|
||||
"ok": False,
|
||||
"version": VERSION,
|
||||
"device": dev or DEVICE,
|
||||
"deviceConfigured": DEVICE,
|
||||
"deviceExists": dev is not None,
|
||||
"writable": False,
|
||||
"lastError": _last_error,
|
||||
"lastPrintAt": _last_ok,
|
||||
}
|
||||
if dev:
|
||||
status["writable"] = os.access(dev, os.W_OK)
|
||||
status["ok"] = status["writable"]
|
||||
return status
|
||||
|
||||
|
||||
def write_to_printer(data):
|
||||
"""Schreibt einen Auftrag auf den Drucker. Wirft bei Fehlern OSError."""
|
||||
global _last_error, _last_ok
|
||||
|
||||
with _print_lock:
|
||||
dev = find_device()
|
||||
if dev is None:
|
||||
raise OSError("Kein Druckergeraet gefunden (/dev/usb/lp*). Kabel? Strom?")
|
||||
|
||||
# os.open statt open(): kein Puffer dazwischen, und Fehler des Druckers
|
||||
# (Papier leer, Deckel offen) kommen direkt als OSError zurueck.
|
||||
fd = os.open(dev, os.O_WRONLY)
|
||||
try:
|
||||
written = 0
|
||||
while written < len(data):
|
||||
written += os.write(fd, data[written:])
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
_last_error = None
|
||||
_last_ok = time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
return written
|
||||
|
||||
|
||||
def token_ok(supplied):
|
||||
if not TOKEN:
|
||||
return True
|
||||
return hmac.compare_digest(TOKEN, (supplied or "").strip())
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
server_version = "bondrucker/" + VERSION
|
||||
protocol_version = "HTTP/1.1"
|
||||
|
||||
def _send(self, code, payload):
|
||||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||||
self.send_response(code)
|
||||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_GET(self):
|
||||
if self.path.rstrip("/") in ("/status", "/health", ""):
|
||||
st = device_status()
|
||||
self._send(200 if st["ok"] else 503, st)
|
||||
else:
|
||||
self._send(404, {"ok": False, "error": "Unbekannter Pfad"})
|
||||
|
||||
def do_POST(self):
|
||||
global _last_error
|
||||
|
||||
if self.path.rstrip("/") != "/print":
|
||||
return self._send(404, {"ok": False, "error": "Unbekannter Pfad"})
|
||||
|
||||
if not token_ok(self.headers.get("X-Print-Token")):
|
||||
log(f"[abgelehnt] falsches Token von {self.client_address[0]}")
|
||||
return self._send(403, {"ok": False, "error": "Token ungueltig"})
|
||||
|
||||
try:
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
except ValueError:
|
||||
return self._send(400, {"ok": False, "error": "Content-Length fehlt"})
|
||||
|
||||
if length <= 0:
|
||||
return self._send(400, {"ok": False, "error": "Leerer Druckauftrag"})
|
||||
if length > MAX_BYTES:
|
||||
return self._send(413, {"ok": False, "error": f"Auftrag groesser als {MAX_BYTES} Bytes"})
|
||||
|
||||
data = self.rfile.read(length)
|
||||
|
||||
try:
|
||||
written = write_to_printer(data)
|
||||
except OSError as err:
|
||||
_last_error = str(err)
|
||||
log(f"[fehler] {self.client_address[0]}: {err}")
|
||||
return self._send(502, {"ok": False, "error": f"Drucker meldet: {err}"})
|
||||
|
||||
log(f"[gedruckt] {written} Bytes von {self.client_address[0]}")
|
||||
self._send(200, {"ok": True, "bytes": written})
|
||||
|
||||
# Zugriffszeilen selbst formatieren; das Standardformat von
|
||||
# BaseHTTPRequestHandler geht auf stderr und verrauscht das Journal.
|
||||
def log_message(self, fmt, *args):
|
||||
pass
|
||||
|
||||
|
||||
class Server(ThreadingHTTPServer):
|
||||
daemon_threads = True
|
||||
allow_reuse_address = True
|
||||
|
||||
|
||||
def main():
|
||||
st = device_status()
|
||||
log(f"Bondrucker-Bruecke {VERSION}")
|
||||
log(f" Geraet: {st['device']} (vorhanden: {st['deviceExists']}, beschreibbar: {st['writable']})")
|
||||
log(f" Lauscht: {HOST}:{PORT}")
|
||||
log(f" Token: {'gesetzt' if TOKEN else 'NICHT gesetzt - jeder im Netz darf drucken'}")
|
||||
|
||||
if not st["deviceExists"]:
|
||||
log(" Hinweis: Drucker gerade nicht da. Der Dienst laeuft trotzdem und")
|
||||
log(" findet das Geraet, sobald es wieder angesteckt ist.")
|
||||
|
||||
try:
|
||||
Server((HOST, PORT), Handler).serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
log("Beendet.")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user