Apply formatting

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.
 

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"
+python-versions = ">=3.13"
-content-hash = "18e1a5c8b085aa639b665279856f9e3c51a734a6a8d1d4dd135ca7e4f67e86ae"
+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 seconde is booth mode, where a Raspberry Pi is connected to
-The web interface exists but may not be used, as the press of the button with take
+a thermal printer, a button and a flash.
- a picture and activate the flash while simply informing the web page.
+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(
+rpi = Raspberry(print_queue, app, configuration_file)
-    print_queue,
-    app,
-    configuration_file
-)
 RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
 
 
 #  Web & API management
 web = Web(app, print_queue)
 
-# Start worker thread
+# Create a pool of printers
-# When created, the worker will try to find printers connected to the system
 try:
-    worker = PrintWorker(app, print_queue)
+    # comment: Find all printers connected to the system
+    printers_pool = Printers(app)
+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))
@@ -125,6 +133,7 @@ limiter = Limiter(
 
 app.logger.info("🖶 Welcome to LittlePrynter !")
 
+
 # General routes
 @app.route("/")
 @limiter.limit("1/second", override_defaults=False)
@@ -164,9 +173,13 @@ 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)
+        web.print_sms(txt, sign, do_cut)
     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")
@@ -210,9 +223,19 @@ 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:
         app.logger.debug("Sending the image to the printer.")
-        web.print_image(file, sign)
+        web.print_image(file, sign, do_cut, process_image)
     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")
@@ -257,9 +280,15 @@ 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:
         # comment: We try to print the SMS
-        web.print_sms(txt, sign)
+        web.print_sms(txt, sign, do_cut)
     except RuntimeError as e:
         return str(e), 500
     # end try
@@ -292,9 +321,19 @@ 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:
         app.logger.debug("Sending the image to the printer.")
-        web.print_image(file, sign)
+        web.print_image(file, sign, do_cut, process_image)
     except RuntimeError as e:
         return str(e), 500
 
@@ -350,6 +389,12 @@ def api_worker_stop():
     return jsonify(worker.current_state())
 
 
+@app.route("/api/printers")
+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

@@ -11,16 +11,19 @@ from enum import Enum
 import uuid
 import threading
 import usb.core
-from PIL import Image, ImageEnhance
+from pathlib import Path
+from uuid import uuid4
+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
@@ -97,6 +101,7 @@ class Printer(ABC):
         """Marks the printer as being not used, it's available."""
         self.used = False
 
+
 class EscPosPrinter(Printer):
     """
     Create a new ESC/POS based printer.
@@ -124,8 +129,10 @@ class EscPosPrinter(Printer):
                 "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():
@@ -183,7 +190,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."
@@ -195,6 +201,7 @@ class EscPosPrinter(Printer):
         clean_signature = str(signature)
 
         # Make checks on the size of the message being printed
+        # TODO: make the max length configurable
         if len(clean_msg) > 4096:
             self.app.logger.warning(
                 "Could not print message of this length: " + str(len(clean_msg))
@@ -204,7 +211,7 @@ class EscPosPrinter(Printer):
                 + str(len(clean_msg))
                 + ", needs to be below 4096 caracters long."
             )
-
+        # TODO: make the max length configurable
         if len(signature) > 256:
             self.app.logger.warning(
                 "Could not print signature of this length: " + str(len(clean_signature))
@@ -235,9 +242,9 @@ class EscPosPrinter(Printer):
         self.ready = True
 
     def _print_img(self, path, signature="", center=True, process=False):
-        self.ready = False
         clean_signature = str(signature)
 
+        # TODO: make the max length configurable
         if len(signature) > 256:
             self.app.logger.warning(
                 "Could not print signature of this length: " + str(len(clean_signature))
@@ -302,7 +309,6 @@ class EscPosPrinter(Printer):
         self.ready = True
 
     def _qr(self, content):
-        self.ready = False
         try:
             self.printer.open(self.usb_args)
             self.printer.qr(content, center=True)
@@ -317,7 +323,6 @@ class EscPosPrinter(Printer):
         self.ready = True
 
     def _cut(self):
-        self.ready = False
         try:
             self.printer.open(self.usb_args)
             self.printer.cut()
@@ -342,13 +347,15 @@ class EscPosPrinter(Printer):
     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")
 
             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))
+                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)
 
@@ -386,9 +393,8 @@ class BrotherPrinter(Printer):
     """
 
     def __init__(self, app, vendor_id, device_id):
-        super().__init__(
+
-            app, vendor_id, device_id, printer_type=PrinterType.BROTHER
+        super().__init__(app, vendor_id, device_id, printer_type=PrinterType.BROTHER)
-        )
         self.printer = None
         self.usb_args = {}
         self.usb_args["idVendor"] = self.vendor_id
@@ -440,6 +446,7 @@ class BrotherPrinter(Printer):
             vendor_id=vendor_id,
             product_id=self.device_id,
             serial_number=serial_number,
+            label_type="62",
         )
         self.app.logger.debug("Created a Brother printer !")
         self.ready = True
@@ -450,165 +457,249 @@ 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
         """
-        label_type = "102"
+        if process:
-        rotate = 0
+            try:
-        dither = False
+                self.app.logger.debug("Proccessing the image")
-
+                processed_path = _process_image(self, data)
-        data = _process_image(self, path=data, max_width=720, max_height=720)
+            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:
+            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:
-            # Prepare the image for printing
+            im = Image.open(processed_path)
-            qlr = BrotherQLRaster(self.printer_info["model"])
+            qlr = BrotherQLRaster(self.printer_info.model)
-
+            create_label(qlr, im, self.printer_info.label_type, cut=False, dither=True)
-            instructions = convert(
-                qlr=qlr,
-                images=[data],
-                label=label_type,
-                rotate=rotate,
-                threshold=70,
-                dither=dither,
-                compress=True,
-                red=False,
-                dpi_600=False,
-                hq=False,
-                cut=True,
-            )
-
-            # Debug logging
-            if os.getenv("FLASK_DEBUG"):
-                self.app.logger.debug(f"""
-                Print parameters:
-                - Label type: {label_type}
-                - Rotate: {rotate}
-                - Dither: {dither}
-                - Model: {self.printer_info['model']}
-                - Backend: {self.printer_info['backend']}
-                - Identifier: {self.printer_info['identifier']}
-                """)
-
-            # Try to print using Python API
-            # send() =  status = {
-            #     'instructions_sent': True, # The instructions were sent to the printer.
-            #     'outcome': 'unknown', # String description of the outcome of the sending operation like: 'unknown', 'sent', 'printed', 'error'
-            #     'printer_state': None, # If the selected backend supports reading back the printer state, this key will contain it.
-            #     'did_print': False, # If True, a print was produced. It defaults to False if the outcome is uncertain (due to a backend without read-back capability).
-            #     'ready_for_next_job': False, # If True, the printer is ready to receive the next instructions. It defaults to False if the state is unknown.
-            # }
-            status = send(
-                instructions=instructions,
-                printer_identifier=self.printer_info["identifier"],
-                backend_identifier="pyusb",
-            )
-            self.app.logger.debug("Status: " + str(status))
-
-            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"])
-            #     self.app.logger.debug("Printer is borky : " + str(self.ready))
-            # else:
-            #     self.ready = True
-
-        except usb.core.USBError as e:
-            # Treat timeout errors as successful since they often occur after print completion
-            if e.errno == 110:  # Operation timed out
-                self.app.logger.debug(
-                    "USB timeout occurred - this is normal and the print likely completed"
-                )
-                self.app.logger.debug("Print completed (timeout is normal)")
-                self.ready = True
-
-            self.app.logger.debug(f"USBError encountered: {e}")
-            raise RuntimeError from e
 
+            # noinspection PyCallingNonCallable
+            be = BACKEND_CLASS(BACKEND_STRING_DESCR)
+            be.write(qlr.data)
+            be.dispose()
+            del be
         except Exception as e:
-            self.app.logger.debug(f"Unexpected error during printing: {str(e)}")
+            raise e
-            raise RuntimeError from e
+
+        if signature != "":
+            try:
+                im = Image.open(text_to_image(signature))
+                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
+            except Exception as e:
+                raise e
+
+    def _cut(self):
+        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
+        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.ready:
+            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._print_img(data["img"], signature=data["sign"])
+                        self.ready = True
                     case "cut":
-                            # The cut happens by default on Brother QL printers.
+                        self._cut()
-                            pass
+                        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")
             except Exception as e:
                 raise e
-                # end try
-            else:
-                raise RuntimeError("The printer is not ready to print yet !")
-
-        # These values are by default for now
-
-        # raise NotImplementedError("This printer type is not implemented yet")
 
 
 def _process_image(self, path, max_width=575, max_height=1000):
-    brightness_factor = 1.5  # Used only if image is too dark
+    brightness_threshold = 100
-    brightness_threshold = 100  # Brightness threshold (0–255)
+    contrast_factor = 2.0
-    contrast_factor = 2  # Less than 1.0 = lower contrast
 
-    with Image.open(path) as original_img:
+    with Image.open(path) as img:
-        # Convert to RGB if needed (JPEG doesn't support alpha)
+        # Preserve aspect ratio while fitting within the maximum dimensions.
-        if original_img.mode in ("RGBA", "P"):
+        img.thumbnail(
-            self.app.logger.debug("Converting the image from RGBA to RGBA")
+            (max_width, max_height),
-            original_img = original_img.convert("RGB")
+            Image.Resampling.LANCZOS,
-
-        # Resize while maintaining aspect ratio
-        original_img = original_img.resize((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)
         )
 
-        # 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
+                max(
-            )  # Clamp between 1.1 and 2.5
+                    1 + (brightness_threshold - avg_brightness) / brightness_threshold,
-            self.app.logger.debug(
+                    1.1,
-                f"Image too dark, increasing brightness by a factor of {brightness_factor:.2f}"
+                ),
+                2.5,
             )
-            enhancer = ImageEnhance.Brightness(grayscale)
+            self.app.logger.debug(
-            grayscale = enhancer.enhance(brightness_factor)
+                "Image is dark; increasing brightness by %.2f",
+                brightness_factor,
+            )
+            grayscale = ImageEnhance.Brightness(grayscale).enhance(brightness_factor)
 
-        # Computer current contrast of grayscale image
+        contrast = float(np.asarray(grayscale, dtype=np.float32).std())
-        contrast = np.clip(np.std(np.array(grayscale)), 0, 255)
+        self.app.logger.debug("Grayscale standard deviation: %.2f", contrast)
-        self.app.logger.debug("Standard deviation of the contrast : %s", contrast)
+
-        # # Enhance contrast
+        # Save the enhanced result.
-        contrast_enhancer = ImageEnhance.Contrast(grayscale)
+        processed = ImageEnhance.Contrast(grayscale).enhance(contrast_factor)
-        original_img = contrast_enhancer.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)
+        processed.save(jpeg_path, format="JPEG", quality=95, optimize=True)
-        self.app.logger.debug("Processed and saved image.")
 
+    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

@@ -7,6 +7,7 @@ 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
@@ -19,10 +20,11 @@ class Printers:
 
     def __init__(self, app):
         """
-        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.printers = list(self._discover_printers())
+        self._lock = threading.Lock()
 
     def _discover_printers(self) -> Set[Printer]:
         """
@@ -40,19 +42,20 @@ class Printers:
 
         printers = set()
 
+        # TODO : this could go in the config ?
+        waiting_time_for_printer = 5
+
+        while True:
             # Find all connected USB devices
-        devices = usb.core.find(find_all=True, custom_match=_FindClass(7))
+            # the find() fnc returns a generator
-        waiting_time_for_printer = 10
+            devices = list(
-        while not list(devices):
+                usb.core.find(find_all=True, custom_match=_FindClass(self.app, 7))
-            self.app.logger.warning(str(waiting_time_for_printer) + " Found no printers, waiting for a printer to connect to the system...")
+            )
-            time.sleep(1)
+
-            waiting_time_for_printer = waiting_time_for_printer - 1
+            if devices:
-            devices = usb.core.find(find_all=True, custom_match=_FindClass(7))
-            if waiting_time_for_printer == 0:
                 break
 
-
+            if waiting_time_for_printer <= 0:
-        if not list(devices):
                 self.app.logger.warning(
                     "No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
                 )
@@ -60,6 +63,13 @@ class Printers:
                     "No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
                 )
 
+            self.app.logger.warning(
+                "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:
@@ -111,7 +121,8 @@ class Printers:
                     )
                 except Exception as e:
                     self.app.logger.error(
-                        "Could not create a %s printer class with %s:%s" , product,
+                        "Could not create a %s printer class with %s:%s",
+                        product,
                         dev.idVendor,
                         dev.idProduct,
                     )
@@ -124,28 +135,34 @@ class 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:
+    def acquire_available_printer(self, start_index=0):
-        """
+        """Atomically acquire and return an available printer, or None."""
-        Return a dict key: UUID, value: Printer, with any connected printer.
+        printers = self.printers
-        """
+        if not printers:
-        if len(self.printers) > 0:
+            self.app.logger.error("No printers available for acquisition")
-            for i in self.printers:
+            return None
-                return i
-        else:
-            raise RuntimeError("No printers available")
 
-    # def get_printer(self, printer_type):
+        with self._lock:
-    #     """
+            for offset in range(len(printers)):
-    #     Return a specific printer
+                index = (start_index + offset) % len(printers)
+                printer = printers[index]
 
-    #     printer_type -- a printer type
+                if not printer.used:
-    #     """
+                    printer.use_printer()
-    #     return NotImplementedError()
+                    return index, printer
+
+        return None
+
+    def printers_info(self):
+        """Return the current list of printers"""
+        return self.printers
 
 
 class _FindClass:
@@ -153,19 +170,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:
+        for configuration in device:
-            # find_descriptor: what's it?
+            # Every device can have one or more configuration,
-            intf = usb.util.find_descriptor(cfg, bInterfaceClass=self._class)
+            interface = usb.util.find_descriptor(
-            if intf is not None:
+                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__(
+    def __init__(self, print_queue, app, configuration_file):
-        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.warning("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." +
+                    "Invalid position."
-                    "Choose from 'bottom_right', 'top_left', " +
+                    + "Choose from 'bottom_right', 'top_left', "
-                    " 'top_right', or 'bottom_left'."
+                    + " 'top_right', or 'bottom_left'."
                 )
 
             # Composite the logo onto the image

src/templates/index.html

@@ -6,9 +6,16 @@
       <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 ou pseudo" maxlength="200"><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>
           <br>
-          <input class="form-control" type="text" name="signature" placeholder="Signature or pseudo" maxlength="200"><br>
           <input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
         </form>
       </div>
@@ -21,7 +28,20 @@
       <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="form-control" type="text" name="signature" placeholder="Signature ou pseudo" maxlength="200"><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>
+          <br>
           <input class="btn btn-primary float-right" type="submit" value="Imprimer" name="imprimer">
         </form>
       </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) -> bool:
         """
         Get text and a signature, prints the text and cuts after that.
         """
@@ -29,14 +29,18 @@ class Web:
             # are added back to back, another task could be
             # inserted between the two.
             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) -> bool:
         """
         Get an image and a signature, prints the image and cuts after that.
         """
@@ -56,16 +60,17 @@ class Web:
                             secure_filename(image.filename),
                         ),
                         signature=sign,
-                        process=True,
+                        process=process_image,
                     )
                 )
+                self.app.logger.info("Added a IMG task at position %s", img)
 
+                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 +105,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 +115,7 @@ class Web:
                 )
                 return True
 
-        self.app.logger.error(
+        self.app.logger.error("Could not save file because the filename is forbidden")
-            "Could not save file because the filename is forbidden"
-        )
         raise RuntimeError("This file type is forbidden.")
 
     def get_queue_state(self):

src/worker.py

@@ -18,9 +18,10 @@ class PrintWorker(threading.Thread):
     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):
         super().__init__(daemon=True)
         self.app = app
         self.print_queue = print_queue
@@ -28,16 +29,28 @@ 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
+        self._printer_index = 0
+
+        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:
+    def _acquire_available_printer(self):
-            self.printers = Printers(self.app)
+        result = self.printers.acquire_available_printer(self._printer_index)
-            self.printers_obj = self.printers.printers # Get the number of printers connected to the system
+        if result is None:
-            self.printers_list = iter(self.printers.printers)
+            return None
 
-        except RuntimeError as e:
+        index, printer = result
-            self.app.logger.warning("Could not get any Printers")
+        self._printer_index = (index + 1) % len(self.printers_obj)
-            raise e
+        return printer
 
     def run(self):
         """Background thread that processes queue items"""
@@ -58,37 +71,22 @@ class PrintWorker(threading.Thread):
             # When we are finished with a printer, we release it to the world.
 
             # If we do not have a printer, we try to find one
-            while not self.printer:
-                time.sleep(1)
-                try:
-                    self.app.logger.debug("Trying to find a available printer for the worker.")
-
-                    # From the list of printer, we get one.
-                    # If no printer in the list is avaible, we try again from the start of the loop, that we reset.
-                    while self.printer.used or not self.printer:
-                        self.printer = next(self.printers_list)
-                        try:
-                            self.printer = next(self.printer)
-                        except StopIteration as e:
-                            self.app.error("We do not have any available printers. Resetting our list of printers and trying again. " + str(e))
-                            self.printers_list = iter(self.printers.printers)
-                            time.sleep(1)
-                            # TODO : Here, in case there is really no printer available, we will stay in a loop forever. Maybe after a number of tries, we should die.
-
-
-                    # Here, we must have found a Printer() that is
-                    # Printer.used = False, which means it's available.
-                    # Therefore, we mark it used, so no other worker uses it.
-                    self.printer.use_printer()
 
+            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",
                 )
-                except Exception as e:
+
-                    self.app.logger.error("No printer detected" + str(e))
+                if self.printer is None:
-                    self.printer = 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:
@@ -116,7 +114,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()