Compare commits
6 Commits
175dd3385a
...
multi-prin
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
09e588c3ff | ||
|
|
0699775d35 | ||
|
|
7d19098b61 | ||
|
|
65e4a2ad9c | ||
|
|
af15ed8754 | ||
|
|
53010987f4 |
@@ -90,7 +90,7 @@ if not os.getenv("FLASK_DEBUG") is None and os.getenv("FLASK_DEBUG") is True:
|
|||||||
app.secret_key = configuration_file["secrets"]["flask_secret_key"]
|
app.secret_key = configuration_file["secrets"]["flask_secret_key"]
|
||||||
app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER
|
app.config["UPLOAD_FOLDER"] = UPLOAD_FOLDER
|
||||||
app.config["ALLOWED_EXTENSIONS"] = ALLOWED_EXTENSIONS
|
app.config["ALLOWED_EXTENSIONS"] = ALLOWED_EXTENSIONS
|
||||||
app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000 # Maximum 3Mb for a file upload
|
app.config["MAX_CONTENT_LENGTH"] = 10 * 1000 * 1000 # Maximum 10Mb for a file upload
|
||||||
app.config["TEMPLATES_AUTO_RELOAD"] = True
|
app.config["TEMPLATES_AUTO_RELOAD"] = True
|
||||||
|
|
||||||
# Queue creation
|
# Queue creation
|
||||||
@@ -123,6 +123,8 @@ limiter = Limiter(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
app.logger.info("🖶 Welcome to LittlePrynter !")
|
||||||
|
|
||||||
# General routes
|
# General routes
|
||||||
@app.route("/")
|
@app.route("/")
|
||||||
@limiter.limit("1/second", override_defaults=False)
|
@limiter.limit("1/second", override_defaults=False)
|
||||||
@@ -189,6 +191,9 @@ def web_print_img():
|
|||||||
"No signature found for this print, using default signature : %s", str(e)
|
"No signature found for this print, using default signature : %s", str(e)
|
||||||
)
|
)
|
||||||
sign = configuration_file["defaults"]["signature"]
|
sign = configuration_file["defaults"]["signature"]
|
||||||
|
except werkzeug.exceptions.RequestEntityTooLarge as e:
|
||||||
|
flash("Whoops, image is too big: " + str(e), "error")
|
||||||
|
return redirect(url_for("index"))
|
||||||
|
|
||||||
# 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:
|
||||||
|
|||||||
114
src/printer.py
114
src/printer.py
@@ -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,39 +320,56 @@ 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) -> bool:
|
||||||
return self.printer.is_online() and self.ready and self._has_paper()
|
has_paper = self._has_paper()
|
||||||
|
is_ready = self.ready
|
||||||
|
|
||||||
|
self.app.logger.debug("Has paper : %s " , has_paper )
|
||||||
|
self.app.logger.debug("Ready : %s " , is_ready )
|
||||||
|
|
||||||
|
return is_ready and has_paper # and is_online
|
||||||
|
|
||||||
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:
|
|
||||||
if self._state:
|
|
||||||
self._state = False
|
|
||||||
try:
|
|
||||||
match (task_type.value):
|
|
||||||
case "text":
|
|
||||||
self._print_txt(data["txt"], signature=data["sign"])
|
|
||||||
self._state = True
|
|
||||||
case "image":
|
|
||||||
self._print_img(
|
|
||||||
data["img"], signature=data["sign"], process=data["process"]
|
|
||||||
)
|
|
||||||
self._state = True
|
|
||||||
case "cut":
|
|
||||||
self._cut()
|
|
||||||
self._state = True
|
|
||||||
case "qr":
|
|
||||||
self._qr(data["txt"])
|
|
||||||
self._state = True
|
|
||||||
case _:
|
|
||||||
raise RuntimeError("This task type is not supported")
|
|
||||||
except Exception as e:
|
|
||||||
self._state = True
|
|
||||||
raise RuntimeError from e
|
|
||||||
|
|
||||||
raise RuntimeError("The printer is not ready to print yet !")
|
|
||||||
|
with self._lock:
|
||||||
|
self.app.logger.debug("Acquired lock to start print")
|
||||||
|
|
||||||
|
i_m_ready = self._state()
|
||||||
|
while not i_m_ready:
|
||||||
|
self.app.logger.debug("Waiting for the printer to become ready, current state %s ", str(i_m_ready))
|
||||||
|
i_m_ready = self._state()
|
||||||
|
time.sleep(0.3)
|
||||||
|
|
||||||
|
self.app.logger.debug("Checked state to start printing : %s", self._state())
|
||||||
|
self.ready = False
|
||||||
|
try:
|
||||||
|
self.app.logger.debug("Checking task type")
|
||||||
|
match (task_type.value):
|
||||||
|
case "text":
|
||||||
|
|
||||||
|
self._print_txt(data["txt"], signature=data["sign"])
|
||||||
|
self.ready = True
|
||||||
|
case "image":
|
||||||
|
self._print_img(
|
||||||
|
data["img"], signature=data["sign"], process=data["process"]
|
||||||
|
)
|
||||||
|
self.ready = True
|
||||||
|
case "cut":
|
||||||
|
self._cut()
|
||||||
|
self.ready = True
|
||||||
|
case "qr":
|
||||||
|
self._qr(data["txt"])
|
||||||
|
self.ready = True
|
||||||
|
case _:
|
||||||
|
raise RuntimeError("This task type is not supported")
|
||||||
|
except Exception as e:
|
||||||
|
self.app.logger.debug("Exception occured while printing %s", str(e))
|
||||||
|
self.ready = True
|
||||||
|
raise RuntimeError from e
|
||||||
|
|
||||||
|
|
||||||
class BrotherPrinter(Printer):
|
class BrotherPrinter(Printer):
|
||||||
@@ -519,14 +542,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 (0–255)
|
brightness_threshold = 100 # Brightness threshold (0–255)
|
||||||
# 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 +579,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
|
||||||
|
|||||||
@@ -111,6 +111,11 @@ class Printers:
|
|||||||
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))
|
||||||
|
|
||||||
|
if len(printers) < 1:
|
||||||
|
self.app.logger.warning("Not printers found ! Please plug in a Printer and restart the program.")
|
||||||
|
raise RuntimeError("No printers found")
|
||||||
|
|
||||||
return printers
|
return printers
|
||||||
|
|
||||||
def any(self) -> Printer:
|
def any(self) -> Printer:
|
||||||
|
|||||||
@@ -68,7 +68,7 @@ class Raspberry:
|
|||||||
self.app.logger.warning(
|
self.app.logger.warning(
|
||||||
"Couldn't get sufficient hardware information from /proc/cpuinfo"
|
"Couldn't get sufficient hardware information from /proc/cpuinfo"
|
||||||
)
|
)
|
||||||
self.app.logger.error("Unable to determine if we are on a Raspberry Pi.")
|
self.app.logger.warning("Unable to determine if we are on a Raspberry Pi.")
|
||||||
return False
|
return False
|
||||||
except IOError:
|
except IOError:
|
||||||
self.app.logger.error("Unable to open `/proc/cpuinfo`.")
|
self.app.logger.error("Unable to open `/proc/cpuinfo`.")
|
||||||
|
|||||||
@@ -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"""
|
||||||
|
|||||||
@@ -64,7 +64,7 @@ class PrintWorker(threading.Thread):
|
|||||||
"Ready" if self.printer.ready else "Not ready",
|
"Ready" if self.printer.ready else "Not ready",
|
||||||
)
|
)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.app.logger.error(str(e))
|
self.app.logger.error("No printer detected" + str(e))
|
||||||
self.printer = None
|
self.printer = None
|
||||||
|
|
||||||
if self.state != "idle":
|
if self.state != "idle":
|
||||||
|
|||||||
Reference in New Issue
Block a user