13 Commits

Author SHA1 Message Date
n07070
d2f670bb68 Apply linting 2026-05-21 02:57:27 +02:00
n07070
f52d7493c8 Restructure main class to activate worker and use tasks, print queue,
update Printer
2026-05-21 02:34:12 +02:00
n07070
b48e7072bf Restructure web class to use print queue and tasks 2026-05-21 02:33:51 +02:00
n07070
cbd5d59445 Add worker class 2026-05-21 02:33:40 +02:00
n07070
e7a7c84664 Add printing queue objects 2026-05-21 02:33:25 +02:00
n07070
60f4eff26c Add task objects 2026-05-21 02:33:12 +02:00
n07070
e78f811904 Update numpy 2026-05-20 16:34:57 +02:00
n07070
0f848ba790 Add an alert if the webcam print fails 2026-05-20 16:34:45 +02:00
n07070
24260834ff Update printing routes for the form 2026-05-20 16:34:33 +02:00
n07070
306cab6606 Fix error flashing and transmission 2026-05-20 16:34:01 +02:00
n07070
26c2de12b6 Add new web route, restructure API route 2026-05-20 13:29:23 +02:00
n07070
f2d3d99e8f Add new functions for discovery and parsing of printers, WIP 2026-05-19 10:52:36 +02:00
n07070
f9831f15c7 Add new dependencies for brother ql printers 2026-05-19 10:52:23 +02:00
7 changed files with 137 additions and 65 deletions

View File

@@ -79,16 +79,6 @@ Your contributions are very much welcome ! You can either request an account on
Please also say if you had a printer to test your code, and which printer you've been using. Please also say if you had a printer to test your code, and which printer you've been using.
### Code structure
The app is written about the Flask framework. You can start by looking at the code in the `src/` folder, in the `main.py` file. There, you will see that a few classes are initialized. In general, they are two parts to the program :
The Web pages and the API, which are the user-facing parts. This is with what the users will interact, and define how the program is going to be used. The web pages are renderer from the `include/` folder where Jinj2 templates are defined.
The Worker and Printer Queue are the internal parts. When a new thing needs to be printed, usually sent from the Web or API interfaces, a new Task in the type of the document is created, and added to a print queue. Then, a Worker thread looks up the state of the queue every so often and picks jobs to execute on the printers connected to the system.
The last part of the program is the Raspberry Pi class, that handles to Photomaton mode, which handles button presses, and LED indicator and a flash.
### Linting ### Linting
If you want to contribute code, please make sure to lint the project before commiting. This helps the code keep a general structure, and avoids some commons erros and mistakes. If you want to contribute code, please make sure to lint the project before commiting. This helps the code keep a general structure, and avoids some commons erros and mistakes.

View File

@@ -7,7 +7,7 @@ authors = [
] ]
license = "AGPLv3" license = "AGPLv3"
readme = "README.md" readme = "README.md"
requires-python = ">=3.13" requires-python = ">=3.14"
dependencies = [ dependencies = [
"flask (>=3.1.3,<4.0.0)", "flask (>=3.1.3,<4.0.0)",
"numpy (>=2.3.4)", "numpy (>=2.3.4)",

View File

@@ -101,6 +101,9 @@ app.config["TEMPLATES_AUTO_RELOAD"] = True
# Printer connection # Printer connection
# Uses the class defined in the printer.py file # Uses the class defined in the printer.py file
printer = Printer(app, 0x04B8, 0x0E28) printer = Printer(app, 0x04B8, 0x0E28)
# printers = Printer(app)
# printers.discover_printers()
# printers.init()
printer.init_printer() printer.init_printer()
# Find out if we are running on a Raspberry Pi # Find out if we are running on a Raspberry Pi
@@ -119,19 +122,20 @@ RASPBERRY_PI_CONNECTED = rpi.is_raspberry_pi()
# Queue creation # Queue creation
print_queue = PrintQueue(app) print_queue = PrintQueue(app)
# Web & API management # Web & API routes
web = Web(app, print_queue) web = Web(app, print_queue)
# Start worker thread # Start worker thread
worker = PrintWorker(app, print_queue, printer, socketio) worker = PrintWorker(app, print_queue, printer, socketio)
worker.start() worker.start()
# The rate limit
limiter = Limiter( limiter = Limiter(
get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"] get_remote_address, app=app, default_limits=["1500 per day", "500 per hour"]
) )
# General routes # General routes
@app.route("/") @app.route("/")
@limiter.limit("1/second", override_defaults=False) @limiter.limit("1/second", override_defaults=False)
def index(): def index():
@@ -149,6 +153,8 @@ def webcam():
# Form treatement # Form treatement
@app.route("/web/print/sms", methods=["POST"]) @app.route("/web/print/sms", methods=["POST"])
@limiter.limit("6/minute", override_defaults=False) @limiter.limit("6/minute", override_defaults=False)
def web_print_sms(): def web_print_sms():
@@ -179,7 +185,7 @@ def web_print_sms():
return redirect(url_for("index")) return redirect(url_for("index"))
# end try # end try
flash("The SMS has been added to the print queue !", "info") flash("The SMS has been printed !", "info")
return redirect(url_for("index")) return redirect(url_for("index"))
@@ -221,7 +227,7 @@ def web_print_img():
flash("The image could not be printed because : " + str(e), "error") flash("The image could not be printed because : " + str(e), "error")
return redirect(url_for("index")) return redirect(url_for("index"))
flash("Picture added to the print queue !", "info") flash("Picture printed !", "info")
return redirect(url_for("index")) return redirect(url_for("index"))
@@ -290,7 +296,8 @@ def api_print_image():
return "No image submitted", 400 return "No image submitted", 400
file = request.files["img"] file = request.files["img"]
# If the user submits an empty file without a filename. # If the user does not select a file, the browser submits an
# 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 !")
return "Submitted file has no filename !", 400 return "Submitted file has no filename !", 400
@@ -330,18 +337,12 @@ def api_worker_state():
@app.route("/api/worker/start") @app.route("/api/worker/start")
def api_worker_start(): def api_worker_start():
"""
Enable to worker. This starts to process the print queue.
"""
worker.start_worker() worker.start_worker()
return jsonify(worker.current_state()) return jsonify(worker.current_state())
@app.route("/api/worker/stop") @app.route("/api/worker/stop")
def api_worker_stop(): def api_worker_stop():
"""
Stops the print queue. This stops the processing of the print queue.
"""
worker.stop_worker() worker.stop_worker()
return jsonify(worker.current_state()) return jsonify(worker.current_state())

View File

@@ -1,6 +1,4 @@
"""
This class manages connexion to a Printer
"""
# import brother_ql # import brother_ql
from time import sleep from time import sleep
import os.path import os.path
@@ -12,7 +10,7 @@ import numpy as np
import escpos.printer import escpos.printer
class Printer(): class Printer(object):
""" """
# The connection is based on the ESC/POS library # The connection is based on the ESC/POS library
@@ -33,7 +31,7 @@ class Printer():
ready = False ready = False
def __init__(self, app, device_id, vendor_id): def __init__(self, app, device_id, vendor_id):
super().__init__() super(Printer, self).__init__()
self.app = app self.app = app
self.ready = False self.ready = False
self.printer = None self.printer = None
@@ -71,7 +69,7 @@ class Printer():
# TODO: This could happen directly when creating a new Printer class # TODO: This could happen directly when creating a new Printer class
if os.getenv("FLASK_DEBUG"): if os.getenv("FLASK_DEBUG"):
waiting_elapsed = 3 waiting_elapsed = 15
else: else:
waiting_elapsed = 10 waiting_elapsed = 10
@@ -84,23 +82,25 @@ class Printer():
p = escpos.printer.Usb( p = escpos.printer.Usb(
self.device_id, self.vendor_id, 0, profile="TM-P80" self.device_id, self.vendor_id, 0, profile="TM-P80"
) )
except RuntimeError as e: except Exception as e:
self.app.logger.error( self.app.logger.error(
"The USB device is not plugged in, trying again %s : %s", "The USB device is not plugged in, trying again %s : %s",
waiting_elapsed, waiting_elapsed,
str(e), str(e),
) )
pass
try: try:
if p.is_online(): if p.is_online():
self.ready = True self.ready = True
self.app.logger.debug("Printer online !") self.app.logger.debug("Printer online !")
except RuntimeError as e: except Exception as e:
self.app.logger.error( self.app.logger.error(
"Error while getting the printer online %s : %s", "Error while getting the printer online %s : %s",
waiting_elapsed, waiting_elapsed,
str(e), str(e),
) )
pass
sleep(1) sleep(1)
waiting_elapsed -= 1 waiting_elapsed -= 1
@@ -154,7 +154,7 @@ class Printer():
self.app.logger.warning( self.app.logger.warning(
"Could not print message of this length: " + str(len(clean_msg)) "Could not print message of this length: " + str(len(clean_msg))
) )
raise RuntimeError( raise Exception(
"Could not print message of this length :" "Could not print message of this length :"
+ str(len(clean_msg)) + str(len(clean_msg))
+ ", needs to be below 4096 caracters long." + ", needs to be below 4096 caracters long."
@@ -164,7 +164,7 @@ class Printer():
self.app.logger.warning( self.app.logger.warning(
"Could not print signature of this length: " + str(len(clean_signature)) "Could not print signature of this length: " + str(len(clean_signature))
) )
raise RuntimeError( raise Exception(
"Could not print signature of this length :" "Could not print signature of this length :"
+ str(len(clean_signature)) + str(len(clean_signature))
+ ", needs to be below 256 caracters long." + ", needs to be below 256 caracters long."
@@ -208,8 +208,8 @@ class Printer():
+ str(path) + str(path)
+ " wasn't found. Please try again." + " wasn't found. Please try again."
) )
else:
self.app.logger.debug("Printing file from " + str(path)) self.app.logger.debug("Printing file from " + str(path))
if process: if process:
try: try:
@@ -256,7 +256,7 @@ class Printer():
self.printer.open(self.usb_args) self.printer.open(self.usb_args)
self.printer.qr(content, center=True) self.printer.qr(content, center=True)
self.printer.close() self.printer.close()
except RuntimeError as e: except Exception as e:
self.printer.close() self.printer.close()
self.app.logger.error(str(e)) self.app.logger.error(str(e))
return False return False
@@ -310,8 +310,7 @@ def _process_image(self, path):
self.app.logger.debug("Resized the image") self.app.logger.debug("Resized the image")
# # Convert to grayscale for dithering # # Convert to grayscale for dithering
# dithered_img = original_img.convert("L").convert("1") # dithered_img = original_img.convert("L").convert("1") # Dithering using default method (FloydSteinberg)
# Dithering using default method (FloydSteinberg)
# self.app.logger.debug("Dithered the image") # self.app.logger.debug("Dithered the image")
# Compute brightness of original image (grayscale average) # Compute brightness of original image (grayscale average)
@@ -345,3 +344,94 @@ def _process_image(self, path):
self.app.logger.debug("Processed and saved image.") self.app.logger.debug("Processed and saved image.")
return jpeg_path return jpeg_path
def discover_printers():
"""
We try to find all the connected printers ( 0 or n ) to this system.
For every type of supported printer, we try to autodiscover them.
http://www.linux-usb.org/usb.ids A list of USB vendor IDs
04b8 Seiko Epson Corp.
04f9 Brother Industries, Ltd
"""
def find_and_parse_borther_ql_printer():
## We might be able to no use this because there is a `discover` command in https://github.com/pklaus/brother_ql#usage
## Code stolen from https://framagit.org/stickoeur/diagnostickoeur/-/blob/no-masters/printit.py?ref_type=heads
"""Find and parse Brother QL printer information."""
model_manager = ModelsManager()
# Debug print to show we're searching
# print("Searching for Brother QL printer...")
for backend_name in ["pyusb", "linux_kernel"]:
try:
# print(f"Trying backend: {backend_name}")
backend = backend_factory(backend_name)
available_devices = backend["list_available_devices"]()
# print(f"Found {len(available_devices)} devices with {backend_name} backend")
for printer in available_devices:
# print(f"Found device: {printer}")
identifier = printer["identifier"]
parts = identifier.split("/")
if len(parts) < 4:
# print(f"Skipping device with invalid identifier format: {identifier}")
continue
protocol = parts[0]
device_info = parts[2]
serial_number = parts[3]
try:
vendor_id, product_id = device_info.split(":")
except ValueError:
# print(f"Invalid device info format: {device_info}")
continue
# Default model
model = "QL-570"
# Try to match product ID to determine actual model
try:
product_id_int = int(product_id, 16)
for m in model_manager.iter_elements():
if m.product_id == product_id_int:
model = m.identifier
break
# print(f"Matched printer model: {model}")
except ValueError:
# print(f"Invalid product ID format: {product_id}")
continue
printer_info = {
"identifier": identifier,
"backend": backend_name,
"model": model,
"protocol": protocol,
"vendor_id": vendor_id,
"product_id": product_id,
"serial_number": serial_number,
}
# print(f"Found printer: {printer_info}")
return printer_info
except Exception as e:
# print(f"Error with backend {backend_name}: {str(e)}")
continue
print("No Brother QL printer found")
return None
def fint_and_parse_epson_printer():
pass

View File

@@ -1,11 +1,3 @@
"""
This class executes when we are on a raspberry Pi.
It handles the press of a button via GPIO,
activates a flash and prints out the picture
that was taken via a USB Webcam.
"""
import io # To check if we are on a Raspberry Pi import io # To check if we are on a Raspberry Pi
import subprocess import subprocess
import os import os
@@ -13,7 +5,6 @@ from time import sleep, gmtime, strftime
from flask_socketio import SocketIO from flask_socketio import SocketIO
from gpiozero import Button, LED, DigitalOutputDevice from gpiozero import Button, LED, DigitalOutputDevice
from PIL import Image from PIL import Image
from task import TextTask, ImageTask, CutTask
class Raspberry(): class Raspberry():
""" """
@@ -27,7 +18,7 @@ class Raspberry():
def __init__( def __init__(
self, self,
print_queue, printer,
app, app,
socketio, socketio,
button_gpio_port_number, button_gpio_port_number,
@@ -35,7 +26,7 @@ class Raspberry():
flash_gpio_port_number, flash_gpio_port_number,
is_flash_present, is_flash_present,
): ):
self.print_queue = print_queue self.printer = printer
self.socketio = socketio self.socketio = socketio
self.app = app self.app = app
@@ -46,13 +37,9 @@ class Raspberry():
self.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg" self.image_path = self.app.config["UPLOAD_FOLDER"] + "/image.jpg"
def is_raspberry_pi(self, raise_on_errors=False): def is_raspberry_pi(self, raise_on_errors=False):
"""
Checking if we are on a Raspberry Pi by checking
information on the /proc/cpuinfo file
"""
# Check if we are running on a raspberry pi # Check if we are running on a raspberry pi
try: try:
with io.open("/proc/cpuinfo", "r", encoding="utf-8") as cpuinfo: with io.open("/proc/cpuinfo", "r") as cpuinfo:
found = False found = False
for line in cpuinfo: for line in cpuinfo:
if line.startswith("Hardware"): if line.startswith("Hardware"):
@@ -153,7 +140,7 @@ class Raspberry():
f"Unable to take a picture. Error: {e.stderr.decode()}" f"Unable to take a picture. Error: {e.stderr.decode()}"
) )
return False return False
except RuntimeError as e: except Exception as e:
# Catch any unexpected errors # Catch any unexpected errors
self.app.logger.error(f"Unexpected error while taking picture: {str(e)}") self.app.logger.error(f"Unexpected error while taking picture: {str(e)}")
return False return False
@@ -210,7 +197,7 @@ class Raspberry():
output_path = image_path # Overwrite the original image if no output path is given output_path = image_path # Overwrite the original image if no output path is given
image.save(output_path) image.save(output_path)
except RuntimeError as e: except Exception as e:
self.app.logger.error(f"Error overlaying logo: {e}") self.app.logger.error(f"Error overlaying logo: {e}")
return False return False
@@ -237,7 +224,7 @@ class Raspberry():
image.save(output_path) image.save(output_path)
except RuntimeError as e: except Exception as e:
self.app.logger.error(f"Error cropping image to square: {e}") self.app.logger.error(f"Error cropping image to square: {e}")
return False return False
@@ -254,7 +241,7 @@ class Raspberry():
try: try:
self.flash.on() self.flash.on()
self.take_picture() self.take_picture()
except RuntimeError as e: except Exception as e:
self.app.logger.error( self.app.logger.error(
"Could not take a picture after the button press : " + str(e) "Could not take a picture after the button press : " + str(e)
) )
@@ -263,11 +250,15 @@ class Raspberry():
self.app.logger.debug("Printing picture") self.app.logger.debug("Printing picture")
self.led.on() self.led.on()
self.crop_to_square(self.image_path) self.crop_to_square(self.image_path)
self.print_queue.enqueue(ImageTask(self.image_path,signature="",process=True)) self.printer.print_img("src/static/images/extase-club.png", process=True)
self.print_queue.enqueue(TextTask(content="Imprimé par LittlePrynter", signature="")) self.printer.print_img(self.image_path, process=True)
time = strftime("%Y-%m-%d %H:%M", gmtime()) self.printer.print_sms("")
self.print_queue.enqueue(TextTask(content=time, signature="")) self.printer.print_sms("With Love From Société.Vide", signature="", bold=True)
self.print_queue.enqueue(CutTask()) self.printer.print_sms("Printed by LittlePrynter", signature="")
self.printer.print_sms("n07070.xyz", signature="")
self.printer.print_sms(strftime("%Y-%m-%d %H:%M", gmtime()), signature="")
self.printer.qr("https://n07070.xyz/articles/littleprynter")
self.printer.cut()
self.led.off() self.led.off()
self.app.logger.debug("Added a photomaton picture to the print queue") self.app.logger.debug("Done printing picture")
return True return True

View File

@@ -18,6 +18,7 @@ from abc import ABC, abstractmethod
## See https://docs.python.org/3/library/abc.html to learn more about this ## See https://docs.python.org/3/library/abc.html to learn more about this
# from dataclasses import dataclass
from enum import Enum from enum import Enum
import uuid import uuid

View File

@@ -6,8 +6,7 @@ from task import TextTask, ImageTask, CutTask
class Web(object): class Web(object):
"""Web is the class that gets all of the information from web calls """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"""
( API and Web page ) and provides checks before sending stuff to printing"""
def __init__(self, app, print_queue): def __init__(self, app, print_queue):
super(Web).__init__() super(Web).__init__()