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

@@ -101,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
@@ -122,20 +119,19 @@ RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
 # Queue creation
 print_queue = PrintQueue(app)
 
-#  Web & API  routes
+#  Web & API management
 web = Web(app, print_queue)
 
 # 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():
@@ -153,8 +149,6 @@ def webcam():
 
 
 # Form treatement
-
-
 @app.route("/web/print/sms", methods=["POST"])
 @limiter.limit("6/minute", override_defaults=False)
 def web_print_sms():
@@ -185,7 +179,7 @@ def web_print_sms():
         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"))
 
 
@@ -227,7 +221,7 @@ def web_print_img():
         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"))
 
 
@@ -296,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
+    # If the user submits an empty file without a filename.
-    # empty file without a filename.
     if file.filename == "":
         app.logger.error("Submitted file has no filename !")
         return "Submitted file has no filename !", 400
@@ -337,12 +330,18 @@ def api_worker_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())
 

src/printer.py

@@ -1,4 +1,6 @@
-
+"""
+This class manages connexion to a Printer
+"""
 # import brother_ql
 from time import sleep
 import os.path
@@ -10,7 +12,7 @@ import numpy as np
 import escpos.printer
 
 
-class Printer(object):
+class Printer():
     """
     # The connection is based on the ESC/POS library
 
@@ -31,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
@@ -69,7 +71,7 @@ class Printer(object):
 
         # TODO: This could happen directly when creating a new Printer class
         if os.getenv("FLASK_DEBUG"):
-            waiting_elapsed = 15
+            waiting_elapsed = 3
         else:
             waiting_elapsed = 10
 
@@ -82,25 +84,23 @@ class Printer(object):
                 p = escpos.printer.Usb(
                     self.device_id, self.vendor_id, 0, profile="TM-P80"
                 )
-            except Exception as e:
+            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:
+            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
@@ -154,7 +154,7 @@ class Printer(object):
             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."
@@ -164,7 +164,7 @@ 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."
@@ -208,8 +208,8 @@ class Printer(object):
                 + str(path)
                 + " wasn't found. Please try again."
             )
-        else:
+
-            self.app.logger.debug("Printing file from " + str(path))
+        self.app.logger.debug("Printing file from " + str(path))
 
         if process:
             try:
@@ -256,7 +256,7 @@ class Printer(object):
             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
@@ -310,7 +310,8 @@ def _process_image(self, path):
         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)
@@ -344,94 +345,3 @@ def _process_image(self, path):
         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,3 +1,11 @@
+"""
+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
@@ -5,6 +13,7 @@ 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():
     """
@@ -18,7 +27,7 @@ class Raspberry():
 
     def __init__(
         self,
-        printer,
+        print_queue,
         app,
         socketio,
         button_gpio_port_number,
@@ -26,7 +35,7 @@ class Raspberry():
         flash_gpio_port_number,
         is_flash_present,
     ):
-        self.printer = printer
+        self.print_queue = print_queue
         self.socketio = socketio
         self.app = app
 
@@ -37,9 +46,13 @@ class Raspberry():
         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"):
@@ -140,7 +153,7 @@ class Raspberry():
                 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
@@ -197,7 +210,7 @@ class Raspberry():
                 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
 
@@ -224,7 +237,7 @@ class Raspberry():
 
             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
 
@@ -241,7 +254,7 @@ class Raspberry():
         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)
             )
@@ -250,15 +263,11 @@ class Raspberry():
         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.print_queue.enqueue(ImageTask(self.image_path,signature="",process=True))
-        self.printer.print_img(self.image_path, process=True)
+        self.print_queue.enqueue(TextTask(content="Imprimé par LittlePrynter", signature=""))
-        self.printer.print_sms("")
+        time = strftime("%Y-%m-%d %H:%M", gmtime())
-        self.printer.print_sms("With Love From Société.Vide", signature="", bold=True)
+        self.print_queue.enqueue(TextTask(content=time, signature=""))
-        self.printer.print_sms("Printed by LittlePrynter", signature="")
+        self.print_queue.enqueue(CutTask())
-        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("Done printing picture")
+        self.app.logger.debug("Added a photomaton picture to the print queue")
         return True

src/task.py

@@ -18,7 +18,6 @@ from abc import ABC, abstractmethod
 
 ## See https://docs.python.org/3/library/abc.html to learn more about this
 
-# from dataclasses import dataclass
 from enum import Enum
 import uuid
 

src/web.py

@@ -6,7 +6,8 @@ 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, print_queue):
         super(Web).__init__()