From 3c490e10b4ddede0af5f5704c84bc080c8ba514a Mon Sep 17 00:00:00 2001 From: n07070 Date: Thu, 4 Jun 2026 00:31:04 +0200 Subject: [PATCH] Apply black formatter --- src/main.py | 3 +++ src/print_queue.py | 5 +++- src/printer.py | 60 ++++++++++++++++++++++++++++------------------ src/printers.py | 49 +++++++++++++++++++++++++------------ src/raspberry.py | 9 ++++--- src/task.py | 6 ++++- src/web.py | 4 ++-- src/worker.py | 16 +++++++++---- 8 files changed, 103 insertions(+), 49 deletions(-) diff --git a/src/main.py b/src/main.py index 75b5cff..f428402 100644 --- a/src/main.py +++ b/src/main.py @@ -126,6 +126,7 @@ limiter = Limiter( get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"] ) + # General routes @app.route("/") @limiter.limit("1/second", override_defaults=False) @@ -317,11 +318,13 @@ def api_queue_status(): """API endpoint for entire queue""" return jsonify(web.get_queue_state()) + @app.route("/api/queue/completed", methods=["GET"]) def api_queue_completed(): """API endpoint that returns the finished tasks""" return jsonify(web.get_queue_completed()) + @app.route("/api/worker", methods=["GET"]) def api_worker_state(): """API endpoint to get the worker state""" diff --git a/src/print_queue.py b/src/print_queue.py index 2922143..0dfc24f 100644 --- a/src/print_queue.py +++ b/src/print_queue.py @@ -76,7 +76,10 @@ class PrintQueue: """Return current queue state""" 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) } for t in self._queue] + return [ + {"task_id": t.task_id, "status": t.status, "type": str(t.task_type)} + for t in self._queue + ] def get_queue_completed(self): """Return completed queue elements""" diff --git a/src/printer.py b/src/printer.py index 5e32dde..222c847 100644 --- a/src/printer.py +++ b/src/printer.py @@ -1,8 +1,10 @@ """ This class manages connexion to a Printer """ + from time import sleep import os.path +import os from abc import ABC, abstractmethod from dataclasses import dataclass @@ -14,7 +16,6 @@ import threading from PIL import Image, ImageEnhance import numpy as np - # Importing the modules needed for each supported printer Type import escpos.printer from brother_ql.models import ModelsManager @@ -24,12 +25,15 @@ from brother_ql.raster import BrotherQLRaster from brother_ql.conversion import convert from brother_ql.backends.helpers import send + class PrinterType(Enum): - """ - What are the capacities of a Printer ? - """ - EPSON = "epson" - BROTHER = "brother" + """ + What are the capacities of a Printer ? + """ + + EPSON = "epson" + BROTHER = "brother" + # For Brother-QL Printers @dataclass @@ -44,7 +48,7 @@ class PrinterInfo: model: str = "QL-570" status: str = "unknown" label_type: str = "unknown" - label_size : str = "unknown" + label_size: str = "unknown" label_width: int = 0 label_height: int = 0 @@ -54,6 +58,7 @@ class PrinterInfo: def __setitem__(self, key, value): setattr(self, key, value) + class Printer(ABC): """ If it outputs printed paper and speaks like a printer, then it must be a printer. @@ -80,7 +85,7 @@ class Printer(ABC): """Reports the state of the Printer""" @abstractmethod - def print_task(self, task_type, data)-> None: + def print_task(self, task_type, data) -> None: """Takes a PrintTask and executes it""" @@ -96,7 +101,7 @@ class EscPosPrinter(Printer): Making sure it has paper, Define default print settings """ - super().__init__(app,vendor_id,device_id, printer_type=PrinterType.EPSON) + super().__init__(app, vendor_id, device_id, printer_type=PrinterType.EPSON) self.printer = None self.usb_args = {} self.usb_args["idVendor"] = self.vendor_id @@ -105,9 +110,7 @@ class EscPosPrinter(Printer): try: # This also calls open(), which we need to close() # or else the device will appear as busy. - p = escpos.printer.Usb( - self.vendor_id, self.device_id, 0, profile="TM-P80" - ) + p = escpos.printer.Usb(self.vendor_id, self.device_id, 0, profile="TM-P80") except escpos.exceptions.DeviceNotFoundError as e: self.app.logger.error( "The USB device is not plugged in : %s", @@ -141,7 +144,7 @@ class EscPosPrinter(Printer): # Beware : if we print every time the printer becomes ready, it means # we are printing before and after every print ! self.printer = p - self.printer.close() # We close the connexion to the Printer + self.printer.close() # We close the connexion to the Printer try: self._has_paper() @@ -339,13 +342,16 @@ class EscPosPrinter(Printer): else: raise RuntimeError("The printer is not ready to print yet !") + class BrotherPrinter(Printer): """ Manages connexion and capabilities of a BrotherQL 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.device_id @@ -363,7 +369,9 @@ class BrotherPrinter(Printer): parts = identifier.split("/") if len(parts) < 4: - self.app.logger.warning(f"Skipping device with invalid identifier format: {identifier}") + self.app.logger.warning( + f"Skipping device with invalid identifier format: {identifier}" + ) continue protocol = parts[0] @@ -378,7 +386,7 @@ class BrotherPrinter(Printer): break self.app.logger.debug(f"Matched printer model: {model}") except ValueError: - self.app.logger.warning(f"Invalid product ID format: {product_id}") + self.app.logger.warning(f"Invalid product ID format: {m.product_id}") self.printer_info = PrinterInfo( identifier=identifier, @@ -392,14 +400,13 @@ class BrotherPrinter(Printer): self.ready = True - def _has_paper(self): raise NotImplementedError("This printer model does not support this.") def _state(self): return self.ready - def _print_img(self,data): + def _print_img(self, data): """ Print a raster image via a Brother QL printer """ @@ -426,7 +433,7 @@ class BrotherPrinter(Printer): ) # Debug logging - if FLASK_DEBUG: + if os.getenv("FLASK_DEBUG"): self.app.logger.debug(f""" Print parameters: - Label type: {label_type} @@ -447,11 +454,15 @@ class BrotherPrinter(Printer): # } status = send( instructions=instructions, - printer_identifier= self.printer_info["identifier"], - backend_identifier="pyusb" + printer_identifier=self.printer_info["identifier"], + backend_identifier="pyusb", ) - if not status["did_print"] or status["outcome"] == "error" or status["outcome"] == "unknown": + if ( + not status["did_print"] + or status["outcome"] == "error" + or status["outcome"] == "unknown" + ): raise RuntimeError("Failed to print using Python API") if status["printer_state"]: @@ -462,7 +473,9 @@ class BrotherPrinter(Printer): except usb.core.USBError as e: # Treat timeout errors as successful since they often occur after print completion if e.errno == 110: # Operation timed out - self.app.logger.debug("USB timeout occurred - this is normal and the print likely completed") + self.app.logger.debug( + "USB timeout occurred - this is normal and the print likely completed" + ) self.app.logger.debug("Print completed (timeout is normal)") self.ready = True @@ -496,6 +509,7 @@ class BrotherPrinter(Printer): # raise NotImplementedError("This printer type is not implemented yet") + def _process_image(self, path): brightness_factor = 1.5 # Used only if image is too dark brightness_threshold = 100 # Brightness threshold (0–255) diff --git a/src/printers.py b/src/printers.py index 7df1d99..379ce18 100644 --- a/src/printers.py +++ b/src/printers.py @@ -3,15 +3,18 @@ A collection of Printers. It has methods to discover printers, and provides an interface for the methods expected from printers. """ + from collections.abc import Mapping, Set import usb.core import usb.util from printer import Printer, EscPosPrinter, BrotherPrinter, PrinterType -class Printers(): + +class Printers: """ Finds and creates a set of Printer that can be used by the Workers to print. """ + def __init__(self, app): """ Discover printers connected to the computer and return a Collection of Printer() @@ -36,10 +39,14 @@ class Printers(): printers = set() # Find all connected USB devices - devices = usb.core.find(find_all=True,custom_match=_FindClass(7)) + devices = usb.core.find(find_all=True, custom_match=_FindClass(7)) if not devices: - self.app.logger.warning("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.") + self.app.logger.warning( + "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." + ) for dev in devices: # Attempt to get the manufacturer and product strings @@ -52,7 +59,13 @@ class Printers(): product = usb.util.get_string(dev, dev.iProduct) except Exception: product = "Unknown" - self.app.logger.debug("Looking at %s %s (%s:%s)", manufacturer, product, hex(dev.idVendor), hex(dev.idProduct)) + self.app.logger.debug( + "Looking at %s %s (%s:%s)", + manufacturer, + product, + hex(dev.idVendor), + hex(dev.idProduct), + ) if manufacturer == "EPSON": try: @@ -60,13 +73,15 @@ class Printers(): self.app.logger.debug("Trying to creat a new EPSON printer") prid = dev.idProduct vendir = dev.idVendor - escpos_printer = EscPosPrinter(self.app, vendor_id=vendir, device_id=prid) + escpos_printer = EscPosPrinter( + self.app, vendor_id=vendir, device_id=prid + ) except Exception as e: raise e # If the object creation is successfull, we add it to the list of Printers printers.add(escpos_printer) - self.app.logger.debug("Found a %s printer" % manufacturer ) + self.app.logger.debug("Found a %s printer" % manufacturer) # We already found the type of printer, # we don't need an extra comparaison. @@ -79,14 +94,20 @@ class Printers(): self.app.logger.debug("Trying to creat a new BROTHER printer") prid = dev.idProduct vendir = dev.idVendor - brother_printer = BrotherPrinter(self.app, vendor_id=vendir,device_id=prid) + brother_printer = BrotherPrinter( + self.app, vendor_id=vendir, device_id=prid + ) except Exception as e: - self.app.logger.error("Could not create a %s printer class with %s:%s" % product, dev.idVendor, dev.idProduct) + self.app.logger.error( + "Could not create a %s printer class with %s:%s" % product, + dev.idVendor, + dev.idProduct, + ) raise e # 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)) return printers @@ -110,9 +131,10 @@ class Printers(): return NotImplementedError() -class _FindClass(): +class _FindClass: def __init__(self, class_): self._class = class_ + def __call__(self, device): # first, let's check the device if device.bDeviceClass == self._class: @@ -121,10 +143,7 @@ class _FindClass(): # interface that matches our class for cfg in device: # find_descriptor: what's it? - intf = usb.util.find_descriptor( - cfg, - bInterfaceClass=self._class - ) + intf = usb.util.find_descriptor(cfg, bInterfaceClass=self._class) if intf is not None: return True diff --git a/src/raspberry.py b/src/raspberry.py index a75b401..7e1841b 100644 --- a/src/raspberry.py +++ b/src/raspberry.py @@ -15,7 +15,8 @@ from gpiozero import Button, LED, DigitalOutputDevice from PIL import Image from task import TextTask, ImageTask, CutTask -class Raspberry(): + +class Raspberry: """ This class will manage three things : - Connecting to a USB webcam @@ -263,8 +264,10 @@ class Raspberry(): self.app.logger.debug("Printing picture") self.led.on() self.crop_to_square(self.image_path) - self.print_queue.enqueue(ImageTask(self.image_path,signature="",process=True)) - self.print_queue.enqueue(TextTask(content="Imprimé par LittlePrynter", signature="")) + self.print_queue.enqueue(ImageTask(self.image_path, signature="", process=True)) + self.print_queue.enqueue( + TextTask(content="Imprimé par LittlePrynter", signature="") + ) time = strftime("%Y-%m-%d %H:%M", gmtime()) self.print_queue.enqueue(TextTask(content=time, signature="")) self.print_queue.enqueue(CutTask()) diff --git a/src/task.py b/src/task.py index 9fc38a9..5b45920 100644 --- a/src/task.py +++ b/src/task.py @@ -14,6 +14,7 @@ to print at the same time. We can also delay and store printing tasks until a printer becomes available if none is online. """ + from abc import ABC, abstractmethod ## See https://docs.python.org/3/library/abc.html to learn more about this @@ -27,6 +28,7 @@ class TaskType(Enum): """ The different tasks supported by the printers """ + TEXT = "text" IMAGE = "image" CUT = "cut" @@ -47,7 +49,6 @@ class PrintTask(ABC): def get_print_data(self): """Return data formatted for printer""" - def _generate_id(self): # Generate unique task ID return str(uuid.uuid4()) @@ -66,8 +67,10 @@ class TextTask(PrintTask): def get_print_data(self): return {"txt": self.content, "sign": self.signature} + class QRTask(TextTask): """This task prints a QR-Code, the signature is ignore and is always the content itself""" + def __init__(self, content): super().__init__(content, signature="") self.content = content @@ -76,6 +79,7 @@ class QRTask(TextTask): def get_print_data(self): return {"txt": self.content, "sign": self.signature} + class ImageTask(PrintTask): """ This tasks represents a image content ( in the form of it's path ), and it's signature. diff --git a/src/web.py b/src/web.py index 1db14dd..b31cd10 100644 --- a/src/web.py +++ b/src/web.py @@ -6,7 +6,7 @@ from task import TextTask, ImageTask, CutTask class Web(object): """Web is the class that gets all of the information from web calls - ( API and Web page ) and provides checks before sending stuff to printing""" + ( API and Web page ) and provides checks before sending stuff to printing""" def __init__(self, app, print_queue): super(Web).__init__() @@ -116,4 +116,4 @@ class Web(object): def get_queue_completed(self): """Return completed queue elements""" - return self.print_queue.get_queue_completed() \ No newline at end of file + return self.print_queue.get_queue_completed() diff --git a/src/worker.py b/src/worker.py index e5a7fec..ad10220 100644 --- a/src/worker.py +++ b/src/worker.py @@ -31,7 +31,7 @@ class PrintWorker(threading.Thread): """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("Threads actives : %s " % threading.enumerate()) while True: @@ -48,7 +48,11 @@ class PrintWorker(threading.Thread): try: self.app.logger.debug("Changing printers") self.printer = next(self.printers) - self.app.logger.debug("The worker got a %s printer and it's %s", self.printer.printer_type, "Ready" if self.printer.ready else "Not ready") + self.app.logger.debug( + "The worker got a %s printer and it's %s", + self.printer.printer_type, + "Ready" if self.printer.ready else "Not ready", + ) except Exception as e: self.app.logger.error(str(e)) self.printer = None @@ -89,7 +93,9 @@ class PrintWorker(threading.Thread): task.status = "completed" self.print_queue.mark_completed(task.task_id, "completed") self._emit_status(task.task_id, "completed") - self.app.logger.debug("Finished printing task %s " % task.task_id) + self.app.logger.debug( + "Finished printing task %s " % task.task_id + ) self.state = "idle" except RuntimeError as e: @@ -97,7 +103,9 @@ class PrintWorker(threading.Thread): self.state = "idle" self.print_queue.mark_completed(task.task_id, "failed") self._emit_status(task.task_id, "failed", error=str(e)) - self.app.logger.error("Could not print task %s because %s " % task.task_id, str(e)) + self.app.logger.error( + "Could not print task %s because %s " % task.task_id, str(e) + ) else: # When they are no new tasks to handle, we put the thread to sleep.