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

Update the way a printer is found, update the configuration of Brother

Browse Source 
Printers
This commit is contained in:
n07070
2026-11-11 16:58:44 +01:00
parent fd0507ff4f
commit 704b7b83cd
5 changed files with 231 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/main.py
View File

@@ -40,6 +40,7 @@ from raspberry import Raspberry # The Raspberry pi control Class
from web import Web # Wrapper for the web routes and API
from print_queue import PrintQueue
from worker import PrintWorker
from printers import Printers
# We create the main Flask object
app = Flask(__name__)
@@ -108,10 +109,18 @@ RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
# Web & API management
web = Web(app, print_queue)
# Start worker thread
# When created, the worker will try to find printers connected to the system
# Create a pool of printers
try:
worker = PrintWorker(app, print_queue)
# comment: Find all printers connected to the system
printers_pool = Printers(app)
except Exception as e:
app.logger.error("Could not create a pool of printers because %s", str(e))
sys.exit(-1)
# end try
# Start worker thread
try:
worker = PrintWorker(app, print_queue, printers_pool=printers_pool)
worker.start()
except Exception as e:
app.logger.error("Could not start the worker because %s ", str(e))

176
src/printer.py
View File

@@ -11,9 +11,12 @@ from enum import Enum
import uuid
import threading
import usb.core
from PIL import Image, ImageEnhance
from pathlib import Path
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
@@ -386,8 +389,9 @@ class BrotherPrinter(Printer):
"""
def __init__(self, app, vendor_id, device_id):
super().__init__(
app, vendor_id, device_id, printer_type=PrinterType.BROTHER
app, vendor_id, device_id, printer_type=PrinterType.BROTHER
)
self.printer = None
self.usb_args = {}
@@ -450,16 +454,21 @@ class BrotherPrinter(Printer):
def _state(self):
return self.ready
def _print_txt(self, msg, signature="", bold=False):
"""Print text after converting it to an image"""
self._print_img(text_to_image(msg))
def _print_img(self, data):
"""
Print a raster image via a Brother QL printer
"""
label_type = "102"
# TODO : make this label type go in the configuration
label_type = "62"
rotate = 0
dither = False
data = _process_image(self, path=data, max_width=720, max_height=720)
self.app.logger.debug("Printing image %s", data)
try:
# Prepare the image for printing
qlr = BrotherQLRaster(self.printer_info["model"])
@@ -545,6 +554,9 @@ class BrotherPrinter(Printer):
case "cut":
# The cut happens by default on Brother QL printers.
pass
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:
@@ -559,56 +571,128 @@ class BrotherPrinter(Printer):
def _process_image(self, path, max_width=575, max_height=1000):
brightness_factor = 1.5 # Used only if image is too dark
brightness_threshold = 100 # Brightness threshold (0–255)
contrast_factor = 2 # Less than 1.0 = lower contrast
brightness_threshold = 100
contrast_factor = 2.0
with Image.open(path) as original_img:
# Convert to RGB if needed (JPEG doesn't support alpha)
if original_img.mode in ("RGBA", "P"):
self.app.logger.debug("Converting the image from RGBA to RGBA")
original_img = original_img.convert("RGB")
# Resize while maintaining aspect ratio
original_img = original_img.resize((max_width, max_height), Image.Resampling.LANCZOS)
self.app.logger.debug("Resized the image")
# # Convert to grayscale for dithering
# dithered_img = original_img.convert("L").convert("1")
# Dithering using default method (Floyd–Steinberg)
# self.app.logger.debug("Dithered the image")
# Compute brightness of original image (grayscale average)
grayscale = original_img.convert("L")
avg_brightness = np.array(grayscale).mean()
self.app.logger.debug(
"Average brightness of the image : " + str(avg_brightness)
with Image.open(path) as img:
# Preserve aspect ratio while fitting within the maximum dimensions.
img.thumbnail(
(max_width, max_height),
Image.Resampling.LANCZOS,
)
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
)
# Dynamically compute brightness factor if too dark
if avg_brightness < brightness_threshold:
brightness_factor = (
1 + (brightness_threshold - avg_brightness) / brightness_threshold
)
brightness_factor = min(
max(brightness_factor, 1.1), 2.5
) # Clamp between 1.1 and 2.5
self.app.logger.debug(
f"Image too dark, increasing brightness by a factor of {brightness_factor:.2f}"
max(
1 + (brightness_threshold - avg_brightness)
/ brightness_threshold,
1.1,
),
2.5,
)
self.app.logger.debug(
"Image is dark; increasing brightness by %.2f",
brightness_factor,
)
grayscale = ImageEnhance.Brightness(grayscale).enhance(
brightness_factor
)
enhancer = ImageEnhance.Brightness(grayscale)
grayscale = enhancer.enhance(brightness_factor)
# Computer current contrast of grayscale image
contrast = np.clip(np.std(np.array(grayscale)), 0, 255)
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)
contrast = float(np.asarray(grayscale, dtype=np.float32).std())
self.app.logger.debug("Grayscale standard deviation: %.2f", contrast)
# Save the enhanced result.
processed = ImageEnhance.Contrast(grayscale).enhance(contrast_factor)
# Convert to JPEG and save
jpeg_path = os.path.splitext(path)[0] + "_processed.jpg"
grayscale.save(jpeg_path, format="JPEG", quality=95, optimize=True)
self.app.logger.debug("Processed and saved image.")
processed.save(jpeg_path, format="JPEG", quality=95, optimize=True)
self.app.logger.debug("Processed and saved image: %s", jpeg_path)
return jpeg_path
def text_to_image(
text,
width=400,
font_path="src/static/font/MonospaceTypewriter.ttf",
font_size=24,
padding=10,
line_spacing=4,
background="white",
fill="black",
output_dir="src/static/uploads",
):
"""Converts a string to an image"""
font = ImageFont.truetype(font_path, font_size)
# Measure and wrap text by pixel width.
measure_img = Image.new("RGB", (1, 1))
draw = ImageDraw.Draw(measure_img)
max_text_width = width - 2 * padding
if max_text_width <= 0:
raise ValueError("width must be greater than twice the padding")
lines = []
for paragraph in text.splitlines() or [""]:
words = paragraph.split()
if not words:
lines.append("")
continue
line = ""
for word in words:
candidate = f"{line} {word}".strip()
if draw.textbbox((0, 0), candidate, font=font)[2] <= max_text_width:
line = candidate
continue
if line:
lines.append(line)
# Split a word that is wider than the available width.
line = ""
piece = ""
for char in word:
if piece and draw.textbbox((0, 0), piece + char, font=font)[2] > max_text_width:
lines.append(piece)
piece = char
else:
piece += char
line = piece
if line:
lines.append(line)
if not lines:
lines = [""]
ascent, descent = font.getmetrics()
line_height = ascent + descent
height = (
2 * padding
+ len(lines) * line_height
+ (len(lines) - 1) * line_spacing
)
image = Image.new("RGB", (width, height), background)
draw = ImageDraw.Draw(image)
y = padding
for line in lines:
draw.text((padding, y), line, font=font, fill=fill)
y += line_height + line_spacing
upload_dir = Path(output_dir)
upload_dir.mkdir(parents=True, exist_ok=True)
image_path = upload_dir / f"{uuid4().hex}.png"
image.save(image_path, format="PNG")
return str(image_path)

92
src/printers.py
View File

@@ -7,6 +7,7 @@ the methods expected from printers.
from collections.abc import Set
import time
import threading
import usb.core
import usb.util
from printer import Printer, EscPosPrinter, BrotherPrinter
@@ -19,10 +20,11 @@ class Printers:
def __init__(self, app):
"""
Discover printers connected to the computer and return a Collection of Printer()
Discovers printers connected to the computer
"""
self.app = app
self.printers = self._discover_printers()
self.printers = list(self._discover_printers())
self._lock = threading.Lock()
def _discover_printers(self) -> Set[Printer]:
"""
@@ -40,25 +42,36 @@ class Printers:
printers = set()
# Find all connected USB devices
devices = usb.core.find(find_all=True, custom_match=_FindClass(7))
waiting_time_for_printer = 10
while not list(devices):
self.app.logger.warning(str(waiting_time_for_printer) + " Found no printers, waiting for a printer to connect to the system...")
time.sleep(1)
waiting_time_for_printer = waiting_time_for_printer - 1
devices = usb.core.find(find_all=True, custom_match=_FindClass(7))
if waiting_time_for_printer == 0:
# TODO : this could go in the config ?
waiting_time_for_printer = 5
while True:
# 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)
)
)
if devices:
break
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."
)
raise RuntimeError(
"No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
)
if not list(devices):
self.app.logger.warning(
"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 printers found; waiting for a printer to connect..."
)
time.sleep(1)
waiting_time_for_printer -= 1
for dev in devices:
# Attempt to get the manufacturer and product strings
@@ -107,8 +120,7 @@ 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,
@@ -129,43 +141,45 @@ class Printers:
return printers
def any(self) -> Printer:
"""
Return a dict key: UUID, value: Printer, with any connected printer.
"""
if len(self.printers) > 0:
for i in self.printers:
return i
else:
raise RuntimeError("No printers available")
def acquire_available_printer(self, start_index=0):
"""Atomically acquire and return an available printer, or None."""
printers = self.printers
if not printers:
self.app.logger.error("No printers available for acquisition")
return None
# def get_printer(self, printer_type):
# """
# Return a specific printer
with self._lock:
for offset in range(len(printers)):
index = (start_index + offset) % len(printers)
printer = printers[index]
# printer_type -- a printer type
# """
# return NotImplementedError()
if not printer.used:
printer.use_printer()
return index, printer
return None
class _FindClass:
"""
Use by usb.core to modify the way USB devices are found
Taken from pyUSB documentation on Github
"""
def __init__(self, class_):
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))
if device.bDeviceClass == self._class:
return True
# ok, transverse all devices to find an
# interface that matches our class
for cfg in device:
# find_descriptor: what's it?
intf = usb.util.find_descriptor(cfg, bInterfaceClass=self._class)
if intf is not None:
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
if interface is not None:
return True
return False

BIN
src/static/font/MonospaceTypewriter.ttf Normal file
View File

Binary file not shown.

73
src/worker.py
View File

@@ -18,9 +18,9 @@ class PrintWorker(threading.Thread):
will choose an available Printer from a list of Printers.
If a print fails, it's not retried, but will be added to a list of completed
tasks.
tasks. TODO : Change that
"""
def __init__(self, app, print_queue):
def __init__(self, app, print_queue, printers_pool):
super().__init__(daemon=True)
self.app = app
self.print_queue = print_queue
@@ -28,16 +28,29 @@ class PrintWorker(threading.Thread):
self._lock = threading.Lock()
self.running = True
self.state = "idle" # idle, printing, dead, drinking-a-beer
# All workers share the same pool of printers.
self.printers = printers_pool
self.printers_obj = self.printers.printers
self._printer_index = 0
self.app.logger.debug("Worker initialized.")
self.app.logger.debug("Printers connected:")
for printer in self.printers_obj:
self.app.logger.debug("Printer type: %s", printer.printer_type)
self.app.logger.debug("Printer state: %s", printer.ready)
self.app.logger.debug("Ho great, I'm alive... I'm ready to work another day...")
try:
self.printers = Printers(self.app)
self.printers_obj = self.printers.printers # Get the number of printers connected to the system
self.printers_list = iter(self.printers.printers)
def _acquire_available_printer(self):
result = self.printers.acquire_available_printer(self._printer_index)
if result is None:
return None
index, printer = result
self._printer_index = (index + 1) % len(self.printers_obj)
return printer
except RuntimeError as e:
self.app.logger.warning("Could not get any Printers")
raise e
def run(self):
"""Background thread that processes queue items"""
@@ -58,37 +71,23 @@ class PrintWorker(threading.Thread):
# When we are finished with a printer, we release it to the world.
# If we do not have a printer, we try to find one
while not self.printer:
time.sleep(1)
try:
self.app.logger.debug("Trying to find a available printer for the worker.")
# From the list of printer, we get one.
# If no printer in the list is avaible, we try again from the start of the loop, that we reset.
while self.printer.used or not self.printer:
self.printer = next(self.printers_list)
try:
self.printer = next(self.printer)
except StopIteration as e:
self.app.error("We do not have any available printers. Resetting our list of printers and trying again. " + str(e))
self.printers_list = iter(self.printers.printers)
time.sleep(1)
# TODO : Here, in case there is really no printer available, we will stay in a loop forever. Maybe after a number of tries, we should die.
while self.running and self.printer is None:
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",
self.printer.printer_type,
"Ready" if self.printer.ready else "Not ready",
)
# Here, we must have found a Printer() that is
# Printer.used = False, which means it's available.
# Therefore, we mark it used, so no other worker uses it.
self.printer.use_printer()
if self.printer is None:
time.sleep(1)
self.printer.use_printer()
self.app.logger.debug(
"The worker got a %s printer and it's %s",
self.printer.printer_type,
"Ready" if self.printer.ready else "Not ready",
)
except Exception as e:
self.app.logger.error("No printer detected" + str(e))
self.printer = None
# If the printer is not ready, we wait for it to become ready.
while not self.printer.ready:
@@ -116,7 +115,7 @@ class PrintWorker(threading.Thread):
if task:
try:
self.app.logger.info("Got a new task")
self.app.logger.debug("Got task %s", 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()