Change Error to warning when not on a Raspberry

because this is a real-time command, we might not get the good answer
and fail to print fast enough. See https://download4.epson.biz/sec_pubs/pos/reference_en/escpos/realtime_commands.html

src/main.py

@@ -83,14 +83,14 @@ except PermissionError:
     sys.exit(77)
 
 # Output the config file
-if os.getenv("FLASK_DEBUG"):
+if not os.getenv("FLASK_DEBUG") is None and os.getenv("FLASK_DEBUG") is True:
     pprint.pprint(configuration_file)
 
 # We define the app module used by Flask
 app.secret_key = configuration_file["secrets"]["flask_secret_key"]
 app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER
 app.config["ALLOWED_EXTENSIONS"] = ALLOWED_EXTENSIONS
-app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000  # Maximum 3Mb for a file upload
+app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000  # Maximum 10Mb for a file upload
 app.config["TEMPLATES_AUTO_RELOAD"] = True
 
 # Queue creation
@@ -100,11 +100,7 @@ print_queue = PrintQueue(app)
 rpi = Raspberry(
     print_queue,
     app,
-    socketio,
-    configuration_file["rpi"]["button_gpio_port_number"],
-    configuration_file["rpi"]["indicator_gpio_port_number"],
-    configuration_file["rpi"]["flash_gpio_port_number"],
-    configuration_file["rpi"]["flash"],
+    configuration_file
 )
 RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
 
@@ -115,7 +111,7 @@ web = Web(app, print_queue)
 # Start worker thread
 # When created, the worker will try to find printers connected to the system
 try:
-    worker = PrintWorker(app, print_queue, socketio)
+    worker = PrintWorker(app, print_queue)
     worker.start()
 except Exception as e:
     app.logger.error("Could not start the worker because %s ", str(e))
@@ -126,6 +122,9 @@ limiter = Limiter(
     get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"]
 )
 
+
+app.logger.info("🖶 Welcome to LittlePrynter !")
+
 # General routes
 @app.route("/")
 @limiter.limit("1/second", override_defaults=False)
@@ -192,6 +191,9 @@ def web_print_img():
             "No signature found for this print, using default signature : %s", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
+    except werkzeug.exceptions.RequestEntityTooLarge as e:
+        flash("Whoops, image is too big: " + str(e), "error")
+        return redirect(url_for("index"))
 
     # check if the post request has the file part
     if "img" not in request.files:
@@ -317,11 +319,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"""
@@ -403,4 +407,4 @@ def camera_status():
 
 
 if __name__ == "__main__":
-    app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")
+    app.run(use_reloader=False, host="0.0.0.0", ssl_context="adhoc")

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), "content": str(t.get_print_data())}
+                for t in self._queue
+            ]
 
     def get_queue_completed(self):
         """Return completed queue elements"""

src/printer.py

@@ -1,39 +1,43 @@
 """
 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
-
+import time
 from enum import Enum
 import uuid
-import usb.core
 import threading
-
+import usb.core
 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
 from brother_ql.backends import backend_factory
-from brother_ql import labels
 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"
 
+
 # For Brother-QL Printers
 @dataclass
 class PrinterInfo:
+    """
+    Brother-QL printer information
+    """
     identifier: str
     backend: str
     protocol: str
@@ -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",
@@ -172,7 +175,7 @@ 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."
@@ -212,6 +215,7 @@ class EscPosPrinter(Printer):
             self.printer.textln(clean_msg)
             if clean_signature:
                 self.printer.textln(clean_signature)
+            self.printer.textln()
             self.printer.close()
         except Exception as e:
             self.app.logger.error("Unable to print because : " + str(e))
@@ -220,9 +224,10 @@ class EscPosPrinter(Printer):
             ) from e
 
         self.app.logger.info("Printed text")
-        return True
+        self.ready = True
 
     def _print_img(self, path, signature="", center=True, process=False):
+        self.ready = False
         clean_signature = str(signature)
 
         if len(signature) > 256:
@@ -248,30 +253,33 @@ class EscPosPrinter(Printer):
         if process:
             try:
                 self.app.logger.debug("Proccessing the image")
-                path = _process_image(self, path)
+                processed_path = _process_image(self, path)
             except RuntimeError as e:
                 self.app.logger.error(
                     "Error while processing the image, aborting print : %s", str(e)
                 )
                 raise e
         else:
+            processed_path = path
             self.app.logger.warning("Not proccessing the image")
 
         try:
             self.printer.open(self.usb_args)
-            self.printer.image(path, center=center)
+            self.printer.image(processed_path, center=center)
             self.printer.textln(signature)
             self.printer.close()
-            self.app.logger.debug("Printed an image : " + str(path))
+            self.app.logger.debug("Printed an image : " + str(processed_path))
         except Exception as e:
             self.app.logger.error(str(e))
             raise RuntimeError("Could not print the picture") from e
         finally:
             try:
                 os.remove(path)
+                os.remove(processed_path)
             except OSError as e:
                 raise e
 
+            self.app.logger.debug("Removed image : " + str(processed_path))
             self.app.logger.debug("Removed image : " + str(path))
 
             try:
@@ -283,9 +291,10 @@ class EscPosPrinter(Printer):
                 raise RuntimeError("Could not close the printer connexion. ") from e
 
         self.app.logger.info("Printed a picture")
-        return True
+        self.ready = True
 
     def _qr(self, content):
+        self.ready = False
         try:
             self.printer.open(self.usb_args)
             self.printer.qr(content, center=True)
@@ -297,8 +306,10 @@ class EscPosPrinter(Printer):
             raise e
 
         self.app.logger.info("Printed a QR")
+        self.ready = True
 
     def _cut(self):
+        self.ready = False
         try:
             self.printer.open(self.usb_args)
             self.printer.cut()
@@ -309,35 +320,57 @@ class EscPosPrinter(Printer):
             raise e
 
         self.app.logger.info("Did a cut")
-        return True
+        self.ready = True
 
-    def _state(self):
-        return self.printer.is_online() and self.ready and self._has_paper()
+    def _state(self) -> bool:
+        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 )
+
+        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:
-            if self._state:
-                self._state = False
+            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)
+
+            self.app.logger.debug("Checked state to start printing : %s", self._state())
+            self.ready = False
+            try:
+                self.app.logger.debug("Checking task type")
                 match (task_type.value):
                     case "text":
+
                         self._print_txt(data["txt"], signature=data["sign"])
-                        self._state = True
+                        self.ready = True
                     case "image":
                         self._print_img(
                             data["img"], signature=data["sign"], process=data["process"]
                         )
-                        self._state = True
+                        self.ready = True
                     case "cut":
                         self._cut()
-                        self._state = True
+                        self.ready = True
                     case "qr":
                         self._qr(data["txt"])
-                        self._state = True
+                        self.ready = True
                     case _:
                         raise RuntimeError("This task type is not supported")
-            else:
-                raise RuntimeError("The printer is not ready to print yet !")
+            except Exception as e:
+                self.app.logger.debug("Exception occured while printing %s", str(e))
+                self.ready = True
+                raise RuntimeError from e
+
 
 class BrotherPrinter(Printer):
     """
@@ -345,7 +378,9 @@ 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.device_id
@@ -363,11 +398,13 @@ 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]
-            device_info = parts[2]
+            # device_info = parts[2]
             serial_number = parts[3]
 
         try:
@@ -378,7 +415,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 +429,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 +462,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 +483,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 +502,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,17 +538,18 @@ 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)
-    contrast_factor = 0.6  # Less than 1.0 = lower contrast
+    contrast_factor = 2  # Less than 1.0 = lower contrast
     max_width = 575
     max_height = 1000
 
     with Image.open(path) as original_img:
         # Convert to RGB if needed (JPEG doesn't support alpha)
         if original_img.mode in ("RGBA", "P"):
-            self.app.logger.debug("Converting the image to RGB from RGBA")
+            self.app.logger.debug("Converting the image from RGBA to RGBA")
             original_img = original_img.convert("RGB")
 
         # Resize while maintaining aspect ratio
@@ -536,16 +579,19 @@ def _process_image(self, path):
             self.app.logger.debug(
                 f"Image too dark, increasing brightness by a factor of {brightness_factor:.2f}"
             )
-            enhancer = ImageEnhance.Brightness(original_img)
-            original_img = enhancer.enhance(brightness_factor)
+            enhancer = ImageEnhance.Brightness(grayscale)
+            grayscale = enhancer.enhance(brightness_factor)
 
-        # # Reduce contrast
-        # contrast_enhancer = ImageEnhance.Contrast(original_img)
-        # original_img = contrast_enhancer.enhance(contrast_factor)
+        # Computer current contrast of grayscale image
+        contrast = np.clip(np.std(np.array(grayscale)), 0, 255)
+        self.app.logger.debug("Standard deviation of the contrast : %s", contrast)
+        # # Enhance contrast
+        contrast_enhancer = ImageEnhance.Contrast(grayscale)
+        original_img = contrast_enhancer.enhance(contrast_factor)
 
         # Convert to JPEG and save
         jpeg_path = os.path.splitext(path)[0] + "_processed.jpg"
-        original_img.save(jpeg_path, format="JPEG", quality=95, optimize=True)
+        grayscale.save(jpeg_path, format="JPEG", quality=95, optimize=True)
         self.app.logger.debug("Processed and saved image.")
 
     return jpeg_path

src/printers.py

@@ -1,17 +1,21 @@
 """
 A collection of Printers.
 
-It has methods to discover printers, and provides an interface for the methods expected from printers.
+It has methods to discover printers, and provides an interface for
+the methods expected from printers.
 """
-from collections.abc import Mapping, Set
+
+from collections.abc import Set
 import usb.core
 import usb.util
-from printer import Printer, EscPosPrinter, BrotherPrinter, PrinterType
+from printer import Printer, EscPosPrinter, BrotherPrinter
 
-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 +40,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 +60,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 +74,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,16 +95,27 @@ 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))
+
+        if len(printers) < 1:
+            self.app.logger.warning("Not printers found ! Please plug in a Printer and restart the program.")
+            raise RuntimeError("No printers found")
 
-        self.app.logger.debug("Found %s printers" % len(printers))
         return printers
 
     def any(self) -> Printer:
@@ -101,18 +128,23 @@ class Printers():
         else:
             raise RuntimeError("No printers available")
 
-    def get_printer(self, printer_type):
+    # def get_printer(self, printer_type):
+    #     """
+    #     Return a specific printer
+
+    #     printer_type -- a printer type
+    #     """
+    #     return NotImplementedError()
+
+
+class _FindClass:
     """
-        Return a specific printer
-
-        printer_type -- a printer type
+    Use by usb.core to modify the way USB devices are found
+    Taken from pyUSB documentation on Github
     """
-        return NotImplementedError()
-
-
-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 +153,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
 

src/raspberry.py

@@ -10,12 +10,12 @@ import io  # To check if we are on a Raspberry Pi
 import subprocess
 import os
 from time import sleep, gmtime, strftime
-from flask_socketio import SocketIO
 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
@@ -23,29 +23,22 @@ class Raspberry():
     - Activating a flash ( or light )
     - Flash an indicator light
 
+    # pylint: disable=too-many-instance-attributes
+    # dede
     """
 
     def __init__(
         self,
         print_queue,
         app,
-        socketio,
-        button_gpio_port_number,
-        indicator_gpio_port_number,
-        flash_gpio_port_number,
-        is_flash_present,
+        configuration_file
     ):
         self.print_queue = print_queue
-        self.socketio = socketio
         self.app = app
-
-        self.flash_gpio = flash_gpio_port_number
-        self.is_flash_present = is_flash_present
-        self.button_gpio = button_gpio_port_number
-        self.led_gpio = indicator_gpio_port_number
+        self.configuration_file = configuration_file
         self.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg"
 
-    def is_raspberry_pi(self, raise_on_errors=False):
+    def is_raspberry_pi(self):
         """
         Checking if we are on a Raspberry Pi by checking
         information on the /proc/cpuinfo file
@@ -72,9 +65,10 @@ class Raspberry():
                             return False
 
                 if not found:
-                    self.app.logger.error(
-                        "Couldn't get sufficient hardware information from /proc/cpuinfo, Unable to determine if we are on a Raspberry Pi."
+                    self.app.logger.warning(
+                        "Couldn't get sufficient hardware information from /proc/cpuinfo"
                     )
+                    self.app.logger.warning("Unable to determine if we are on a Raspberry Pi.")
                     return False
         except IOError:
             self.app.logger.error("Unable to open `/proc/cpuinfo`.")
@@ -83,28 +77,38 @@ class Raspberry():
         self.app.logger.debug("It seems we are on a Raspberry Pi")
 
         try:
-            self.initialise_gpio()
+            self._initialise_gpio()
         except Exception as e:
             self.app.logger.debug("Could not init GPIO : " + str(e))
             raise e
         return True
 
-    def initialise_gpio(self):
+    def _initialise_gpio(self):
+        """
+        Set GPIO ports from configuration and activate them
+        to show the user it's working.
+        """
         self.app.logger.debug("Initializing GPIO")
 
-        self.led = LED(self.led_gpio)
+        self.led = LED(self.configuration_file["rpi"]["indicator_gpio_port_number"])
         self.app.logger.debug("Activated indicator LED")
         self.indicator_countdown(iters=3)
-        self.button = Button(self.button_gpio, pull_up=True, bounce_time=0.1)
+        self.button = Button(
+            self.configuration_file["rpi"]["button_gpio_port_number"],
+            pull_up=True, bounce_time=0.1
+        )
         self.button.when_pressed = self.on_button_pressed
         self.app.logger.debug("Activated button")
 
         # The "flash" is a relay-controlled device ( light bulb for example )
-        self.flash = DigitalOutputDevice(self.flash_gpio)
+        self.flash = DigitalOutputDevice(self.configuration_file["rpi"]["flash_gpio_port_number"])
         self.flash_toggle()
         self.app.logger.debug("Activated flash")
 
     def indicator_countdown(self, iters=10, multi=10):
+        """
+        Activates the LED faster and faster to show a countdown
+        """
         for i in range(iters, 0, -1):
             self.led.on()
             sleep(i / multi)
@@ -112,7 +116,10 @@ class Raspberry():
             sleep(i / multi)
 
     def indicator_led(self, timing=0.2, l=5):
-        for i in range(l):
+        """
+        Turns on the indicator LED for a certain period of time
+        """
+        for _ in range(l):
             self.app.logger.debug("LED turned on")
             self.led.on()
             sleep(timing)
@@ -121,6 +128,9 @@ class Raspberry():
             sleep(timing)
 
     def flash_toggle(self):
+        """
+        Flashes the flash
+        """
         self.app.logger.debug("Flash turned on")
         self.flash.on()
         sleep(0.3)
@@ -128,6 +138,9 @@ class Raspberry():
         self.app.logger.debug("Flash turned off")
 
     def take_picture(self):
+        """
+        Takes a picture via the USB webcam
+        """
         # Validate if the image path is valid
         if not os.path.isdir(os.path.dirname(self.image_path)):
             self.app.logger.error(
@@ -173,6 +186,9 @@ class Raspberry():
         position="bottom_right",
         margin=10,
     ):
+        """
+        Takes an image and overlays it with a another picture.
+        """
         try:
             image = Image.open(image_path).convert("RGBA")
             logo = Image.open(logo_path).convert("RGBA")
@@ -199,7 +215,9 @@ class Raspberry():
                 y = image.height - logo.height - margin
             else:
                 raise ValueError(
-                    "Invalid position. Choose from 'bottom_right', 'top_left', 'top_right', or 'bottom_left'."
+                    "Invalid position." +
+                    "Choose from 'bottom_right', 'top_left', " +
+                    " 'top_right', or 'bottom_left'."
                 )
 
             # Composite the logo onto the image
@@ -217,6 +235,9 @@ class Raspberry():
         return True
 
     def crop_to_square(self, image_path, output_path=None):
+        """
+        Crop an image so that it becomes a square
+        """
         try:
             image = Image.open(image_path)
             width, height = image.size
@@ -244,6 +265,10 @@ class Raspberry():
         return True
 
     def on_button_pressed(self):
+        """
+        When a button press is detected, a picture is taken from the webcam
+        and added to the print queue.
+        """
         self.app.logger.debug("Button has been pressed")
         self.led.on()
         self.app.logger.debug("Counting down")
@@ -263,8 +288,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())

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.

src/templates/index.html

@@ -6,7 +6,8 @@
       <h3 class="card-header">Print a short message</h3>
       <div class="card-body">
         <form class="form-group" action="/web/print/sms" method="post">
-          <input class="form-control" type="text" name="txt" placeholder="200 chars or less " maxlength="200" required><br>
+          <textarea class="form-control" type="text" name="txt" placeholder="4096 chars or less " maxlength="4096"></textarea>
+          <br>
           <input class="form-control" type="text" name="signature" placeholder="Signature or pseudo" maxlength="200"><br>
           <input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
         </form>

src/web.py

@@ -1,10 +1,13 @@
+"""
+Manage all of the inputs from a web source
+"""
+
 import os
-from flask import flash
 from werkzeug.utils import secure_filename
 from task import TextTask, ImageTask, CutTask
 
 
-class Web(object):
+class Web:
     """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"""
 
@@ -41,7 +44,7 @@ class Web(object):
             file_uploaded = self.upload_file(image)
         except Exception as e:
             self.app.logger.error(e)
-            raise RuntimeError("Could not upload file") from e
+            raise RuntimeError("Could not upload file : " + str(e)) from e
 
         if file_uploaded:
             self.app.logger.debug("File has been uploaded, printing...")
@@ -65,16 +68,19 @@ class Web(object):
 
         return True
 
-    def login(self, username: str, password: str) -> bool:
-        """Not implemented"""
-        return
+    # def login(self, username: str, password: str) -> bool:
+    #     """Not implemented"""
+    #     return
 
-    def logout(self, username: str, password: str) -> bool:
-        """Not implemented"""
-        return
+    # def logout(self, username: str, password: str) -> bool:
+    #     """Not implemented"""
+    #     return
 
     def allowed_file(self, filename) -> bool:
-        self.app.logger.debug("Is the filename allowed ?")
+        """
+        Check if the file extension is allowed
+        """
+        self.app.logger.debug("Checking if the file extension is allowed")
         return (
             "." in filename
             and filename.rsplit(".", 1)[1].lower()
@@ -82,8 +88,11 @@ class Web(object):
         )
 
     def upload_file(self, image) -> bool:
+        """
+        Save the file after executing checks on it
+        """
         self.app.logger.debug("Validating file")
-        if image:
+        if not image is None or not image == "":
             if self.allowed_file(image.filename):
                 filename = secure_filename(image.filename)
                 self.app.logger.debug("File valid")
@@ -91,7 +100,7 @@ class Web(object):
                     image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
                 except OSError as e:
                     self.app.logger.error("Could not save file %s", e)
-                    return False
+                    raise RuntimeError("An OS error occured while uploading this file : " + str(e)) from e
 
                 self.app.logger.debug(
                     "File saved to "
@@ -102,13 +111,7 @@ class Web(object):
         self.app.logger.error(
             "Could not save file because the filename is forbidden"
         )
-            return False
-
-        else:
-            self.app.logger.error(
-                "Could not save file, it seems to be null ? : " + str(filename)
-            )
-            return False
+        raise RuntimeError("This file type is forbidden.")
 
     def get_queue_state(self):
         """Return current queue state"""

src/worker.py

@@ -1,6 +1,7 @@
-# This is the main printing thread
-# As explained in the task file, this is where we command
-# printing to happen.
+"""
+This is the main printing thread. A worker thread consums Tasks from
+ a PrintQueue, while trying to find available printers.
+"""
 
 import threading
 import time
@@ -8,13 +9,21 @@ from printers import Printers
 
 
 class PrintWorker(threading.Thread):
-    def __init__(self, app, print_queue, socketio=None):
+    """
+    A thread used to consume Tasks added to a Print Queue.
+
+    On initialisation, the worker will try to find Printers,
+    and on each print, choose an available Printer from a list of Printers.
+
+    If a print fails, it's not retried, but will be added to a list of completed
+    tasks.
+    """
+    def __init__(self, app, print_queue):
         super().__init__(daemon=True)
         self.app = app
         self.print_queue = print_queue
         self.printer = None
         self._lock = threading.Lock()
-        self.socketio = socketio  # Optional
         self.running = True
         self.state = "idle"  # idle, printing, dead, drinking-a-beer
         self.app.logger.debug("Ho great, I'm alive... I'm ready to work another day...")
@@ -30,8 +39,8 @@ class PrintWorker(threading.Thread):
     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:
 
@@ -40,7 +49,8 @@ class PrintWorker(threading.Thread):
                 time.sleep(0.2)
                 continue
 
-            # If we have no available printer, we look at the list printers we know about, and try to find one that is available.
+            # If we have no available printer, we look at the list printers
+            # we know about, and try to find one that is available.
             # When we find a printer, we acquire it
             # When we are finished with a printer, we release it to the world.
             while not self.printer or not self.printer.ready:
@@ -48,9 +58,13 @@ 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.app.logger.error("No printer detected" + str(e))
                     self.printer = None
 
             if self.state != "idle":
@@ -75,7 +89,6 @@ class PrintWorker(threading.Thread):
                         self.app.logger.info("Got a new task")
                         self.app.logger.debug("Got task %s", task.task_id)
                         task.status = "processing"
-                        self._emit_status(task.task_id, "processing")
 
                         print_data = task.get_print_data()
 
@@ -88,39 +101,24 @@ 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:
                         task.status = "failed"
                         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.
                     self.state = "idle"
                     time.sleep(0.1)
 
-    def _emit_status(self, task_id, status, error=None):
-        """Emit status update via Socket.IO if available"""
-        if not self.socketio:
-            return
-
-        room = f"task_{task_id}"
-        data = {
-            "task_id": task_id,
-            "status": status,
-            "position": None,  # Task no longer in queue
-        }
-
-        if error:
-            data["error"] = error
-
-        self.socketio.emit("task_status", data, room=room)
-
     def stop_worker(self):
         """
         Give the worker a break