Commit 19bbdf26f45a0fc90e6ddf0163540fe74b9cb3f6
Commits[COMMIT BEGIN]commit 19bbdf26f45a0fc90e6ddf0163540fe74b9cb3f6 Author: 0x4248 <[email protected]> Date: Tue Feb 3 09:39:07 2026 +0000 escpos: fix IO hangs diff --git a/toolkits/ESCPOS/src/server.py b/toolkits/ESCPOS/src/server.py index 7c832f0..5c82f39 100644 --- a/toolkits/ESCPOS/src/server.py +++ b/toolkits/ESCPOS/src/server.py @@ -1,168 +1,92 @@ -# python http server that lets you send escpos command via the escpos formatting language -from flask import request, Response -from flask import Flask +#!/usr/bin/env python3 + +from flask import Flask, request, Response from subprocess import Popen, PIPE -import shlex import logging import os import datetime +import threading +import queue app = Flask(__name__) -logging.basicConfig(level=logging.DEBUG) -PRINTER = "/dev/usb/lp0" - -STYLE = """ -* { - font-family: monospace; - background-color: #000; - color: #fff; -} -a { - color: #0f0; -} - -input, textarea { - background-color: #222; - color: #fff; - border: 1px solid #555; -} - -input[type="submit"] { - background-color: #444; - color: #fff; - border: 1px solid #888; - padding: 5px 10px; -} - -""" - -def generate_escpos_cmdline(print_id): - return f"./build/escpos < ./data/jobs/{print_id}.epml > {PRINTER}" +logging.basicConfig(level=logging.INFO) + +PRINTER = "/dev/usb/lp0" +JOB_DIR = "./data/jobs" + +# --- sanity checks --------------------------------------------------------- if not os.path.exists(PRINTER): - logging.error(f"Printer device {PRINTER} does not exist. Please check the printer connection.") - exit(1) - -if not os.path.exists("./data/jobs"): - os.makedirs("./data/jobs") -else: - for filename in os.listdir("./data/jobs"): - file_path = os.path.join("./data/jobs", filename) - if os.path.isfile(file_path): - os.remove(file_path) - -class JobList: - def __init__(self): - self.jobs = [] - - def add_job(self, job_id, title="New Job"): - timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") - self.jobs.append((job_id, title, timestamp)) - - def get_jobs(self): - return self.jobs - -print_jobs = JobList() - [email protected]('/print/<print_id>', methods=['POST']) + raise RuntimeError(f"Printer device {PRINTER} does not exist") + +os.makedirs(JOB_DIR, exist_ok=True) + +# --- job queue ------------------------------------------------------------- + +print_queue = queue.Queue() + +def printer_worker(): + while True: + job_path = print_queue.get() + try: + logging.info("printing %s", job_path) + with open(job_path, "rb") as infile, open(PRINTER, "wb") as outfile: + proc = Popen( + ["./build/escpos"], + stdin=infile, + stdout=outfile, + stderr=PIPE, + close_fds=True, + ) + _, stderr = proc.communicate() + if proc.returncode != 0: + logging.error("print failed: %s", stderr.decode()) + finally: + print_queue.task_done() + +threading.Thread(target=printer_worker, daemon=True).start() + +# --- routes ---------------------------------------------------------------- + [email protected]("/print/<print_id>", methods=["POST"]) def print_escpos(print_id): - escpos_data = request.data.decode('utf-8') - job_file_path = f"./data/jobs/{print_id}.epml" - - with open(job_file_path, 'w') as job_file: - job_file.write(escpos_data) - - cmdline = generate_escpos_cmdline(print_id) - logging.debug(f"Executing command: {cmdline}") - - process = Popen(cmdline, shell=True, stderr=PIPE) - process.communicate() - - - if process.returncode != 0: - logging.error(f"Error printing job {print_id}: {stderr.decode('utf-8')}") - return Response(f"Error printing job {print_id}: {stderr.decode('utf-8')}", status=500) - - logging.info(f"Successfully printed job {print_id}") - return Response(f"Successfully printed job {print_id}", status=200) - - [email protected]('/print/submit_job', methods=['POST']) + job_path = f"{JOB_DIR}/{print_id}.epml" + with open(job_path, "wb") as f: + f.write(request.data) + + print_queue.put(job_path) + return Response("queued\n", status=202) + + [email protected]("/print/submit_job", methods=["POST"]) def submit_print_job(): - escpos_data = request.form.get('escpos_data', '') - job_title = request.form.get('job_title', 'New Job') - print_now = request.form.get('print_now', 'no') == 'yes' - copies = int(request.form.get('copies', '1')) - + escpos_data = request.form.get("escpos_data", "") + copies = int(request.form.get("copies", "1")) + job_id = datetime.datetime.now().strftime("%Y%m%d%H%M%S") - job_file_path = f"./data/jobs/{job_id}.epml" - - with open(job_file_path, 'w') as job_file: - job_file.write(escpos_data) - - print_jobs.add_job(job_id, job_title) - - if print_now: - for _ in range(copies): - cmdline = generate_escpos_cmdline(job_id) - logging.debug(f"Executing command: {cmdline}") - - process = Popen(shlex.split(cmdline), stdout=PIPE, stderr=PIPE) - stdout, stderr = process.communicate() - - if process.returncode != 0: - logging.error(f"Error printing job {job_id}: {stderr.decode('utf-8')}") - return Response(f"Error printing job {job_id}: {stderr.decode('utf-8')}", status=500) - - logging.info(f"Successfully printed job {job_id} ({copies} copies)") - return Response(f"Successfully printed job {job_id} ({copies} copies)", status=200) - - return Response(f"Job {job_id} submitted successfully", status=200) - [email protected]('/', methods=['GET']) + job_path = f"{JOB_DIR}/{job_id}.epml" + + with open(job_path, "w") as f: + f.write(escpos_data) + + for _ in range(copies): + print_queue.put(job_path) + + return Response("queued\n", status=202) + + [email protected]("/", methods=["GET"]) def index(): - job_list_html = "<h2>Print Jobs</h2><ul>" - for job in print_jobs.get_jobs(): - job_id, title, timestamp = job - job_list_html += f"<li><a href='/print/{job_id}'>{title} - {timestamp}</a></li>" - job_list_html += "</ul>" - - head_html = f""" - <head> - <style> - {STYLE} - </style> - </head> - """ - form_html = """ + return """ <h1>ESC/POS Print Server</h1> - <p>Use the buttons below to insert ESC/POS commands into the textarea.</p> - <a onclick="appendText('[NORMAL]')">[NORMAL]</a> - <a onclick="appendText('[CENTER]')">[CENTER]</a> - <a onclick="appendText('[BOLD]')">[BOLD]</a> - <a onclick="appendText('[CUT]')">[CUT]</a> - <a onclick="appendText('[INVERT ON]')">[INVERT ON]</a> - <a onclick="appendText('[INVERT OFF]')">[INVERT OFF]</a> - <a onclick="appendText('[WIDE]')">[WIDE]</a> - <a onclick="appendText('[HR]')">[HORIZONTAL RULE]</a> - <a onclick="appendText('[NL]')">[NEWLINE]</a> - <br><br> <form action="/print/submit_job" method="post"> - <input type="text" name="job_title" placeholder="Job Title"><br> - <textarea id="escpos_data" name="escpos_data" rows="10" cols="50" placeholder="Enter ESC/POS commands here..."></textarea><br> - <input type="checkbox" name="print_now" value="yes" checked> Print Now<br> - <input type="number" name="copies" value="1" min="1" max="10"> Copies<br> - <input type="submit" value="Send Job to Server"> + <textarea name="escpos_data" rows="10" cols="50"></textarea><br> + <input type="number" name="copies" value="1" min="1" max="10"><br> + <input type="submit"> </form> - <script> - function appendText(text) { - var textarea = document.getElementById('escpos_data'); - textarea.value += text; - } - </script> """ - - return head_html + form_html + job_list_html -if __name__ == '__main__': +# --- main ------------------------------------------------------------------ + +if __name__ == "__main__": app.run(host="0.0.0.0", port=80, debug=False, use_reloader=False)[COMMIT END](C) 2025 0x4248 (C) 2025 4248 Media and 4248 Systems, All part of 0x4248 See LICENCE files for more information. Not all files are by 0x4248 always check Licencing.