Apply formating, light linting

printers

README.md

@@ -15,6 +15,7 @@ To make this project work, you will need :
 - A Thermal Printer :
     - I used *Adafruit's thermal printer* at first, but it broke down. You can get it [here](https://www.adafruit.com/product/597).
     - I then used an *EPSON TM-T2OIII*. It's expensive, but it's also high quality, and has better CUPS and specific library support.
+    - Sometime after, a friend leant me a *Brother QL-650TD* to add support for this type of printer.
 - A Raspberry Pi or any computer with USB or Serial.
 - Some electric wires.
 - Some knowledge of the command line,
@@ -37,13 +38,15 @@ You can also get some information from [here](https://learn.adafruit.com/mini-th
 
 LittlePrynter itself does not require any other software than the ones installed with it, i.e the ones listed in the `pyproject.toml` file.
 
-The version for the EPSON TM-T20III uses a library called `python-escpos`, which doesn't need a CUPS driver or anything else. It's included in the Python dependencies.
+The project supports EPSON label printers as well as Brother label printers, and you can try to adapt it for other printers. For example, using `adafruit-thermal`.
 
-The project only supports the EPSON printer, but you can try to adapt it for other printers. For example, using `adafruit-thermal`, or `brother-ql`.
+For the EPSON TM-T20III, LittlePrynter uses a library called `python-escpos`, which doesn't need a CUPS driver or anything else. It's included in the Python dependencies. For the Brother-QL, we use the `brother-ql` library, which is also included.
 
 ### Install & setup the project :
 
-Theses commands will copy the software on your computer, go into the directory, then activate a virtual environnement and install all of the project's dependecies.
+We use [Poetry](https://python-poetry.org/) to manage the Python environnement and packages. You will need to install it on your computer to install LittlePrynter.
+
+The following commands will copy the software on your computer, go into the directory, then activate a virtual environnement and install all of the project's dependecies.
 
 ```
 $ git clone https://git.n07070.xyz/n07070/LittlePrynter
@@ -69,6 +72,8 @@ $ flask run --cert=adhoc
 
 The `--cert=adhoc` argument will make LittlePrinter accessible in HTTPS with a [self-signed SSL certificate](https://en.wikipedia.org/wiki/Self-signed_certificate). This provides a free and easy way to use HTTPS, but does not provide any trust value. It is, on the other hand, a good way to access the webcam in developpement mode or on a local network, because a HTTPS connexion is required by browsers to access the needed APIs.
 
+You can also use the `run.sh` for this.
+
 This command should start a web server with which you can test your configuration. If you plan on exposing your printer to the Internet, and give it an IP / URL, _please, please, please_, don't run it this way. Look at Flask's documentation and read about running a production server. It's a little more work, but it will prevent your computer/server being hacked in too easily.
 
 Voilà !
@@ -81,11 +86,11 @@ Please also say if you had a printer to test your code, and which printer you've
 
 ### 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 app is written around 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 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 of 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.
 
@@ -130,7 +135,7 @@ If you liked this project, feel free to support my work !
 - [A link to buy one in Europe](https://rlx.sk/sk/various-boards/1829-mini-thermal-receipt-printer-adafruit-597.html)
 - [Another link to buy one, direct from factory](https://www.cashinotech.com/csn-a2-58mm-mini-panel-thermal-receipt-printer_p11.html)
 - [Diagnostickoeur](https://framagit.org/stickoeur/diagnostickoeur), another printer software mainly around Brother QL printers.
-
+- [FaxTheSystem](https://github.com/hackstub/faxthesystem/tree/main), another printer software, also around an Epson TM printer
 
 ## Licence
 

configuration/config.toml

@@ -1,11 +1,28 @@
 # Configuration file the LittlePrynter
 
 [defaults]
-signature = "Anonymous"
+signature = ""
+
 
 # Printer settings
 [printer]
 upload_folder = "src/static/uploads"
+# How long do we wait before giving up on waiting for a printer connexion
+waiting_time = 5
+
+[printer.epson]
+# Configurations for EPSON printers
+profile = "TM-T20III"
+label_type = 80 # 80 or 58 ( label width in mm )
+align_text = "center"
+text_max_length = 4096
+signature_max_length = 256
+
+[printer.brother]
+# Configurations for Brother printers
+label_type = "62" # See https://github.com/pklaus/brother_ql#usage for label types
+text_max_length = 4096
+signature_max_length = 256
 
 # Raspberry Pi Configuration
 [rpi]

configuration/config.toml.sample

@@ -5,8 +5,25 @@ signature = "Anonymous"
 
 # Printer settings
 [printer]
-vendor_id = "0x04b8"
-device_id = "0x0e28"
+# Where do we upload the pictures and such
+upload_folder = "src/static/uploads"
+
+# How long do we wait before giving up on waiting for a printer connexion
+waiting_time = 5
+
+[printer.epson]
+# Configurations for EPSON printers
+profile = "TM-T20III"
+label_type = 80 # 80 or 58 ( label width in mm )
+align_text = "center"
+text_max_length = 4096
+signature_max_length = 256
+
+[printer.brother]
+# Configurations for Brother printers
+label_type = 62 # See https://github.com/pklaus/brother_ql#usage for label types
+text_max_length = 4096
+signature_max_length = 256
 
 # Raspberry Pi Configuration
 [rpi]

poetry.lock

@@ -651,6 +651,18 @@ enabler = ["pytest-enabler (>=3.4)"]
 test = ["jaraco.test (>=5.4)", "pytest (>=6,!=8.1.*)", "zipp (>=3.17)"]
 type = ["pytest-mypy (>=1.0.1) ; platform_python_implementation != \"PyPy\""]
 
+[[package]]
+name = "iniconfig"
+version = "2.3.1"
+description = "brain-dead simple config-ini parsing"
+optional = false
+python-versions = ">=3.10"
+groups = ["main"]
+files = [
+    {file = "iniconfig-2.3.1-py3-none-any.whl", hash = "sha256:9121e2c1fdb355232495be3194c8dfe87ccc2d5dee45947b78e68f499790d7a7"},
+    {file = "iniconfig-2.3.1.tar.gz", hash = "sha256:67f4b9c50da0dedf52af349e7749a80a9057a5031199791b906c3bb3ae878960"},
+]
+
 [[package]]
 name = "isort"
 version = "8.0.1"
@@ -1137,6 +1149,18 @@ files = [
     {file = "platformdirs-4.9.6.tar.gz", hash = "sha256:3bfa75b0ad0db84096ae777218481852c0ebc6c727b3168c1b9e0118e458cf0a"},
 ]
 
+[[package]]
+name = "pluggy"
+version = "1.7.0"
+description = "plugin and hook calling mechanisms for python"
+optional = false
+python-versions = ">=3.10"
+groups = ["main"]
+files = [
+    {file = "pluggy-1.7.0-py3-none-any.whl", hash = "sha256:7dd7b0d8832ba3cb632c306926ded123429211b83641b35dc5c41ad2d34f9bec"},
+    {file = "pluggy-1.7.0.tar.gz", hash = "sha256:d1eaa46ebb595891b860ab086b4d09c8588af65ebd4361b8e8f4bb8920b90ba8"},
+]
+
 [[package]]
 name = "pycparser"
 version = "3.0"
@@ -1201,6 +1225,21 @@ files = [
     {file = "pycryptodomex-3.23.0.tar.gz", hash = "sha256:71909758f010c82bc99b0abf4ea12012c98962fbf0583c2164f8b84533c2e4da"},
 ]
 
+[[package]]
+name = "pygments"
+version = "2.21.0"
+description = "Pygments is a syntax highlighting package written in Python."
+optional = false
+python-versions = ">=3.9"
+groups = ["main"]
+files = [
+    {file = "pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9"},
+    {file = "pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c"},
+]
+
+[package.extras]
+windows-terminal = ["colorama (>=0.4.6)"]
+
 [[package]]
 name = "pylint"
 version = "4.0.5"
@@ -1253,6 +1292,28 @@ files = [
 [package.extras]
 cp2110 = ["hidapi"]
 
+[[package]]
+name = "pytest"
+version = "9.2.0"
+description = "pytest: simple powerful testing with Python"
+optional = false
+python-versions = ">=3.10"
+groups = ["main"]
+files = [
+    {file = "pytest-9.2.0-py3-none-any.whl", hash = "sha256:07acdcf88ec174f907dc01af70bf52d49b014bbab03faeeade31c96b18c2e588"},
+    {file = "pytest-9.2.0.tar.gz", hash = "sha256:97c4f3b1f3c7414aa36465e09b8220f1d03627993bd51b1282e193753c80557b"},
+]
+
+[package.dependencies]
+colorama = {version = ">=0.4", markers = "sys_platform == \"win32\""}
+iniconfig = ">=2"
+packaging = ">=24"
+pluggy = ">=1.6,<2"
+pygments = ">=2.15"
+
+[package.extras]
+dev = ["argcomplete", "asynctest ; python_version < \"3.11\"", "attrs (>=23.1)", "coverage (>=7.5)", "hypothesis (>=6.75)", "mock", "numpy (>=1.26)", "pexpect (>=4.9)", "pytest-xdist (>=3.5)", "pyyaml", "requests", "setuptools", "xmlschema"]
+
 [[package]]
 name = "python-barcode"
 version = "0.16.1"
@@ -1882,5 +1943,5 @@ h11 = ">=0.16.0,<1"
 
 [metadata]
 lock-version = "2.1"
-python-versions = ">=3.14"
-content-hash = "18e1a5c8b085aa639b665279856f9e3c51a734a6a8d1d4dd135ca7e4f67e86ae"
+python-versions = ">=3.13"
+content-hash = "7f8ddfbcbf88df2634e01dd4daf62ad54d1b97c279e3c4b478822fb5d4485647"

pyproject.toml

@@ -43,6 +43,7 @@ dependencies = [
     "pep8 (>=1.7.1,<2.0.0)",
     "pylint (>=4.0.5,<5.0.0)",
     "brother-ql-inventree (>=1.3,<2.0)",
+    "pytest (>=9.2.0,<10.0.0)",
 ]
 
 [tool.poetry]

run.sh

@@ -0,0 +1,2 @@
+eval $(poetry env activate)
+export FLASK_DEBUG=True; export FLASK_APP=src/main.py && flask run  --cert=adhoc --debug --no-debugger --host 0.0.0.0 --without-threads

src/main.py

@@ -1,7 +1,8 @@
 """
 Welcome to the LittlePrynter's source code.
-This program expose a web interface, with user authentification,
+This program expose a web interface,
 that makes it possible to print messages from the web.
+
 It also exposes a API, making it possible to print and interface
 with much of the printer's abilities.
 
@@ -10,12 +11,14 @@ Then we build the API around Flask,
 Then we build the web interface, using the simple Jinja2 templating.
 
 We support two modes :
+
 The first is a simple mode, where a computer, connected to a thermal printer,
 runs this program and exposes a web interface that makes use of the client's camera
-The seconde is booth mode, where a Raspberry Pi is connected to a thermal printer,
-a button and a flash.
-The web interface exists but may not be used, as the press of the button with take
- a picture and activate the flash while simply informing the web page.
+
+The seconde is booth mode, where a Raspberry Pi is connected to
+a thermal printer, a button and a flash.
+The web interface exists but it's not needed, as the press of the button
+takes a picture and activates the flash while simply informing the web page.
 """
 
 # Following are the librairies we import,
@@ -40,6 +43,7 @@ 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
+from printers import Printers
 
 # We create the main Flask object
 app = Flask(__name__)
@@ -97,21 +101,25 @@ app.config["TEMPLATES_AUTO_RELOAD"] = True
 print_queue = PrintQueue(app)
 
 # Find out if we are running on a Raspberry Pi
-rpi = Raspberry(
-    print_queue,
-    app,
-    configuration_file
-)
+rpi = Raspberry(print_queue, app, configuration_file)
 RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
 
 
 #  Web & API management
 web = Web(app, print_queue)
 
-# Start worker thread
-# When created, the worker will try to find printers connected to the system
+# Create a pool of printers
 try:
-    worker = PrintWorker(app, print_queue)
+    # comment: Find all printers connected to the system
+    printers_pool = Printers(app, configuration_file)
+except Exception as e:
+    app.logger.error("Could not create a pool of printers because %s", str(e))
+    sys.exit(-1)
+# end try
+
+# Start worker thread
+try:
+    worker = PrintWorker(app, print_queue, printers_pool=printers_pool)
     worker.start()
 except Exception as e:
     app.logger.error("Could not start the worker because %s ", str(e))
@@ -123,6 +131,9 @@ limiter = Limiter(
 )
 
 
+app.logger.info("🖶 Welcome to LittlePrynter !")
+
+
 # General routes
 @app.route("/")
 @limiter.limit("1/second", override_defaults=False)
@@ -162,9 +173,19 @@ def web_print_sms():
             "No signature found for this print, using default signature : %s ", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
+    try:
+        do_cut = bool(request.form["cut"])
+    except werkzeug.exceptions.BadRequestKeyError:
+        do_cut = False
 
     try:
-        web.print_sms(txt, sign)
+        # Clamp value between 1 and
+        many = max(1, min(int(request.form["many"]), 20))
+    except werkzeug.exceptions.BadRequestKeyError:
+        many = 1
+
+    try:
+        web.print_sms(txt, sign, do_cut, many)
     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")
@@ -208,9 +229,25 @@ def web_print_img():
         flash("Submitted file has no filename !", "error")
         return redirect(url_for("index"))
 
+    try:
+        do_cut = bool(request.form["cut"])
+    except werkzeug.exceptions.BadRequestKeyError:
+        do_cut = False
+
+    try:
+        process_image = bool(request.form["process"])
+    except werkzeug.exceptions.BadRequestKeyError:
+        process_image = False
+
+    try:
+        # Clamp value between 1 and
+        many = max(1, min(int(request.form["many"]), 20))
+    except werkzeug.exceptions.BadRequestKeyError:
+        many = 1
+
     try:
         app.logger.debug("Sending the image to the printer.")
-        web.print_image(file, sign)
+        web.print_image(file, sign, do_cut, process_image, many)
     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")
@@ -255,9 +292,21 @@ def api_print_sms():
             "No signature found for this print, using default signature. : %s", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
+
+    try:
+        do_cut = bool(request.form["cut"])
+    except werkzeug.exceptions.BadRequestKeyError:
+        do_cut = False
+
+    try:
+        # Clamp value between 1 and
+        many = max(1, min(int(request.form["many"]), 20))
+    except werkzeug.exceptions.BadRequestKeyError:
+        many = 1
+
     try:
         # comment: We try to print the SMS
-        web.print_sms(txt, sign)
+        web.print_sms(txt, sign, do_cut, many)
     except RuntimeError as e:
         return str(e), 500
     # end try
@@ -290,9 +339,25 @@ def api_print_image():
         app.logger.error("Submitted file has no filename !")
         return "Submitted file has no filename !", 400
 
+    try:
+        do_cut = bool(request.form["cut"])
+    except werkzeug.exceptions.BadRequestKeyError:
+        do_cut = False
+
+    try:
+        process_image = bool(request.form["process"])
+    except werkzeug.exceptions.BadRequestKeyError:
+        process_image = False
+
+    try:
+        # Clamp value between 1 and
+        many = max(1, min(int(request.form["many"]), 20))
+    except werkzeug.exceptions.BadRequestKeyError:
+        many = 1
+
     try:
         app.logger.debug("Sending the image to the printer.")
-        web.print_image(file, sign)
+        web.print_image(file, sign, do_cut, process_image, many)
     except RuntimeError as e:
         return str(e), 500
 
@@ -301,6 +366,7 @@ def api_print_image():
 
 # TODO: This might not depend on the Raspberry Pi
 @app.route("/api/camera/picture", methods=["GET"])
+@limiter.limit("6/minute", override_defaults=False)
 def camera_picture():
     """Returns a picture taken by the camera on a raspberry pi"""
     if RASPBERRY_PI_CONNECTED:
@@ -313,24 +379,28 @@ def camera_picture():
 
 
 @app.route("/api/queue", methods=["GET"])
+@limiter.limit("6/minute", override_defaults=False)
 def api_queue_status():
     """API endpoint for entire queue"""
     return jsonify(web.get_queue_state())
 
 
 @app.route("/api/queue/completed", methods=["GET"])
+@limiter.limit("6/minute", override_defaults=False)
 def api_queue_completed():
     """API endpoint that returns the finished tasks"""
     return jsonify(web.get_queue_completed())
 
 
 @app.route("/api/worker", methods=["GET"])
+@limiter.limit("6/minute", override_defaults=False)
 def api_worker_state():
     """API endpoint to get the worker state"""
     return jsonify(worker.current_state())
 
 
 @app.route("/api/worker/start")
+@limiter.limit("6/minute", override_defaults=False)
 def api_worker_start():
     """
     Enable to worker. This starts to process the print queue.
@@ -340,6 +410,7 @@ def api_worker_start():
 
 
 @app.route("/api/worker/stop")
+@limiter.limit("6/minute", override_defaults=False)
 def api_worker_stop():
     """
     Stops the print queue. This stops the processing of the print queue.
@@ -348,6 +419,13 @@ def api_worker_stop():
     return jsonify(worker.current_state())
 
 
+@app.route("/api/printers")
+@limiter.limit("6/minute", override_defaults=False)
+def api_printers_info():
+    """Returns the current state of every printers"""
+    return jsonify(printers_pool.printers_info())
+
+
 ## Authentification
 
 

src/print_queue.py

@@ -77,7 +77,12 @@ class PrintQueue:
         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())}
+                {
+                    "task_id": t.task_id,
+                    "status": t.status,
+                    "type": str(t.task_type),
+                    "content": str(t.get_print_data()),
+                }
                 for t in self._queue
             ]
 

src/printer.py

@@ -10,17 +10,20 @@ import time
 from enum import Enum
 import uuid
 import threading
+from pathlib import Path
+from uuid import uuid4
 import usb.core
-from PIL import Image, ImageEnhance
+from PIL import Image, ImageEnhance, ImageDraw, ImageFont
 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.backends import backend_factory, guess_backend
 from brother_ql.raster import BrotherQLRaster
 from brother_ql.conversion import convert
 from brother_ql.backends.helpers import send
+from brother_ql import create_label
 
 
 class PrinterType(Enum):
@@ -38,6 +41,7 @@ class PrinterInfo:
     """
     Brother-QL printer information
     """
+
     identifier: str
     backend: str
     protocol: str
@@ -75,6 +79,7 @@ class Printer(ABC):
         self.ready = False
         self.printer_type = printer_type
         self._lock = threading.Lock()
+        self.used = False
 
     @abstractmethod
     def _has_paper(self) -> bool:
@@ -88,13 +93,21 @@ class Printer(ABC):
     def print_task(self, task_type, data) -> None:
         """Takes a PrintTask and executes it"""
 
+    def use_printer(self) -> None:
+        """Marks the printer as being used"""
+        self.used = True
+
+    def free_printer(self) -> None:
+        """Marks the printer as being not used, it's available."""
+        self.used = False
+
 
 class EscPosPrinter(Printer):
     """
     Create a new ESC/POS based printer.
     """
 
-    def __init__(self, app, vendor_id, device_id):
+    def __init__(self, app, vendor_id, device_id, configuration_file):
         """
         Create a connexion to a ESC/POS Printer via USB,
         Making sure the printer is alive,
@@ -106,18 +119,23 @@ class EscPosPrinter(Printer):
         self.usb_args = {}
         self.usb_args["idVendor"] = self.vendor_id
         self.usb_args["idProduct"] = self.device_id
+        self.configuration_file = configuration_file
 
         try:
             # This also calls open(), which we need to close()
             # or else the device will appear as busy.
-            p = escpos.printer.Usb(self.vendor_id, self.device_id, 0, profile="TM-P80")
+            p = escpos.printer.Usb(
+                self.vendor_id, self.device_id, 0, profile=configuration_file["profile"]
+            )
         except escpos.exceptions.DeviceNotFoundError as e:
             self.app.logger.error(
                 "The USB device is not plugged in : %s",
                 str(e),
             )
+            raise e
         except Exception as e:
             self.app.logger.error("Printer could not be connected : %s ", str(e))
+            raise e
 
         try:
             if p.is_online():
@@ -125,9 +143,21 @@ class EscPosPrinter(Printer):
         except Exception as e:
             raise e
 
+        self.printer_info = PrinterInfo(
+            identifier=configuration_file["profile"],
+            backend="pyusb",
+            model=configuration_file["profile"],
+            protocol="ESC/POS",
+            vendor_id=self.vendor_id,
+            product_id=self.device_id,
+            serial_number="",
+            label_type=self.configuration_file["label_type"],
+        )
+
         # Setting up the printing options.
+        # TODO : Expose this to the web interface
         p.set(
-            align="center",
+            align=self.configuration_file["align_text"],
             font="a",
             bold=False,
             underline=0,
@@ -175,7 +205,6 @@ class EscPosPrinter(Printer):
                 return True
 
     def _print_txt(self, msg, signature="", bold=False):
-        self.ready = False
         if not isinstance(msg, str):
             self.app.logger.error(
                 "It is not possible to print a " + str(type(msg)) + ", only strings."
@@ -187,24 +216,25 @@ class EscPosPrinter(Printer):
         clean_signature = str(signature)
 
         # Make checks on the size of the message being printed
-        if len(clean_msg) > 4096:
+        if len(clean_msg) > int(self.configuration_file["text_max_length"]):
             self.app.logger.warning(
                 "Could not print message of this length: " + str(len(clean_msg))
             )
             raise RuntimeError(
                 "Could not print message of this length :"
                 + str(len(clean_msg))
-                + ", needs to be below 4096 caracters long."
+                + ", needs to be below %s caracters long.",
+                str(self.configuration_file["text_max_length"]),
             )
-
-        if len(signature) > 256:
+        if len(signature) > int(self.configuration_file["signature_max_length"]):
             self.app.logger.warning(
                 "Could not print signature of this length: " + str(len(clean_signature))
             )
             raise RuntimeError(
                 "Could not print signature of this length :"
                 + str(len(clean_signature))
-                + ", needs to be below 256 caracters long."
+                + ", needs to be below %s caracters long.",
+                str(self.configuration_file["signature_max_length"]),
             )
 
         # Do the actual printing
@@ -227,17 +257,17 @@ class EscPosPrinter(Printer):
         self.ready = True
 
     def _print_img(self, path, signature="", center=True, process=False):
-        self.ready = False
         clean_signature = str(signature)
 
-        if len(signature) > 256:
+        if len(signature) > int(self.configuration_file["signature_max_length"]):
             self.app.logger.warning(
                 "Could not print signature of this length: " + str(len(clean_signature))
             )
-            raise ValueError(
+            raise RuntimeError(
                 "Could not print signature of this length :"
                 + str(len(clean_signature))
-                + ", needs to be below 256 caracters long."
+                + ", needs to be below %s caracters long.",
+                str(self.configuration_file["signature_max_length"]),
             )
 
         if not os.path.isfile(str(path)):
@@ -272,29 +302,25 @@ class EscPosPrinter(Printer):
         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
+        # finally:
+        #     try:
+        #         os.remove(path)
+        #         os.remove(processed_path)
+        #         self.app.logger.debug("Removed image : " + str(processed_path))
+        #         self.app.logger.debug("Removed image : " + str(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)
-                )
-                raise RuntimeError("Could not close the printer connexion. ") from e
+        try:
+            self.printer.close()
+        except Exception as 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")
         self.ready = True
 
     def _qr(self, content):
-        self.ready = False
         try:
             self.printer.open(self.usb_args)
             self.printer.qr(content, center=True)
@@ -309,7 +335,6 @@ class EscPosPrinter(Printer):
         self.ready = True
 
     def _cut(self):
-        self.ready = False
         try:
             self.printer.open(self.usb_args)
             self.printer.cut()
@@ -322,22 +347,31 @@ class EscPosPrinter(Printer):
         self.app.logger.info("Did a cut")
         self.ready = True
 
-    def _state(self):
-        self.app.logger.debug("Online : %s " , self.printer.is_online())
-        self.app.logger.debug("Has paper : %s " , self._has_paper())
-        self.app.logger.debug("Ready : %s " , self.ready)
-        return self.printer.is_online() and self.ready and self._has_paper()
+    def _state(self) -> bool:
+        has_paper = self._has_paper()
+        is_ready = self.ready
+
+        self.app.logger.debug("Has paper : %s ", has_paper)
+        self.app.logger.debug("Ready : %s ", is_ready)
+
+        return is_ready and has_paper  # and is_online
 
     def print_task(self, task_type, data):
         """Execute actual print based on task type"""
+
         with self._lock:
             self.app.logger.debug("Acquired lock to start print")
 
-            while not self._state():
-                self.app.logger.debug("Waiting for the printer to become ready..")
+            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.app.logger.debug("Checked state to start printing : %s", self._state())
             self.ready = False
             try:
                 self.app.logger.debug("Checking task type")
@@ -370,19 +404,20 @@ 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
-        )
+    def __init__(self, app, vendor_id, device_id, configuration_file):
+
+        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.usb_args["idVendor"] = self.vendor_id
+        self.usb_args["idProduct"] = self.device_id
         self.model_manager = ModelsManager()
 
         # Code taken from https://github.com/5shekel/printit/blob/master/printer_utils.py
 
+        self.app.logger.debug("Creating backend for Brother")
         backend = backend_factory("pyusb")
+
         available_devices = backend["list_available_devices"]()
 
         for printer in available_devices:
@@ -401,14 +436,19 @@ class BrotherPrinter(Printer):
             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:
+                if m.product_id == self.device_id:
                     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}")
+        except ValueError as ve:
+            self.app.logger.warning("Invalid product ID format:" + str(ve))
+            raise ve
+        except Exception as e:
+            self.app.logger.error("Error while finding a model ? : " + str(e))
+            raise e
 
         self.printer_info = PrinterInfo(
             identifier=identifier,
@@ -418,8 +458,11 @@ class BrotherPrinter(Printer):
             vendor_id=vendor_id,
             product_id=self.device_id,
             serial_number=serial_number,
+            label_type=configuration_file[
+                "label_type"
+            ],  # TODO: Implement automatic label type detection : https://framagit.org/stickoeur/diagnostickoeur/-/blob/no-masters/printit.py?ref_type=heads#L239
         )
-
+        self.app.logger.debug("Created a Brother printer !")
         self.ready = True
 
     def _has_paper(self):
@@ -428,163 +471,317 @@ class BrotherPrinter(Printer):
     def _state(self):
         return self.ready
 
-    def _print_img(self, data):
+    def _print_txt(self, msg, signature="", bold=False):
+        """Print text after converting it to an image"""
+        self._print_img(text_to_image(msg))
+        self._print_img(text_to_image(signature))
+
+    def _print_img(
+        self,
+        data,
+        signature="",
+        process=False,
+    ):
         """
         Print a raster image via a Brother QL printer
         """
-        self.ready = False
-        label_type = "102"
-        rotate = 0
-        dither = False
+        self.app.logger.debug("Printing an image %s via Brother printer", data)
+        if process:
+            try:
+                self.app.logger.debug("Proccessing the image")
+                processed_path = _process_image(self, data)
+            except RuntimeError as e:
+                self.app.logger.error(
+                    "Error while processing the image, aborting print : %s", str(e)
+                )
+                raise e
+        else:
+            processed_path = data
+            self.app.logger.warning("Not proccessing the image")
+
         try:
-            # Prepare the image for printing
-            qlr = BrotherQLRaster(self.printer_info["model"])
+            selected_backend = guess_backend(self.printer_info.identifier)
+            BACKEND_CLASS = backend_factory(selected_backend)["backend_class"]
+            BACKEND_STRING_DESCR = self.printer_info.identifier
+        except Exception as e:
+            self.app.logger.error(
+                "Couldn't guess the backend to use from the printer string descriptor"
+            )
+            raise e
+
+        try:
+            im = Image.open(processed_path)
+            qlr = BrotherQLRaster(self.printer_info.model)
+            # create_label(qlr, im, self.printer_info.label_type, cut=False, dither=True)
+
+            # noinspection PyCallingNonCallable
+            # be = BACKEND_CLASS(BACKEND_STRING_DESCR)
+            # be.write(qlr.data)
+            # be.dispose()
+            # del be
 
             instructions = convert(
                 qlr=qlr,
-                images=[data["img"]],
-                label=label_type,
-                rotate=rotate,
-                threshold=70,
-                dither=dither,
-                compress=True,
-                red=False,
+                images=[im],
+                label=self.printer_info.label_type,
+                dither=True,
+                compress=False,
                 dpi_600=False,
-                hq=False,
-                cut=True,
+                cut=False,
             )
 
-            # 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(
+            success = send(
                 instructions=instructions,
-                printer_identifier=self.printer_info["identifier"],
+                printer_identifier=BACKEND_STRING_DESCR,
                 backend_identifier="pyusb",
+                blocking=False,
             )
 
-            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
+            self.app.logger.debug(str(success))
+            self.app.logger.debug("Printed image")
+            time.sleep(2)
 
         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
-
+            if "timeout error" in str(e):
+                self.app.logger.error("USB Timeout Error %s", str(e))
         except Exception as e:
-            error_msg = f"Unexpected error during printing: {str(e)}"
-            self.app.logger.debug(error_msg)
-            raise RuntimeError from e
+            raise e
+        # finally:
+        #     try:
+        #         os.remove(processed_path)
+        #         self.app.logger.debug("Removed image : " + str(processed_path))
+        #     except OSError as e:
+        #         raise e
+
+        if signature != "":
+            self.app.logger.debug("Printing signature for the image %s", data)
+            try:
+                signature_file = text_to_image(signature)
+                im = Image.open(signature_file)
+                qlr = BrotherQLRaster(self.printer_info.model)
+                # create_label(
+                #     qlr, im, self.printer_info.label_type, cut=False, dither=True
+                # )
+
+                # noinspection PyCallingNonCallable
+                # be = BACKEND_CLASS(BACKEND_STRING_DESCR)
+                # be.write(qlr.data)
+                # be.dispose()
+                # del be
+                instructions = convert(
+                    qlr=qlr,
+                    images=[signature_file],
+                    label=self.printer_info.label_type,
+                    dither=True,
+                    compress=False,
+                    dpi_600=False,
+                    cut=False,
+                )
+
+                success = send(
+                    instructions=instructions,
+                    printer_identifier=BACKEND_STRING_DESCR,
+                    backend_identifier="pyusb",
+                    blocking=False,
+                )
+                self.app.logger.debug(str(success))
+                self.app.logger.debug("Printed signature")
+                time.sleep(2)
+
+            except Exception as e:
+                raise e
+            finally:
+                try:
+                    # os.remove(signature_file)
+                    self.app.logger.debug("Removed image : " + str(signature_file))
+                except OSError as e:
+                    raise e
+
+    def _cut(self):
+        self.app.logger.debug("Cutting via Brother Printer")
+        try:
+            selected_backend = guess_backend(self.printer_info.identifier)
+            BACKEND_CLASS = backend_factory(selected_backend)["backend_class"]
+            BACKEND_STRING_DESCR = self.printer_info.identifier
+        except Exception as e:
+            self.app.logger.error(
+                "Couldn't guess the backend to use from the printer string descriptor"
+            )
+            raise e
+
+        data = "src/static/images/1x1.png"
+        try:
+            im = Image.open(data)
+            qlr = BrotherQLRaster(self.printer_info.model)
+            create_label(
+                qlr, im, self.printer_info.label_type, threshold=70, cut=True, rotate=0
+            )
+
+            # noinspection PyCallingNonCallable
+            be = BACKEND_CLASS(BACKEND_STRING_DESCR)
+            be.write(qlr.data)
+            be.dispose()
+            del be
+            self.app.logger.debug("Finished cutting via Brother Printer")
+        except Exception as e:
+            raise e
+
+    def _qr(self):
+        # This should not be reached by a print on a BrotherQL
+        raise NotImplementedError()
 
     def print_task(self, task_type, data):
         """Execute actual print based on task type"""
         with self._lock:
-            if self._state:
-                self._state = False
+            self.app.logger.debug("Acquired lock to start print")
+            i_m_ready = self._state()
+            while not i_m_ready:
+                self.app.logger.debug(
+                    "Waiting for the printer to become ready, current state %s ",
+                    str(i_m_ready),
+                )
+                i_m_ready = self._state()
+                time.sleep(0.3)
+
+            self.app.logger.debug("Checked state to start printing : %s", self._state())
+            self.ready = False
+
+            try:
                 match (task_type.value):
                     case "image":
-                        self._print_img(data["img"])
-                        self._state = True
+                        self._print_img(data["img"], signature=data["sign"])
+                        self.ready = True
                     case "cut":
-                        # The cut happens by default on Brother QL printers.
-                        self._state = True
+                        self._cut()
+                        self.ready = True
+                    case "text":
+                        self._print_txt(data["txt"], signature=data["sign"])
+                        self.ready = 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")
+            except Exception as e:
+                raise e
 
 
-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
-    max_width = 575
-    max_height = 1000
+def _process_image(self, path, max_width=575, max_height=1000):
+    brightness_threshold = 100
+    contrast_factor = 2.0
 
-    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")
-            original_img = original_img.convert("RGB")
-
-        # Resize while maintaining aspect ratio
-        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)
-        # self.app.logger.debug("Dithered the image")
-
-        # Compute brightness of original image (grayscale average)
-        grayscale = original_img.convert("L")
-        avg_brightness = np.array(grayscale).mean()
-        self.app.logger.debug(
-            "Average brightness of the image : " + str(avg_brightness)
+    with Image.open(path) as img:
+        # Preserve aspect ratio while fitting within the maximum dimensions.
+        img.thumbnail(
+            (max_width, max_height),
+            Image.Resampling.LANCZOS,
         )
 
-        # Dynamically compute brightness factor if too dark
+        grayscale = img.convert("L")
+
+        avg_brightness = np.asarray(grayscale, dtype=np.float32).mean()
+        self.app.logger.debug("Average brightness of the image: %.2f", avg_brightness)
+
         if avg_brightness < brightness_threshold:
-            brightness_factor = (
-                1 + (brightness_threshold - avg_brightness) / brightness_threshold
-            )
             brightness_factor = min(
-                max(brightness_factor, 1.1), 2.5
-            )  # Clamp between 1.1 and 2.5
-            self.app.logger.debug(
-                f"Image too dark, increasing brightness by a factor of {brightness_factor:.2f}"
+                max(
+                    1 + (brightness_threshold - avg_brightness) / brightness_threshold,
+                    1.1,
+                ),
+                2.5,
             )
-            enhancer = ImageEnhance.Brightness(grayscale)
-            grayscale = enhancer.enhance(brightness_factor)
+            self.app.logger.debug(
+                "Image is dark; increasing brightness by %.2f",
+                brightness_factor,
+            )
+            grayscale = ImageEnhance.Brightness(grayscale).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)
+        contrast = float(np.asarray(grayscale, dtype=np.float32).std())
+        self.app.logger.debug("Grayscale standard deviation: %.2f", contrast)
+
+        # Save the enhanced result.
+        processed = ImageEnhance.Contrast(grayscale).enhance(contrast_factor)
 
-        # 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.")
+        processed.save(jpeg_path, format="JPEG", quality=95, optimize=True)
 
+    self.app.logger.debug("Processed and saved image: %s", jpeg_path)
     return jpeg_path
+
+
+def text_to_image(
+    text,
+    width=400,
+    font_path="src/static/font/MonospaceTypewriter.ttf",
+    font_size=24,
+    padding=10,
+    line_spacing=4,
+    background="white",
+    fill="black",
+    output_dir="src/static/uploads",
+):
+    """Converts a string to an image"""
+
+    font = ImageFont.truetype(font_path, font_size)
+
+    # Measure and wrap text by pixel width.
+    measure_img = Image.new("RGB", (1, 1))
+    draw = ImageDraw.Draw(measure_img)
+    max_text_width = width - 2 * padding
+    if max_text_width <= 0:
+        raise ValueError("width must be greater than twice the padding")
+
+    lines = []
+    for paragraph in text.splitlines() or [""]:
+        words = paragraph.split()
+        if not words:
+            lines.append("")
+            continue
+
+        line = ""
+        for word in words:
+            candidate = f"{line} {word}".strip()
+            if draw.textbbox((0, 0), candidate, font=font)[2] <= max_text_width:
+                line = candidate
+                continue
+
+            if line:
+                lines.append(line)
+
+            # Split a word that is wider than the available width.
+            line = ""
+            piece = ""
+            for char in word:
+                if (
+                    piece
+                    and draw.textbbox((0, 0), piece + char, font=font)[2]
+                    > max_text_width
+                ):
+                    lines.append(piece)
+                    piece = char
+                else:
+                    piece += char
+            line = piece
+
+        if line:
+            lines.append(line)
+
+    if not lines:
+        lines = [""]
+
+    ascent, descent = font.getmetrics()
+    line_height = ascent + descent
+    height = 2 * padding + len(lines) * line_height + (len(lines) - 1) * line_spacing
+
+    image = Image.new("RGB", (width, height), background)
+    draw = ImageDraw.Draw(image)
+
+    y = padding
+    for line in lines:
+        draw.text((padding, y), line, font=font, fill=fill)
+        y += line_height + line_spacing
+
+    upload_dir = Path(output_dir)
+    upload_dir.mkdir(parents=True, exist_ok=True)
+
+    image_path = upload_dir / f"{uuid4().hex}.png"
+    image.save(image_path, format="PNG")
+
+    return str(image_path)

src/printers.py

@@ -6,6 +6,8 @@ the methods expected from printers.
 """
 
 from collections.abc import Set
+import time
+import threading
 import usb.core
 import usb.util
 from printer import Printer, EscPosPrinter, BrotherPrinter
@@ -16,12 +18,14 @@ class Printers:
     Finds and creates a set of Printer that can be used by the Workers to print.
     """
 
-    def __init__(self, app):
+    def __init__(self, app, configuration_file):
         """
-        Discover printers connected to the computer and return a Collection of Printer()
+        Discovers printers connected to the computer
         """
         self.app = app
-        self.printers = self._discover_printers()
+        self.configuration_file = configuration_file
+        self.printers = list(self._discover_printers())
+        self._lock = threading.Lock()
 
     def _discover_printers(self) -> Set[Printer]:
         """
@@ -39,16 +43,33 @@ class Printers:
 
         printers = set()
 
-        # Find all connected USB devices
-        devices = usb.core.find(find_all=True, custom_match=_FindClass(7))
-        if not devices:
+        waiting_time_for_printer = self.configuration_file["printer"]["waiting_time"]
+
+        while True:
+            # Find all connected USB devices
+            # the find() fnc returns a generator
+            devices = list(
+                usb.core.find(find_all=True, custom_match=_FindClass(self.app, 7))
+            )
+
+            if devices:
+                break
+
+            if waiting_time_for_printer <= 0:
+                self.app.logger.warning(
+                    "No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
+                )
+                raise RuntimeError(
+                    "No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
+                )
+
             self.app.logger.warning(
-                "No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
-            )
-            raise RuntimeError(
-                "No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
+                "No printers found; waiting for a printer to connect..."
             )
 
+            time.sleep(1)
+            waiting_time_for_printer -= 1
+
         for dev in devices:
             # Attempt to get the manufacturer and product strings
             try:
@@ -60,6 +81,7 @@ class Printers:
                 product = usb.util.get_string(dev, dev.iProduct)
             except Exception:
                 product = "Unknown"
+
             self.app.logger.debug(
                 "Looking at %s %s (%s:%s)",
                 manufacturer,
@@ -68,73 +90,98 @@ class Printers:
                 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
+            match (manufacturer):
+                case "EPSON":
+                    try:
+                        # We create a new EscPosPrinter()
+                        self.app.logger.debug("Trying to create a new EPSON printer")
+                        prid = dev.idProduct
+                        vendir = dev.idVendor
+                        escpos_printer = EscPosPrinter(
+                            self.app,
+                            vendir,
+                            prid,
+                            self.configuration_file["printer"]["epson"],
+                        )
+                    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)
+                case "Brother":
+                    try:
+                        # We create a new BrotherPrinter()
+                        self.app.logger.debug("Trying to create a new BROTHER printer")
+
+                        brother_printer = BrotherPrinter(
+                            self.app,
+                            dev.idVendor,
+                            dev.idProduct,
+                            self.configuration_file["printer"]["brother"],
+                        )
+                    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)
+                case _:
+                    self.app.logger.debug(
+                        "Found a printer of unknown manufacturer : %s", manufacturer
                     )
-                except Exception as e:
-                    raise e
+            # end match
 
-                # 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))
+        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.")
+            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 acquire_available_printer(self, start_index=0):
+        """Atomically acquire and return an available printer, or None."""
+        printers = self.printers
+        if not printers:
+            self.app.logger.error("No printers available for acquisition")
+            return None
 
-    # def get_printer(self, printer_type):
-    #     """
-    #     Return a specific printer
+        with self._lock:
+            # So don't always look at the first printer available if need be
+            # this allows for a round-robin lookup
+            for offset in range(len(printers)):
+                index = (start_index + offset) % len(printers)
+                printer = printers[index]
 
-    #     printer_type -- a printer type
-    #     """
-    #     return NotImplementedError()
+                if not printer.used:
+                    printer.use_printer()
+                    return index, printer
+
+        return None
+
+    def printers_info(self):
+        """Return the current list of printers"""
+        printers_info_list = []
+        for i in self.printers:
+            printers_info_list.append(
+                [
+                    i.printer_type.name,
+                    "Used: " + str(i.used),
+                    "Ready: " + str(i.ready),
+                    i.printer_info,
+                ]
+            )
+
+        return printers_info_list
 
 
 class _FindClass:
@@ -142,19 +189,27 @@ class _FindClass:
     Use by usb.core to modify the way USB devices are found
     Taken from pyUSB documentation on Github
     """
-    def __init__(self, class_):
+
+    def __init__(self, app, class_):
         self._class = class_
+        self.app = app
 
     def __call__(self, device):
         # first, let's check the device
+        self.app.logger.debug(
+            "Found a USB device of class %s", str(device.bDeviceClass)
+        )
         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:
+        for configuration in device:
+            # Every device can have one or more configuration,
+            interface = usb.util.find_descriptor(
+                configuration, bInterfaceClass=self._class
+            )
+            # each config can have several interfaces
+            if interface is not None:
                 return True
 
         return False

src/raspberry.py

@@ -27,12 +27,7 @@ class Raspberry:
     # dede
     """
 
-    def __init__(
-        self,
-        print_queue,
-        app,
-        configuration_file
-    ):
+    def __init__(self, print_queue, app, configuration_file):
         self.print_queue = print_queue
         self.app = app
         self.configuration_file = configuration_file
@@ -68,7 +63,9 @@ class Raspberry:
                     self.app.logger.warning(
                         "Couldn't get sufficient hardware information from /proc/cpuinfo"
                     )
-                    self.app.logger.error("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`.")
@@ -95,13 +92,16 @@ class Raspberry:
         self.indicator_countdown(iters=3)
         self.button = Button(
             self.configuration_file["rpi"]["button_gpio_port_number"],
-            pull_up=True, bounce_time=0.1
+            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.configuration_file["rpi"]["flash_gpio_port_number"]
+        )
         self.flash_toggle()
         self.app.logger.debug("Activated flash")
 
@@ -215,9 +215,9 @@ class Raspberry:
                 y = image.height - logo.height - margin
             else:
                 raise ValueError(
-                    "Invalid position." +
-                    "Choose from 'bottom_right', 'top_left', " +
-                    " 'top_right', or 'bottom_left'."
+                    "Invalid position."
+                    + "Choose from 'bottom_right', 'top_left', "
+                    + " 'top_right', or 'bottom_left'."
                 )
 
             # Composite the logo onto the image

src/templates/api.html

@@ -5,6 +5,6 @@
     <title>LittlePrynter API</title>
   </head>
   <body>
-    "Welcome to the printing software's API. Please see the index page to get started.
+    Welcome to the printing software's API. Please see the index page to get started.
   </body>
 </html>

src/templates/base.html

@@ -53,7 +53,7 @@
         <hr>
 
         <footer class="row">
-        <p class=" text-center">Little Prynter is built by <a href="https://n07070.xyz/about-me/">n07070</a> because it's fun :) | <a href="https://git.n07070.xyz/n07070/littleprynter">Source code - AGPLv3</a></p>
+        <p class=" text-center">Little Prynter is built by <a href="https://n07070.xyz/about-me/">n07070</a> because it's fun :) | <a href="/api">API interface</a> | <a href="https://git.n07070.xyz/n07070/littleprynter">Source code - AGPLv3</a></p>
         </footer>
     </body>
             <script src="https://cdn.socket.io/4.0.0/socket.io.min.js"></script>

src/templates/footer.html

@@ -1,5 +0,0 @@
-<hr>
-<footer>Little Prynter is built by <a href="https://n07070.xyz/about-me/">n07070</a> because it's fun :) | <a href="https://git.n07070.xyz/n07070/littleprynter/src/branch/epson-tm-t20iii">Source code - AGPLv3</a></footer>
-</div>
-</body>
-</html>

src/templates/index.html

@@ -6,10 +6,23 @@
       <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>
+          <textarea class="form-control" type="text" name="txt" placeholder="4096 caractères maximum " maxlength="4096"></textarea>
           <br>
-          <input class="form-control" type="text" name="signature" placeholder="Signature or pseudo" maxlength="200"><br>
-          <input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
+          <input class="form-control" type="text" name="signature" placeholder="Signature ou pseudo" maxlength="200"><br>
+          <div class="input-group">
+            <span class="input-group-text">Combien ?</span>
+            <input type="number" aria-label="how-many" class="form-control" name="many" max="20" min="1" value="1">
+            <input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
+          </div>
+          <br>
+          <div class="form-check">
+            <input class="form-check-input" type="checkbox" value="True" id="flexCheckDefault" name="cut" checked>
+            <label class="form-check-label" for="flexCheckDefault">
+              Couper après impression ?
+            </label>
+          </div>
+
+
         </form>
       </div>
     </div>
@@ -21,8 +34,26 @@
       <div class="card-body">
         <form enctype="multipart/form-data" class="form-group" action="/web/print/img" method="post">
           <input class="form-control" type="file" name="img" 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">
+          <input class="form-control" type="text" name="signature" placeholder="Signature ou pseudo" maxlength="200"><br>
+          <div class="input-group">
+            <span class="input-group-text">Combien ?</span>
+            <input type="number" aria-label="how-many" class="form-control" name="many" max="20" min="1" value="1">
+            <input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
+          </div>
+          <br>
+          <div class="form-check .form-check-inline">
+            <input class="form-check-input" type="checkbox" value="True" id="flexCheckDefault" name="cut" checked>
+            <label class="form-check-label" for="flexCheckDefault">
+              Couper après impression ?
+            </label>
+          </div>
+          <div class="form-check .form-check-inline">
+            <input class="form-check-input" type="checkbox" value="True" id="flexCheckDefault" name="process">
+            <label class="form-check-label" for="flexCheckDefault">
+              Traiter l'image ?
+            </label>
+          </div>
+
         </form>
       </div>
     </div>

src/templates/password.html

@@ -1,30 +0,0 @@
-<!DOCTYPE html>
-<html>
-    <head>
-        <meta charset="utf-8">
-        <title>Log in</title>
-        <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
-        <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.1.3/css/bootstrap.min.css" integrity="sha384-MCw98/SFnGE8fJT3GXwEOngsV7Zt27NXFoaoApmYm81iuXoPkFOJwJ8ERdknLPMO" crossorigin="anonymous">
-        <link rel="stylesheet" href="{{ url_for('static',filename='css/style.css') }}">
-    </head>
-    <body class="container">
-        <div class="col-md-6 offset-md-3">
-            <h1>Login</h1>
-            {% with messages = get_flashed_messages(with_categories=true) %}
-              {% if messages %}
-                {% for category, message in messages %}
-                  <div class="alert alert-{{ category }}" role="alert">{{ message }}</div>
-                {% endfor %}
-              {% endif %}
-            {% endwith %}
-
-            <form action="/login" method="post">
-                <div class="form-group">
-                <input type="text" name="username" class="form-control" placeholder="Pseudo">
-                <input type="password" name="password" class="form-control" placeholder="Password">
-                <button type="submit" class="btn btn-primary">Submit</button>
-                </div>
-            </form>
-        </div>
-    </body>
-</html>

src/web.py

@@ -16,7 +16,7 @@ class Web:
         self.print_queue = print_queue
         self.app = app
 
-    def print_sms(self, texte, sign: str) -> bool:
+    def print_sms(self, texte, sign: str, do_cut: bool, many: int = 1) -> bool:
         """
         Get text and a signature, prints the text and cuts after that.
         """
@@ -28,15 +28,22 @@ class Web:
             # !! 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())
+            for i in range(many):
+                sms = self.print_queue.enqueue(TextTask(content=texte, signature=sign))
+                self.app.logger.info("Added a SMS task at position %s", sms)
+                if do_cut:
+                    cut = self.print_queue.enqueue(CutTask())
+                    self.app.logger.info("Added a CUT task at position %s", 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)
+
         return True
 
-    def print_image(self, image, sign: str) -> bool:
+    def print_image(
+        self, image, sign: str, do_cut: bool, process_image: bool, many: int = 1
+    ) -> bool:
         """
         Get an image and a signature, prints the image and cuts after that.
         """
@@ -49,23 +56,25 @@ class Web:
         if file_uploaded:
             self.app.logger.debug("File has been uploaded, printing...")
             try:
-                img = self.print_queue.enqueue(
-                    ImageTask(
-                        os.path.join(
-                            self.app.config["UPLOAD_FOLDER"],
-                            secure_filename(image.filename),
-                        ),
-                        signature=sign,
-                        process=True,
+                for i in range(many):
+                    img = self.print_queue.enqueue(
+                        ImageTask(
+                            os.path.join(
+                                self.app.config["UPLOAD_FOLDER"],
+                                secure_filename(image.filename),
+                            ),
+                            signature=sign,
+                            process=process_image,
+                        )
                     )
-                )
+                    self.app.logger.info("Added a IMG task at position %s", img)
 
-                cut = self.print_queue.enqueue(CutTask())
+                    if do_cut:
+                        cut = self.print_queue.enqueue(CutTask())
+                        self.app.logger.info("Added a CUT task at position %s", 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)
-
         return True
 
     # def login(self, username: str, password: str) -> bool:
@@ -100,7 +109,9 @@ class Web:
                     image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
                 except OSError as e:
                     self.app.logger.error("Could not save file %s", e)
-                    raise RuntimeError("An OS error occured while uploading this file : " + str(e)) from e
+                    raise RuntimeError(
+                        "An OS error occured while uploading this file : " + str(e)
+                    ) from e
 
                 self.app.logger.debug(
                     "File saved to "
@@ -108,9 +119,7 @@ class Web:
                 )
                 return True
 
-        self.app.logger.error(
-            "Could not save file because the filename is forbidden"
-        )
+        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):

src/worker.py

@@ -5,7 +5,7 @@ This is the main printing thread. A worker thread consums Tasks from
 
 import threading
 import time
-from printers import Printers
+import traceback
 
 
 class PrintWorker(threading.Thread):
@@ -13,12 +13,14 @@ 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.
+    and on each print, if the choosen Printer becomes unavailable, it
+    will 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.
+    tasks. TODO : Change that
     """
-    def __init__(self, app, print_queue):
+
+    def __init__(self, app, print_queue, printers_pool, printer_index=0):
         super().__init__(daemon=True)
         self.app = app
         self.print_queue = print_queue
@@ -26,21 +28,40 @@ class PrintWorker(threading.Thread):
         self._lock = threading.Lock()
         self.running = True
         self.state = "idle"  # idle, printing, dead, drinking-a-beer
+
+        # All workers share the same pool of printers.
+        self.printers = printers_pool
+        self.printers_obj = self.printers.printers
+        # If we create several workers, we can assign an index
+        # so that we lookup the nth printer for the nth worker
+        self._printer_index = printer_index
+
+        self.app.logger.debug("Worker initialized.")
+        self.app.logger.debug("Printers connected:")
+        for printer in self.printers_obj:
+            self.app.logger.debug("Printer type: %s", printer.printer_type)
+            self.app.logger.debug("Printer state: %s", printer.ready)
+
         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 _acquire_available_printer(self):
+        result = self.printers.acquire_available_printer(
+            start_index=self._printer_index
+        )
+        if result is None:
+            return None
+
+        index, printer = result
+        # In case we need to lookup another printer, we don't
+        # look at the same printer in the printer list
+        self._printer_index = (index + 1) % len(self.printers_obj)
+        return printer
 
     def run(self):
         """Background thread that processes queue items"""
         self.app.logger.debug("Worker %s started working.", threading.get_ident())
-        self.app.logger.debug("Current threads : %s" , threading.active_count())
-        self.app.logger.debug("Threads actives : %s " , threading.enumerate())
+        self.app.logger.debug("Current threads : %s", threading.active_count())
+        self.app.logger.debug("Threads actives : %s ", threading.enumerate())
 
         while True:
 
@@ -53,19 +74,29 @@ class PrintWorker(threading.Thread):
             # 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:
+
+            # If we do not have a printer, we try to find one
+
+            while self.running and self.printer is None:
+                self.app.logger.debug(
+                    "Trying to find a available printer for the worker."
+                )
+                self.printer = self._acquire_available_printer()
+                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",
+                )
+
+                if self.printer is None:
+                    time.sleep(1)
+
+            self.printer.use_printer()
+
+            # If the printer is not ready, we wait for it to become ready.
+            while not self.printer.ready:
+                self.app.debug("Waiting for our printer to become 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...")
@@ -80,6 +111,7 @@ class PrintWorker(threading.Thread):
                 except Exception as e:
                     self.app.logger.error("Could not get a new task ! %s ", str(e))
                     self.state = "idle"
+                    self.printer.free_printer()
                     raise RuntimeError(
                         "We could not get a new task because " + str(e)
                     ) from e
@@ -87,7 +119,9 @@ class PrintWorker(threading.Thread):
                 if task:
                     try:
                         self.app.logger.info("Got a new task")
-                        self.app.logger.debug("Got task %s", task.task_id)
+                        self.app.logger.debug(
+                            "Got task %s %s", task.task_type, task.task_id
+                        )
                         task.status = "processing"
 
                         print_data = task.get_print_data()
@@ -97,12 +131,13 @@ class PrintWorker(threading.Thread):
                         except RuntimeError as e:
                             self.state = "idle"
                             self.app.logger.error("Could not print : %s", str(e))
-                            raise e
+                            self.app.logger.error(traceback.format_exc())
+                            self.printer.free_printer()
 
                         task.status = "completed"
                         self.print_queue.mark_completed(task.task_id, "completed")
                         self.app.logger.debug(
-                            "Finished printing task %s " , task.task_id
+                            "Finished printing task %s ", task.task_id
                         )
                         self.state = "idle"
 
@@ -111,7 +146,7 @@ class PrintWorker(threading.Thread):
                         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)
+                            "Could not print task %s because %s ", task.task_id, str(e)
                         )
 
                 else:
@@ -125,6 +160,10 @@ class PrintWorker(threading.Thread):
         """
         self.app.logger.debug("Giving the worker a break")
         self.state = "drinking-a-beer"
+        # Free the printer during the worker break
+        # so it can be used by someone else
+        self.printer.free_printer()
+        self.printer = None
         self.running = False
 
     def start_worker(self):