Update Error raising in uploads and image processing

This commit is contained in:
n07070
2026-06-04 19:26:21 +02:00
parent 175dd3385a
commit 53010987f4
2 changed files with 63 additions and 44 deletions

View File

@@ -6,12 +6,11 @@ import os.path
import os import os
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from dataclasses import dataclass from dataclasses import dataclass
import time
from enum import Enum from enum import Enum
import uuid import uuid
import threading import threading
import usb.core import usb.core
from PIL import Image, ImageEnhance from PIL import Image, ImageEnhance
import numpy as np import numpy as np
@@ -176,7 +175,7 @@ class EscPosPrinter(Printer):
return True return True
def _print_txt(self, msg, signature="", bold=False): def _print_txt(self, msg, signature="", bold=False):
self.ready = False
if not isinstance(msg, str): if not isinstance(msg, str):
self.app.logger.error( self.app.logger.error(
"It is not possible to print a " + str(type(msg)) + ", only strings." "It is not possible to print a " + str(type(msg)) + ", only strings."
@@ -225,9 +224,10 @@ class EscPosPrinter(Printer):
) from e ) from e
self.app.logger.info("Printed text") self.app.logger.info("Printed text")
return True self.ready = True
def _print_img(self, path, signature="", center=True, process=False): def _print_img(self, path, signature="", center=True, process=False):
self.ready = False
clean_signature = str(signature) clean_signature = str(signature)
if len(signature) > 256: if len(signature) > 256:
@@ -253,30 +253,33 @@ class EscPosPrinter(Printer):
if process: if process:
try: try:
self.app.logger.debug("Proccessing the image") self.app.logger.debug("Proccessing the image")
path = _process_image(self, path) processed_path = _process_image(self, path)
except RuntimeError as e: except RuntimeError as e:
self.app.logger.error( self.app.logger.error(
"Error while processing the image, aborting print : %s", str(e) "Error while processing the image, aborting print : %s", str(e)
) )
raise e raise e
else: else:
processed_path = path
self.app.logger.warning("Not proccessing the image") self.app.logger.warning("Not proccessing the image")
try: try:
self.printer.open(self.usb_args) 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.textln(signature)
self.printer.close() 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: except Exception as e:
self.app.logger.error(str(e)) self.app.logger.error(str(e))
raise RuntimeError("Could not print the picture") from e raise RuntimeError("Could not print the picture") from e
finally: finally:
try: try:
os.remove(path) os.remove(path)
os.remove(processed_path)
except OSError as e: except OSError as e:
raise e raise e
self.app.logger.debug("Removed image : " + str(processed_path))
self.app.logger.debug("Removed image : " + str(path)) self.app.logger.debug("Removed image : " + str(path))
try: try:
@@ -288,9 +291,10 @@ class EscPosPrinter(Printer):
raise RuntimeError("Could not close the printer connexion. ") from e raise RuntimeError("Could not close the printer connexion. ") from e
self.app.logger.info("Printed a picture") self.app.logger.info("Printed a picture")
return True self.ready = True
def _qr(self, content): def _qr(self, content):
self.ready = False
try: try:
self.printer.open(self.usb_args) self.printer.open(self.usb_args)
self.printer.qr(content, center=True) self.printer.qr(content, center=True)
@@ -302,8 +306,10 @@ class EscPosPrinter(Printer):
raise e raise e
self.app.logger.info("Printed a QR") self.app.logger.info("Printed a QR")
self.ready = True
def _cut(self): def _cut(self):
self.ready = False
try: try:
self.printer.open(self.usb_args) self.printer.open(self.usb_args)
self.printer.cut() self.printer.cut()
@@ -314,40 +320,50 @@ class EscPosPrinter(Printer):
raise e raise e
self.app.logger.info("Did a cut") self.app.logger.info("Did a cut")
return True self.ready = True
def _state(self): def _state(self):
self.app.logger.debug("Online : %s " , self.printer.is_online())
self.app.logger.debug("Has paper : %s " , self._has_paper())
self.app.logger.debug("Ready : %s " , self.ready)
return self.printer.is_online() and self.ready and self._has_paper() return self.printer.is_online() and self.ready and self._has_paper()
def print_task(self, task_type, data): def print_task(self, task_type, data):
"""Execute actual print based on task type""" """Execute actual print based on task type"""
with self._lock: with self._lock:
if self._state: self.app.logger.debug("Acquired lock to start print")
self._state = False
while not self._state():
self.app.logger.debug("Waiting for the printer to become ready..")
time.sleep(0.3)
self.app.logger.debug("Checked state to start printing %s", self._state())
self.ready = False
try: try:
self.app.logger.debug("Checking task type")
match (task_type.value): match (task_type.value):
case "text": case "text":
self._print_txt(data["txt"], signature=data["sign"]) self._print_txt(data["txt"], signature=data["sign"])
self._state = True self.ready = True
case "image": case "image":
self._print_img( self._print_img(
data["img"], signature=data["sign"], process=data["process"] data["img"], signature=data["sign"], process=data["process"]
) )
self._state = True self.ready = True
case "cut": case "cut":
self._cut() self._cut()
self._state = True self.ready = True
case "qr": case "qr":
self._qr(data["txt"]) self._qr(data["txt"])
self._state = True self.ready = True
case _: case _:
raise RuntimeError("This task type is not supported") raise RuntimeError("This task type is not supported")
except Exception as e: except Exception as e:
self._state = True self.app.logger.debug("Exception occured while printing %s", str(e))
self.ready = True
raise RuntimeError from e raise RuntimeError from e
raise RuntimeError("The printer is not ready to print yet !")
class BrotherPrinter(Printer): class BrotherPrinter(Printer):
""" """
@@ -519,14 +535,14 @@ class BrotherPrinter(Printer):
def _process_image(self, path): def _process_image(self, path):
brightness_factor = 1.5 # Used only if image is too dark brightness_factor = 1.5 # Used only if image is too dark
brightness_threshold = 100 # Brightness threshold (0255) brightness_threshold = 100 # Brightness threshold (0255)
# contrast_factor = 0.6 # Less than 1.0 = lower contrast contrast_factor = 2 # Less than 1.0 = lower contrast
max_width = 575 max_width = 575
max_height = 1000 max_height = 1000
with Image.open(path) as original_img: with Image.open(path) as original_img:
# Convert to RGB if needed (JPEG doesn't support alpha) # Convert to RGB if needed (JPEG doesn't support alpha)
if original_img.mode in ("RGBA", "P"): 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") original_img = original_img.convert("RGB")
# Resize while maintaining aspect ratio # Resize while maintaining aspect ratio
@@ -556,16 +572,19 @@ def _process_image(self, path):
self.app.logger.debug( self.app.logger.debug(
f"Image too dark, increasing brightness by a factor of {brightness_factor:.2f}" 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 # Convert to JPEG and save
jpeg_path = os.path.splitext(path)[0] + "_processed.jpg" 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.") self.app.logger.debug("Processed and saved image.")
return jpeg_path return jpeg_path

View File

@@ -44,7 +44,7 @@ class Web:
file_uploaded = self.upload_file(image) file_uploaded = self.upload_file(image)
except Exception as e: except Exception as e:
self.app.logger.error(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: if file_uploaded:
self.app.logger.debug("File has been uploaded, printing...") 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)) image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
except OSError as e: except OSError as e:
self.app.logger.error("Could not save file %s", 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( self.app.logger.debug(
"File saved to " "File saved to "
@@ -111,7 +111,7 @@ class Web:
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"
) )
return False raise RuntimeError("This file type is forbidden.")
def get_queue_state(self): def get_queue_state(self):
"""Return current queue state""" """Return current queue state"""