Restructure main class to activate worker and use tasks, print queue,
update Printer
This commit is contained in:
n07070
@@ -1,12 +1,11 @@
|
||||
# Importing the module to manage the connection to the printer.
|
||||
import escpos.printer
|
||||
import brother_ql
|
||||
# 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):
|
||||
"""
|
||||
# The connection is based on the ESC/POS library
|
||||
@@ -39,7 +38,9 @@ class Printer(object):
|
||||
self.usb_args["idProduct"] = self.vendor_id
|
||||
|
||||
def check_paper(self) -> bool:
|
||||
# Let's check paper status
|
||||
"""
|
||||
On printers that support it, we check that the printer has paper
|
||||
"""
|
||||
self.app.logger.debug("Checking paper status...")
|
||||
self.printer.open(self.usb_args)
|
||||
status = self.printer.paper_status()
|
||||
@@ -58,10 +59,13 @@ class Printer(object):
|
||||
self.printer.close()
|
||||
|
||||
def init_printer(self):
|
||||
"""
|
||||
Check if the printer online ? Is the communication with the printer successfull ?
|
||||
"""
|
||||
|
||||
# Is the printer online ? Is the communication with the printer successfull ?
|
||||
# TODO: This could happen directly when creating a new Printer class
|
||||
if os.getenv("FLASK_DEBUG"):
|
||||
waiting_elapsed = 1
|
||||
waiting_elapsed = 15
|
||||
else:
|
||||
waiting_elapsed = 10
|
||||
|
||||
@@ -122,10 +126,17 @@ class Printer(object):
|
||||
|
||||
return True
|
||||
|
||||
def print_sms(self, msg, signature="", bold=False):
|
||||
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.")
|
||||
raise ValueError
|
||||
|
||||
# We make sure that the signature is not something too goofy
|
||||
clean_msg = str(msg) + "\n"
|
||||
clean_signature = str(signature)
|
||||
|
||||
# Make checks on the size of the message being printed
|
||||
if len(clean_msg) > 4096:
|
||||
self.app.logger.warning(
|
||||
"Could not print message of this length: " + str(len(clean_msg))
|
||||
@@ -146,6 +157,8 @@ class Printer(object):
|
||||
+ ", needs to be below 256 caracters long."
|
||||
)
|
||||
|
||||
# Do the actual printing
|
||||
# We would pop the next element in the queue here, if it's a sms type
|
||||
try:
|
||||
self.printer.open(self.usb_args)
|
||||
self.printer.set(align="center", font="a", bold=bold)
|
||||
@@ -160,14 +173,14 @@ class Printer(object):
|
||||
self.app.logger.info("Printed text")
|
||||
return True
|
||||
|
||||
def print_img(self, path, sign="", center=True, process=False):
|
||||
clean_signature = str(sign)
|
||||
def _print_img(self, path, signature="", center=True, process=False):
|
||||
clean_signature = str(signature)
|
||||
|
||||
if len(sign) > 256:
|
||||
if len(signature) > 256:
|
||||
self.app.logger.warning(
|
||||
"Could not print signature of this length: " + str(len(clean_signature))
|
||||
)
|
||||
raise Exception(
|
||||
raise ValueError(
|
||||
"Could not print signature of this length :"
|
||||
+ str(len(clean_signature))
|
||||
+ ", needs to be below 256 caracters long."
|
||||
@@ -175,7 +188,7 @@ class Printer(object):
|
||||
|
||||
if not os.path.isfile(str(path)):
|
||||
self.app.logger.warning("File does not exist : " + str(path))
|
||||
raise Exception(
|
||||
raise OSError(
|
||||
"The file path for this image :"
|
||||
+ str(path)
|
||||
+ " wasn't found. Please try again."
|
||||
@@ -186,34 +199,40 @@ class Printer(object):
|
||||
if process:
|
||||
try:
|
||||
self.app.logger.debug("Proccessing the image")
|
||||
path = process_image(self, path)
|
||||
except Exception as e:
|
||||
self.app.logger.error(str(e))
|
||||
return False
|
||||
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:
|
||||
self.app.logger.warning("Not proccessing the image")
|
||||
|
||||
try:
|
||||
self.printer.open(self.usb_args)
|
||||
self.printer.image(path, center=center)
|
||||
self.printer.textln(signature)
|
||||
self.printer.close()
|
||||
self.app.logger.debug("Printed an image : " + str(path))
|
||||
os.remove(path)
|
||||
self.app.logger.debug("Removed image : " + str(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)
|
||||
except OSError as e:
|
||||
raise e
|
||||
|
||||
self.app.logger.debug("Removed image : " + str(path))
|
||||
|
||||
try:
|
||||
self.printer.close()
|
||||
except Exception as e:
|
||||
self.app.logger.error(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")
|
||||
return True
|
||||
|
||||
def qr(self, content):
|
||||
def _qr(self, content):
|
||||
try:
|
||||
self.printer.open(self.usb_args)
|
||||
self.printer.qr(content, center=True)
|
||||
@@ -226,7 +245,7 @@ class Printer(object):
|
||||
self.app.logger.info("Printed a QR")
|
||||
return True
|
||||
|
||||
def cut(self):
|
||||
def _cut(self):
|
||||
try:
|
||||
self.printer.open(self.usb_args)
|
||||
self.printer.cut()
|
||||
@@ -239,8 +258,20 @@ class Printer(object):
|
||||
self.app.logger.info("Did a cut")
|
||||
return True
|
||||
|
||||
def print_task(self, task_type, data):
|
||||
"""Execute actual print based on task type"""
|
||||
match (task_type.value):
|
||||
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"):
|
||||
self._cut()
|
||||
case _:
|
||||
raise RuntimeError("This task type is not supported")
|
||||
|
||||
def process_image(self, path):
|
||||
|
||||
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
|
||||
@@ -254,7 +285,7 @@ def process_image(self, path):
|
||||
original_img = original_img.convert("RGB")
|
||||
|
||||
# Resize while maintaining aspect ratio
|
||||
original_img.thumbnail((max_width, max_height), Image.LANCZOS)
|
||||
original_img.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)
|
||||
self.app.logger.debug("Resized the image")
|
||||
|
||||
# # Convert to grayscale for dithering
|
||||
@@ -286,18 +317,10 @@ def process_image(self, path):
|
||||
# contrast_enhancer = ImageEnhance.Contrast(original_img)
|
||||
# original_img = contrast_enhancer.enhance(contrast_factor)
|
||||
|
||||
# Final resize check
|
||||
if original_img.height > max_height:
|
||||
raise ValueError("Image is too long, sorry! Keep it below 575×1000 pixels.")
|
||||
self.app.logger.error(
|
||||
"Image is too long, sorry! Keep it below 575×1000 pixels."
|
||||
)
|
||||
return False
|
||||
|
||||
# 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)
|
||||
app.logger.debug("Processed and saved image.")
|
||||
self.app.logger.debug("Processed and saved image.")
|
||||
|
||||
return jpeg_path
|
||||
|
||||
@@ -387,6 +410,5 @@ def find_and_parse_borther_ql_printer():
|
||||
print("No Brother QL printer found")
|
||||
return None
|
||||
|
||||
|
||||
def fint_and_parse_epson_printer():
|
||||
pass
|
||||
Reference in New Issue
Block a user