Merge pull request 'Restructure the code and implement a printing queue' (#29) from restructure-printing-queue into master

Reviewed-on: #29

README.md

@@ -79,6 +79,16 @@ 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.

pyproject.toml

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

src/main.py

@@ -39,11 +39,14 @@ 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
 
-# Variables
-
+# We create the main Flask object
 app = Flask(__name__)
-socketio = SocketIO(app)
+socketio = SocketIO(app, cors_allowed_origins="*")
+
+# Global variables
 ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
 
 # Load the configuration file
@@ -82,7 +85,7 @@ except FileExistsError:
     app.logger.debug("Directory %s already exists.", UPLOAD_FOLDER)
 except PermissionError:
     app.logger.error("Permission denied: Unable to create %s", UPLOAD_FOLDER)
-    exit(77)
+    sys.exit(77)
 
 # Output the config file
 if os.getenv("FLASK_DEBUG"):
@@ -98,9 +101,6 @@ app.config["TEMPLATES_AUTO_RELOAD"] = True
 # 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
@@ -116,19 +116,22 @@ rpi = Raspberry(
 
 RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
 
+# Queue creation
+print_queue = PrintQueue(app)
 
-#  Web & API  routes
+#  Web & API management
+web = Web(app, print_queue)
 
-web = Web(app, printer)
-
-if __name__ == "__main__":
-    app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")
+# Start worker thread
+worker = PrintWorker(app, print_queue, printer, socketio)
+worker.start()
 
+# The rate limit
 limiter = Limiter(
     get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"]
 )
 
-
+# General routes
 @app.route("/")
 @limiter.limit("1/second", override_defaults=False)
 def index():
@@ -144,6 +147,8 @@ 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():
@@ -154,7 +159,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:
@@ -162,19 +167,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.", str(e)
+            "No signature found for this print, using default signature : %s ", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
 
     try:
         web.print_sms(txt, sign)
-    except Exception as e:
+    except RuntimeError 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 printed !", 'info')
+    flash("The SMS has been added to the print queue !", "info")
     return redirect(url_for("index"))
 
 
@@ -189,7 +194,7 @@ 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.", str(e)
+            "No signature found for this print, using default signature : %s", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
 
@@ -197,7 +202,7 @@ def web_print_img():
     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"]
@@ -205,20 +210,21 @@ 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 Exception as e:
+    except RuntimeError 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 printed !", 'info')
+    flash("Picture added to the print queue !", "info")
     return redirect(url_for("index"))
 
+
 # API routes
 # The api has the following methods
 # api/print/{sms,img,letter,qr,barcode}
@@ -257,11 +263,12 @@ def api_print_sms():
     try:
         # comment: We try to print the SMS
         web.print_sms(txt, sign)
-    except Exception as e:
+    except RuntimeError 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():
@@ -283,8 +290,7 @@ def api_print_image():
         return "No image submitted", 400
 
     file = request.files["img"]
-    # If the user does not select a file, the browser submits an
-    # empty file without a filename.
+    # If the user 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
@@ -292,25 +298,57 @@ def api_print_image():
     try:
         app.logger.debug("Sending the image to the printer.")
         web.print_image(file, sign)
-    except Exception as e:
+    except RuntimeError as e:
         return str(e), 500
 
     return "OK", 200
 
+
 @app.route("/api/camera/picture", methods=["GET"])
 def camera_picture():
-    """Returns a picture taken by the camera"""
+    """Returns a picture taken by the camera on a raspberry pi"""
     if RASPBERRY_PI_CONNECTED:
         try:
             return rpi.camera_picture()
-        except Exception as e:
+        except RuntimeError 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/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():
@@ -362,3 +400,7 @@ def camera_status():
         socketio.emit("camera_status", True)
     else:
         socketio.emit("camera_status", False)
+
+
+if __name__ == "__main__":
+    app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")

src/print_queue.py

@@ -0,0 +1,135 @@
+"""
+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} for t in self._queue]
+
+    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,13 +1,18 @@
-# Importing the module to manage the connection to the printer.
-import escpos.printer
-import brother_ql
-from time import sleep, gmtime, strftime
+"""
+This class manages connexion to a Printer
+"""
+# import brother_ql
+from time import sleep
 import os.path
-from PIL import Image, ImageEnhance, ImageOps
+
+from PIL import Image, ImageEnhance
 import numpy as np
 
+# Importing the module to manage the connection to the printer.
+import escpos.printer
 
-class Printer(object):
+
+class Printer():
     """
     # The connection is based on the ESC/POS library
 
@@ -28,7 +33,7 @@ class Printer(object):
     ready = False
 
     def __init__(self, app, device_id, vendor_id):
-        super(Printer, self).__init__()
+        super().__init__()
         self.app = app
         self.ready = False
         self.printer = None
@@ -39,7 +44,9 @@ class Printer(object):
         self.usb_args["idProduct"] = self.vendor_id
 
     def check_paper(self) -> bool:
-        # Let's check paper status
+        """
+        On printers that support it, we check that the printer has paper
+        """
         self.app.logger.debug("Checking paper status...")
         self.printer.open(self.usb_args)
         status = self.printer.paper_status()
@@ -58,10 +65,13 @@ class Printer(object):
                 self.printer.close()
 
     def init_printer(self):
+        """
+        Check if the printer online ? Is the communication with the printer successfull ?
+        """
 
-        # Is the printer online ? Is the communication with the printer successfull ?
+        # TODO: This could happen directly when creating a new Printer class
         if os.getenv("FLASK_DEBUG"):
-            waiting_elapsed = 1
+            waiting_elapsed = 3
         else:
             waiting_elapsed = 10
 
@@ -71,21 +81,26 @@ class Printer(object):
             try:
                 # This also calls open(), which we need to close()
                 # or else the device will appear as busy.
-                p = escpos.printer.Usb(self.device_id, self.vendor_id, 0, profile="TM-P80")
-            except Exception as e:
-                self.app.logger.error(
-                    "The USB device is not plugged in, trying again %s : %s",waiting_elapsed, str(e)
+                p = escpos.printer.Usb(
+                    self.device_id, self.vendor_id, 0, profile="TM-P80"
+                )
+            except RuntimeError as 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:
-                self.app.logger.error("Error while getting the printer online %s : %s",waiting_elapsed, str(e)
+            except RuntimeError as e:
+                self.app.logger.error(
+                    "Error while getting the printer online %s : %s",
+                    waiting_elapsed,
+                    str(e),
                 )
-                pass
 
             sleep(1)
             waiting_elapsed -= 1
@@ -122,15 +137,24 @@ class Printer(object):
 
         return True
 
-    def print_sms(self, msg, signature="", bold=False):
+    def _print_sms(self, msg, signature="", bold=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
         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 Exception(
+            raise RuntimeError(
                 "Could not print message of this length :"
                 + str(len(clean_msg))
                 + ", needs to be below 4096 caracters long."
@@ -140,12 +164,14 @@ class Printer(object):
             self.app.logger.warning(
                 "Could not print signature of this length: " + str(len(clean_signature))
             )
-            raise Exception(
+            raise RuntimeError(
                 "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)
@@ -155,19 +181,21 @@ class Printer(object):
             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")
         return True
 
-    def print_img(self, path, sign="", center=True, process=False):
-        clean_signature = str(sign)
+    def _print_img(self, path, signature="", center=True, process=False):
+        clean_signature = str(signature)
 
-        if len(sign) > 256:
+        if len(signature) > 256:
             self.app.logger.warning(
                 "Could not print signature of this length: " + str(len(clean_signature))
             )
-            raise Exception(
+            raise ValueError(
                 "Could not print signature of this length :"
                 + str(len(clean_signature))
                 + ", needs to be below 256 caracters long."
@@ -175,50 +203,60 @@ class Printer(object):
 
         if not os.path.isfile(str(path)):
             self.app.logger.warning("File does not exist : " + str(path))
-            raise Exception(
+            raise OSError(
                 "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")
-                path = process_image(self, path)
-            except Exception as e:
-                self.app.logger.error(str(e))
-                return False
+                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:
             self.app.logger.warning("Not proccessing the image")
 
         try:
             self.printer.open(self.usb_args)
             self.printer.image(path, center=center)
+            self.printer.textln(signature)
             self.printer.close()
             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)
+            except OSError as e:
+                raise e
+
+            self.app.logger.debug("Removed image : " + str(path))
+
             try:
                 self.printer.close()
             except Exception as e:
-                self.app.logger.error(str(e))
+                self.app.logger.error(
+                    "Could not close the printer connexion %s", str(e)
+                )
                 raise RuntimeError("Could not close the printer connexion. ") from e
 
         self.app.logger.info("Printed a picture")
         return True
 
-    def qr(self, content):
+    def _qr(self, content):
         try:
             self.printer.open(self.usb_args)
             self.printer.qr(content, center=True)
             self.printer.close()
-        except Exception as e:
+        except RuntimeError as e:
             self.printer.close()
             self.app.logger.error(str(e))
             return False
@@ -226,7 +264,7 @@ class Printer(object):
         self.app.logger.info("Printed a QR")
         return True
 
-    def cut(self):
+    def _cut(self):
         try:
             self.printer.open(self.usb_args)
             self.printer.cut()
@@ -239,8 +277,22 @@ class Printer(object):
         self.app.logger.info("Did a cut")
         return True
 
+    def print_task(self, task_type, data):
+        """Execute actual print based on task type"""
+        match (task_type.value):
+            case "text":
+                self._print_sms(data["txt"], signature=data["sign"])
+            case "image":
+                self._print_img(
+                    data["img"], signature=data["sign"], process=data["process"]
+                )
+            case "cut":
+                self._cut()
+            case _:
+                raise RuntimeError("This task type is not supported")
 
-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 = 0.6  # Less than 1.0 = lower contrast
@@ -254,11 +306,12 @@ def process_image(self, path):
             original_img = original_img.convert("RGB")
 
         # Resize while maintaining aspect ratio
-        original_img.thumbnail((max_width, max_height), Image.LANCZOS)
+        original_img.thumbnail((max_width, max_height), Image.Resampling.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)
@@ -286,107 +339,9 @@ def process_image(self, path):
         # 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"
         original_img.save(jpeg_path, format="JPEG", quality=95, optimize=True)
-        app.logger.debug("Processed and saved image.")
+        self.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/raspberry.py

@@ -1,13 +1,21 @@
-from flask_socketio import SocketIO
-from gpiozero import Button, LED, DigitalOutputDevice
-from time import sleep, gmtime, strftime
-from PIL import Image
+"""
+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.
+"""
+
 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(object):
+class Raspberry():
     """
     This class will manage three things :
     - Connecting to a USB webcam
@@ -19,7 +27,7 @@ class Raspberry(object):
 
     def __init__(
         self,
-        printer,
+        print_queue,
         app,
         socketio,
         button_gpio_port_number,
@@ -27,7 +35,7 @@ class Raspberry(object):
         flash_gpio_port_number,
         is_flash_present,
     ):
-        self.printer = printer
+        self.print_queue = print_queue
         self.socketio = socketio
         self.app = app
 
@@ -38,9 +46,13 @@ class Raspberry(object):
         self.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg"
 
     def is_raspberry_pi(self, raise_on_errors=False):
+        """
+        Checking if we are on a Raspberry Pi by checking
+        information on the /proc/cpuinfo file
+        """
         # Check if we are running on a raspberry pi
         try:
-            with io.open("/proc/cpuinfo", "r") as cpuinfo:
+            with io.open("/proc/cpuinfo", "r", encoding="utf-8") as cpuinfo:
                 found = False
                 for line in cpuinfo:
                     if line.startswith("Hardware"):
@@ -141,7 +153,7 @@ class Raspberry(object):
                 f"Unable to take a picture. Error: {e.stderr.decode()}"
             )
             return False
-        except Exception as e:
+        except RuntimeError as e:
             # Catch any unexpected errors
             self.app.logger.error(f"Unexpected error while taking picture: {str(e)}")
             return False
@@ -198,7 +210,7 @@ class Raspberry(object):
                 output_path = image_path  # Overwrite the original image if no output path is given
             image.save(output_path)
 
-        except Exception as e:
+        except RuntimeError as e:
             self.app.logger.error(f"Error overlaying logo: {e}")
             return False
 
@@ -225,7 +237,7 @@ class Raspberry(object):
 
             image.save(output_path)
 
-        except Exception as e:
+        except RuntimeError as e:
             self.app.logger.error(f"Error cropping image to square: {e}")
             return False
 
@@ -242,7 +254,7 @@ class Raspberry(object):
         try:
             self.flash.on()
             self.take_picture()
-        except Exception as e:
+        except RuntimeError as e:
             self.app.logger.error(
                 "Could not take a picture after the button press : " + str(e)
             )
@@ -251,15 +263,11 @@ class Raspberry(object):
         self.app.logger.debug("Printing picture")
         self.led.on()
         self.crop_to_square(self.image_path)
-        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.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.led.off()
-        self.app.logger.debug("Done printing picture")
+        self.app.logger.debug("Added a photomaton picture to the print queue")
         return True

src/task.py

@@ -0,0 +1,98 @@
+"""
+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"
+
+
+class PrintTask(ABC):
+    """
+    A print task holds information about what we are looking to print.
+    """
+
+    def __init__(self, task_type):
+        self.task_id = self._generate_id()
+        self.task_type = task_type
+        self.status = "pending"  # pending, processing, completed, failed
+
+        print("Created a new " + str(self.task_type) + " with ID " + self.task_id)
+
+    @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 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/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,16 +1,17 @@
-from flask import Flask, request, flash
+from flask import flash
 from werkzeug.utils import secure_filename
-from printer import Printer
 import time
 import os
+from task import TextTask, ImageTask, CutTask
 
 
 class Web(object):
-    """Web is the class that gets all of the information from web calls ( API and Web page ) and provides checks before sending stuff to printing"""
+    """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, printer):
+    def __init__(self, app, print_queue):
         super(Web).__init__()
-        self.printer = printer
+        self.print_queue = print_queue
         self.app = app
 
     def print_sms(self, texte, sign: str) -> bool:
@@ -18,13 +19,19 @@ class Web(object):
         Get text and a signature, prints the text and cuts after that.
         """
         self.app.logger.debug("Printing : " + str(texte) + " from " + str(sign))
+
         try:
-            self.printer.print_sms(texte, sign)
-            self.printer.cut()
+            # 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())
         except Exception as e:
             self.app.logger.error(e)
-            raise RuntimeError("Could not print SMS, " + str(e)) from 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)
         return True
 
     def print_image(self, image, sign: str) -> bool:
@@ -32,28 +39,31 @@ class Web(object):
         Get an image and a signature, prints the image and cuts after that.
         """
         try:
-            self.upload_file(image)
+            file_uploaded = self.upload_file(image)
         except Exception as e:
             self.app.logger.error(e)
             raise RuntimeError("Could not upload file") from e
 
+        if file_uploaded:
             self.app.logger.debug("File has been uploaded, printing...")
-
-
             try:
-            self.printer.print_img(
+                img = self.print_queue.enqueue(
+                    ImageTask(
                         os.path.join(
                             self.app.config["UPLOAD_FOLDER"],
                             secure_filename(image.filename),
                         ),
-                sign=sign,
+                        signature=sign,
                         process=True,
                     )
-            self.printer.cut()
-        except Exception as e:
-            raise RuntimeError("Could not print file") from e
+                )
+
+                cut = self.print_queue.enqueue(CutTask())
+            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)
 
-        self.app.logger.debug("Image printed and cut !")
         return True
 
     def login(self, username: str, password: str) -> bool:
@@ -74,12 +84,13 @@ class Web(object):
 
     def upload_file(self, image) -> bool:
         self.app.logger.debug("Validating file")
-        if image and self.allowed_file(image.filename):
+        if image:
+            if 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 Exception as e:
+                except OSError as e:
                     self.app.logger.error("Could not save file %s", e)
                     return False
 
@@ -89,5 +100,16 @@ class Web(object):
                 )
                 return True
             else:
-            self.app.logger.error("Could not save file " + str(filename))
+                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
+
+    def get_queue_state(self):
+        """Return current queue state"""
+        return self.print_queue.get_queue_state()

src/worker.py

@@ -0,0 +1,107 @@
+# This is the main printing thread
+# As explained in the task file, this is where we command
+# printing to happen.
+
+import threading
+import time
+
+
+class PrintWorker(threading.Thread):
+    def __init__(self, app, print_queue, printer, socketio=None):
+        super().__init__(daemon=True)
+        self.app = app
+        self.print_queue = print_queue
+        self.printer = printer
+        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...")
+
+    def run(self):
+        """Background thread that processes queue items"""
+        self.app.logger.info("Worker started working.")
+        while True:
+            if not self.running or not self.printer.ready:
+                time.sleep(0.2)
+                continue
+
+            try:
+                task = self.print_queue.dequeue()
+            except Exception as e:
+                self.app.logger.error("Could not get a new task ! %s ", str(e))
+                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)
+                    self.state = "printing"
+                    task.status = "processing"
+                    self._emit_status(task.task_id, "processing")
+
+                    print_data = task.get_print_data()
+                    try:
+                        self.printer.print_task(task.task_type, print_data)
+                    except RuntimeError as e:
+                        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._emit_status(task.task_id, "completed")
+
+                except RuntimeError as e:
+                    task.status = "failed"
+                    self.print_queue.mark_completed(task.task_id, "failed")
+                    self._emit_status(task.task_id, "failed", error=str(e))
+                    print(f"Print task {task.task_id} failed: {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
+        """
+        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,
+        }