195 lines
6.6 KiB
Python
195 lines
6.6 KiB
Python
"""
|
|
A collection of Printers.
|
|
|
|
It has methods to discover printers, and provides an interface for
|
|
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
|
|
|
|
|
|
class Printers:
|
|
"""
|
|
Finds and creates a set of Printer that can be used by the Workers to print.
|
|
"""
|
|
|
|
def __init__(self, app, configuration_file):
|
|
"""
|
|
Discovers printers connected to the computer
|
|
"""
|
|
self.app = app
|
|
self.configuration_file = configuration_file
|
|
self.printers = list(self._discover_printers())
|
|
self._lock = threading.Lock()
|
|
|
|
def _discover_printers(self) -> Set[Printer]:
|
|
"""
|
|
Gets connected USB printer devices using the pyusb library.
|
|
|
|
We analyse the USB devices, get the ones that match
|
|
the printer class ( 7 ) and for Brother and EPSON printers,
|
|
try to create a Printer object that can be used by the Worker class
|
|
to execute prints.
|
|
|
|
Returns a set of Printer
|
|
"""
|
|
|
|
self.app.logger.debug("Discovering USB Devices connected to this system")
|
|
|
|
printers = set()
|
|
|
|
waiting_time_for_printer = self.configuration_file["printer"]["waiting_time"]
|
|
|
|
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."
|
|
)
|
|
|
|
self.app.logger.warning(
|
|
"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
|
|
try:
|
|
manufacturer = usb.util.get_string(dev, dev.iManufacturer)
|
|
except Exception:
|
|
manufacturer = "Unknown"
|
|
|
|
try:
|
|
product = usb.util.get_string(dev, dev.iProduct)
|
|
except Exception:
|
|
product = "Unknown"
|
|
|
|
self.app.logger.debug(
|
|
"Looking at %s %s (%s:%s)",
|
|
manufacturer,
|
|
product,
|
|
hex(dev.idVendor),
|
|
hex(dev.idProduct),
|
|
)
|
|
|
|
match (manufacturer):
|
|
case ("EPSON"):
|
|
try:
|
|
# We create a new EscPosPrinter()
|
|
self.app.logger.debug("Trying to create a new EPSON printer")
|
|
prid = dev.idProduct
|
|
vendir = dev.idVendor
|
|
escpos_printer = EscPosPrinter(self.app, vendir, prid, self.configuration_file["printer"]["epson"] )
|
|
except Exception as e:
|
|
raise e
|
|
|
|
# If the object creation is successfull, we add it to the list of Printers
|
|
printers.add(escpos_printer)
|
|
self.app.logger.debug("Found a %s printer", manufacturer)
|
|
case ("Brother"):
|
|
try:
|
|
# We create a new BrotherPrinter()
|
|
self.app.logger.debug("Trying to create a new BROTHER printer")
|
|
|
|
brother_printer = BrotherPrinter(self.app, dev.idVendor, dev.idProduct, self.configuration_file["printer"]["brother"] )
|
|
except Exception as e:
|
|
self.app.logger.error(
|
|
"Could not create a %s printer class with %s:%s",
|
|
product,
|
|
dev.idVendor,
|
|
dev.idProduct,
|
|
)
|
|
raise e
|
|
|
|
# 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)
|
|
case (_):
|
|
self.app.logger.debug("Found a printer of unknown manufacturer : %s", manufacturer)
|
|
# end match
|
|
|
|
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
|
|
|
|
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
|
|
|
|
with self._lock:
|
|
# So don't always look at the first printer available if need be
|
|
# this allows for a round-robin lookup
|
|
for offset in range(len(printers)):
|
|
index = (start_index + offset) % len(printers)
|
|
printer = printers[index]
|
|
|
|
if not printer.used:
|
|
printer.use_printer()
|
|
return index, printer
|
|
|
|
return None
|
|
|
|
# def printers_info(self):
|
|
# """Return the current list of printers"""
|
|
# printers_info_list = {}
|
|
# for i in self.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)
|
|
)
|
|
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
|
|
if interface is not None:
|
|
return True
|
|
|
|
return False
|