Apply linting to web

src/main.py

@@ -100,11 +100,7 @@ print_queue = PrintQueue(app)
 rpi = Raspberry(
     print_queue,
     app,
-    socketio,
+    configuration_file
-    configuration_file["rpi"]["button_gpio_port_number"],
-    configuration_file["rpi"]["indicator_gpio_port_number"],
-    configuration_file["rpi"]["flash_gpio_port_number"],
-    configuration_file["rpi"]["flash"],
 )
 RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
 
@@ -126,6 +122,7 @@ limiter = Limiter(
     get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"]
 )
 
+
 # General routes
 @app.route("/")
 @limiter.limit("1/second", override_defaults=False)
@@ -317,11 +314,13 @@ def api_queue_status():
     """API endpoint for entire queue"""
     return jsonify(web.get_queue_state())
 
+
 @app.route("/api/queue/completed", methods=["GET"])
 def api_queue_completed():
     """API endpoint that returns the finished tasks"""
     return jsonify(web.get_queue_completed())
 
+
 @app.route("/api/worker", methods=["GET"])
 def api_worker_state():
     """API endpoint to get the worker state"""

src/print_queue.py

@@ -76,7 +76,10 @@ class PrintQueue:
         """Return current queue state"""
         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) } for t in self._queue]
+            return [
+                {"task_id": t.task_id, "status": t.status, "type": str(t.task_type)}
+                for t in self._queue
+            ]
 
     def get_queue_completed(self):
         """Return completed queue elements"""

src/printer.py

@@ -1,39 +1,44 @@
 """
 This class manages connexion to a Printer
 """
-from time import sleep
+
 import os.path
+import os
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 
 from enum import Enum
 import uuid
-import usb.core
 import threading
+import usb.core
 
 from PIL import Image, ImageEnhance
 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 import labels
 from brother_ql.raster import BrotherQLRaster
 from brother_ql.conversion import convert
 from brother_ql.backends.helpers import send
 
+
 class PrinterType(Enum):
     """
     What are the capacities of a Printer ?
     """
+
     EPSON = "epson"
     BROTHER = "brother"
 
+
 # For Brother-QL Printers
 @dataclass
 class PrinterInfo:
+    """
+    Brother-QL printer information
+    """
     identifier: str
     backend: str
     protocol: str
@@ -44,7 +49,7 @@ class PrinterInfo:
     model: str = "QL-570"
     status: str = "unknown"
     label_type: str = "unknown"
-    label_size : str = "unknown"
+    label_size: str = "unknown"
     label_width: int = 0
     label_height: int = 0
 
@@ -54,6 +59,7 @@ class PrinterInfo:
     def __setitem__(self, key, value):
         setattr(self, key, value)
 
+
 class Printer(ABC):
     """
     If it outputs printed paper and speaks like a printer, then it must be a printer.
@@ -80,7 +86,7 @@ class Printer(ABC):
         """Reports the state of the Printer"""
 
     @abstractmethod
-    def print_task(self, task_type, data)-> None:
+    def print_task(self, task_type, data) -> None:
         """Takes a PrintTask and executes it"""
 
 
@@ -96,7 +102,7 @@ class EscPosPrinter(Printer):
         Making sure it has paper,
         Define default print settings
         """
-        super().__init__(app,vendor_id,device_id, printer_type=PrinterType.EPSON)
+        super().__init__(app, vendor_id, device_id, printer_type=PrinterType.EPSON)
         self.printer = None
         self.usb_args = {}
         self.usb_args["idVendor"] = self.vendor_id
@@ -105,9 +111,7 @@ class EscPosPrinter(Printer):
         try:
             # This also calls open(), which we need to close()
             # or else the device will appear as busy.
-            p = escpos.printer.Usb(
+            p = escpos.printer.Usb(self.vendor_id, self.device_id, 0, profile="TM-P80")
-                self.vendor_id, self.device_id, 0, profile="TM-P80"
-            )
         except escpos.exceptions.DeviceNotFoundError as e:
             self.app.logger.error(
                 "The USB device is not plugged in : %s",
@@ -339,13 +343,16 @@ class EscPosPrinter(Printer):
             else:
                 raise RuntimeError("The printer is not ready to print yet !")
 
+
 class BrotherPrinter(Printer):
     """
     Manages connexion and capabilities of a BrotherQL Printer
     """
 
     def __init__(self, app, vendor_id, device_id):
-        super().__init__(app, vendor_id="",device_id="", printer_type=PrinterType.BROTHER)
+        super().__init__(
+            app, vendor_id="", device_id="", printer_type=PrinterType.BROTHER
+        )
         self.printer = None
         self.usb_args = {}
         self.usb_args["idVendor"] = self.device_id
@@ -363,11 +370,13 @@ class BrotherPrinter(Printer):
             parts = identifier.split("/")
 
             if len(parts) < 4:
-                self.app.logger.warning(f"Skipping device with invalid identifier format: {identifier}")
+                self.app.logger.warning(
+                    f"Skipping device with invalid identifier format: {identifier}"
+                )
                 continue
 
             protocol = parts[0]
-            device_info = parts[2]
+            # device_info = parts[2]
             serial_number = parts[3]
 
         try:
@@ -378,7 +387,7 @@ class BrotherPrinter(Printer):
                     break
             self.app.logger.debug(f"Matched printer model: {model}")
         except ValueError:
-            self.app.logger.warning(f"Invalid product ID format: {product_id}")
+            self.app.logger.warning(f"Invalid product ID format: {m.product_id}")
 
         self.printer_info = PrinterInfo(
             identifier=identifier,
@@ -392,14 +401,13 @@ class BrotherPrinter(Printer):
 
         self.ready = True
 
-
     def _has_paper(self):
         raise NotImplementedError("This printer model does not support this.")
 
     def _state(self):
         return self.ready
 
-    def _print_img(self,data):
+    def _print_img(self, data):
         """
         Print a raster image via a Brother QL printer
         """
@@ -426,7 +434,7 @@ class BrotherPrinter(Printer):
             )
 
             # Debug logging
-            if FLASK_DEBUG:
+            if os.getenv("FLASK_DEBUG"):
                 self.app.logger.debug(f"""
                 Print parameters:
                 - Label type: {label_type}
@@ -447,11 +455,15 @@ class BrotherPrinter(Printer):
             # }
             status = send(
                 instructions=instructions,
-                printer_identifier= self.printer_info["identifier"],
+                printer_identifier=self.printer_info["identifier"],
-                backend_identifier="pyusb"
+                backend_identifier="pyusb",
             )
 
-            if not status["did_print"] or status["outcome"] == "error" or status["outcome"] == "unknown":
+            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"]:
@@ -462,7 +474,9 @@ class BrotherPrinter(Printer):
         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(
+                    "USB timeout occurred - this is normal and the print likely completed"
+                )
                 self.app.logger.debug("Print completed (timeout is normal)")
                 self.ready = True
 
@@ -496,10 +510,11 @@ class BrotherPrinter(Printer):
 
         # raise NotImplementedError("This printer type is not implemented yet")
 
+
 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 = 0.6  # Less than 1.0 = lower contrast
     max_width = 575
     max_height = 1000
 

src/printers.py

@@ -1,17 +1,21 @@
 """
 A collection of Printers.
 
-It has methods to discover printers, and provides an interface for the methods expected from printers.
+It has methods to discover printers, and provides an interface for
+the methods expected from printers.
 """
-from collections.abc import Mapping, Set
+
+from collections.abc import Set
 import usb.core
 import usb.util
-from printer import Printer, EscPosPrinter, BrotherPrinter, PrinterType
+from printer import Printer, EscPosPrinter, BrotherPrinter
 
-class Printers():
+
+class Printers:
     """
     Finds and creates a set of Printer that can be used by the Workers to print.
     """
+
     def __init__(self, app):
         """
         Discover printers connected to the computer and return a Collection of Printer()
@@ -36,10 +40,14 @@ class Printers():
         printers = set()
 
         # Find all connected USB devices
-        devices = usb.core.find(find_all=True,custom_match=_FindClass(7))
+        devices = usb.core.find(find_all=True, custom_match=_FindClass(7))
         if not devices:
-            self.app.logger.warning("No USB devices of class 7 ( printers ) found or pyusb could not access the bus.")
+            self.app.logger.warning(
-            raise RuntimeError("No USB devices of class 7 ( printers ) found or pyusb could not access the bus.")
+                "No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
+            )
+            raise RuntimeError(
+                "No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
+            )
 
         for dev in devices:
             # Attempt to get the manufacturer and product strings
@@ -52,7 +60,13 @@ class Printers():
                 product = usb.util.get_string(dev, dev.iProduct)
             except Exception:
                 product = "Unknown"
-            self.app.logger.debug("Looking at %s %s (%s:%s)", manufacturer, product, hex(dev.idVendor), hex(dev.idProduct))
+            self.app.logger.debug(
+                "Looking at %s %s (%s:%s)",
+                manufacturer,
+                product,
+                hex(dev.idVendor),
+                hex(dev.idProduct),
+            )
 
             if manufacturer == "EPSON":
                 try:
@@ -60,13 +74,15 @@ class Printers():
                     self.app.logger.debug("Trying to creat a new EPSON printer")
                     prid = dev.idProduct
                     vendir = dev.idVendor
-                    escpos_printer = EscPosPrinter(self.app, vendor_id=vendir, device_id=prid)
+                    escpos_printer = EscPosPrinter(
+                        self.app, vendor_id=vendir, device_id=prid
+                    )
                 except Exception as e:
                     raise e
 
                 # If the object creation is successfull, we add it to the list of Printers
                 printers.add(escpos_printer)
-                self.app.logger.debug("Found a %s printer" % manufacturer )
+                self.app.logger.debug("Found a %s printer", manufacturer)
 
                 # We already found the type of printer,
                 # we don't need an extra comparaison.
@@ -79,16 +95,22 @@ class Printers():
                     self.app.logger.debug("Trying to creat a new BROTHER printer")
                     prid = dev.idProduct
                     vendir = dev.idVendor
-                    brother_printer = BrotherPrinter(self.app, vendor_id=vendir,device_id=prid)
+                    brother_printer = BrotherPrinter(
+                        self.app, vendor_id=vendir, device_id=prid
+                    )
                 except Exception as e:
-                    self.app.logger.error("Could not create a %s printer class with %s:%s" % product, dev.idVendor, dev.idProduct)
+                    self.app.logger.error(
+                        "Could not create a %s printer class with %s:%s" , product,
+                        dev.idVendor,
+                        dev.idProduct,
+                    )
                     raise e
 
                 # If the object creation is successfull, we add it to the list of Printers
                 printers.add(brother_printer)
-                self.app.logger.debug("Found a %s printer" % manufacturer )
+                self.app.logger.debug("Found a %s printer" , manufacturer)
 
-        self.app.logger.debug("Found %s printers" % len(printers))
+        self.app.logger.debug("Found %s printers" , len(printers))
         return printers
 
     def any(self) -> Printer:
@@ -101,18 +123,23 @@ class Printers():
         else:
             raise RuntimeError("No printers available")
 
-    def get_printer(self, printer_type):
+    # def get_printer(self, printer_type):
+    #     """
+    #     Return a specific printer
+
+    #     printer_type -- a printer type
+    #     """
+    #     return NotImplementedError()
+
+
+class _FindClass:
     """
-        Return a specific printer
+    Use by usb.core to modify the way USB devices are found
-
+    Taken from pyUSB documentation on Github
-        printer_type -- a printer type
     """
-        return NotImplementedError()
-
-
-class _FindClass():
     def __init__(self, class_):
         self._class = class_
+
     def __call__(self, device):
         # first, let's check the device
         if device.bDeviceClass == self._class:
@@ -121,10 +148,7 @@ class _FindClass():
         # interface that matches our class
         for cfg in device:
             # find_descriptor: what's it?
-            intf = usb.util.find_descriptor(
+            intf = usb.util.find_descriptor(cfg, bInterfaceClass=self._class)
-                                        cfg,
-                                        bInterfaceClass=self._class
-                                )
             if intf is not None:
                 return True
 

src/raspberry.py

@@ -10,12 +10,12 @@ 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
 from task import TextTask, ImageTask, CutTask
 
-class Raspberry():
+
+class Raspberry:
     """
     This class will manage three things :
     - Connecting to a USB webcam
@@ -23,29 +23,22 @@ class Raspberry():
     - Activating a flash ( or light )
     - Flash an indicator light
 
+    # pylint: disable=too-many-instance-attributes
+    # dede
     """
 
     def __init__(
         self,
         print_queue,
         app,
-        socketio,
+        configuration_file
-        button_gpio_port_number,
-        indicator_gpio_port_number,
-        flash_gpio_port_number,
-        is_flash_present,
     ):
         self.print_queue = print_queue
-        self.socketio = socketio
         self.app = app
-
+        self.configuration_file = configuration_file
-        self.flash_gpio = flash_gpio_port_number
-        self.is_flash_present = is_flash_present
-        self.button_gpio = button_gpio_port_number
-        self.led_gpio = indicator_gpio_port_number
         self.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg"
 
-    def is_raspberry_pi(self, raise_on_errors=False):
+    def is_raspberry_pi(self):
         """
         Checking if we are on a Raspberry Pi by checking
         information on the /proc/cpuinfo file
@@ -72,9 +65,10 @@ class Raspberry():
                             return False
 
                 if not found:
-                    self.app.logger.error(
+                    self.app.logger.warning(
-                        "Couldn't get sufficient hardware information from /proc/cpuinfo, Unable to determine if we are on a Raspberry Pi."
+                        "Couldn't get sufficient hardware information from /proc/cpuinfo"
                     )
+                    self.app.logger.error("Unable to determine if we are on a Raspberry Pi.")
                     return False
         except IOError:
             self.app.logger.error("Unable to open `/proc/cpuinfo`.")
@@ -83,28 +77,38 @@ class Raspberry():
         self.app.logger.debug("It seems we are on a Raspberry Pi")
 
         try:
-            self.initialise_gpio()
+            self._initialise_gpio()
         except Exception as e:
             self.app.logger.debug("Could not init GPIO : " + str(e))
             raise e
         return True
 
-    def initialise_gpio(self):
+    def _initialise_gpio(self):
+        """
+        Set GPIO ports from configuration and activate them
+        to show the user it's working.
+        """
         self.app.logger.debug("Initializing GPIO")
 
-        self.led = LED(self.led_gpio)
+        self.led = LED(self.configuration_file["rpi"]["indicator_gpio_port_number"])
         self.app.logger.debug("Activated indicator LED")
         self.indicator_countdown(iters=3)
-        self.button = Button(self.button_gpio, pull_up=True, bounce_time=0.1)
+        self.button = Button(
+            self.configuration_file["rpi"]["button_gpio_port_number"],
+            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.flash_gpio)
+        self.flash = DigitalOutputDevice(self.configuration_file["rpi"]["flash_gpio_port_number"])
         self.flash_toggle()
         self.app.logger.debug("Activated flash")
 
     def indicator_countdown(self, iters=10, multi=10):
+        """
+        Activates the LED faster and faster to show a countdown
+        """
         for i in range(iters, 0, -1):
             self.led.on()
             sleep(i / multi)
@@ -112,7 +116,10 @@ class Raspberry():
             sleep(i / multi)
 
     def indicator_led(self, timing=0.2, l=5):
-        for i in range(l):
+        """
+        Turns on the indicator LED for a certain period of time
+        """
+        for _ in range(l):
             self.app.logger.debug("LED turned on")
             self.led.on()
             sleep(timing)
@@ -121,6 +128,9 @@ class Raspberry():
             sleep(timing)
 
     def flash_toggle(self):
+        """
+        Flashes the flash
+        """
         self.app.logger.debug("Flash turned on")
         self.flash.on()
         sleep(0.3)
@@ -128,6 +138,9 @@ class Raspberry():
         self.app.logger.debug("Flash turned off")
 
     def take_picture(self):
+        """
+        Takes a picture via the USB webcam
+        """
         # Validate if the image path is valid
         if not os.path.isdir(os.path.dirname(self.image_path)):
             self.app.logger.error(
@@ -173,6 +186,9 @@ class Raspberry():
         position="bottom_right",
         margin=10,
     ):
+        """
+        Takes an image and overlays it with a another picture.
+        """
         try:
             image = Image.open(image_path).convert("RGBA")
             logo = Image.open(logo_path).convert("RGBA")
@@ -199,7 +215,9 @@ class Raspberry():
                 y = image.height - logo.height - margin
             else:
                 raise ValueError(
-                    "Invalid position. Choose from 'bottom_right', 'top_left', 'top_right', or 'bottom_left'."
+                    "Invalid position." +
+                    "Choose from 'bottom_right', 'top_left', " +
+                    " 'top_right', or 'bottom_left'."
                 )
 
             # Composite the logo onto the image
@@ -217,6 +235,9 @@ class Raspberry():
         return True
 
     def crop_to_square(self, image_path, output_path=None):
+        """
+        Crop an image so that it becomes a square
+        """
         try:
             image = Image.open(image_path)
             width, height = image.size
@@ -244,6 +265,10 @@ class Raspberry():
         return True
 
     def on_button_pressed(self):
+        """
+        When a button press is detected, a picture is taken from the webcam
+        and added to the print queue.
+        """
         self.app.logger.debug("Button has been pressed")
         self.led.on()
         self.app.logger.debug("Counting down")
@@ -263,8 +288,10 @@ class Raspberry():
         self.app.logger.debug("Printing picture")
         self.led.on()
         self.crop_to_square(self.image_path)
-        self.print_queue.enqueue(ImageTask(self.image_path,signature="",process=True))
+        self.print_queue.enqueue(ImageTask(self.image_path, signature="", process=True))
-        self.print_queue.enqueue(TextTask(content="Imprimé par LittlePrynter", signature=""))
+        self.print_queue.enqueue(
+            TextTask(content="Imprimé par LittlePrynter", signature="")
+        )
         time = strftime("%Y-%m-%d %H:%M", gmtime())
         self.print_queue.enqueue(TextTask(content=time, signature=""))
         self.print_queue.enqueue(CutTask())

src/task.py

@@ -14,6 +14,7 @@ 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
@@ -27,6 +28,7 @@ class TaskType(Enum):
     """
     The different tasks supported by the printers
     """
+
     TEXT = "text"
     IMAGE = "image"
     CUT = "cut"
@@ -47,7 +49,6 @@ class PrintTask(ABC):
     def get_print_data(self):
         """Return data formatted for printer"""
 
-
     def _generate_id(self):
         # Generate unique task ID
         return str(uuid.uuid4())
@@ -66,8 +67,10 @@ class TextTask(PrintTask):
     def get_print_data(self):
         return {"txt": self.content, "sign": self.signature}
 
+
 class QRTask(TextTask):
     """This task prints a QR-Code, the signature is ignore and is always the content itself"""
+
     def __init__(self, content):
         super().__init__(content, signature="")
         self.content = content
@@ -76,6 +79,7 @@ class QRTask(TextTask):
     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.

src/web.py

@@ -1,10 +1,13 @@
+"""
+Manage all of the inputs from a web source
+"""
+
 import os
-from flask import flash
 from werkzeug.utils import secure_filename
 from task import TextTask, ImageTask, CutTask
 
 
-class Web(object):
+class Web:
     """Web is the class that gets all of the information from web calls
     ( API and Web page ) and provides checks before sending stuff to printing"""
 
@@ -65,16 +68,19 @@ class Web(object):
 
         return True
 
-    def login(self, username: str, password: str) -> bool:
+    # def login(self, username: str, password: str) -> bool:
-        """Not implemented"""
+    #     """Not implemented"""
-        return
+    #     return
 
-    def logout(self, username: str, password: str) -> bool:
+    # def logout(self, username: str, password: str) -> bool:
-        """Not implemented"""
+    #     """Not implemented"""
-        return
+    #     return
 
     def allowed_file(self, filename) -> bool:
-        self.app.logger.debug("Is the filename allowed ?")
+        """
+        Check if the file extension is allowed
+        """
+        self.app.logger.debug("Checking if the file extension is allowed")
         return (
             "." in filename
             and filename.rsplit(".", 1)[1].lower()
@@ -82,8 +88,11 @@ class Web(object):
         )
 
     def upload_file(self, image) -> bool:
+        """
+        Save the file after executing checks on it
+        """
         self.app.logger.debug("Validating file")
-        if image:
+        if not image is None or not image == "":
             if self.allowed_file(image.filename):
                 filename = secure_filename(image.filename)
                 self.app.logger.debug("File valid")
@@ -104,12 +113,6 @@ class Web(object):
         )
         return False
 
-        else:
-            self.app.logger.error(
-                "Could not save file, it seems to be null ? : " + str(filename)
-            )
-            return False
-
     def get_queue_state(self):
         """Return current queue state"""
         return self.print_queue.get_queue_state()

src/worker.py

@@ -48,7 +48,11 @@ class PrintWorker(threading.Thread):
                 try:
                     self.app.logger.debug("Changing printers")
                     self.printer = next(self.printers)
-                    self.app.logger.debug("The worker got a %s printer and it's %s", self.printer.printer_type, "Ready" if self.printer.ready else "Not ready")
+                    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(str(e))
                     self.printer = None
@@ -89,7 +93,9 @@ 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)
+                        self.app.logger.debug(
+                            "Finished printing task %s " % task.task_id
+                        )
                         self.state = "idle"
 
                     except RuntimeError as e:
@@ -97,7 +103,9 @@ class PrintWorker(threading.Thread):
                         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))
+                        self.app.logger.error(
+                            "Could not print task %s because %s " % task.task_id, str(e)
+                        )
 
                 else:
                     # When they are no new tasks to handle, we put the thread to sleep.