n07070/littleprynter
2
0
Fork 0
Code Issues 21 Pull Requests 1 Projects Releases Activity

Add support for signature on brother printers

Browse Source
This commit is contained in:
n07070
2026-11-13 02:45:11 +01:00
parent 0a6431eecf
commit e2eb3d3825
5 changed files with 141 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/main.py
View File

@@ -1,7 +1,8 @@
"""
Welcome to the LittlePrynter's source code.
This program expose a web interface, with user authentification,
This program expose a web interface,
that makes it possible to print messages from the web.
It also exposes a API, making it possible to print and interface
with much of the printer's abilities.
@@ -10,12 +11,14 @@ Then we build the API around Flask,
Then we build the web interface, using the simple Jinja2 templating.
We support two modes :
The first is a simple mode, where a computer, connected to a thermal printer,
runs this program and exposes a web interface that makes use of the client's camera
The seconde is booth mode, where a Raspberry Pi is connected to a thermal printer,
a button and a flash.
The web interface exists but may not be used, as the press of the button with take
a picture and activate the flash while simply informing the web page.
The seconde is booth mode, where a Raspberry Pi is connected to
a thermal printer, a button and a flash.
The web interface exists but it's not needed, as the press of the button
takes a picture and activates the flash while simply informing the web page.
"""
# Following are the librairies we import,
@@ -98,11 +101,7 @@ app.config["TEMPLATES_AUTO_RELOAD"] = True
print_queue = PrintQueue(app)
# Find out if we are running on a Raspberry Pi
rpi = Raspberry(
print_queue,
app,
configuration_file
)
rpi = Raspberry(print_queue, app, configuration_file)
RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
@@ -134,6 +133,7 @@ limiter = Limiter(
app.logger.info("🖶 Welcome to LittlePrynter !")
# General routes
@app.route("/")
@limiter.limit("1/second", override_defaults=False)
@@ -233,7 +233,6 @@ def web_print_img():
except werkzeug.exceptions.BadRequestKeyError:
process_image = False
try:
app.logger.debug("Sending the image to the printer.")
web.print_image(file, sign, do_cut, process_image)
@@ -287,7 +286,6 @@ def api_print_sms():
except werkzeug.exceptions.BadRequestKeyError:
do_cut = False
try:
# comment: We try to print the SMS
web.print_sms(txt, sign, do_cut)
@@ -390,11 +388,13 @@ def api_worker_stop():
worker.stop_worker()
return jsonify(worker.current_state())
@app.route("/api/printers")
def api_printers_info():
"""Returns the current state of every printers"""
return jsonify(printers_pool.printers_info())
## Authentification

7
src/print_queue.py
View File

@@ -77,7 +77,12 @@ class PrintQueue:
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), "content": str(t.get_print_data())}
{
"task_id": t.task_id,
"status": t.status,
"type": str(t.task_type),
"content": str(t.get_print_data()),
}
for t in self._queue
]

150
src/printer.py
View File

@@ -16,7 +16,6 @@ from uuid import uuid4
from PIL import Image, ImageEnhance, ImageDraw, ImageFont
import numpy as np
# Importing the modules needed for each supported printer Type
import escpos.printer
from brother_ql.models import ModelsManager
@@ -42,6 +41,7 @@ class PrinterInfo:
"""
Brother-QL printer information
"""
identifier: str
backend: str
protocol: str
@@ -94,13 +94,14 @@ class Printer(ABC):
"""Takes a PrintTask and executes it"""
def use_printer(self) -> None:
"""Marks the printer as being used """
"""Marks the printer as being used"""
self.used = True
def free_printer(self) -> None:
"""Marks the printer as being not used, it's available."""
self.used = False
class EscPosPrinter(Printer):
"""
Create a new ESC/POS based printer.
@@ -128,8 +129,10 @@ class EscPosPrinter(Printer):
"The USB device is not plugged in : %s",
str(e),
)
raise e
except Exception as e:
self.app.logger.error("Printer could not be connected : %s ", str(e))
raise e
try:
if p.is_online():
@@ -187,7 +190,6 @@ class EscPosPrinter(Printer):
return True
def _print_txt(self, msg, signature="", bold=False):
self.ready = False
if not isinstance(msg, str):
self.app.logger.error(
"It is not possible to print a " + str(type(msg)) + ", only strings."
@@ -199,6 +201,7 @@ class EscPosPrinter(Printer):
clean_signature = str(signature)
# Make checks on the size of the message being printed
# TODO: make the max length configurable
if len(clean_msg) > 4096:
self.app.logger.warning(
"Could not print message of this length: " + str(len(clean_msg))
@@ -208,7 +211,7 @@ class EscPosPrinter(Printer):
+ str(len(clean_msg))
+ ", needs to be below 4096 caracters long."
)
# TODO: make the max length configurable
if len(signature) > 256:
self.app.logger.warning(
"Could not print signature of this length: " + str(len(clean_signature))
@@ -239,9 +242,9 @@ class EscPosPrinter(Printer):
self.ready = True
def _print_img(self, path, signature="", center=True, process=False):
self.ready = False
clean_signature = str(signature)
# TODO: make the max length configurable
if len(signature) > 256:
self.app.logger.warning(
"Could not print signature of this length: " + str(len(clean_signature))
@@ -306,7 +309,6 @@ class EscPosPrinter(Printer):
self.ready = True
def _qr(self, content):
self.ready = False
try:
self.printer.open(self.usb_args)
self.printer.qr(content, center=True)
@@ -321,7 +323,6 @@ class EscPosPrinter(Printer):
self.ready = True
def _cut(self):
self.ready = False
try:
self.printer.open(self.usb_args)
self.printer.cut()
@@ -338,21 +339,23 @@ class EscPosPrinter(Printer):
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 )
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
return is_ready and has_paper # and is_online
def print_task(self, task_type, data):
"""Execute actual print based on task type"""
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))
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)
@@ -391,9 +394,7 @@ class BrotherPrinter(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.vendor_id
@@ -431,7 +432,7 @@ class BrotherPrinter(Printer):
self.app.logger.debug(f"Matched printer model: {model}")
except ValueError as ve:
self.app.logger.warning("Invalid product ID format:"+ str(ve))
self.app.logger.warning("Invalid product ID format:" + str(ve))
raise ve
except Exception as e:
self.app.logger.error("Error while finding a model ? : " + str(e))
@@ -459,13 +460,12 @@ class BrotherPrinter(Printer):
def _print_txt(self, msg, signature="", bold=False):
"""Print text after converting it to an image"""
self._print_img(text_to_image(msg))
self._print_img(text_to_image(signature))
def _print_img(self, data, process=False):
def _print_img(self, data, signature="", process=False):
"""
Print a raster image via a Brother QL printer
"""
self.ready = False
if process:
try:
self.app.logger.debug("Proccessing the image")
@@ -480,10 +480,13 @@ class BrotherPrinter(Printer):
self.app.logger.warning("Not proccessing the image")
try:
selected_backend = guess_backend(self.printer_info.identifier)
BACKEND_CLASS = backend_factory(selected_backend)['backend_class']
BACKEND_CLASS = backend_factory(selected_backend)["backend_class"]
BACKEND_STRING_DESCR = self.printer_info.identifier
except Exception:
self.app.logger.error("Couldn't guess the backend to use from the printer string descriptor")
except Exception as e:
self.app.logger.error(
"Couldn't guess the backend to use from the printer string descriptor"
)
raise e
try:
im = Image.open(processed_path)
@@ -498,20 +501,40 @@ class BrotherPrinter(Printer):
except Exception as e:
raise e
if signature != "":
try:
im = Image.open(text_to_image(signature))
qlr = BrotherQLRaster(self.printer_info.model)
create_label(
qlr, im, self.printer_info.label_type, cut=False, dither=True
)
# noinspection PyCallingNonCallable
be = BACKEND_CLASS(BACKEND_STRING_DESCR)
be.write(qlr.data)
be.dispose()
del be
except Exception as e:
raise e
def _cut(self):
self.ready = False
try:
selected_backend = guess_backend(self.printer_info.identifier)
BACKEND_CLASS = backend_factory(selected_backend)['backend_class']
BACKEND_CLASS = backend_factory(selected_backend)["backend_class"]
BACKEND_STRING_DESCR = self.printer_info.identifier
except Exception as e:
self.app.logger.error("Couldn't guess the backend to use from the printer string descriptor")
self.app.logger.error(
"Couldn't guess the backend to use from the printer string descriptor"
)
raise e
data = "src/static/images/1x1.png"
try:
im = Image.open(data)
qlr = BrotherQLRaster(self.printer_info.model)
create_label(qlr, im, self.printer_info.label_type, threshold=70, cut=True, rotate=0)
create_label(
qlr, im, self.printer_info.label_type, threshold=70, cut=True, rotate=0
)
# noinspection PyCallingNonCallable
be = BACKEND_CLASS(BACKEND_STRING_DESCR)
@@ -521,32 +544,41 @@ class BrotherPrinter(Printer):
except Exception as e:
raise e
def _qr(self):
# This should not be reached by a print on a BrotherQL
raise NotImplementedError
def print_task(self, task_type, data):
"""Execute actual print based on task type"""
with self._lock:
if self.ready:
try:
match (task_type.value):
case "image":
self._print_img(data["img"])
self.ready = True
case "cut":
self._cut()
self.ready = True
case "text":
self._print_txt(data["txt"], signature=data["sign"])
self.ready = True
case _:
raise RuntimeError("This task type is not supported")
except Exception as e:
raise e
# end try
else:
raise RuntimeError("The printer is not ready to print yet !")
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)
# These values are by default for now
self.app.logger.debug("Checked state to start printing : %s", self._state())
self.ready = False
# raise NotImplementedError("This printer type is not implemented yet")
try:
match (task_type.value):
case "image":
self._print_img(data["img"], signature=data["sign"])
self.ready = True
case "cut":
self._cut()
self.ready = True
case "text":
self._print_txt(data["txt"], signature=data["sign"])
self.ready = True
case _:
raise RuntimeError("This task type is not supported")
except Exception as e:
raise e
def _process_image(self, path, max_width=575, max_height=1000):
@@ -563,15 +595,12 @@ def _process_image(self, path, max_width=575, max_height=1000):
grayscale = img.convert("L")
avg_brightness = np.asarray(grayscale, dtype=np.float32).mean()
self.app.logger.debug(
"Average brightness of the image: %.2f", avg_brightness
)
self.app.logger.debug("Average brightness of the image: %.2f", avg_brightness)
if avg_brightness < brightness_threshold:
brightness_factor = min(
max(
1 + (brightness_threshold - avg_brightness)
/ brightness_threshold,
1 + (brightness_threshold - avg_brightness) / brightness_threshold,
1.1,
),
2.5,
@@ -580,9 +609,7 @@ def _process_image(self, path, max_width=575, max_height=1000):
"Image is dark; increasing brightness by %.2f",
brightness_factor,
)
grayscale = ImageEnhance.Brightness(grayscale).enhance(
brightness_factor
)
grayscale = ImageEnhance.Brightness(grayscale).enhance(brightness_factor)
contrast = float(np.asarray(grayscale, dtype=np.float32).std())
self.app.logger.debug("Grayscale standard deviation: %.2f", contrast)
@@ -596,6 +623,7 @@ def _process_image(self, path, max_width=575, max_height=1000):
self.app.logger.debug("Processed and saved image: %s", jpeg_path)
return jpeg_path
def text_to_image(
text,
width=400,
@@ -639,7 +667,11 @@ def text_to_image(
line = ""
piece = ""
for char in word:
if piece and draw.textbbox((0, 0), piece + char, font=font)[2] > max_text_width:
if (
piece
and draw.textbbox((0, 0), piece + char, font=font)[2]
> max_text_width
):
lines.append(piece)
piece = char
else:
@@ -654,11 +686,7 @@ def text_to_image(
ascent, descent = font.getmetrics()
line_height = ascent + descent
height = (
2 * padding
+ len(lines) * line_height
+ (len(lines) - 1) * line_spacing
)
height = 2 * padding + len(lines) * line_height + (len(lines) - 1) * line_spacing
image = Image.new("RGB", (width, height), background)
draw = ImageDraw.Draw(image)

37
src/printers.py
View File

@@ -49,10 +49,7 @@ class Printers:
# Find all connected USB devices
# the find() fnc returns a generator
devices = list(
usb.core.find(
find_all=True,
custom_match=_FindClass(self.app,7)
)
usb.core.find(find_all=True, custom_match=_FindClass(self.app, 7))
)
if devices:
@@ -60,15 +57,15 @@ class Printers:
if waiting_time_for_printer <= 0:
self.app.logger.warning(
"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."
"No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
)
self.app.logger.warning(
"No printers found; waiting for a printer to connect..."
)
)
time.sleep(1)
waiting_time_for_printer -= 1
@@ -120,10 +117,12 @@ class Printers:
self.app.logger.debug("Trying to create a new BROTHER printer")
brother_printer = BrotherPrinter(
self.app, vendor_id=dev.idVendor, device_id=dev.idProduct)
self.app, vendor_id=dev.idVendor, device_id=dev.idProduct
)
except Exception as e:
self.app.logger.error(
"Could not create a %s printer class with %s:%s" , product,
"Could not create a %s printer class with %s:%s",
product,
dev.idVendor,
dev.idProduct,
)
@@ -131,12 +130,14 @@ class Printers:
# 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))
if len(printers) < 1:
self.app.logger.warning("Not printers found ! Please plug in a Printer and restart the program.")
self.app.logger.warning(
"Not printers found ! Please plug in a Printer and restart the program."
)
raise RuntimeError("No printers found")
return printers
@@ -163,26 +164,32 @@ class Printers:
"""Return the current list of printers"""
return self.printers
class _FindClass:
"""
Use by usb.core to modify the way USB devices are found
Taken from pyUSB documentation on Github
"""
def __init__(self, app, class_):
self._class = class_
self.app = app
def __call__(self, device):
# first, let's check the device
self.app.logger.debug("Found a USB device of class %s", str(device.bDeviceClass))
self.app.logger.debug(
"Found a USB device of class %s", str(device.bDeviceClass)
)
if device.bDeviceClass == self._class:
return True
# ok, transverse all devices to find an
# interface that matches our class
for configuration in device:
# Every device can have one or more configuration,
interface = usb.util.find_descriptor(configuration, bInterfaceClass=self._class)
# each config can have several interfaces
interface = usb.util.find_descriptor(
configuration, bInterfaceClass=self._class
)
# each config can have several interfaces
if interface is not None:
return True

21
src/worker.py
View File

@@ -20,6 +20,7 @@ class PrintWorker(threading.Thread):
If a print fails, it's not retried, but will be added to a list of completed
tasks. TODO : Change that
"""
def __init__(self, app, print_queue, printers_pool):
super().__init__(daemon=True)
self.app = app
@@ -51,12 +52,11 @@ class PrintWorker(threading.Thread):
self._printer_index = (index + 1) % len(self.printers_obj)
return printer
def run(self):
"""Background thread that processes queue items"""
self.app.logger.debug("Worker %s started working.", threading.get_ident())
self.app.logger.debug("Current threads : %s" , threading.active_count())
self.app.logger.debug("Threads actives : %s " , threading.enumerate())
self.app.logger.debug("Current threads : %s", threading.active_count())
self.app.logger.debug("Threads actives : %s ", threading.enumerate())
while True:
@@ -73,7 +73,9 @@ class PrintWorker(threading.Thread):
# If we do not have a printer, we try to find one
while self.running and self.printer is None:
self.app.logger.debug("Trying to find a available printer for the worker.")
self.app.logger.debug(
"Trying to find a available printer for the worker."
)
self.printer = self._acquire_available_printer()
self.app.logger.debug(
"The worker got a %s printer and it's %s",
@@ -81,14 +83,11 @@ class PrintWorker(threading.Thread):
"Ready" if self.printer.ready else "Not ready",
)
if self.printer is None:
time.sleep(1)
self.printer.use_printer()
# If the printer is not ready, we wait for it to become ready.
while not self.printer.ready:
self.app.debug("Waiting for our printer to become ready...")
@@ -115,7 +114,9 @@ class PrintWorker(threading.Thread):
if task:
try:
self.app.logger.info("Got a new task")
self.app.logger.debug("Got task %s %s", task.task_type, task.task_id)
self.app.logger.debug(
"Got task %s %s", task.task_type, task.task_id
)
task.status = "processing"
print_data = task.get_print_data()
@@ -132,7 +133,7 @@ class PrintWorker(threading.Thread):
task.status = "completed"
self.print_queue.mark_completed(task.task_id, "completed")
self.app.logger.debug(
"Finished printing task %s " , task.task_id
"Finished printing task %s ", task.task_id
)
self.state = "idle"
@@ -141,7 +142,7 @@ class PrintWorker(threading.Thread):
self.state = "idle"
self.print_queue.mark_completed(task.task_id, "failed")
self.app.logger.error(
"Could not print task %s because %s " , task.task_id, str(e)
"Could not print task %s because %s ", task.task_id, str(e)
)
else: