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

Change the way a Printer is used by the worker

Browse Source
This commit is contained in:
n07070
2026-10-11 21:42:29 +02:00
parent 31213eeeea
commit fd0507ff4f
4 changed files with 59 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main.py
View File

@@ -115,8 +115,6 @@ try:
worker.start()
except Exception as e:
app.logger.error("Could not start the worker because %s ", str(e))
raise e
sys.exit(-1)
# The rate limit

8
src/printer.py
View File

@@ -75,6 +75,7 @@ class Printer(ABC):
self.ready = False
self.printer_type = printer_type
self._lock = threading.Lock()
self.used = False
@abstractmethod
def _has_paper(self) -> bool:
@@ -88,6 +89,13 @@ class Printer(ABC):
def print_task(self, task_type, data) -> None:
"""Takes a PrintTask and executes it"""
def use_printer(self) -> None:
"""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):
"""

14
src/printers.py
View File

@@ -6,6 +6,7 @@ the methods expected from printers.
"""
from collections.abc import Set
import time
import usb.core
import usb.util
from printer import Printer, EscPosPrinter, BrotherPrinter
@@ -41,7 +42,17 @@ class Printers:
# Find all connected USB devices
devices = usb.core.find(find_all=True, custom_match=_FindClass(7))
if not devices:
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:
break
if not list(devices):
self.app.logger.warning(
"No USB devices of class 7 ( printers ) found or pyusb could not access the bus."
)
@@ -60,6 +71,7 @@ class Printers:
product = usb.util.get_string(dev, dev.iProduct)
except Exception:
product = "Unknown"
self.app.logger.debug(
"Looking at %s %s (%s:%s)",
manufacturer,

44
src/worker.py
View File

@@ -5,6 +5,7 @@ This is the main printing thread. A worker thread consums Tasks from
import threading
import time
import traceback
from printers import Printers
@@ -13,7 +14,8 @@ class PrintWorker(threading.Thread):
A thread used to consume Tasks added to a Print Queue.
On initialisation, the worker will try to find Printers,
and on each print, choose an available Printer from a list of Printers.
and on each print, if the choosen Printer becomes unavailable, it
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.
@@ -30,8 +32,9 @@ class PrintWorker(threading.Thread):
try:
self.printers = Printers(self.app)
self.printers_obj = self.printers.printers
self.printers = iter(self.printers.printers)
self.printers_obj = self.printers.printers # Get the number of printers connected to the system
self.printers_list = iter(self.printers.printers)
except RuntimeError as e:
self.app.logger.warning("Could not get any Printers")
raise e
@@ -53,11 +56,31 @@ class PrintWorker(threading.Thread):
# we know about, and try to find one that is available.
# When we find a printer, we acquire it
# When we are finished with a printer, we release it to the world.
while not self.printer or not self.printer.ready:
# If we do not have a printer, we try to find one
while not self.printer:
time.sleep(1)
try:
self.app.logger.debug("Changing printers")
self.printer = next(self.printers)
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.
# 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()
self.app.logger.debug(
"The worker got a %s printer and it's %s",
self.printer.printer_type,
@@ -67,6 +90,11 @@ class PrintWorker(threading.Thread):
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:
self.app.debug("Waiting for our printer to become ready...")
time.sleep(1)
if self.state != "idle":
self.app.logger("We are not idle, waiting...")
time.sleep(1)
@@ -80,6 +108,7 @@ class PrintWorker(threading.Thread):
except Exception as e:
self.app.logger.error("Could not get a new task ! %s ", str(e))
self.state = "idle"
self.printer.free_printer()
raise RuntimeError(
"We could not get a new task because " + str(e)
) from e
@@ -97,6 +126,8 @@ class PrintWorker(threading.Thread):
except RuntimeError as e:
self.state = "idle"
self.app.logger.error("Could not print : %s", str(e))
self.app.logger.error(traceback.format_exc())
self.printer.free_printer()
raise e
task.status = "completed"
@@ -125,6 +156,7 @@ class PrintWorker(threading.Thread):
"""
self.app.logger.debug("Giving the worker a break")
self.state = "drinking-a-beer"
self.printer.free_printer()
self.running = False
def start_worker(self):