Restructure the code and implement a printing queue #29

Merged
n07070 merged 21 commits from restructure-printing-queue into master 2026-05-27 00:00:56 +02:00
3 changed files with 107 additions and 103 deletions
Showing only changes of commit 3f915a1b25 - Show all commits

View File

@@ -85,7 +85,7 @@ except PermissionError:
exit(77) exit(77)
# Output the config file # Output the config file
if os.getenv("LIPY_DEBUG") is True: if os.getenv("FLASK_DEBUG"):
pprint.pprint(configuration_file) pprint.pprint(configuration_file)
# We define the app module used by Flask # We define the app module used by Flask
@@ -154,7 +154,7 @@ def web_print_sms():
txt = request.form["txt"] txt = request.form["txt"]
except werkzeug.exceptions.BadRequestKeyError as e: except werkzeug.exceptions.BadRequestKeyError as e:
app.logger.error("Whoops, we are missing the txt input field. : %s ", str(e)) app.logger.error("Whoops, we are missing the txt input field. : %s ", str(e))
flash("Whoops, no forms submitted or missing signature : %s", str(e)) flash("Whoops, no forms submitted or missing signature : " + str(e), 'error')
return redirect(url_for("index")) return redirect(url_for("index"))
try: try:
@@ -170,15 +170,15 @@ def web_print_sms():
web.print_sms(txt, sign) web.print_sms(txt, sign)
except Exception as e: except Exception as e:
app.logger.error("Whoops, we could not print an SMS because : %s ", str(e)) app.logger.error("Whoops, we could not print an SMS because : %s ", str(e))
flash("Whoops, we could not print an SMS because : %s ", str(e)) flash("Whoops, we could not print an SMS because :" + str(e), 'error')
return redirect(url_for("index")) return redirect(url_for("index"))
# end try # end try
flash("The SMS has been printed !") flash("The SMS has been printed !", 'info')
return redirect(url_for("index")) return redirect(url_for("index"))
@app.route("/web/print/img") @app.route("/web/print/img", methods=["POST"])
@limiter.limit("1/second", override_defaults=False) @limiter.limit("1/second", override_defaults=False)
def web_print_img(): def web_print_img():
"""Prints an image on a printer""" """Prints an image on a printer"""
@@ -193,12 +193,11 @@ def web_print_img():
) )
sign = configuration_file["defaults"]["signature"] sign = configuration_file["defaults"]["signature"]
if request.method == "POST":
# check if the post request has the file part # check if the post request has the file part
if "img" not in request.files: if "img" not in request.files:
app.logger.error("Whoops, no images submitted : %s ", str(e)) app.logger.error("Whoops, no images submitted : %s ", str(e))
app.logger.error("Error getting the files : %s", str(e)) app.logger.error("Error getting the files : %s", str(e))
flash("Whoops, no images submitted : %s", str(e)) flash("Whoops, no images submitted : " + str(e), 'error')
return redirect(url_for("index")) return redirect(url_for("index"))
file = request.files["img"] file = request.files["img"]
@@ -206,7 +205,7 @@ def web_print_img():
# empty file without a filename. # empty file without a filename.
if file.filename == "": if file.filename == "":
app.logger.error("Submitted file has no filename !") app.logger.error("Submitted file has no filename !")
flash("Submitted file has no filename !") flash("Submitted file has no filename !", 'error')
return redirect(url_for("index")) return redirect(url_for("index"))
try: try:
@@ -214,15 +213,10 @@ def web_print_img():
web.print_image(file, sign) web.print_image(file, sign)
except Exception as e: except Exception as e:
app.logger.error("The image could not be printed because : %s ", str(e)) app.logger.error("The image could not be printed because : %s ", str(e))
flash("The image could not be printed because : %s ", str(e)) flash("The image could not be printed because : " + str(e), 'error')
return redirect(url_for("index")) return redirect(url_for("index"))
else: flash("Picture printed !", 'info')
app.logger.error("Method not allowed, please POST")
flash("Method not allowed, please POST")
return redirect(url_for("index"))
flash("Picture printed !")
return redirect(url_for("index")) return redirect(url_for("index"))
# API routes # API routes
@@ -246,7 +240,7 @@ def api_index():
@limiter.limit("6/minute", override_defaults=False) @limiter.limit("6/minute", override_defaults=False)
def api_print_sms(): def api_print_sms():
"""Prints a short message on a printer""" """Prints a short message on a printer"""
app.logger.debug("Printing an sms") app.logger.debug("Printing an sms via API")
try: try:
txt = request.form["txt"] txt = request.form["txt"]
except werkzeug.exceptions.BadRequestKeyError as e: except werkzeug.exceptions.BadRequestKeyError as e:
@@ -257,7 +251,7 @@ def api_print_sms():
sign = request.form["signature"] sign = request.form["signature"]
except werkzeug.exceptions.BadRequestKeyError as e: except werkzeug.exceptions.BadRequestKeyError as e:
app.logger.warning( 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"] sign = configuration_file["defaults"]["signature"]
try: try:
@@ -272,51 +266,38 @@ def api_print_sms():
@limiter.limit("6/minute", override_defaults=False) @limiter.limit("6/minute", override_defaults=False)
def api_print_image(): def api_print_image():
"""Prints an image on a printer""" """Prints an image on a printer"""
app.logger.debug("Printing an image") app.logger.debug("Printing an image via API")
try: try:
# comment: We try to get a signature # comment: We try to get a signature
sign = request.form["signature"] sign = request.form["signature"]
except werkzeug.exceptions.BadRequestKeyError as e: except werkzeug.exceptions.BadRequestKeyError as e:
app.logger.warning( 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"] sign = configuration_file["defaults"]["signature"]
if request.method == "POST":
# check if the post request has the file part # check if the post request has the file part
if "img" not in request.files: if "img" not in request.files:
app.logger.error("Whoops, no images submitted : %s ", str(e)) app.logger.error("Whoops, no images submitted.")
app.logger.error("Error getting the files : %s", str(e)) return "No image submitted", 400
flash("Whoops, no images submitted : %s", str(e))
return redirect(url_for("index"))
file = request.files["img"] file = request.files["img"]
# If the user does not select a file, the browser submits an # If the user does not select a file, the browser submits an
# empty file without a filename. # empty file without a filename.
if file.filename == "": if file.filename == "":
app.logger.error("Submitted file has no filename !") app.logger.error("Submitted file has no filename !")
flash("Submitted file has no filename !") return "Submitted file has no filename !", 400
return redirect(url_for("index"))
try: try:
app.logger.debug("Sending the image to the printer.") app.logger.debug("Sending the image to the printer.")
web.print_image(file, sign) web.print_image(file, sign)
except Exception as e: except Exception as e:
app.logger.error("The image could not be printed because : %s ", str(e)) return str(e), 500
flash("The image could not be printed because : %s ", str(e))
return redirect(url_for("index"))
else: return "OK", 200
app.logger.error("Method not allowed")
flash("Method not allowed")
return redirect(url_for("index"))
flash("Picture printed ! ") @app.route("/api/camera/picture", methods=["GET"])
return redirect(url_for("index"))
@app.route("/api/camera/picture")
def camera_picture(): def camera_picture():
"""Returns a picture taken by the camera""" """Returns a picture taken by the camera"""
if RASPBERRY_PI_CONNECTED: if RASPBERRY_PI_CONNECTED:

View File

@@ -1,6 +1,6 @@
# Importing the module to manage the connection to the printer. # Importing the module to manage the connection to the printer.
import escpos.printer as escp import escpos.printer
import brother-ql-inventree import brother_ql
from time import sleep, gmtime, strftime from time import sleep, gmtime, strftime
import os.path import os.path
from PIL import Image, ImageEnhance, ImageOps from PIL import Image, ImageEnhance, ImageOps
@@ -47,7 +47,7 @@ class Printer(object):
case 0: case 0:
self.app.logger.error("Printer has no more paper, aborting...") self.app.logger.error("Printer has no more paper, aborting...")
self.printer.close() self.printer.close()
raise Exception("No more paper in the printer") raise RuntimeError("No more paper in the printer")
case 1: case 1:
self.app.logger.warning( self.app.logger.warning(
"Printer needs paper to be changed very soon ! " "Printer needs paper to be changed very soon ! "
@@ -60,17 +60,21 @@ class Printer(object):
def init_printer(self): def init_printer(self):
# Is the printer online ? Is the communication with the printer successfull ? # Is the printer online ? Is the communication with the printer successfull ?
waiting_elapsed = 30 if os.getenv("FLASK_DEBUG"):
waiting_elapsed = 1
else:
waiting_elapsed = 10
self.app.logger.debug("Waiting for printer to get online...") self.app.logger.debug("Waiting for printer to get online...")
while not self.ready: while not self.ready:
try: try:
# This also calls open(), which we need to close() # This also calls open(), which we need to close()
# or else the device will appear as busy. # or else the device will appear as busy.
p = escp.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: except Exception as e:
self.app.logger.error( self.app.logger.error(
"The USB device is not plugged in, trying again : " + str(e) "The USB device is not plugged in, trying again %s : %s",waiting_elapsed, str(e)
) )
pass pass
@@ -79,6 +83,8 @@ class Printer(object):
self.ready = True self.ready = True
self.app.logger.debug("Printer online !") self.app.logger.debug("Printer online !")
except Exception as e: except Exception as e:
self.app.logger.error("Error while getting the printer online %s : %s",waiting_elapsed, str(e)
)
pass pass
sleep(1) sleep(1)
@@ -149,6 +155,7 @@ class Printer(object):
self.printer.close() self.printer.close()
except Exception as e: except Exception as e:
self.app.logger.error("Unable to print because : " + str(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
self.app.logger.info("Printed text") self.app.logger.info("Printed text")
return True return True
@@ -194,9 +201,14 @@ class Printer(object):
os.remove(path) os.remove(path)
self.app.logger.debug("Removed image : " + str(path)) self.app.logger.debug("Removed image : " + str(path))
except Exception as e: except Exception as e:
self.printer.close()
self.app.logger.error(str(e)) self.app.logger.error(str(e))
return False raise RuntimeError("Could not print the picture") from e
finally:
try:
self.printer.close()
except Exception as e:
self.app.logger.error(str(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 return True
@@ -285,6 +297,7 @@ def process_image(self, path):
# 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) original_img.save(jpeg_path, format="JPEG", quality=95, optimize=True)
app.logger.debug("Processed and saved image.")
return jpeg_path return jpeg_path

View File

@@ -6,31 +6,41 @@ import os
class Web(object): class Web(object):
"""docstring for 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"""
def __init__(self, app, printer): def __init__(self, app, printer):
super(Web).__init__() super(Web).__init__()
self.printer = printer self.printer = printer
self.app = app self.app = app
def print_sms(self, texte, sign: str): def print_sms(self, texte, sign: str) -> bool:
# TODO: verify the texte before printing it here ? """
Get text and a signature, prints the text and cuts after that.
"""
self.app.logger.debug("Printing : " + str(texte) + " from " + str(sign)) self.app.logger.debug("Printing : " + str(texte) + " from " + str(sign))
try: try:
self.printer.print_sms(texte, sign) self.printer.print_sms(texte, sign)
self.printer.cut() self.printer.cut()
except Exception as e: except Exception as e:
self.app.logger.error(e) self.app.logger.error(e)
flash("Error while printing the SMS : " + str(e)) raise RuntimeError("Could not print SMS, " + str(e)) from e
flash("You message " + str(texte) + " has been printed :)") return True
def print_image(self, image, sign): def print_image(self, image, sign: str) -> bool:
self.app.logger.debug("Uploading file") """
Get an image and a signature, prints the image and cuts after that.
"""
try: try:
self.app.logger.debug("Uploading file from " + str(sign)) self.upload_file(image)
if self.upload_file(image): except Exception as e:
self.app.logger.error(e)
raise RuntimeError("Could not upload file") from e
self.app.logger.debug("File has been uploaded, printing...") self.app.logger.debug("File has been uploaded, printing...")
try:
self.printer.print_img( self.printer.print_img(
os.path.join( os.path.join(
self.app.config["UPLOAD_FOLDER"], self.app.config["UPLOAD_FOLDER"],
@@ -41,16 +51,18 @@ class Web(object):
) )
self.printer.cut() self.printer.cut()
except Exception as e: except Exception as e:
self.app.logger.error(e) raise RuntimeError("Could not print file") from e
flash("Could not upload file." + str(e))
flash("Your image has been printed :)") self.app.logger.debug("Image printed and cut !")
return True
def login(username: str, password: str) -> bool: def login(self, username: str, password: str) -> bool:
pass """Not implemented"""
return
def logout(username: str, password: str) -> bool: def logout(self, username: str, password: str) -> bool:
pass """Not implemented"""
return
def allowed_file(self, filename) -> bool: def allowed_file(self, filename) -> bool:
self.app.logger.debug("Is the filename allowed ?") self.app.logger.debug("Is the filename allowed ?")
@@ -67,10 +79,8 @@ class Web(object):
self.app.logger.debug("File valid") self.app.logger.debug("File valid")
try: try:
image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename)) image.save(os.path.join(self.app.config["UPLOAD_FOLDER"], filename))
self.app.logger.debug("File saved")
except Exception as e: except Exception as e:
self.app.logger.error("Could not save file") self.app.logger.error("Could not save file %s", e)
flash(str(e), "error")
return False return False
self.app.logger.debug( self.app.logger.debug(