Change Error to warning when not on a Raspberry

because this is a real-time command, we might not get the good answer
and fail to print fast enough. See https://download4.epson.biz/sec_pubs/pos/reference_en/escpos/realtime_commands.html

src/main.py

@@ -83,14 +83,14 @@ except PermissionError:
     sys.exit(77)
 
 # Output the config file
-if os.getenv("FLASK_DEBUG"):
+if not os.getenv("FLASK_DEBUG") is None and os.getenv("FLASK_DEBUG") is True:
     pprint.pprint(configuration_file)
 
 # We define the app module used by Flask
 app.secret_key = configuration_file["secrets"]["flask_secret_key"]
 app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER
 app.config["ALLOWED_EXTENSIONS"] = ALLOWED_EXTENSIONS
-app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000  # Maximum 3Mb for a file upload
+app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000  # Maximum 10Mb for a file upload
 app.config["TEMPLATES_AUTO_RELOAD"] = True
 
 # Queue creation
@@ -111,7 +111,7 @@ web = Web(app, print_queue)
 # Start worker thread
 # When created, the worker will try to find printers connected to the system
 try:
-    worker = PrintWorker(app, print_queue, socketio)
+    worker = PrintWorker(app, print_queue)
     worker.start()
 except Exception as e:
     app.logger.error("Could not start the worker because %s ", str(e))
@@ -123,6 +123,8 @@ limiter = Limiter(
 )
 
 
+app.logger.info("🖶 Welcome to LittlePrynter !")
+
 # General routes
 @app.route("/")
 @limiter.limit("1/second", override_defaults=False)
@@ -189,6 +191,9 @@ def web_print_img():
             "No signature found for this print, using default signature : %s", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
+    except werkzeug.exceptions.RequestEntityTooLarge as e:
+        flash("Whoops, image is too big: " + str(e), "error")
+        return redirect(url_for("index"))
 
     # check if the post request has the file part
     if "img" not in request.files:
@@ -402,4 +407,4 @@ def camera_status():
 
 
 if __name__ == "__main__":
-    app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")
+    app.run(use_reloader=False, host="0.0.0.0", ssl_context="adhoc")

src/print_queue.py

@@ -77,7 +77,7 @@ 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)}
+                {"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

@@ -6,12 +6,11 @@ import os.path
 import os
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-
+import time
 from enum import Enum
 import uuid
 import threading
 import usb.core
-
 from PIL import Image, ImageEnhance
 import numpy as np
 
@@ -176,7 +175,7 @@ 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."
@@ -216,6 +215,7 @@ class EscPosPrinter(Printer):
             self.printer.textln(clean_msg)
             if clean_signature:
                 self.printer.textln(clean_signature)
+            self.printer.textln()
             self.printer.close()
         except Exception as e:
             self.app.logger.error("Unable to print because : " + str(e))
@@ -224,9 +224,10 @@ class EscPosPrinter(Printer):
             ) from e
 
         self.app.logger.info("Printed text")
-        return True
+        self.ready = True
 
     def _print_img(self, path, signature="", center=True, process=False):
+        self.ready = False
         clean_signature = str(signature)
 
         if len(signature) > 256:
@@ -252,30 +253,33 @@ class EscPosPrinter(Printer):
         if process:
             try:
                 self.app.logger.debug("Proccessing the image")
-                path = _process_image(self, path)
+                processed_path = _process_image(self, path)
             except RuntimeError as e:
                 self.app.logger.error(
                     "Error while processing the image, aborting print : %s", str(e)
                 )
                 raise e
         else:
+            processed_path = path
             self.app.logger.warning("Not proccessing the image")
 
         try:
             self.printer.open(self.usb_args)
-            self.printer.image(path, center=center)
+            self.printer.image(processed_path, center=center)
             self.printer.textln(signature)
             self.printer.close()
-            self.app.logger.debug("Printed an image : " + str(path))
+            self.app.logger.debug("Printed an image : " + str(processed_path))
         except Exception as e:
             self.app.logger.error(str(e))
             raise RuntimeError("Could not print the picture") from e
         finally:
             try:
                 os.remove(path)
+                os.remove(processed_path)
             except OSError as e:
                 raise e
 
+            self.app.logger.debug("Removed image : " + str(processed_path))
             self.app.logger.debug("Removed image : " + str(path))
 
             try:
@@ -287,9 +291,10 @@ class EscPosPrinter(Printer):
                 raise RuntimeError("Could not close the printer connexion. ") from e
 
         self.app.logger.info("Printed a picture")
-        return True
+        self.ready = True
 
     def _qr(self, content):
+        self.ready = False
         try:
             self.printer.open(self.usb_args)
             self.printer.qr(content, center=True)
@@ -301,8 +306,10 @@ class EscPosPrinter(Printer):
             raise e
 
         self.app.logger.info("Printed a QR")
+        self.ready = True
 
     def _cut(self):
+        self.ready = False
         try:
             self.printer.open(self.usb_args)
             self.printer.cut()
@@ -313,35 +320,56 @@ class EscPosPrinter(Printer):
             raise e
 
         self.app.logger.info("Did a cut")
-        return True
+        self.ready = True
 
-    def _state(self):
+    def _state(self) -> bool:
-        return self.printer.is_online() and self.ready and self._has_paper()
+        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:
-            if self._state:
+            self.app.logger.debug("Acquired lock to start print")
-                self._state = False
+
+            i_m_ready = self._state()
+            while not i_m_ready:
+                self.app.logger.debug("Waiting for the printer to become ready, current state %s ", str(i_m_ready))
+                i_m_ready = self._state()
+                time.sleep(0.3)
+
+            self.app.logger.debug("Checked state to start printing : %s", self._state())
+            self.ready = False
+            try:
+                self.app.logger.debug("Checking task type")
                 match (task_type.value):
                     case "text":
+
                         self._print_txt(data["txt"], signature=data["sign"])
-                        self._state = True
+                        self.ready = True
                     case "image":
                         self._print_img(
                             data["img"], signature=data["sign"], process=data["process"]
                         )
-                        self._state = True
+                        self.ready = True
                     case "cut":
                         self._cut()
-                        self._state = True
+                        self.ready = True
                     case "qr":
                         self._qr(data["txt"])
-                        self._state = True
+                        self.ready = True
                     case _:
                         raise RuntimeError("This task type is not supported")
-            else:
+            except Exception as e:
-                raise RuntimeError("The printer is not ready to print yet !")
+                self.app.logger.debug("Exception occured while printing %s", str(e))
+                self.ready = True
+                raise RuntimeError from e
 
 
 class BrotherPrinter(Printer):
@@ -514,14 +542,14 @@ class BrotherPrinter(Printer):
 def _process_image(self, path):
     brightness_factor = 1.5  # Used only if image is too dark
     brightness_threshold = 100  # Brightness threshold (0–255)
-    # contrast_factor = 0.6  # Less than 1.0 = lower contrast
+    contrast_factor = 2  # Less than 1.0 = lower contrast
     max_width = 575
     max_height = 1000
 
     with Image.open(path) as original_img:
         # Convert to RGB if needed (JPEG doesn't support alpha)
         if original_img.mode in ("RGBA", "P"):
-            self.app.logger.debug("Converting the image to RGB from RGBA")
+            self.app.logger.debug("Converting the image from RGBA to RGBA")
             original_img = original_img.convert("RGB")
 
         # Resize while maintaining aspect ratio
@@ -551,16 +579,19 @@ def _process_image(self, path):
             self.app.logger.debug(
                 f"Image too dark, increasing brightness by a factor of {brightness_factor:.2f}"
             )
-            enhancer = ImageEnhance.Brightness(original_img)
+            enhancer = ImageEnhance.Brightness(grayscale)
-            original_img = enhancer.enhance(brightness_factor)
+            grayscale = enhancer.enhance(brightness_factor)
 
-        # # Reduce contrast
+        # Computer current contrast of grayscale image
-        # contrast_enhancer = ImageEnhance.Contrast(original_img)
+        contrast = np.clip(np.std(np.array(grayscale)), 0, 255)
-        # original_img = contrast_enhancer.enhance(contrast_factor)
+        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)
 
         # Convert to JPEG and save
         jpeg_path = os.path.splitext(path)[0] + "_processed.jpg"
-        original_img.save(jpeg_path, format="JPEG", quality=95, optimize=True)
+        grayscale.save(jpeg_path, format="JPEG", quality=95, optimize=True)
         self.app.logger.debug("Processed and saved image.")
 
     return jpeg_path

src/printers.py

@@ -111,6 +111,11 @@ class Printers:
                 self.app.logger.debug("Found a %s printer" , manufacturer)
 
         self.app.logger.debug("Found %s printers" , len(printers))
+
+        if len(printers) < 1:
+            self.app.logger.warning("Not printers found ! Please plug in a Printer and restart the program.")
+            raise RuntimeError("No printers found")
+
         return printers
 
     def any(self) -> Printer:

src/raspberry.py

@@ -68,7 +68,7 @@ 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`.")

src/templates/index.html

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

src/web.py

@@ -44,7 +44,7 @@ class Web:
             file_uploaded = self.upload_file(image)
         except Exception as e:
             self.app.logger.error(e)
-            raise RuntimeError("Could not upload file") from e
+            raise RuntimeError("Could not upload file : " + str(e)) from e
 
         if file_uploaded:
             self.app.logger.debug("File has been uploaded, printing...")
@@ -100,7 +100,7 @@ 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)
-                    return False
+                    raise RuntimeError("An OS error occured while uploading this file : " + str(e)) from e
 
                 self.app.logger.debug(
                     "File saved to "
@@ -111,7 +111,7 @@ class Web:
         self.app.logger.error(
             "Could not save file because the filename is forbidden"
         )
-        return False
+        raise RuntimeError("This file type is forbidden.")
 
     def get_queue_state(self):
         """Return current queue state"""

src/worker.py

@@ -1,6 +1,7 @@
-# This is the main printing thread
+"""
-# As explained in the task file, this is where we command
+This is the main printing thread. A worker thread consums Tasks from
-# printing to happen.
+ a PrintQueue, while trying to find available printers.
+"""
 
 import threading
 import time
@@ -8,13 +9,21 @@ from printers import Printers
 
 
 class PrintWorker(threading.Thread):
-    def __init__(self, app, print_queue, socketio=None):
+    """
+    A thread used to consume Tasks added to a Print Queue.
+
+    On initialisation, the worker will try to find Printers,
+    and on each print, choose an available Printer from a list of Printers.
+
+    If a print fails, it's not retried, but will be added to a list of completed
+    tasks.
+    """
+    def __init__(self, app, print_queue):
         super().__init__(daemon=True)
         self.app = app
         self.print_queue = print_queue
         self.printer = None
         self._lock = threading.Lock()
-        self.socketio = socketio  # Optional
         self.running = True
         self.state = "idle"  # idle, printing, dead, drinking-a-beer
         self.app.logger.debug("Ho great, I'm alive... I'm ready to work another day...")
@@ -30,8 +39,8 @@ class PrintWorker(threading.Thread):
     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("Current threads : %s" , threading.active_count())
-        self.app.logger.debug("Threads actives : %s " % threading.enumerate())
+        self.app.logger.debug("Threads actives : %s " , threading.enumerate())
 
         while True:
 
@@ -40,7 +49,8 @@ class PrintWorker(threading.Thread):
                 time.sleep(0.2)
                 continue
 
-            # If we have no available printer, we look at the list printers we know about, and try to find one that is available.
+            # If we have no available printer, we look at the list printers
+            # we know about, and try to find one that is available.
             # When we find a printer, we acquire it
             # When we are finished with a printer, we release it to the world.
             while not self.printer or not self.printer.ready:
@@ -54,7 +64,7 @@ class PrintWorker(threading.Thread):
                         "Ready" if self.printer.ready else "Not ready",
                     )
                 except Exception as e:
-                    self.app.logger.error(str(e))
+                    self.app.logger.error("No printer detected" + str(e))
                     self.printer = None
 
             if self.state != "idle":
@@ -79,7 +89,6 @@ class PrintWorker(threading.Thread):
                         self.app.logger.info("Got a new task")
                         self.app.logger.debug("Got task %s", task.task_id)
                         task.status = "processing"
-                        self._emit_status(task.task_id, "processing")
 
                         print_data = task.get_print_data()
 
@@ -92,9 +101,8 @@ class PrintWorker(threading.Thread):
 
                         task.status = "completed"
                         self.print_queue.mark_completed(task.task_id, "completed")
-                        self._emit_status(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"
 
@@ -102,9 +110,8 @@ class PrintWorker(threading.Thread):
                         task.status = "failed"
                         self.state = "idle"
                         self.print_queue.mark_completed(task.task_id, "failed")
-                        self._emit_status(task.task_id, "failed", error=str(e))
                         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:
@@ -112,23 +119,6 @@ class PrintWorker(threading.Thread):
                     self.state = "idle"
                     time.sleep(0.1)
 
-    def _emit_status(self, task_id, status, error=None):
-        """Emit status update via Socket.IO if available"""
-        if not self.socketio:
-            return
-
-        room = f"task_{task_id}"
-        data = {
-            "task_id": task_id,
-            "status": status,
-            "position": None,  # Task no longer in queue
-        }
-
-        if error:
-            data["error"] = error
-
-        self.socketio.emit("task_status", data, room=room)
-
     def stop_worker(self):
         """
         Give the worker a break