Atlas - server.py

Atlas - server.py
Home / toolkits / ESCPOS / src
Lines: 3 | Size: 4177 bytes
[Download] [Show on GitHub] [Search similar files] [Raw] [Raw (proxy)] 
[FILE BEGIN]
1#!/usr/bin/env python3
2
3from flask import Flask, request, Response
4from subprocess import Popen, PIPE
5import logging
6import os
7import datetime
8import threading
9import queue
10
11app = Flask(__name__)
12logging.basicConfig(level=logging.INFO)
13
14PRINTER = "/dev/usb/lp0"
15JOB_DIR = "./data/jobs"
16
17STYLE = """
18* {
19    font-family: monospace;
20    background-color: #000;
21    color: #fff;
22}
23a {
24    color: #0f0;
25    cursor: pointer;
26    margin-right: 8px;
27}
28
29input, textarea {
30    background-color: #222;
31    color: #fff;
32    border: 1px solid #555;
33}
34
35input[type="submit"] {
36    background-color: #444;
37    border: 1px solid #888;
38    padding: 5px 10px;
39}
40"""
41
42# --- sanity checks ---------------------------------------------------------
43
44if not os.path.exists(PRINTER):
45    raise RuntimeError(f"Printer device {PRINTER} does not exist")
46
47os.makedirs(JOB_DIR, exist_ok=True)
48
49# --- job queue -------------------------------------------------------------
50
51print_queue = queue.Queue()
52
53def printer_worker():
54    while True:
55        job_path = print_queue.get()
56        try:
57            logging.info("printing %s", job_path)
58            with open(job_path, "rb") as infile, open(PRINTER, "wb") as outfile:
59                proc = Popen(
60                    ["./build/escpos"],
61                    stdin=infile,
62                    stdout=outfile,
63                    stderr=PIPE,
64                    close_fds=True,
65                )
66                _, stderr = proc.communicate()
67                if proc.returncode != 0:
68                    logging.error("print failed: %s", stderr.decode())
69        finally:
70            print_queue.task_done()
71
72threading.Thread(target=printer_worker, daemon=True).start()
73
74# --- routes ----------------------------------------------------------------
75
76@app.route("/print/<print_id>", methods=["POST"])
77def print_escpos(print_id):
78    job_path = f"{JOB_DIR}/{print_id}.epml"
79    with open(job_path, "wb") as f:
80        f.write(request.data)
81
82    print_queue.put(job_path)
83    return Response("queued\n", status=202)
84
85
86@app.route("/print/submit_job", methods=["POST"])
87def submit_print_job():
88    escpos_data = request.form.get("escpos_data", "")
89    copies = int(request.form.get("copies", "1"))
90
91    job_id = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
92    job_path = f"{JOB_DIR}/{job_id}.epml"
93
94    with open(job_path, "w") as f:
95        f.write(escpos_data)
96
97    for _ in range(copies):
98        print_queue.put(job_path)
99
100    return Response("queued\n", status=202)
101
102
103@app.route("/", methods=["GET"])
104def index():
105    return f"""
106    <html>
107    <head>
108        <style>{STYLE}</style>
109    </head>
110    <body>
111        <h1>ESC/POS Print Server</h1>
112
113        <p>Insert ESC/POS formatting tokens:</p>
114        <div>
115            <a onclick="appendText('[NORMAL]')">[NORMAL]</a>
116            <a onclick="appendText('[CENTER]')">[CENTER]</a>
117            <a onclick="appendText('[BOLD]')">[BOLD]</a>
118            <a onclick="appendText('[CUT]')">[CUT]</a>
119            <a onclick="appendText('[INVERT ON]')">[INVERT ON]</a>
120            <a onclick="appendText('[INVERT OFF]')">[INVERT OFF]</a>
121            <a onclick="appendText('[WIDE]')">[WIDE]</a>
122            <a onclick="appendText('[HR]')">[HR]</a>
123            <a onclick="appendText('[NL]')">[NL]</a>
124        </div>
125
126        <br>
127
128        <form action="/print/submit_job" method="post">
129            <textarea id="escpos_data"
130                      name="escpos_data"
131                      rows="12"
132                      cols="60"
133                      placeholder="Enter ESC/POS commands here..."></textarea>
134            <br><br>
135            Copies:
136            <input type="number" name="copies" value="1" min="1" max="10">
137            <br><br>
138            <input type="submit" value="Print">
139        </form>
140
141        <script>
142        function appendText(text) {{
143            var textarea = document.getElementById('escpos_data');
144            textarea.value += text;
145        }}
146        </script>
147    </body>
148    </html>
149    """
150
151# --- main ------------------------------------------------------------------
152
153if __name__ == "__main__":
154    app.run(host="0.0.0.0", port=8001, debug=False, use_reloader=False)
155
[FILE 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.