Update numpy

README.md

@@ -79,16 +79,6 @@ Your contributions are very much welcome ! You can either request an account on
 
 Please also say if you had a printer to test your code, and which printer you've been using.
 
-### Code structure
-
-The app is written about the Flask framework. You can start by looking at the code in the `src/` folder, in the `main.py` file. There, you will see that a few classes are initialized. In general, they are two parts to the program :
-
-The Web pages and the API, which are the user-facing parts. This is with what the users will interact, and define how the program is going to be used. The web pages are renderer from the `include/` folder where Jinj2 templates are defined.
-
-The Worker and Printer Queue are the internal parts. When a new thing needs to be printed, usually sent from the Web or API interfaces, a new Task in the type of the document is created, and added to a print queue. Then, a Worker thread looks up the state of the queue every so often and picks jobs to execute on the printers connected to the system.
-
-The last part of the program is the Raspberry Pi class, that handles to Photomaton mode, which handles button presses, and LED indicator and a flash.
-
 ### Linting
 
 If you want to contribute code, please make sure to lint the project before commiting. This helps the code keep a general structure, and avoids some commons erros and mistakes.

configuration/config.toml

@@ -5,6 +5,8 @@ signature = "Anonymous"
 
 # Printer settings
 [printer]
+vendor_id = 0x04b8
+device_id = 0x0e28
 upload_folder = "src/static/uploads"
 
 # Raspberry Pi Configuration

pyproject.toml

@@ -7,7 +7,7 @@ authors = [
 ]
 license = "AGPLv3"
 readme = "README.md"
-requires-python = ">=3.13"
+requires-python = ">=3.14"
 dependencies = [
     "flask (>=3.1.3,<4.0.0)",
     "numpy (>=2.3.4)",

src/main.py

@@ -36,16 +36,14 @@ import werkzeug.exceptions
 from flask_socketio import SocketIO
 from flask_limiter import Limiter
 from flask_limiter.util import get_remote_address
+from printer import Printer  # The wrapper for the printer class
 from raspberry import Raspberry  # The Raspberry pi control Class
 from web import Web  # Wrapper for the web routes and API
-from print_queue import PrintQueue
-from worker import PrintWorker
 
-# We create the main Flask object
+# Variables
+
 app = Flask(__name__)
-socketio = SocketIO(app, cors_allowed_origins="*")
-
-# Global variables
+socketio = SocketIO(app)
 ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
 
 # Load the configuration file
@@ -72,7 +70,11 @@ except OSError as e:
 
 app.logger.debug("Config file loaded !")
 
+# Define the USB connections here.
+vendor_id = configuration_file["printer"]["vendor_id"]
+device_id = configuration_file["printer"]["device_id"]
 UPLOAD_FOLDER = str(configuration_file["printer"]["upload_folder"])
+
 try:
     os.mkdir(UPLOAD_FOLDER)
     app.logger.debug("Directory %s created successfully.", UPLOAD_FOLDER)
@@ -80,52 +82,53 @@ except FileExistsError:
     app.logger.debug("Directory %s already exists.", UPLOAD_FOLDER)
 except PermissionError:
     app.logger.error("Permission denied: Unable to create %s", UPLOAD_FOLDER)
-    sys.exit(77)
+    exit(77)
 
 # Output the config file
-if not os.getenv("FLASK_DEBUG") is None and os.getenv("FLASK_DEBUG") is True:
+if os.getenv("FLASK_DEBUG"):
     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 10Mb for a file upload
+app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000  # Maximum 3Mb for a file upload
 app.config["TEMPLATES_AUTO_RELOAD"] = True
 
-# Queue creation
-print_queue = PrintQueue(app)
+# Printer connection
+# Uses the class defined in the printer.py file
+printer = Printer(app, 0x04B8, 0x0E28)
+# printers = Printer(app)
+# printers.discover_printers()
+# printers.init()
+printer.init_printer()
 
 # Find out if we are running on a Raspberry Pi
 rpi = Raspberry(
-    print_queue,
+    printer,
     app,
-    configuration_file
+    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"],
 )
+
 RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
 
 
-#  Web & API management
-web = Web(app, print_queue)
+#  Web & API  routes
 
-# Start worker thread
-# When created, the worker will try to find printers connected to the system
-try:
-    worker = PrintWorker(app, print_queue)
-    worker.start()
-except Exception as e:
-    app.logger.error("Could not start the worker because %s ", str(e))
-    sys.exit(-1)
+web = Web(app, printer)
+
+if __name__ == "__main__":
+    app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")
 
-# The rate limit
 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)
 def index():
@@ -141,8 +144,6 @@ def webcam():
     app.logger.debug("Loading webcam interface")
     return render_template("webcam.html")
 
-
-# Form treatement
 @app.route("/web/print/sms", methods=["POST"])
 @limiter.limit("6/minute", override_defaults=False)
 def web_print_sms():
@@ -153,7 +154,7 @@ def web_print_sms():
         txt = request.form["txt"]
     except werkzeug.exceptions.BadRequestKeyError as e:
         app.logger.error("Whoops, we are missing the txt input field. : %s ", str(e))
-        flash("Whoops, no forms submitted or missing signature : " + str(e), "error")
+        flash("Whoops, no forms submitted or missing signature : " + str(e), 'error')
         return redirect(url_for("index"))
 
     try:
@@ -161,19 +162,19 @@ def web_print_sms():
         sign = request.form["signature"]
     except werkzeug.exceptions.BadRequestKeyError as e:
         app.logger.warning(
-            "No signature found for this print, using default signature : %s ", str(e)
+            "No signature found for this print, using default signature.", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
 
     try:
         web.print_sms(txt, sign)
-    except RuntimeError as e:
+    except Exception as e:
         app.logger.error("Whoops, we could not print an SMS because : %s ", str(e))
-        flash("Whoops, we could not print an SMS because :" + str(e), "error")
+        flash("Whoops, we could not print an SMS because :" + str(e), 'error')
         return redirect(url_for("index"))
 
     # end try
-    flash("The SMS has been added to the print queue !", "info")
+    flash("The SMS has been printed !", 'info')
     return redirect(url_for("index"))
 
 
@@ -188,18 +189,15 @@ def web_print_img():
         sign = request.form["signature"]
     except werkzeug.exceptions.BadRequestKeyError as e:
         app.logger.warning(
-            "No signature found for this print, using default signature : %s", str(e)
+            "No signature found for this print, using default signature.", 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:
         app.logger.error("Whoops, no images submitted  : %s ", str(e))
         app.logger.error("Error getting the files : %s", str(e))
-        flash("Whoops, no images submitted  : " + str(e), "error")
+        flash("Whoops, no images submitted  : " + str(e), 'error')
         return redirect(url_for("index"))
 
     file = request.files["img"]
@@ -207,21 +205,20 @@ def web_print_img():
     # empty file without a filename.
     if file.filename == "":
         app.logger.error("Submitted file has no filename !")
-        flash("Submitted file has no filename !", "error")
+        flash("Submitted file has no filename !", 'error')
         return redirect(url_for("index"))
 
     try:
         app.logger.debug("Sending the image to the printer.")
         web.print_image(file, sign)
-    except RuntimeError as e:
+    except Exception as e:
         app.logger.error("The image could not be printed because : %s ", str(e))
-        flash("The image could not be printed because : " + str(e), "error")
+        flash("The image could not be printed because : " + str(e), 'error')
         return redirect(url_for("index"))
 
-    flash("Picture added to the print queue !", "info")
+    flash("Picture printed !", 'info')
     return redirect(url_for("index"))
 
-
 # API routes
 # The api has the following methods
 # api/print/{sms,img,letter,qr,barcode}
@@ -260,12 +257,11 @@ def api_print_sms():
     try:
         # comment: We try to print the SMS
         web.print_sms(txt, sign)
-    except RuntimeError as e:
+    except Exception as e:
         return str(e), 500
     # end try
     return "OK", 200
 
-
 @app.route("/api/print/img", methods=["POST"])
 @limiter.limit("6/minute", override_defaults=False)
 def api_print_image():
@@ -287,7 +283,8 @@ def api_print_image():
         return "No image submitted", 400
 
     file = request.files["img"]
-    # If the user submits an empty file without a filename.
+    # If the user does not select a file, the browser submits an
+    # empty file without a filename.
     if file.filename == "":
         app.logger.error("Submitted file has no filename !")
         return "Submitted file has no filename !", 400
@@ -295,64 +292,25 @@ def api_print_image():
     try:
         app.logger.debug("Sending the image to the printer.")
         web.print_image(file, sign)
-    except RuntimeError as e:
+    except Exception as e:
         return str(e), 500
 
     return "OK", 200
 
-
-# TODO: This might not depend on the Raspberry Pi
 @app.route("/api/camera/picture", methods=["GET"])
 def camera_picture():
-    """Returns a picture taken by the camera on a raspberry pi"""
+    """Returns a picture taken by the camera"""
     if RASPBERRY_PI_CONNECTED:
         try:
             return rpi.camera_picture()
-        except RuntimeError as e:
+        except Exception as e:
             return jsonify({"message": "Error getting the stream : " + e}), 500
     else:
         return jsonify({"message": "No camera present"}), 500
 
 
-@app.route("/api/queue", methods=["GET"])
-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"""
-    return jsonify(worker.current_state())
-
-
-@app.route("/api/worker/start")
-def api_worker_start():
-    """
-    Enable to worker. This starts to process the print queue.
-    """
-    worker.start_worker()
-    return jsonify(worker.current_state())
-
-
-@app.route("/api/worker/stop")
-def api_worker_stop():
-    """
-    Stops the print queue. This stops the processing of the print queue.
-    """
-    worker.stop_worker()
-    return jsonify(worker.current_state())
-
-
 ## Authentification
 
-
 @app.route("/login")
 @limiter.limit("1/second", override_defaults=False)
 def login_page():
@@ -404,7 +362,3 @@ def camera_status():
         socketio.emit("camera_status", True)
     else:
         socketio.emit("camera_status", False)
-
-
-if __name__ == "__main__":
-    app.run(use_reloader=False, host="0.0.0.0", ssl_context="adhoc")

src/print_queue.py

@@ -1,144 +0,0 @@
-"""
-This class has the method by which we manage the Tasks
-It's a printing queue, so we need to add, remove and get information on where
-the queue is
-"""
-
-from collections import deque
-
-# Because actually printing and adding new print job requests happen at
-# diffrent times, the print queue is managed by it's own thread.
-import threading
-
-from datetime import datetime
-from task import TaskType
-
-
-class PrintQueue:
-    """
-    A Double-ended Queue to manage the printing Tasks
-    """
-
-    def __init__(self, app):
-        self.app = app
-        self._queue = deque()
-        self._lock = threading.Lock()
-        self._completed_tasks = {}  # Store completed task info
-        self._task_counter = 0
-        self.app.logger.debug("Created a new PrintQueue")
-
-    def __len__(self) -> int:
-        return len(self._queue)
-
-    def enqueue(self, task):
-        """Add task to right of the queue and return position"""
-        with self._lock:
-            try:
-                self.app.logger.info("Add task %s to queue ", task.task_id)
-                self._queue.append(task)
-                position = self._queue.index(task)
-                # We return the current position of the task if it was added
-                self.app.logger.debug(
-                    "Added a new task %s to the queue at position %s",
-                    task.task_id,
-                    position,
-                )
-                return position
-            except Exception as e:
-                self.app.logger.error("Could not add a task to the queue : %s ", e)
-                raise e
-
-    def dequeue(self):
-        """Remove and return next task ( from the left of the queue ) (thread-safe)"""
-        with self._lock:
-            return self._queue.popleft() if len(self._queue) > 0 else None
-
-    def get_position(self, task):
-        """Get current position of task in queue (1-indexed)"""
-        with self._lock:
-            if task.task_id in self._completed_tasks:
-                return None  # Task already completed
-
-        try:
-            # Try to get the position of a Task
-            return self._queue.index(task)
-        except ValueError as e:
-            raise e
-        # end try
-
-    def is_empty(self):
-        """Check if queue is empty"""
-        with self._lock:
-            self.app.logger.debug("Checking if queue is empty")
-            return len(self._queue) == 0
-
-    def get_queue_state(self):
-        """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), "content": str(t.get_print_data())}
-                for t in self._queue
-            ]
-
-    def get_queue_completed(self):
-        """Return completed queue elements"""
-        with self._lock:
-            self.app.logger.debug("Return completed queue elements")
-            return self._completed_tasks
-
-    def get_status(self, task_id):
-        """Get full status info for a task"""
-        with self._lock:
-
-            if task_id in self._completed_tasks:
-                return self._completed_tasks[task_id]
-
-            # Check in queue if it exists
-            for index, task in enumerate(self._queue):
-                if task.task_id == task_id:
-                    # Depending on it's type, we return more info
-                    if task.task_type == TaskType.IMAGE:
-                        return {
-                            "task_id": task_id,
-                            "status": task.status,
-                            "type": task.task_type,
-                            "position": index,
-                            "in_queue": True,
-                            "content": task.content,
-                            "signature": task.signature,
-                        }
-
-                    if task.task_type == TaskType.TEXT:
-                        return {
-                            "task_id": task_id,
-                            "status": task.status,
-                            "type": task.task_type,
-                            "position": index,
-                            "in_queue": True,
-                            "image_path": str(task.image_path),
-                            "signature": task.signature,
-                            "process": str(task.process),
-                        }
-
-                    if task.task_type == TaskType.CUT:
-                        return {
-                            "task_id": task_id,
-                            "status": task.status,
-                            "type": task.task_type,
-                            "position": index,
-                            "in_queue": True,
-                        }
-
-            return None
-
-    def mark_completed(self, task_id, task_status):
-        """Mark task as completed and remove from queue"""
-        with self._lock:
-            self._completed_tasks[task_id] = {
-                "task_id": task_id,
-                "status": task_status,
-                "position": None,
-                "in_queue": False,
-                "completed_at": datetime.now().isoformat(),
-            }

src/printer.py

@@ -1,129 +1,100 @@
-"""
-This class manages connexion to a Printer
-"""
-
+# Importing the module to manage the connection to the printer.
+import escpos.printer
+import brother_ql
+from time import sleep, gmtime, strftime
 import os.path
-import os
-from abc import ABC, abstractmethod
-from dataclasses import dataclass
-import time
-from enum import Enum
-import uuid
-import threading
-import usb.core
-from PIL import Image, ImageEnhance
+from PIL import Image, ImageEnhance, ImageOps
 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.raster import BrotherQLRaster
-from brother_ql.conversion import convert
-from brother_ql.backends.helpers import send
 
-
-class PrinterType(Enum):
+class Printer(object):
     """
-    What are the capacities of a Printer ?
+    # The connection is based on the ESC/POS library
+
+    ## Connection to the USB printer
+
+    ## Making sure the printer is alive
+
+    ## Making sure it has paper
+
+    ## Define default print settings
+
+    ## Print starting message, log time of first print, cut.
+
+    ## Annonce readyness : return a positive pong message.
     """
 
-    EPSON = "epson"
-    BROTHER = "brother"
+    # Is the printer ready to accept a new print ?
+    ready = False
 
-
-# For Brother-QL Printers
-@dataclass
-class PrinterInfo:
-    """
-    Brother-QL printer information
-    """
-    identifier: str
-    backend: str
-    protocol: str
-    vendor_id: str
-    product_id: str
-    serial_number: str
-    name: str = "Brother QL Printer"
-    model: str = "QL-570"
-    status: str = "unknown"
-    label_type: str = "unknown"
-    label_size: str = "unknown"
-    label_width: int = 0
-    label_height: int = 0
-
-    def __getitem__(self, item):
-        return getattr(self, item)
-
-    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.
-    """
-
-    def __init__(self, app, vendor_id, device_id, printer_type: PrinterType):
-        """
-        We initialize a Printer via it's USB connexion, and generate a unique ID
-        """
-        self.id = uuid.uuid4()
+    def __init__(self, app, device_id, vendor_id):
+        super(Printer, self).__init__()
         self.app = app
-        self.vendor_id = vendor_id
-        self.device_id = device_id
         self.ready = False
-        self.printer_type = printer_type
-        self._lock = threading.Lock()
-
-    @abstractmethod
-    def _has_paper(self) -> bool:
-        """Check if the printer has papier"""
-
-    @abstractmethod
-    def _state(self) -> bool:
-        """Reports the state of the Printer"""
-
-    @abstractmethod
-    def print_task(self, task_type, data) -> None:
-        """Takes a PrintTask and executes it"""
-
-
-class EscPosPrinter(Printer):
-    """
-    Create a new ESC/POS based printer.
-    """
-
-    def __init__(self, app, vendor_id, device_id):
-        """
-        Create a connexion to a ESC/POS Printer via USB,
-        Making sure the printer is alive,
-        Making sure it has paper,
-        Define default print settings
-        """
-        super().__init__(app, vendor_id, device_id, printer_type=PrinterType.EPSON)
         self.printer = None
+        self.device_id = device_id
+        self.vendor_id = vendor_id
         self.usb_args = {}
-        self.usb_args["idVendor"] = self.vendor_id
-        self.usb_args["idProduct"] = self.device_id
+        self.usb_args["idVendor"] = self.device_id
+        self.usb_args["idProduct"] = self.vendor_id
 
+    def check_paper(self) -> bool:
+        # Let's check paper status
+        self.app.logger.debug("Checking paper status...")
+        self.printer.open(self.usb_args)
+        status = self.printer.paper_status()
+        match status:
+            case 0:
+                self.app.logger.error("Printer has no more paper, aborting...")
+                self.printer.close()
+                raise RuntimeError("No more paper in the printer")
+            case 1:
+                self.app.logger.warning(
+                    "Printer needs paper to be changed very soon ! "
+                )
+                self.printer.close()
+            case 2:
+                self.app.logger.debug("Printer has paper, good to go")
+                self.printer.close()
+
+    def init_printer(self):
+
+        # Is the printer online ? Is the communication with the printer successfull ?
+        if os.getenv("FLASK_DEBUG"):
+            waiting_elapsed = 1
+        else:
+            waiting_elapsed = 10
+
+        self.app.logger.debug("Waiting for printer to get online...")
+
+        while not self.ready:
             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")
-        except escpos.exceptions.DeviceNotFoundError as e:
-            self.app.logger.error(
-                "The USB device is not plugged in : %s",
-                str(e),
-            )
+                p = escpos.printer.Usb(self.device_id, self.vendor_id, 0, profile="TM-P80")
             except Exception as e:
-            self.app.logger.error("Printer could not be connected : %s ", str(e))
+                self.app.logger.error(
+                    "The USB device is not plugged in, trying again %s : %s",waiting_elapsed, str(e)
+                )
+                pass
 
             try:
                 if p.is_online():
+                    self.ready = True
                     self.app.logger.debug("Printer online !")
             except Exception as e:
-            raise e
+                self.app.logger.error("Error while getting the printer online %s : %s",waiting_elapsed, str(e)
+                )
+                pass
+
+            sleep(1)
+            waiting_elapsed -= 1
+            if waiting_elapsed < 1:
+                self.app.logger.error(
+                    "Printer took more than 30 seconds to get online, aborting..."
+                )
+                waiting_elapsed = 1  # Reset the waiting time for the next print.
+                return False
 
         # Setting up the printing options.
         p.set(
@@ -144,54 +115,22 @@ 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
-
-        try:
-            self._has_paper()
-        except Exception as e:
-            raise e
-
         self.ready = True
+        self.printer.close()
+
+        self.check_paper()
 
-    def _has_paper(self):
-        """Check if the printer has paper left"""
-        self.app.logger.debug("Checking paper status...")
-        self.printer.open(self.usb_args)
-        status = self.printer.paper_status()
-        match status:
-            case 0:
-                self.app.logger.error("Printer has no more paper, aborting...")
-                self.printer.close()
-                raise RuntimeError("No more paper in the printer")
-            case 1:
-                self.app.logger.warning(
-                    "Printer needs paper to be changed very soon ! "
-                )
-                self.printer.close()
-                return True
-            case 2:
-                self.app.logger.debug("Printer has paper, good to go")
-                self.printer.close()
         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."
-            )
-            raise ValueError
-
-        # We make sure that the signature is not something too goofy
+    def print_sms(self, msg, signature="", bold=False):
         clean_msg = str(msg) + "\n"
         clean_signature = str(signature)
 
-        # Make checks on the size of the message being printed
         if len(clean_msg) > 4096:
             self.app.logger.warning(
                 "Could not print message of this length: " + str(len(clean_msg))
             )
-            raise RuntimeError(
+            raise Exception(
                 "Could not print message of this length :"
                 + str(len(clean_msg))
                 + ", needs to be below 4096 caracters long."
@@ -201,40 +140,34 @@ class EscPosPrinter(Printer):
             self.app.logger.warning(
                 "Could not print signature of this length: " + str(len(clean_signature))
             )
-            raise RuntimeError(
+            raise Exception(
                 "Could not print signature of this length :"
                 + str(len(clean_signature))
                 + ", needs to be below 256 caracters long."
             )
 
-        # Do the actual printing
-        # We would pop the next element in the queue here, if it's a sms type
         try:
             self.printer.open(self.usb_args)
             self.printer.set(align="center", font="a", bold=bold)
             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))
-            raise RuntimeError(
-                "Unable to print a SMS, the printer couldn't do it."
-            ) from e
+            raise RuntimeError("Unable to print a SMS, the printer couldn't do it.") from e
 
         self.app.logger.info("Printed text")
-        self.ready = True
+        return True
 
-    def _print_img(self, path, signature="", center=True, process=False):
-        self.ready = False
-        clean_signature = str(signature)
+    def print_img(self, path, sign="", center=True, process=False):
+        clean_signature = str(sign)
 
-        if len(signature) > 256:
+        if len(sign) > 256:
             self.app.logger.warning(
                 "Could not print signature of this length: " + str(len(clean_signature))
             )
-            raise ValueError(
+            raise Exception(
                 "Could not print signature of this length :"
                 + str(len(clean_signature))
                 + ", needs to be below 256 caracters long."
@@ -242,74 +175,58 @@ class EscPosPrinter(Printer):
 
         if not os.path.isfile(str(path)):
             self.app.logger.warning("File does not exist : " + str(path))
-            raise OSError(
+            raise Exception(
                 "The file path for this image :"
                 + str(path)
                 + " wasn't found. Please try again."
             )
-
+        else:
             self.app.logger.debug("Printing file from " + str(path))
 
         if process:
             try:
                 self.app.logger.debug("Proccessing the image")
-                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
+                path = process_image(self, path)
+            except Exception as e:
+                self.app.logger.error(str(e))
+                return False
         else:
-            processed_path = path
             self.app.logger.warning("Not proccessing the image")
 
         try:
             self.printer.open(self.usb_args)
-            self.printer.image(processed_path, center=center)
-            self.printer.textln(signature)
+            self.printer.image(path, center=center)
             self.printer.close()
-            self.app.logger.debug("Printed an image : " + str(processed_path))
+            self.app.logger.debug("Printed an image : " + str(path))
+            os.remove(path)
+            self.app.logger.debug("Removed image : " + str(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:
                 self.printer.close()
             except Exception as e:
-                self.app.logger.error(
-                    "Could not close the printer connexion %s", str(e)
-                )
+                self.app.logger.error(str(e))
                 raise RuntimeError("Could not close the printer connexion. ") from e
 
         self.app.logger.info("Printed a picture")
-        self.ready = True
+        return True
 
-    def _qr(self, content):
-        self.ready = False
+    def qr(self, content):
         try:
             self.printer.open(self.usb_args)
             self.printer.qr(content, center=True)
-            self.printer.textln(content)
             self.printer.close()
-        except RuntimeError as e:
+        except Exception as e:
             self.printer.close()
             self.app.logger.error(str(e))
-            raise e
+            return False
 
         self.app.logger.info("Printed a QR")
-        self.ready = True
+        return True
 
-    def _cut(self):
-        self.ready = False
+    def cut(self):
         try:
             self.printer.open(self.usb_args)
             self.printer.cut()
@@ -320,245 +237,28 @@ class EscPosPrinter(Printer):
             raise e
 
         self.app.logger.info("Did a cut")
-        self.ready = True
-
-    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"""
+        return True
 
 
-        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))
-                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.ready = True
-                    case "image":
-                        self._print_img(
-                            data["img"], signature=data["sign"], process=data["process"]
-                        )
-                        self.ready = True
-                    case "cut":
-                        self._cut()
-                        self.ready = True
-                    case "qr":
-                        self._qr(data["txt"])
-                        self.ready = True
-                    case _:
-                        raise RuntimeError("This task type is not supported")
-            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):
-    """
-    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
-        )
-        self.printer = None
-        self.usb_args = {}
-        self.usb_args["idVendor"] = self.device_id
-        self.usb_args["idProduct"] = self.vendor_id
-        self.model_manager = ModelsManager()
-
-        # Code taken from https://github.com/5shekel/printit/blob/master/printer_utils.py
-
-        backend = backend_factory("pyusb")
-        available_devices = backend["list_available_devices"]()
-
-        for printer in available_devices:
-            self.app.logger.debug(f"Found device: {printer}")
-            identifier = printer["identifier"]
-            parts = identifier.split("/")
-
-            if len(parts) < 4:
-                self.app.logger.warning(
-                    f"Skipping device with invalid identifier format: {identifier}"
-                )
-                continue
-
-            protocol = parts[0]
-            # device_info = parts[2]
-            serial_number = parts[3]
-
-        try:
-            product_id_int = int(self.device_id, 16)
-            for m in self.model_manager.iter_elements():
-                if m.product_id == product_id_int:
-                    model = m.identifier
-                    break
-            self.app.logger.debug(f"Matched printer model: {model}")
-        except ValueError:
-            self.app.logger.warning(f"Invalid product ID format: {m.product_id}")
-
-        self.printer_info = PrinterInfo(
-            identifier=identifier,
-            backend="pyusb",
-            model=model,
-            protocol=protocol,
-            vendor_id=vendor_id,
-            product_id=self.device_id,
-            serial_number=serial_number,
-        )
-
-        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):
-        """
-        Print a raster image via a Brother QL printer
-        """
-        self.ready = False
-        label_type = "102"
-        rotate = 0
-        dither = False
-        try:
-            # Prepare the image for printing
-            qlr = BrotherQLRaster(self.printer_info["model"])
-
-            instructions = convert(
-                qlr=qlr,
-                images=[data["img"]],
-                label=label_type,
-                rotate=rotate,
-                threshold=70,
-                dither=dither,
-                compress=True,
-                red=False,
-                dpi_600=False,
-                hq=False,
-                cut=True,
-            )
-
-            # Debug logging
-            if os.getenv("FLASK_DEBUG"):
-                self.app.logger.debug(f"""
-                Print parameters:
-                - Label type: {label_type}
-                - Rotate: {rotate}
-                - Dither: {dither}
-                - Model: {self.printer_info['model']}
-                - Backend: {self.printer_info['backend']}
-                - Identifier: {self.printer_info['identifier']}
-                """)
-
-            # Try to print using Python API
-            # send() =  status = {
-            #     'instructions_sent': True, # The instructions were sent to the printer.
-            #     'outcome': 'unknown', # String description of the outcome of the sending operation like: 'unknown', 'sent', 'printed', 'error'
-            #     'printer_state': None, # If the selected backend supports reading back the printer state, this key will contain it.
-            #     'did_print': False, # If True, a print was produced. It defaults to False if the outcome is uncertain (due to a backend without read-back capability).
-            #     'ready_for_next_job': False, # If True, the printer is ready to receive the next instructions. It defaults to False if the state is unknown.
-            # }
-            status = send(
-                instructions=instructions,
-                printer_identifier=self.printer_info["identifier"],
-                backend_identifier="pyusb",
-            )
-
-            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"]:
-                self.ready = bool(status["printer_state"])
-            else:
-                self.ready = True
-
-        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("Print completed (timeout is normal)")
-                self.ready = True
-
-            error_msg = f"USBError encountered: {e}"
-            self.app.logger.debug(error_msg)
-            raise RuntimeError from e
-
-        except Exception as e:
-            error_msg = f"Unexpected error during printing: {str(e)}"
-            self.app.logger.debug(error_msg)
-            raise RuntimeError from e
-
-    def print_task(self, task_type, data):
-        """Execute actual print based on task type"""
-        with self._lock:
-            if self._state:
-                self._state = False
-                match (task_type.value):
-                    case "image":
-                        self._print_img(data["img"])
-                        self._state = True
-                    case "cut":
-                        # The cut happens by default on Brother QL printers.
-                        self._state = True
-                    case _:
-                        raise RuntimeError("This task type is not supported")
-            else:
-                raise RuntimeError("The printer is not ready to print yet !")
-
-        # These values are by default for now
-
-        # raise NotImplementedError("This printer type is not implemented yet")
-
-
-def _process_image(self, path):
+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 = 2  # Less than 1.0 = lower contrast
+    contrast_factor = 0.6  # 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 from RGBA to RGBA")
+            self.app.logger.debug("Converting the image to RGB from RGBA")
             original_img = original_img.convert("RGB")
 
         # Resize while maintaining aspect ratio
-        original_img.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)
+        original_img.thumbnail((max_width, max_height), Image.LANCZOS)
         self.app.logger.debug("Resized the image")
 
         # # Convert to grayscale for dithering
-        # dithered_img = original_img.convert("L").convert("1")
-        # Dithering using default method (Floyd–Steinberg)
+        # dithered_img = original_img.convert("L").convert("1")  # Dithering using default method (Floyd–Steinberg)
         # self.app.logger.debug("Dithered the image")
 
         # Compute brightness of original image (grayscale average)
@@ -579,19 +279,114 @@ 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(grayscale)
-            grayscale = enhancer.enhance(brightness_factor)
+            enhancer = ImageEnhance.Brightness(original_img)
+            original_img = enhancer.enhance(brightness_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)
+        # # Reduce contrast
+        # contrast_enhancer = ImageEnhance.Contrast(original_img)
+        # original_img = contrast_enhancer.enhance(contrast_factor)
+
+        # Final resize check
+        if original_img.height > max_height:
+            raise ValueError("Image is too long, sorry! Keep it below 575×1000 pixels.")
+            self.app.logger.error(
+                "Image is too long, sorry! Keep it below 575×1000 pixels."
+            )
+            return False
 
         # Convert to JPEG and save
         jpeg_path = os.path.splitext(path)[0] + "_processed.jpg"
-        grayscale.save(jpeg_path, format="JPEG", quality=95, optimize=True)
-        self.app.logger.debug("Processed and saved image.")
+        original_img.save(jpeg_path, format="JPEG", quality=95, optimize=True)
+        app.logger.debug("Processed and saved image.")
 
     return jpeg_path
+
+def discover_printers():
+    """
+    We try to find all the connected printers ( 0 or n ) to this system.
+
+    For every type of supported printer, we try to autodiscover them.
+
+    http://www.linux-usb.org/usb.ids A list of USB vendor IDs
+
+    04b8  Seiko Epson Corp.
+    04f9  Brother Industries, Ltd
+    """
+
+def find_and_parse_borther_ql_printer():
+
+
+    ## We might be able to no use this because there is a `discover` command in https://github.com/pklaus/brother_ql#usage
+
+    ## Code stolen from https://framagit.org/stickoeur/diagnostickoeur/-/blob/no-masters/printit.py?ref_type=heads
+
+    """Find and parse Brother QL printer information."""
+
+    model_manager = ModelsManager()
+
+    # Debug print to show we're searching
+    # print("Searching for Brother QL printer...")
+
+    for backend_name in ["pyusb", "linux_kernel"]:
+        try:
+            #print(f"Trying backend: {backend_name}")
+            backend = backend_factory(backend_name)
+            available_devices = backend["list_available_devices"]()
+            #print(f"Found {len(available_devices)} devices with {backend_name} backend")
+
+            for printer in available_devices:
+                #print(f"Found device: {printer}")
+                identifier = printer["identifier"]
+                parts = identifier.split("/")
+
+                if len(parts) < 4:
+                    #print(f"Skipping device with invalid identifier format: {identifier}")
+                    continue
+
+                protocol = parts[0]
+                device_info = parts[2]
+                serial_number = parts[3]
+
+                try:
+                    vendor_id, product_id = device_info.split(":")
+                except ValueError:
+                    #print(f"Invalid device info format: {device_info}")
+                    continue
+
+                # Default model
+                model = "QL-570"
+
+                # Try to match product ID to determine actual model
+                try:
+                    product_id_int = int(product_id, 16)
+                    for m in model_manager.iter_elements():
+                        if m.product_id == product_id_int:
+                            model = m.identifier
+                            break
+                    #print(f"Matched printer model: {model}")
+                except ValueError:
+                    #print(f"Invalid product ID format: {product_id}")
+                    continue
+
+                printer_info = {
+                    "identifier": identifier,
+                    "backend": backend_name,
+                    "model": model,
+                    "protocol": protocol,
+                    "vendor_id": vendor_id,
+                    "product_id": product_id,
+                    "serial_number": serial_number,
+                }
+                #print(f"Found printer: {printer_info}")
+                return printer_info
+
+        except Exception as e:
+            #print(f"Error with backend {backend_name}: {str(e)}")
+            continue
+
+    print("No Brother QL printer found")
+    return None
+
+
+def fint_and_parse_epson_printer():
+    pass

src/printers.py

@@ -1,160 +0,0 @@
-"""
-A collection of Printers.
-
-It has methods to discover printers, and provides an interface for
-the methods expected from printers.
-"""
-
-from collections.abc import Set
-import usb.core
-import usb.util
-from printer import Printer, EscPosPrinter, BrotherPrinter
-
-
-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()
-        """
-        self.app = app
-        self.printers = self._discover_printers()
-
-    def _discover_printers(self) -> Set[Printer]:
-        """
-        Gets connected USB printer devices using the pyusb library.
-
-        We analyse the USB devices, get the ones that match
-        the printer class ( 7 ) and for Brother and EPSON printers,
-        try to create a Printer object that can be used by the Worker class
-        to execute prints.
-
-        Returns a set of Printer
-        """
-
-        self.app.logger.debug("Discovering USB Devices connected to this system")
-
-        printers = set()
-
-        # Find all connected USB devices
-        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."
-            )
-
-        for dev in devices:
-            # Attempt to get the manufacturer and product strings
-            try:
-                manufacturer = usb.util.get_string(dev, dev.iManufacturer)
-            except Exception:
-                manufacturer = "Unknown"
-
-            try:
-                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),
-            )
-
-            if manufacturer == "EPSON":
-                try:
-                    # We create a new EscPosPrinter()
-                    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
-                    )
-                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)
-
-                # We already found the type of printer,
-                # we don't need an extra comparaison.
-                continue
-
-            # or a Brother Printer
-            if manufacturer == "Brother":
-                try:
-                    # We create a new BrotherPrinter()
-                    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
-                    )
-                except Exception as e:
-                    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 %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")
-
-        return printers
-
-    def any(self) -> Printer:
-        """
-        Return a dict key: UUID, value: Printer, with any connected printer.
-        """
-        if len(self.printers) > 0:
-            for i in self.printers:
-                return i
-        else:
-            raise RuntimeError("No printers available")
-
-    # def get_printer(self, printer_type):
-    #     """
-    #     Return a specific printer
-
-    #     printer_type -- a printer type
-    #     """
-    #     return NotImplementedError()
-
-
-class _FindClass:
-    """
-    Use by usb.core to modify the way USB devices are found
-    Taken from pyUSB documentation on Github
-    """
-    def __init__(self, class_):
-        self._class = class_
-
-    def __call__(self, device):
-        # first, let's check the device
-        if device.bDeviceClass == self._class:
-            return True
-        # ok, transverse all devices to find an
-        # interface that matches our class
-        for cfg in device:
-            # find_descriptor: what's it?
-            intf = usb.util.find_descriptor(cfg, bInterfaceClass=self._class)
-            if intf is not None:
-                return True
-
-        return False

src/raspberry.py

@@ -1,21 +1,13 @@
-"""
-This class executes when we are on a raspberry Pi.
-
-It handles the press of a button via GPIO,
-activates a flash and prints out the picture
-that was taken via a USB Webcam.
-"""
-
+from flask_socketio import SocketIO
+from gpiozero import Button, LED, DigitalOutputDevice
+from time import sleep, gmtime, strftime
+from PIL import Image
 import io  # To check if we are on a Raspberry Pi
 import subprocess
 import os
-from time import sleep, gmtime, strftime
-from gpiozero import Button, LED, DigitalOutputDevice
-from PIL import Image
-from task import TextTask, ImageTask, CutTask
 
 
-class Raspberry:
+class Raspberry(object):
     """
     This class will manage three things :
     - Connecting to a USB webcam
@@ -23,29 +15,32 @@ class Raspberry:
     - Activating a flash ( or light )
     - Flash an indicator light
 
-    # pylint: disable=too-many-instance-attributes
-    # dede
     """
 
     def __init__(
         self,
-        print_queue,
+        printer,
         app,
-        configuration_file
+        socketio,
+        button_gpio_port_number,
+        indicator_gpio_port_number,
+        flash_gpio_port_number,
+        is_flash_present,
     ):
-        self.print_queue = print_queue
+        self.printer = printer
+        self.socketio = socketio
         self.app = app
-        self.configuration_file = configuration_file
+
+        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.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg"
 
-    def is_raspberry_pi(self):
-        """
-        Checking if we are on a Raspberry Pi by checking
-        information on the /proc/cpuinfo file
-        """
+    def is_raspberry_pi(self, raise_on_errors=False):
         # Check if we are running on a raspberry pi
         try:
-            with io.open("/proc/cpuinfo", "r", encoding="utf-8") as cpuinfo:
+            with io.open("/proc/cpuinfo", "r") as cpuinfo:
                 found = False
                 for line in cpuinfo:
                     if line.startswith("Hardware"):
@@ -65,10 +60,9 @@ class Raspberry:
                             return False
 
                 if not found:
-                    self.app.logger.warning(
-                        "Couldn't get sufficient hardware information from /proc/cpuinfo"
+                    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("Unable to determine if we are on a Raspberry Pi.")
                     return False
         except IOError:
             self.app.logger.error("Unable to open `/proc/cpuinfo`.")
@@ -77,38 +71,28 @@ 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):
-        """
-        Set GPIO ports from configuration and activate them
-        to show the user it's working.
-        """
+    def initialise_gpio(self):
         self.app.logger.debug("Initializing GPIO")
 
-        self.led = LED(self.configuration_file["rpi"]["indicator_gpio_port_number"])
+        self.led = LED(self.led_gpio)
         self.app.logger.debug("Activated indicator LED")
         self.indicator_countdown(iters=3)
-        self.button = Button(
-            self.configuration_file["rpi"]["button_gpio_port_number"],
-            pull_up=True, bounce_time=0.1
-        )
+        self.button = Button(self.button_gpio, 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.configuration_file["rpi"]["flash_gpio_port_number"])
+        self.flash = DigitalOutputDevice(self.flash_gpio)
         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)
@@ -116,10 +100,7 @@ class Raspberry:
             sleep(i / multi)
 
     def indicator_led(self, timing=0.2, l=5):
-        """
-        Turns on the indicator LED for a certain period of time
-        """
-        for _ in range(l):
+        for i in range(l):
             self.app.logger.debug("LED turned on")
             self.led.on()
             sleep(timing)
@@ -128,9 +109,6 @@ class Raspberry:
             sleep(timing)
 
     def flash_toggle(self):
-        """
-        Flashes the flash
-        """
         self.app.logger.debug("Flash turned on")
         self.flash.on()
         sleep(0.3)
@@ -138,9 +116,6 @@ 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(
@@ -166,7 +141,7 @@ class Raspberry:
                 f"Unable to take a picture. Error: {e.stderr.decode()}"
             )
             return False
-        except RuntimeError as e:
+        except Exception as e:
             # Catch any unexpected errors
             self.app.logger.error(f"Unexpected error while taking picture: {str(e)}")
             return False
@@ -186,9 +161,6 @@ 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")
@@ -215,9 +187,7 @@ 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
@@ -228,16 +198,13 @@ class Raspberry:
                 output_path = image_path  # Overwrite the original image if no output path is given
             image.save(output_path)
 
-        except RuntimeError as e:
+        except Exception as e:
             self.app.logger.error(f"Error overlaying logo: {e}")
             return False
 
         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
@@ -258,17 +225,13 @@ class Raspberry:
 
             image.save(output_path)
 
-        except RuntimeError as e:
+        except Exception as e:
             self.app.logger.error(f"Error cropping image to square: {e}")
             return False
 
         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")
@@ -279,7 +242,7 @@ class Raspberry:
         try:
             self.flash.on()
             self.take_picture()
-        except RuntimeError as e:
+        except Exception as e:
             self.app.logger.error(
                 "Could not take a picture after the button press : " + str(e)
             )
@@ -288,13 +251,15 @@ 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="")
-        )
-        time = strftime("%Y-%m-%d %H:%M", gmtime())
-        self.print_queue.enqueue(TextTask(content=time, signature=""))
-        self.print_queue.enqueue(CutTask())
+        self.printer.print_img("src/static/images/extase-club.png", process=True)
+        self.printer.print_img(self.image_path, process=True)
+        self.printer.print_sms("")
+        self.printer.print_sms("With Love From Société.Vide", signature="", bold=True)
+        self.printer.print_sms("Printed by LittlePrynter", signature="")
+        self.printer.print_sms("n07070.xyz", signature="")
+        self.printer.print_sms(strftime("%Y-%m-%d %H:%M", gmtime()), signature="")
+        self.printer.qr("https://n07070.xyz/articles/littleprynter")
+        self.printer.cut()
         self.led.off()
-        self.app.logger.debug("Added a photomaton picture to the print queue")
+        self.app.logger.debug("Done printing picture")
         return True

src/task.py

@@ -1,110 +0,0 @@
-"""
-Here we define the types of tasks
-We are using Abstract Base Classes,
-like this we can define types of tasks ( text, images, ... )
-that all work with the same basic options
-
-The tasks are going to be injected into a Queue.
-It's a usefull way of storing information in our
-program, while making sure that things are indeed printed.
-It's also a way to prevent two concurrent connexions creating
-a access conflict on a single printer, like two people wanting
-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
-
-from enum import Enum
-import uuid
-
-
-## You can expand this if you want to take other types of tasks into account
-class TaskType(Enum):
-    """
-    The different tasks supported by the printers
-    """
-
-    TEXT = "text"
-    IMAGE = "image"
-    CUT = "cut"
-    QR = "qr"
-
-
-class PrintTask(ABC):
-    """
-    A print task holds information about what we are looking to print.
-    """
-
-    def __init__(self, task_type: TaskType):
-        self.task_id = self._generate_id()
-        self.task_type = task_type
-        self.status = "pending"  # pending, processing, completed, failed
-
-    @abstractmethod
-    def get_print_data(self):
-        """Return data formatted for printer"""
-
-    def _generate_id(self):
-        # Generate unique task ID
-        return str(uuid.uuid4())
-
-
-class TextTask(PrintTask):
-    """
-    This tasks represents a texte content, and it's signature.
-    """
-
-    def __init__(self, content, signature):
-        super().__init__(TaskType.TEXT)
-        self.content = content
-        self.signature = signature
-
-    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
-        self.signature = content
-
-    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.
-    """
-
-    def __init__(self, image_path, signature, process):
-        super().__init__(TaskType.IMAGE)
-        self.image_path = image_path
-        self.signature = signature
-        self.process = process
-
-    def get_print_data(self):
-        # Return image data in printer-compatible format
-        return {"img": self.image_path, "sign": self.signature, "process": self.process}
-
-
-class CutTask(PrintTask):
-    """
-    This class activates the cutter on the printer if it exists
-    """
-
-    def __init__(self):
-        super().__init__(TaskType.CUT)
-
-    # There is no print data,
-    # the task existence in itself is indication of what to do
-    def get_print_data(self):
-        return None

src/templates/index.html

@@ -6,8 +6,7 @@
       <h3 class="card-header">Print a short message</h3>
       <div class="card-body">
         <form class="form-group" action="/web/print/sms" method="post">
-          <textarea class="form-control" type="text" name="txt" placeholder="4096 chars or less " maxlength="4096"></textarea>
-          <br>
+          <input class="form-control" type="text" name="txt" placeholder="200 chars or less " maxlength="200" required><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/user.py

@@ -1,41 +1,41 @@
-# class User(object):
-#     """docstring for User."""
+class User(object):
+    """docstring for User."""
 
-#     def __init__(self, arg):
-#         super(User, self).__init__()
-#         self.arg = arg
+    def __init__(self, arg):
+        super(User, self).__init__()
+        self.arg = arg
 
 
-# #         @app.route('/login', methods=['POST','GET'])
-# # @limiter.limit("100 per minute", error_message=error_handler_limiter)
-# def login():
-#     if request.method == "POST":
-#         if not session.get("logged_in"):
-#             if request.form["username"] and request.form["password"]:
-#                 # Get the json
-#                 with open("users.json") as f:
-#                     users_file = json.load(f)
-#                 for user in users_file["users"]:
-#                     if users_file["users"][user] == request.form["password"]:
-#                         session["logged_in"] = True
-#                         session["user"] = request.form["username"]
+#         @app.route('/login', methods=['POST','GET'])
+# @limiter.limit("100 per minute", error_message=error_handler_limiter)
+def login():
+    if request.method == "POST":
+        if not session.get("logged_in"):
+            if request.form["username"] and request.form["password"]:
+                # Get the json
+                with open("users.json") as f:
+                    users_file = json.load(f)
+                for user in users_file["users"]:
+                    if users_file["users"][user] == request.form["password"]:
+                        session["logged_in"] = True
+                        session["user"] = request.form["username"]
 
-#                 if not session.get("logged_in"):
-#                     flash("Mot de passe ou pseudo invalide.", "danger")
-#                     return redirect(url_for("login"))
-#                 else:
-#                     return redirect(url_for("display_index_page"))
-#             else:
-#                 flash("Incorrect logins")
-#                 return render_template("password.html")
-#         else:
-#             return render_template("password.html")
-#     else:
-#         return render_template("password.html")
+                if not session.get("logged_in"):
+                    flash("Mot de passe ou pseudo invalide.", "danger")
+                    return redirect(url_for("login"))
+                else:
+                    return redirect(url_for("display_index_page"))
+            else:
+                flash("Incorrect logins")
+                return render_template("password.html")
+        else:
+            return render_template("password.html")
+    else:
+        return render_template("password.html")
 
 
-# @app.route("/logout")
-# def logout():
-#     session["logged_in"] = False
-#     flash("Tu est déconnecté", "info")
-#     return redirect(url_for("login"))
+@app.route("/logout")
+def logout():
+    session["logged_in"] = False
+    flash("Tu est déconnecté", "info")
+    return redirect(url_for("login"))

src/web.py

@@ -1,19 +1,16 @@
-"""
-Manage all of the inputs from a web source
-"""
-
-import os
+from flask import Flask, request, flash
 from werkzeug.utils import secure_filename
-from task import TextTask, ImageTask, CutTask
+from printer import Printer
+import time
+import os
 
 
-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"""
+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"""
 
-    def __init__(self, app, print_queue):
+    def __init__(self, app, printer):
         super(Web).__init__()
-        self.print_queue = print_queue
+        self.printer = printer
         self.app = app
 
     def print_sms(self, texte, sign: str) -> bool:
@@ -21,19 +18,13 @@ class Web:
         Get text and a signature, prints the text and cuts after that.
         """
         self.app.logger.debug("Printing : " + str(texte) + " from " + str(sign))
-
         try:
-            # We create two new tasks and add them directly to the queue
-            # TODO: this might need to be improved because
-            # !! there is no garantee !! that both the SMS task and the Cut task
-            # are added back to back, another task could be
-            # inserted between the two.
-            sms = self.print_queue.enqueue(TextTask(content=texte, signature=sign))
-            cut = self.print_queue.enqueue(CutTask())
+            self.printer.print_sms(texte, sign)
+            self.printer.cut()
         except Exception as e:
             self.app.logger.error(e)
-            raise RuntimeError("Could not add SMS to queue, " + str(e)) from e
-        self.app.logger.info("Added two new tasks at position %s and %s", sms, cut)
+            raise RuntimeError("Could not print SMS, " + str(e)) from e
+
         return True
 
     def print_image(self, image, sign: str) -> bool:
@@ -41,46 +32,40 @@ class Web:
         Get an image and a signature, prints the image and cuts after that.
         """
         try:
-            file_uploaded = self.upload_file(image)
+            self.upload_file(image)
         except Exception as e:
             self.app.logger.error(e)
-            raise RuntimeError("Could not upload file : " + str(e)) from e
+            raise RuntimeError("Could not upload file") from e
 
-        if file_uploaded:
         self.app.logger.debug("File has been uploaded, printing...")
+
+
         try:
-                img = self.print_queue.enqueue(
-                    ImageTask(
+            self.printer.print_img(
                 os.path.join(
                     self.app.config["UPLOAD_FOLDER"],
                     secure_filename(image.filename),
                 ),
-                        signature=sign,
+                sign=sign,
                 process=True,
             )
-                )
-
-                cut = self.print_queue.enqueue(CutTask())
+            self.printer.cut()
         except Exception as e:
-                raise RuntimeError("Could not add IMG to queue" + str(e)) from e
-
-            self.app.logger.info("Added two new tasks at position %s and %s", img, cut)
+            raise RuntimeError("Could not print file") from e
 
+        self.app.logger.debug("Image printed and cut !")
         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:
-        """
-        Check if the file extension is allowed
-        """
-        self.app.logger.debug("Checking if the file extension is allowed")
+        self.app.logger.debug("Is the filename allowed ?")
         return (
             "." in filename
             and filename.rsplit(".", 1)[1].lower()
@@ -88,35 +73,21 @@ class Web:
         )
 
     def upload_file(self, image) -> bool:
-        """
-        Save the file after executing checks on it
-        """
         self.app.logger.debug("Validating file")
-        if not image is None or not image == "":
-            if self.allowed_file(image.filename):
+        if image and self.allowed_file(image.filename):
             filename = secure_filename(image.filename)
             self.app.logger.debug("File valid")
             try:
                 image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
-                except OSError as e:
+            except Exception as e:
                 self.app.logger.error("Could not save file %s", e)
-                    raise RuntimeError("An OS error occured while uploading this file : " + str(e)) from e
+                return False
 
             self.app.logger.debug(
                 "File saved to "
                 + str(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
             )
             return True
-
-        self.app.logger.error(
-            "Could not save file because the filename is forbidden"
-        )
-        raise RuntimeError("This file type is forbidden.")
-
-    def get_queue_state(self):
-        """Return current queue state"""
-        return self.print_queue.get_queue_state()
-
-    def get_queue_completed(self):
-        """Return completed queue elements"""
-        return self.print_queue.get_queue_completed()
+        else:
+            self.app.logger.error("Could not save file " + str(filename))
+            return False

src/worker.py

@@ -1,147 +0,0 @@
-"""
-This is the main printing thread. A worker thread consums Tasks from
- a PrintQueue, while trying to find available printers.
-"""
-
-import threading
-import time
-from printers import Printers
-
-
-class PrintWorker(threading.Thread):
-    """
-    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.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...")
-
-        try:
-            self.printers = Printers(self.app)
-            self.printers_obj = self.printers.printers
-            self.printers = iter(self.printers.printers)
-        except RuntimeError as e:
-            self.app.logger.warning("Could not get any Printers")
-            raise e
-
-    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())
-
-        while True:
-
-            # If the printer is dead or asleep, it can't work.
-            if not self.running:
-                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.
-            # 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:
-                time.sleep(1)
-                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",
-                    )
-                except Exception as e:
-                    self.app.logger.error("No printer detected" + str(e))
-                    self.printer = None
-
-            if self.state != "idle":
-                self.app.logger("We are not idle, waiting...")
-                time.sleep(1)
-                continue
-
-            self.state = "printing"
-
-            with self._lock:
-                try:
-                    task = self.print_queue.dequeue()
-                except Exception as e:
-                    self.app.logger.error("Could not get a new task ! %s ", str(e))
-                    self.state = "idle"
-                    raise RuntimeError(
-                        "We could not get a new task because " + str(e)
-                    ) from e
-
-                if task:
-                    try:
-                        self.app.logger.info("Got a new task")
-                        self.app.logger.debug("Got task %s", task.task_id)
-                        task.status = "processing"
-
-                        print_data = task.get_print_data()
-
-                        try:
-                            self.printer.print_task(task.task_type, print_data)
-                        except RuntimeError as e:
-                            self.state = "idle"
-                            self.app.logger.error("Could not print : %s", str(e))
-                            raise e
-
-                        task.status = "completed"
-                        self.print_queue.mark_completed(task.task_id, "completed")
-                        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.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 stop_worker(self):
-        """
-        Give the worker a break
-        """
-        self.app.logger.debug("Giving the worker a break")
-        self.state = "drinking-a-beer"
-        self.running = False
-
-    def start_worker(self):
-        """
-        Get the worker back to it
-        """
-        self.app.logger.debug("Time to work !")
-        self.state = "idle"
-        self.running = True
-
-    def current_state(self):
-        """
-        Return the worker state
-        """
-        return {
-            "is_running": self.running,
-            "queue_size": len(self.print_queue),
-            "state": self.state,
-            "printers": len(self.printers_obj),
-        }