src/main.py

@@ -23,7 +23,6 @@ import sys
 import os  # For VARS from the shell.
 import pprint  # To pretty print JSON
 import toml  # Used for the config file parsing
-import threading
 from flask import (
     Flask,
     request,
@@ -43,7 +42,6 @@ from web import Web  # Wrapper for the web routes and API
 from print_queue import PrintQueue
 from worker import PrintWorker
 
-
 # We create the main Flask object
 app = Flask(__name__)
 socketio = SocketIO(app, cors_allowed_origins="*")
@@ -87,7 +85,7 @@ except FileExistsError:
     app.logger.debug("Directory %s already exists.", UPLOAD_FOLDER)
 except PermissionError:
     app.logger.error("Permission denied: Unable to create %s", UPLOAD_FOLDER)
-    exit(77)
+    sys.exit(77)
 
 # Output the config file
 if os.getenv("FLASK_DEBUG"):
@@ -137,6 +135,7 @@ limiter = Limiter(
 
 # General routes
 
+
 @app.route("/")
 @limiter.limit("1/second", override_defaults=False)
 def index():
@@ -152,8 +151,10 @@ def webcam():
     app.logger.debug("Loading webcam interface")
     return render_template("webcam.html")
 
+
 # Form treatement
 
+
 @app.route("/web/print/sms", methods=["POST"])
 @limiter.limit("6/minute", override_defaults=False)
 def web_print_sms():
@@ -164,7 +165,7 @@ def web_print_sms():
         txt = request.form["txt"]
     except werkzeug.exceptions.BadRequestKeyError as e:
         app.logger.error("Whoops, we are missing the txt input field. : %s ", str(e))
-        flash("Whoops, no forms submitted or missing signature : " + str(e), 'error')
+        flash("Whoops, no forms submitted or missing signature : " + str(e), "error")
         return redirect(url_for("index"))
 
     try:
@@ -172,19 +173,19 @@ def web_print_sms():
         sign = request.form["signature"]
     except werkzeug.exceptions.BadRequestKeyError as e:
         app.logger.warning(
-            "No signature found for this print, using default signature.", str(e)
+            "No signature found for this print, using default signature : %s ", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
 
     try:
         web.print_sms(txt, sign)
-    except Exception as e:
+    except RuntimeError as e:
         app.logger.error("Whoops, we could not print an SMS because : %s ", str(e))
-        flash("Whoops, we could not print an SMS because :" + str(e), 'error')
+        flash("Whoops, we could not print an SMS because :" + str(e), "error")
         return redirect(url_for("index"))
 
     # end try
-    flash("The SMS has been printed !", 'info')
+    flash("The SMS has been printed !", "info")
     return redirect(url_for("index"))
 
 
@@ -199,7 +200,7 @@ def web_print_img():
         sign = request.form["signature"]
     except werkzeug.exceptions.BadRequestKeyError as e:
         app.logger.warning(
-            "No signature found for this print, using default signature.", str(e)
+            "No signature found for this print, using default signature : %s", str(e)
         )
         sign = configuration_file["defaults"]["signature"]
 
@@ -207,7 +208,7 @@ def web_print_img():
     if "img" not in request.files:
         app.logger.error("Whoops, no images submitted  : %s ", str(e))
         app.logger.error("Error getting the files : %s", str(e))
-        flash("Whoops, no images submitted  : " + str(e), 'error')
+        flash("Whoops, no images submitted  : " + str(e), "error")
         return redirect(url_for("index"))
 
     file = request.files["img"]
@@ -215,20 +216,21 @@ def web_print_img():
     # empty file without a filename.
     if file.filename == "":
         app.logger.error("Submitted file has no filename !")
-        flash("Submitted file has no filename !", 'error')
+        flash("Submitted file has no filename !", "error")
         return redirect(url_for("index"))
 
     try:
         app.logger.debug("Sending the image to the printer.")
         web.print_image(file, sign)
-    except Exception as e:
+    except RuntimeError as e:
         app.logger.error("The image could not be printed because : %s ", str(e))
-        flash("The image could not be printed because : " + str(e), 'error')
+        flash("The image could not be printed because : " + str(e), "error")
         return redirect(url_for("index"))
 
-    flash("Picture printed !", 'info')
+    flash("Picture printed !", "info")
     return redirect(url_for("index"))
 
+
 # API routes
 # The api has the following methods
 # api/print/{sms,img,letter,qr,barcode}
@@ -267,11 +269,12 @@ def api_print_sms():
     try:
         # comment: We try to print the SMS
         web.print_sms(txt, sign)
-    except Exception as e:
+    except RuntimeError as e:
         return str(e), 500
     # end try
     return "OK", 200
 
+
 @app.route("/api/print/img", methods=["POST"])
 @limiter.limit("6/minute", override_defaults=False)
 def api_print_image():
@@ -302,38 +305,43 @@ def api_print_image():
     try:
         app.logger.debug("Sending the image to the printer.")
         web.print_image(file, sign)
-    except Exception as e:
+    except RuntimeError as e:
         return str(e), 500
 
     return "OK", 200
 
+
 @app.route("/api/camera/picture", methods=["GET"])
 def camera_picture():
     """Returns a picture taken by the camera on a raspberry pi"""
     if RASPBERRY_PI_CONNECTED:
         try:
             return rpi.camera_picture()
-        except Exception as e:
+        except RuntimeError as e:
             return jsonify({"message": "Error getting the stream : " + e}), 500
     else:
         return jsonify({"message": "No camera present"}), 500
 
-@app.route('/api/queue', methods=["GET"])
+
+@app.route("/api/queue", methods=["GET"])
 def api_queue_status():
     """API endpoint for entire queue"""
     return jsonify(web.get_queue_state())
 
-@app.route('/api/worker', methods=["GET"])
+
+@app.route("/api/worker", methods=["GET"])
 def api_worker_state():
     """API endpoint to get the worker state"""
     return jsonify(worker.current_state())
 
-@app.route('/api/worker/start')
+
+@app.route("/api/worker/start")
 def api_worker_start():
     worker.start_worker()
     return jsonify(worker.current_state())
 
-@app.route('/api/worker/stop')
+
+@app.route("/api/worker/stop")
 def api_worker_stop():
     worker.stop_worker()
     return jsonify(worker.current_state())
@@ -341,6 +349,7 @@ def api_worker_stop():
 
 ## Authentification
 
+
 @app.route("/login")
 @limiter.limit("1/second", override_defaults=False)
 def login_page():
@@ -393,5 +402,6 @@ def camera_status():
     else:
         socketio.emit("camera_status", False)
 
+
 if __name__ == "__main__":
     app.run(debug=True, use_reloader=False, host="0.0.0.0", ssl_context="adhoc")

src/print_queue.py

@@ -1,6 +1,8 @@
-# This class has the method by which we manage the Tasks
-# It's a printing queue, so we need to add, remove and get information on where
-# the queue is
+"""
+This class has the method by which we manage the Tasks
+It's a printing queue, so we need to add, remove and get information on where
+the queue is
+"""
 
 from collections import deque
 
@@ -9,12 +11,14 @@ from collections import deque
 import threading
 
 from datetime import datetime
-from task import TaskType, CutTask
+from task import TaskType
+
 
 class PrintQueue:
     """
     A Double-ended Queue to manage the printing Tasks
     """
+
     def __init__(self, app):
         self.app = app
         self._queue = deque()
@@ -34,7 +38,11 @@ class PrintQueue:
                 self._queue.append(task)
                 position = self._queue.index(task)
                 # We return the current position of the task if it was added
-                self.app.logger.debug("Added a new task %s to the queue at position %s", task.task_id, position)
+                self.app.logger.debug(
+                    "Added a new task %s to the queue at position %s",
+                    task.task_id,
+                    position,
+                )
                 return position
             except Exception as e:
                 self.app.logger.error("Could not add a task to the queue : %s ", e)
@@ -89,7 +97,7 @@ class PrintQueue:
                             "position": index,
                             "in_queue": True,
                             "content": task.content,
-                            "signature": task.signature
+                            "signature": task.signature,
                         }
 
                     if task.task_type == TaskType.TEXT:
@@ -101,7 +109,7 @@ class PrintQueue:
                             "in_queue": True,
                             "image_path": str(task.image_path),
                             "signature": task.signature,
-                            "process" : str(task.process)
+                            "process": str(task.process),
                         }
 
                     if task.task_type == TaskType.CUT:
@@ -110,7 +118,7 @@ class PrintQueue:
                             "status": task.status,
                             "type": task.task_type,
                             "position": index,
-                            "in_queue": True
+                            "in_queue": True,
                         }
 
             return None
@@ -123,5 +131,5 @@ class PrintQueue:
                 "status": task_status,
                 "position": None,
                 "in_queue": False,
-                "completed_at": datetime.now().isoformat()
+                "completed_at": datetime.now().isoformat(),
             }

src/printer.py

@@ -1,10 +1,14 @@
+
+# import brother_ql
+from time import sleep
+import os.path
+
+from PIL import Image, ImageEnhance
+import numpy as np
+
 # Importing the module to manage the connection to the printer.
 import escpos.printer
-# import brother_ql
-from time import sleep, gmtime, strftime
-import os.path
-from PIL import Image, ImageEnhance, ImageOps
-import numpy as np
+
 
 class Printer(object):
     """
@@ -75,10 +79,14 @@ class Printer(object):
             try:
                 # This also calls open(), which we need to close()
                 # or else the device will appear as busy.
-                p = escpos.printer.Usb(self.device_id, self.vendor_id, 0, profile="TM-P80")
+                p = escpos.printer.Usb(
+                    self.device_id, self.vendor_id, 0, profile="TM-P80"
+                )
             except Exception as e:
                 self.app.logger.error(
-                    "The USB device is not plugged in, trying again %s : %s",waiting_elapsed, str(e)
+                    "The USB device is not plugged in, trying again %s : %s",
+                    waiting_elapsed,
+                    str(e),
                 )
                 pass
 
@@ -87,7 +95,10 @@ class Printer(object):
                     self.ready = True
                     self.app.logger.debug("Printer online !")
             except Exception as e:
-                self.app.logger.error("Error while getting the printer online %s : %s",waiting_elapsed, str(e)
+                self.app.logger.error(
+                    "Error while getting the printer online %s : %s",
+                    waiting_elapsed,
+                    str(e),
                 )
                 pass
 
@@ -129,7 +140,9 @@ class Printer(object):
     def _print_sms(self, msg, signature="", bold=False):
 
         if not isinstance(msg, str):
-            self.app.logger.error("It is not possible to print a " + str(type(msg)) + ", only strings.")
+            self.app.logger.error(
+                "It is not possible to print a " + str(type(msg)) + ", only strings."
+            )
             raise ValueError
 
         # We make sure that the signature is not something too goofy
@@ -168,7 +181,9 @@ class Printer(object):
             self.printer.close()
         except Exception as e:
             self.app.logger.error("Unable to print because : " + str(e))
-            raise RuntimeError("Unable to print a SMS, the printer couldn't do it.") from e
+            raise RuntimeError(
+                "Unable to print a SMS, the printer couldn't do it."
+            ) from e
 
         self.app.logger.info("Printed text")
         return True
@@ -201,7 +216,9 @@ class Printer(object):
                 self.app.logger.debug("Proccessing the image")
                 path = _process_image(self, path)
             except RuntimeError as e:
-                self.app.logger.error("Error while processing the image, aborting print : %s",str(e))
+                self.app.logger.error(
+                    "Error while processing the image, aborting print : %s", str(e)
+                )
                 raise e
         else:
             self.app.logger.warning("Not proccessing the image")
@@ -226,7 +243,9 @@ class Printer(object):
             try:
                 self.printer.close()
             except Exception as e:
-                self.app.logger.error("Could not close the printer connexion %s", str(e))
+                self.app.logger.error(
+                    "Could not close the printer connexion %s", str(e)
+                )
                 raise RuntimeError("Could not close the printer connexion. ") from e
 
         self.app.logger.info("Printed a picture")
@@ -261,11 +280,13 @@ class Printer(object):
     def print_task(self, task_type, data):
         """Execute actual print based on task type"""
         match (task_type.value):
-            case ("text"):
+            case "text":
                 self._print_sms(data["txt"], signature=data["sign"])
-            case ("image"):
-                self._print_img(data["img"], signature=data["sign"],process=data["process"])
-            case ("cut"):
+            case "image":
+                self._print_img(
+                    data["img"], signature=data["sign"], process=data["process"]
+                )
+            case "cut":
                 self._cut()
             case _:
                 raise RuntimeError("This task type is not supported")
@@ -324,6 +345,7 @@ def _process_image(self, path):
 
     return jpeg_path
 
+
 def discover_printers():
     """
     We try to find all the connected printers ( 0 or n ) to this system.
@@ -336,8 +358,8 @@ def discover_printers():
     04f9  Brother Industries, Ltd
     """
 
-def find_and_parse_borther_ql_printer():
 
+def find_and_parse_borther_ql_printer():
 
     ## We might be able to no use this because there is a `discover` command in https://github.com/pklaus/brother_ql#usage
 
@@ -410,5 +432,6 @@ def find_and_parse_borther_ql_printer():
     print("No Brother QL printer found")
     return None
 
+
 def fint_and_parse_epson_printer():
     pass

src/raspberry.py

@@ -1,13 +1,12 @@
-from flask_socketio import SocketIO
-from gpiozero import Button, LED, DigitalOutputDevice
-from time import sleep, gmtime, strftime
-from PIL import Image
 import io  # To check if we are on a Raspberry Pi
 import subprocess
 import os
+from time import sleep, gmtime, strftime
+from flask_socketio import SocketIO
+from gpiozero import Button, LED, DigitalOutputDevice
+from PIL import Image
 
-
-class Raspberry(object):
+class Raspberry():
     """
     This class will manage three things :
     - Connecting to a USB webcam

src/task.py

@@ -1,27 +1,33 @@
-# Here we define the types of tasks
-# We are using Abstract Base Classes,
-# like this we can define types of tasks ( text, images, ... )
-# that all work with the same basic options
+"""
+Here we define the types of tasks
+We are using Abstract Base Classes,
+like this we can define types of tasks ( text, images, ... )
+that all work with the same basic options
 
-# The tasks are going to be injected into a Queue.
-# It's a usefull way of storing information in our
-# program, while making sure that things are indeed printed.
-# It's also a way to prevent two concurrent connexions creating
-# a access conflict on a single printer, like two people wanting
-# to print at the same time.
-
-# We can also delay and store printing tasks until a printer becomes
-# available if none is online.
+The tasks are going to be injected into a Queue.
+It's a usefull way of storing information in our
+program, while making sure that things are indeed printed.
+It's also a way to prevent two concurrent connexions creating
+a access conflict on a single printer, like two people wanting
+to print at the same time.
 
+We can also delay and store printing tasks until a printer becomes
+available if none is online.
+"""
 from abc import ABC, abstractmethod
+
 ## See https://docs.python.org/3/library/abc.html to learn more about this
 
 # from dataclasses import dataclass
 from enum import Enum
 import uuid
 
+
 ## You can expand this if you want to take other types of tasks into account
 class TaskType(Enum):
+    """
+    The different tasks supported by the printers
+    """
     TEXT = "text"
     IMAGE = "image"
     CUT = "cut"
@@ -31,6 +37,7 @@ class PrintTask(ABC):
     """
     A print task holds information about what we are looking to print.
     """
+
     def __init__(self, task_type):
         self.task_id = self._generate_id()
         self.task_type = task_type
@@ -41,7 +48,7 @@ class PrintTask(ABC):
     @abstractmethod
     def get_print_data(self):
         """Return data formatted for printer"""
-        pass
+
 
     def _generate_id(self):
         # Generate unique task ID
@@ -52,6 +59,7 @@ class TextTask(PrintTask):
     """
     This tasks represents a texte content, and it's signature.
     """
+
     def __init__(self, content, signature):
         super().__init__(TaskType.TEXT)
         self.content = content
@@ -60,10 +68,12 @@ class TextTask(PrintTask):
     def get_print_data(self):
         return {"txt": self.content, "sign": self.signature}
 
+
 class ImageTask(PrintTask):
     """
     This tasks represents a image content ( in the form of it's path ), and it's signature.
     """
+
     def __init__(self, image_path, signature, process):
         super().__init__(TaskType.IMAGE)
         self.image_path = image_path
@@ -74,6 +84,7 @@ class ImageTask(PrintTask):
         # Return image data in printer-compatible format
         return {"img": self.image_path, "sign": self.signature, "process": self.process}
 
+
 class CutTask(PrintTask):
     """
     This class activates the cutter on the printer if it exists

src/user.py

@@ -1,41 +1,41 @@
-class User(object):
-    """docstring for User."""
+# class User(object):
+#     """docstring for User."""
 
-    def __init__(self, arg):
-        super(User, self).__init__()
-        self.arg = arg
+#     def __init__(self, arg):
+#         super(User, self).__init__()
+#         self.arg = arg
 
 
-#         @app.route('/login', methods=['POST','GET'])
-# @limiter.limit("100 per minute", error_message=error_handler_limiter)
-def login():
-    if request.method == "POST":
-        if not session.get("logged_in"):
-            if request.form["username"] and request.form["password"]:
-                # Get the json
-                with open("users.json") as f:
-                    users_file = json.load(f)
-                for user in users_file["users"]:
-                    if users_file["users"][user] == request.form["password"]:
-                        session["logged_in"] = True
-                        session["user"] = request.form["username"]
+# #         @app.route('/login', methods=['POST','GET'])
+# # @limiter.limit("100 per minute", error_message=error_handler_limiter)
+# def login():
+#     if request.method == "POST":
+#         if not session.get("logged_in"):
+#             if request.form["username"] and request.form["password"]:
+#                 # Get the json
+#                 with open("users.json") as f:
+#                     users_file = json.load(f)
+#                 for user in users_file["users"]:
+#                     if users_file["users"][user] == request.form["password"]:
+#                         session["logged_in"] = True
+#                         session["user"] = request.form["username"]
 
-                if not session.get("logged_in"):
-                    flash("Mot de passe ou pseudo invalide.", "danger")
-                    return redirect(url_for("login"))
-                else:
-                    return redirect(url_for("display_index_page"))
-            else:
-                flash("Incorrect logins")
-                return render_template("password.html")
-        else:
-            return render_template("password.html")
-    else:
-        return render_template("password.html")
+#                 if not session.get("logged_in"):
+#                     flash("Mot de passe ou pseudo invalide.", "danger")
+#                     return redirect(url_for("login"))
+#                 else:
+#                     return redirect(url_for("display_index_page"))
+#             else:
+#                 flash("Incorrect logins")
+#                 return render_template("password.html")
+#         else:
+#             return render_template("password.html")
+#     else:
+#         return render_template("password.html")
 
 
-@app.route("/logout")
-def logout():
-    session["logged_in"] = False
-    flash("Tu est déconnecté", "info")
-    return redirect(url_for("login"))
+# @app.route("/logout")
+# def logout():
+#     session["logged_in"] = False
+#     flash("Tu est déconnecté", "info")
+#     return redirect(url_for("login"))

src/web.py

@@ -46,12 +46,16 @@ class Web(object):
         if file_uploaded:
             self.app.logger.debug("File has been uploaded, printing...")
             try:
-                img = self.print_queue.enqueue(ImageTask(os.path.join(
+                img = self.print_queue.enqueue(
+                    ImageTask(
+                        os.path.join(
                             self.app.config["UPLOAD_FOLDER"],
                             secure_filename(image.filename),
                         ),
                         signature=sign,
-                process=True))
+                        process=True,
+                    )
+                )
 
                 cut = self.print_queue.enqueue(CutTask())
             except Exception as e:
@@ -59,7 +63,6 @@ class Web(object):
 
             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:
@@ -96,10 +99,14 @@ class Web(object):
                 )
                 return True
             else:
-                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"
+                )
                 return False
         else:
-            self.app.logger.error("Could not save file, it seems to be null ? : " + str(filename))
+            self.app.logger.error(
+                "Could not save file, it seems to be null ? : " + str(filename)
+            )
             return False
 
     def get_queue_state(self):

src/worker.py

@@ -4,7 +4,7 @@
 
 import threading
 import time
-from task import TaskType
+
 
 class PrintWorker(threading.Thread):
     def __init__(self, app, print_queue, printer, socketio=None):
@@ -22,23 +22,17 @@ class PrintWorker(threading.Thread):
         """Background thread that processes queue items"""
         self.app.logger.info("Worker started working.")
         while True:
-            if not self.running:
+            if not self.running or not self.printer.ready:
                 time.sleep(0.2)
                 continue
 
-            # TODO: This could be improved to simply no start
-            # the while loop as long as the printer is not ready.
-            # and maybe get out of it when the printer is not ready anymore ?
-            if not self.printer.ready:
-                self.app.logger.debug("Waiting for the printer to be ready...")
-                time.sleep(1)
-                continue
-
             try:
                 task = self.print_queue.dequeue()
             except Exception as e:
                 self.app.logger.error("Could not get a new task ! %s ", str(e))
-                raise RuntimeError("We could not get a new task because " + str(e)) from e
+                raise RuntimeError(
+                    "We could not get a new task because " + str(e)
+                ) from e
 
             if task:
                 try:
@@ -51,7 +45,7 @@ class PrintWorker(threading.Thread):
                     print_data = task.get_print_data()
                     try:
                         self.printer.print_task(task.task_type, print_data)
-                    except Exception as e:
+                    except RuntimeError as e:
                         self.app.logger.error("Could not print : %s", str(e))
                         raise e
 
@@ -59,7 +53,7 @@ class PrintWorker(threading.Thread):
                     self.print_queue.mark_completed(task.task_id, "completed")
                     self._emit_status(task.task_id, "completed")
 
-                except Exception as e:
+                except RuntimeError as e:
                     task.status = "failed"
                     self.print_queue.mark_completed(task.task_id, "failed")
                     self._emit_status(task.task_id, "failed", error=str(e))
@@ -78,18 +72,19 @@ class PrintWorker(threading.Thread):
         data = {
             "task_id": task_id,
             "status": status,
-            "position": None  # Task no longer in queue
+            "position": None,  # Task no longer in queue
         }
 
         if error:
             data["error"] = error
 
-        self.socketio.emit('task_status', data, room=room)
+        self.socketio.emit("task_status", data, room=room)
 
     def stop_worker(self):
         """
         Give the worker a break
         """
+        self.app.logger.debug("Giving the worker a break")
         self.state = "drinking-a-beer"
         self.running = False
 
@@ -97,6 +92,7 @@ class PrintWorker(threading.Thread):
         """
         Get the worker back to it
         """
+        self.app.logger.debug("Time to work !")
         self.state = "idle"
         self.running = True
 
@@ -107,5 +103,5 @@ class PrintWorker(threading.Thread):
         return {
             "is_running": self.running,
             "queue_size": len(self.print_queue),
-            "state" : self.state
+            "state": self.state,
         }