From e2eb3d3825a5c94a194a26984381ca42c47b5172 Mon Sep 17 00:00:00 2001 From: n07070 Date: Fri, 13 Nov 2026 02:45:11 +0100 Subject: [PATCH] Add support for signature on brother printers --- src/main.py | 24 +++---- src/print_queue.py | 7 ++- src/printer.py | 152 +++++++++++++++++++++++++++------------------ src/printers.py | 37 ++++++----- src/worker.py | 21 ++++--- 5 files changed, 141 insertions(+), 100 deletions(-) diff --git a/src/main.py b/src/main.py index 9f20795..31b459d 100644 --- a/src/main.py +++ b/src/main.py @@ -1,7 +1,8 @@ """ Welcome to the LittlePrynter's source code. -This program expose a web interface, with user authentification, +This program expose a web interface, that makes it possible to print messages from the web. + It also exposes a API, making it possible to print and interface with much of the printer's abilities. @@ -10,12 +11,14 @@ Then we build the API around Flask, Then we build the web interface, using the simple Jinja2 templating. We support two modes : + The first is a simple mode, where a computer, connected to a thermal printer, runs this program and exposes a web interface that makes use of the client's camera -The seconde is booth mode, where a Raspberry Pi is connected to a thermal printer, -a button and a flash. -The web interface exists but may not be used, as the press of the button with take - a picture and activate the flash while simply informing the web page. + +The seconde is booth mode, where a Raspberry Pi is connected to +a thermal printer, a button and a flash. +The web interface exists but it's not needed, as the press of the button +takes a picture and activates the flash while simply informing the web page. """ # Following are the librairies we import, @@ -98,11 +101,7 @@ app.config["TEMPLATES_AUTO_RELOAD"] = True print_queue = PrintQueue(app) # Find out if we are running on a Raspberry Pi -rpi = Raspberry( - print_queue, - app, - configuration_file -) +rpi = Raspberry(print_queue, app, configuration_file) RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi() @@ -134,6 +133,7 @@ limiter = Limiter( app.logger.info("🖶 Welcome to LittlePrynter !") + # General routes @app.route("/") @limiter.limit("1/second", override_defaults=False) @@ -233,7 +233,6 @@ def web_print_img(): except werkzeug.exceptions.BadRequestKeyError: process_image = False - try: app.logger.debug("Sending the image to the printer.") web.print_image(file, sign, do_cut, process_image) @@ -287,7 +286,6 @@ def api_print_sms(): except werkzeug.exceptions.BadRequestKeyError: do_cut = False - try: # comment: We try to print the SMS web.print_sms(txt, sign, do_cut) @@ -390,11 +388,13 @@ def api_worker_stop(): worker.stop_worker() return jsonify(worker.current_state()) + @app.route("/api/printers") def api_printers_info(): """Returns the current state of every printers""" return jsonify(printers_pool.printers_info()) + ## Authentification diff --git a/src/print_queue.py b/src/print_queue.py index 38e3c3e..9a1ca56 100644 --- a/src/print_queue.py +++ b/src/print_queue.py @@ -77,7 +77,12 @@ class PrintQueue: with self._lock: self.app.logger.debug("Return current queue state") return [ - {"task_id": t.task_id, "status": t.status, "type": str(t.task_type), "content": str(t.get_print_data())} + { + "task_id": t.task_id, + "status": t.status, + "type": str(t.task_type), + "content": str(t.get_print_data()), + } for t in self._queue ] diff --git a/src/printer.py b/src/printer.py index ee337cb..fc250e9 100644 --- a/src/printer.py +++ b/src/printer.py @@ -16,7 +16,6 @@ from uuid import uuid4 from PIL import Image, ImageEnhance, ImageDraw, ImageFont import numpy as np - # Importing the modules needed for each supported printer Type import escpos.printer from brother_ql.models import ModelsManager @@ -42,6 +41,7 @@ class PrinterInfo: """ Brother-QL printer information """ + identifier: str backend: str protocol: str @@ -94,13 +94,14 @@ class Printer(ABC): """Takes a PrintTask and executes it""" def use_printer(self) -> None: - """Marks the printer as being used """ + """Marks the printer as being used""" self.used = True def free_printer(self) -> None: """Marks the printer as being not used, it's available.""" self.used = False + class EscPosPrinter(Printer): """ Create a new ESC/POS based printer. @@ -128,8 +129,10 @@ class EscPosPrinter(Printer): "The USB device is not plugged in : %s", str(e), ) + raise e except Exception as e: self.app.logger.error("Printer could not be connected : %s ", str(e)) + raise e try: if p.is_online(): @@ -187,7 +190,6 @@ class EscPosPrinter(Printer): return True def _print_txt(self, msg, signature="", bold=False): - self.ready = False if not isinstance(msg, str): self.app.logger.error( "It is not possible to print a " + str(type(msg)) + ", only strings." @@ -199,6 +201,7 @@ class EscPosPrinter(Printer): clean_signature = str(signature) # Make checks on the size of the message being printed + # TODO: make the max length configurable if len(clean_msg) > 4096: self.app.logger.warning( "Could not print message of this length: " + str(len(clean_msg)) @@ -208,7 +211,7 @@ class EscPosPrinter(Printer): + str(len(clean_msg)) + ", needs to be below 4096 caracters long." ) - + # TODO: make the max length configurable if len(signature) > 256: self.app.logger.warning( "Could not print signature of this length: " + str(len(clean_signature)) @@ -239,9 +242,9 @@ class EscPosPrinter(Printer): self.ready = True def _print_img(self, path, signature="", center=True, process=False): - self.ready = False clean_signature = str(signature) + # TODO: make the max length configurable if len(signature) > 256: self.app.logger.warning( "Could not print signature of this length: " + str(len(clean_signature)) @@ -306,7 +309,6 @@ class EscPosPrinter(Printer): self.ready = True def _qr(self, content): - self.ready = False try: self.printer.open(self.usb_args) self.printer.qr(content, center=True) @@ -321,7 +323,6 @@ class EscPosPrinter(Printer): self.ready = True def _cut(self): - self.ready = False try: self.printer.open(self.usb_args) self.printer.cut() @@ -338,21 +339,23 @@ class EscPosPrinter(Printer): has_paper = self._has_paper() is_ready = self.ready - self.app.logger.debug("Has paper : %s " , has_paper ) - self.app.logger.debug("Ready : %s " , is_ready ) + self.app.logger.debug("Has paper : %s ", has_paper) + self.app.logger.debug("Ready : %s ", is_ready) - return is_ready and has_paper # and is_online + return is_ready and has_paper # and is_online def print_task(self, task_type, data): """Execute actual print based on task type""" - with self._lock: self.app.logger.debug("Acquired lock to start print") i_m_ready = self._state() while not i_m_ready: - self.app.logger.debug("Waiting for the printer to become ready, current state %s ", str(i_m_ready)) + self.app.logger.debug( + "Waiting for the printer to become ready, current state %s ", + str(i_m_ready), + ) i_m_ready = self._state() time.sleep(0.3) @@ -391,9 +394,7 @@ class BrotherPrinter(Printer): def __init__(self, app, vendor_id, device_id): - super().__init__( - app, vendor_id, device_id, printer_type=PrinterType.BROTHER - ) + super().__init__(app, vendor_id, device_id, printer_type=PrinterType.BROTHER) self.printer = None self.usb_args = {} self.usb_args["idVendor"] = self.vendor_id @@ -431,7 +432,7 @@ class BrotherPrinter(Printer): self.app.logger.debug(f"Matched printer model: {model}") except ValueError as ve: - self.app.logger.warning("Invalid product ID format:"+ str(ve)) + self.app.logger.warning("Invalid product ID format:" + str(ve)) raise ve except Exception as e: self.app.logger.error("Error while finding a model ? : " + str(e)) @@ -459,13 +460,12 @@ class BrotherPrinter(Printer): def _print_txt(self, msg, signature="", bold=False): """Print text after converting it to an image""" self._print_img(text_to_image(msg)) + self._print_img(text_to_image(signature)) - def _print_img(self, data, process=False): + def _print_img(self, data, signature="", process=False): """ Print a raster image via a Brother QL printer """ - self.ready = False - if process: try: self.app.logger.debug("Proccessing the image") @@ -480,10 +480,13 @@ class BrotherPrinter(Printer): self.app.logger.warning("Not proccessing the image") try: selected_backend = guess_backend(self.printer_info.identifier) - BACKEND_CLASS = backend_factory(selected_backend)['backend_class'] + BACKEND_CLASS = backend_factory(selected_backend)["backend_class"] BACKEND_STRING_DESCR = self.printer_info.identifier - except Exception: - self.app.logger.error("Couldn't guess the backend to use from the printer string descriptor") + except Exception as e: + self.app.logger.error( + "Couldn't guess the backend to use from the printer string descriptor" + ) + raise e try: im = Image.open(processed_path) @@ -498,20 +501,40 @@ class BrotherPrinter(Printer): except Exception as e: raise e + if signature != "": + try: + im = Image.open(text_to_image(signature)) + qlr = BrotherQLRaster(self.printer_info.model) + create_label( + qlr, im, self.printer_info.label_type, cut=False, dither=True + ) + + # noinspection PyCallingNonCallable + be = BACKEND_CLASS(BACKEND_STRING_DESCR) + be.write(qlr.data) + be.dispose() + del be + except Exception as e: + raise e + def _cut(self): - self.ready = False try: selected_backend = guess_backend(self.printer_info.identifier) - BACKEND_CLASS = backend_factory(selected_backend)['backend_class'] + BACKEND_CLASS = backend_factory(selected_backend)["backend_class"] BACKEND_STRING_DESCR = self.printer_info.identifier except Exception as e: - self.app.logger.error("Couldn't guess the backend to use from the printer string descriptor") + self.app.logger.error( + "Couldn't guess the backend to use from the printer string descriptor" + ) + raise e data = "src/static/images/1x1.png" try: im = Image.open(data) qlr = BrotherQLRaster(self.printer_info.model) - create_label(qlr, im, self.printer_info.label_type, threshold=70, cut=True, rotate=0) + create_label( + qlr, im, self.printer_info.label_type, threshold=70, cut=True, rotate=0 + ) # noinspection PyCallingNonCallable be = BACKEND_CLASS(BACKEND_STRING_DESCR) @@ -521,32 +544,41 @@ class BrotherPrinter(Printer): except Exception as e: raise e + def _qr(self): + # This should not be reached by a print on a BrotherQL + raise NotImplementedError + def print_task(self, task_type, data): """Execute actual print based on task type""" with self._lock: - if self.ready: - try: - match (task_type.value): - case "image": - self._print_img(data["img"]) - self.ready = True - case "cut": - self._cut() - self.ready = True - case "text": - self._print_txt(data["txt"], signature=data["sign"]) - self.ready = True - case _: - raise RuntimeError("This task type is not supported") - except Exception as e: - raise e - # end try - else: - raise RuntimeError("The printer is not ready to print yet !") + self.app.logger.debug("Acquired lock to start print") + i_m_ready = self._state() + while not i_m_ready: + self.app.logger.debug( + "Waiting for the printer to become ready, current state %s ", + str(i_m_ready), + ) + i_m_ready = self._state() + time.sleep(0.3) - # These values are by default for now + self.app.logger.debug("Checked state to start printing : %s", self._state()) + self.ready = False - # raise NotImplementedError("This printer type is not implemented yet") + try: + match (task_type.value): + case "image": + self._print_img(data["img"], signature=data["sign"]) + self.ready = True + case "cut": + self._cut() + self.ready = True + case "text": + self._print_txt(data["txt"], signature=data["sign"]) + self.ready = True + case _: + raise RuntimeError("This task type is not supported") + except Exception as e: + raise e def _process_image(self, path, max_width=575, max_height=1000): @@ -563,15 +595,12 @@ def _process_image(self, path, max_width=575, max_height=1000): grayscale = img.convert("L") avg_brightness = np.asarray(grayscale, dtype=np.float32).mean() - self.app.logger.debug( - "Average brightness of the image: %.2f", avg_brightness - ) + self.app.logger.debug("Average brightness of the image: %.2f", avg_brightness) if avg_brightness < brightness_threshold: brightness_factor = min( max( - 1 + (brightness_threshold - avg_brightness) - / brightness_threshold, + 1 + (brightness_threshold - avg_brightness) / brightness_threshold, 1.1, ), 2.5, @@ -580,9 +609,7 @@ def _process_image(self, path, max_width=575, max_height=1000): "Image is dark; increasing brightness by %.2f", brightness_factor, ) - grayscale = ImageEnhance.Brightness(grayscale).enhance( - brightness_factor - ) + grayscale = ImageEnhance.Brightness(grayscale).enhance(brightness_factor) contrast = float(np.asarray(grayscale, dtype=np.float32).std()) self.app.logger.debug("Grayscale standard deviation: %.2f", contrast) @@ -596,6 +623,7 @@ def _process_image(self, path, max_width=575, max_height=1000): self.app.logger.debug("Processed and saved image: %s", jpeg_path) return jpeg_path + def text_to_image( text, width=400, @@ -639,7 +667,11 @@ def text_to_image( line = "" piece = "" for char in word: - if piece and draw.textbbox((0, 0), piece + char, font=font)[2] > max_text_width: + if ( + piece + and draw.textbbox((0, 0), piece + char, font=font)[2] + > max_text_width + ): lines.append(piece) piece = char else: @@ -654,11 +686,7 @@ def text_to_image( ascent, descent = font.getmetrics() line_height = ascent + descent - height = ( - 2 * padding - + len(lines) * line_height - + (len(lines) - 1) * line_spacing - ) + height = 2 * padding + len(lines) * line_height + (len(lines) - 1) * line_spacing image = Image.new("RGB", (width, height), background) draw = ImageDraw.Draw(image) @@ -674,4 +702,4 @@ def text_to_image( image_path = upload_dir / f"{uuid4().hex}.png" image.save(image_path, format="PNG") - return str(image_path) \ No newline at end of file + return str(image_path) diff --git a/src/printers.py b/src/printers.py index 913bcfb..e69333f 100644 --- a/src/printers.py +++ b/src/printers.py @@ -49,10 +49,7 @@ class Printers: # Find all connected USB devices # the find() fnc returns a generator devices = list( - usb.core.find( - find_all=True, - custom_match=_FindClass(self.app,7) - ) + usb.core.find(find_all=True, custom_match=_FindClass(self.app, 7)) ) if devices: @@ -60,15 +57,15 @@ class Printers: if waiting_time_for_printer <= 0: self.app.logger.warning( - "No USB devices of class 7 ( printers ) found or pyusb could not access the bus." + "No USB devices of class 7 ( printers ) found or pyusb could not access the bus." ) raise RuntimeError( - "No USB devices of class 7 ( printers ) found or pyusb could not access the bus." + "No USB devices of class 7 ( printers ) found or pyusb could not access the bus." ) self.app.logger.warning( "No printers found; waiting for a printer to connect..." - ) + ) time.sleep(1) waiting_time_for_printer -= 1 @@ -120,10 +117,12 @@ class Printers: self.app.logger.debug("Trying to create a new BROTHER printer") brother_printer = BrotherPrinter( - self.app, vendor_id=dev.idVendor, device_id=dev.idProduct) + self.app, vendor_id=dev.idVendor, device_id=dev.idProduct + ) except Exception as e: self.app.logger.error( - "Could not create a %s printer class with %s:%s" , product, + "Could not create a %s printer class with %s:%s", + product, dev.idVendor, dev.idProduct, ) @@ -131,12 +130,14 @@ class Printers: # If the object creation is successfull, we add it to the list of Printers printers.add(brother_printer) - self.app.logger.debug("Found a %s printer" , manufacturer) + self.app.logger.debug("Found a %s printer", manufacturer) - self.app.logger.debug("Found %s printers" , len(printers)) + self.app.logger.debug("Found %s printers", len(printers)) if len(printers) < 1: - self.app.logger.warning("Not printers found ! Please plug in a Printer and restart the program.") + self.app.logger.warning( + "Not printers found ! Please plug in a Printer and restart the program." + ) raise RuntimeError("No printers found") return printers @@ -163,26 +164,32 @@ class Printers: """Return the current list of printers""" return self.printers + class _FindClass: """ Use by usb.core to modify the way USB devices are found Taken from pyUSB documentation on Github """ + def __init__(self, app, class_): self._class = class_ self.app = app def __call__(self, device): # first, let's check the device - self.app.logger.debug("Found a USB device of class %s", str(device.bDeviceClass)) + self.app.logger.debug( + "Found a USB device of class %s", str(device.bDeviceClass) + ) if device.bDeviceClass == self._class: return True # ok, transverse all devices to find an # interface that matches our class for configuration in device: # Every device can have one or more configuration, - interface = usb.util.find_descriptor(configuration, bInterfaceClass=self._class) - # each config can have several interfaces + interface = usb.util.find_descriptor( + configuration, bInterfaceClass=self._class + ) + # each config can have several interfaces if interface is not None: return True diff --git a/src/worker.py b/src/worker.py index 42ae3c1..04dbc8e 100644 --- a/src/worker.py +++ b/src/worker.py @@ -20,6 +20,7 @@ class PrintWorker(threading.Thread): If a print fails, it's not retried, but will be added to a list of completed tasks. TODO : Change that """ + def __init__(self, app, print_queue, printers_pool): super().__init__(daemon=True) self.app = app @@ -51,12 +52,11 @@ class PrintWorker(threading.Thread): self._printer_index = (index + 1) % len(self.printers_obj) return printer - def run(self): """Background thread that processes queue items""" self.app.logger.debug("Worker %s started working.", threading.get_ident()) - self.app.logger.debug("Current threads : %s" , threading.active_count()) - self.app.logger.debug("Threads actives : %s " , threading.enumerate()) + self.app.logger.debug("Current threads : %s", threading.active_count()) + self.app.logger.debug("Threads actives : %s ", threading.enumerate()) while True: @@ -73,7 +73,9 @@ class PrintWorker(threading.Thread): # If we do not have a printer, we try to find one while self.running and self.printer is None: - self.app.logger.debug("Trying to find a available printer for the worker.") + self.app.logger.debug( + "Trying to find a available printer for the worker." + ) self.printer = self._acquire_available_printer() self.app.logger.debug( "The worker got a %s printer and it's %s", @@ -81,14 +83,11 @@ class PrintWorker(threading.Thread): "Ready" if self.printer.ready else "Not ready", ) - if self.printer is None: time.sleep(1) - self.printer.use_printer() - # If the printer is not ready, we wait for it to become ready. while not self.printer.ready: self.app.debug("Waiting for our printer to become ready...") @@ -115,7 +114,9 @@ class PrintWorker(threading.Thread): if task: try: self.app.logger.info("Got a new task") - self.app.logger.debug("Got task %s %s", task.task_type, task.task_id) + self.app.logger.debug( + "Got task %s %s", task.task_type, task.task_id + ) task.status = "processing" print_data = task.get_print_data() @@ -132,7 +133,7 @@ class PrintWorker(threading.Thread): task.status = "completed" self.print_queue.mark_completed(task.task_id, "completed") self.app.logger.debug( - "Finished printing task %s " , task.task_id + "Finished printing task %s ", task.task_id ) self.state = "idle" @@ -141,7 +142,7 @@ class PrintWorker(threading.Thread): self.state = "idle" self.print_queue.mark_completed(task.task_id, "failed") self.app.logger.error( - "Could not print task %s because %s " , task.task_id, str(e) + "Could not print task %s because %s ", task.task_id, str(e) ) else: