Compare commits
1 Commits
adb5e9422d
...
multi-prin
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
070f037449 |
@@ -10,9 +10,9 @@ import time
|
||||
from enum import Enum
|
||||
import uuid
|
||||
import threading
|
||||
import usb.core
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
import usb.core
|
||||
from PIL import Image, ImageEnhance, ImageDraw, ImageFont
|
||||
import numpy as np
|
||||
|
||||
@@ -124,7 +124,9 @@ class EscPosPrinter(Printer):
|
||||
try:
|
||||
# This also calls open(), which we need to close()
|
||||
# or else the device will appear as busy.
|
||||
p = escpos.printer.Usb(self.vendor_id, self.device_id, 0, profile=configuration_file["profile"])
|
||||
p = escpos.printer.Usb(
|
||||
self.vendor_id, self.device_id, 0, profile=configuration_file["profile"]
|
||||
)
|
||||
except escpos.exceptions.DeviceNotFoundError as e:
|
||||
self.app.logger.error(
|
||||
"The USB device is not plugged in : %s",
|
||||
@@ -221,7 +223,8 @@ class EscPosPrinter(Printer):
|
||||
raise RuntimeError(
|
||||
"Could not print message of this length :"
|
||||
+ str(len(clean_msg))
|
||||
+ ", needs to be below %s caracters long.", str(self.configuration_file["text_max_length"])
|
||||
+ ", needs to be below %s caracters long.",
|
||||
str(self.configuration_file["text_max_length"]),
|
||||
)
|
||||
if len(signature) > int(self.configuration_file["signature_max_length"]):
|
||||
self.app.logger.warning(
|
||||
@@ -230,7 +233,8 @@ class EscPosPrinter(Printer):
|
||||
raise RuntimeError(
|
||||
"Could not print signature of this length :"
|
||||
+ str(len(clean_signature))
|
||||
+ ", needs to be below %s caracters long.", str(self.configuration_file["signature_max_length"])
|
||||
+ ", needs to be below %s caracters long.",
|
||||
str(self.configuration_file["signature_max_length"]),
|
||||
)
|
||||
|
||||
# Do the actual printing
|
||||
@@ -262,10 +266,10 @@ class EscPosPrinter(Printer):
|
||||
raise RuntimeError(
|
||||
"Could not print signature of this length :"
|
||||
+ str(len(clean_signature))
|
||||
+ ", needs to be below %s caracters long.", str(self.configuration_file["signature_max_length"])
|
||||
+ ", needs to be below %s caracters long.",
|
||||
str(self.configuration_file["signature_max_length"]),
|
||||
)
|
||||
|
||||
|
||||
if not os.path.isfile(str(path)):
|
||||
self.app.logger.warning("File does not exist : " + str(path))
|
||||
raise OSError(
|
||||
@@ -310,9 +314,7 @@ class EscPosPrinter(Printer):
|
||||
try:
|
||||
self.printer.close()
|
||||
except Exception as e:
|
||||
self.app.logger.error(
|
||||
"Could not close the printer connexion %s", str(e)
|
||||
)
|
||||
self.app.logger.error("Could not close the printer connexion %s", str(e))
|
||||
raise RuntimeError("Could not close the printer connexion. ") from e
|
||||
|
||||
self.app.logger.info("Printed a picture")
|
||||
@@ -456,7 +458,9 @@ class BrotherPrinter(Printer):
|
||||
vendor_id=vendor_id,
|
||||
product_id=self.device_id,
|
||||
serial_number=serial_number,
|
||||
label_type=configuration_file["label_type"], # TODO: Implement automatic label type detection : https://framagit.org/stickoeur/diagnostickoeur/-/blob/no-masters/printit.py?ref_type=heads#L239
|
||||
label_type=configuration_file[
|
||||
"label_type"
|
||||
], # TODO: Implement automatic label type detection : https://framagit.org/stickoeur/diagnostickoeur/-/blob/no-masters/printit.py?ref_type=heads#L239
|
||||
)
|
||||
self.app.logger.debug("Created a Brother printer !")
|
||||
self.ready = True
|
||||
@@ -472,7 +476,12 @@ class BrotherPrinter(Printer):
|
||||
self._print_img(text_to_image(msg))
|
||||
self._print_img(text_to_image(signature))
|
||||
|
||||
def _print_img(self, data, signature="", process=False,):
|
||||
def _print_img(
|
||||
self,
|
||||
data,
|
||||
signature="",
|
||||
process=False,
|
||||
):
|
||||
"""
|
||||
Print a raster image via a Brother QL printer
|
||||
"""
|
||||
@@ -512,20 +521,20 @@ class BrotherPrinter(Printer):
|
||||
# del be
|
||||
|
||||
instructions = convert(
|
||||
qlr=qlr,
|
||||
images=[im],
|
||||
label=self.printer_info.label_type,
|
||||
dither=True,
|
||||
compress=False,
|
||||
dpi_600=False,
|
||||
cut=False,
|
||||
)
|
||||
qlr=qlr,
|
||||
images=[im],
|
||||
label=self.printer_info.label_type,
|
||||
dither=True,
|
||||
compress=False,
|
||||
dpi_600=False,
|
||||
cut=False,
|
||||
)
|
||||
|
||||
success = send(
|
||||
instructions=instructions,
|
||||
printer_identifier=BACKEND_STRING_DESCR,
|
||||
backend_identifier="pyusb",
|
||||
blocking=False
|
||||
blocking=False,
|
||||
)
|
||||
|
||||
self.app.logger.debug(str(success))
|
||||
@@ -534,7 +543,7 @@ class BrotherPrinter(Printer):
|
||||
|
||||
except usb.core.USBError as e:
|
||||
if "timeout error" in str(e):
|
||||
self.app.logger.error("USB Timeout Error %s", str(e) )
|
||||
self.app.logger.error("USB Timeout Error %s", str(e))
|
||||
except Exception as e:
|
||||
raise e
|
||||
# finally:
|
||||
@@ -560,27 +569,25 @@ class BrotherPrinter(Printer):
|
||||
# be.dispose()
|
||||
# del be
|
||||
instructions = convert(
|
||||
qlr=qlr,
|
||||
images=[signature_file],
|
||||
label=self.printer_info.label_type,
|
||||
dither=True,
|
||||
compress=False,
|
||||
dpi_600=False,
|
||||
cut=False,
|
||||
)
|
||||
|
||||
qlr=qlr,
|
||||
images=[signature_file],
|
||||
label=self.printer_info.label_type,
|
||||
dither=True,
|
||||
compress=False,
|
||||
dpi_600=False,
|
||||
cut=False,
|
||||
)
|
||||
|
||||
success = send(
|
||||
instructions=instructions,
|
||||
printer_identifier=BACKEND_STRING_DESCR,
|
||||
backend_identifier="pyusb",
|
||||
blocking=False
|
||||
blocking=False,
|
||||
)
|
||||
self.app.logger.debug(str(success))
|
||||
self.app.logger.debug("Printed signature")
|
||||
time.sleep(2)
|
||||
|
||||
|
||||
except Exception as e:
|
||||
raise e
|
||||
finally:
|
||||
@@ -619,7 +626,6 @@ 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()
|
||||
|
||||
@@ -91,25 +91,35 @@ class Printers:
|
||||
)
|
||||
|
||||
match (manufacturer):
|
||||
case ("EPSON"):
|
||||
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"] )
|
||||
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"):
|
||||
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"] )
|
||||
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",
|
||||
@@ -122,8 +132,10 @@ 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)
|
||||
case (_):
|
||||
self.app.logger.debug("Found a printer of unknown manufacturer : %s", 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))
|
||||
@@ -160,7 +172,14 @@ class Printers:
|
||||
"""Return the current list of printers"""
|
||||
printers_info_list = []
|
||||
for i in self.printers:
|
||||
printers_info_list.append([i.printer_type.name,"Used: " + str(i.used), "Ready: " + str(i.ready),i.printer_info])
|
||||
printers_info_list.append(
|
||||
[
|
||||
i.printer_type.name,
|
||||
"Used: " + str(i.used),
|
||||
"Ready: " + str(i.ready),
|
||||
i.printer_info,
|
||||
]
|
||||
)
|
||||
|
||||
return printers_info_list
|
||||
|
||||
|
||||
@@ -41,7 +41,9 @@ class Web:
|
||||
|
||||
return True
|
||||
|
||||
def print_image(self, image, sign: str, do_cut: bool, process_image: bool, many: int = 1) -> bool:
|
||||
def print_image(
|
||||
self, image, sign: str, do_cut: bool, process_image: bool, many: int = 1
|
||||
) -> bool:
|
||||
"""
|
||||
Get an image and a signature, prints the image and cuts after that.
|
||||
"""
|
||||
|
||||
@@ -45,7 +45,9 @@ class PrintWorker(threading.Thread):
|
||||
self.app.logger.debug("Ho great, I'm alive... I'm ready to work another day...")
|
||||
|
||||
def _acquire_available_printer(self):
|
||||
result = self.printers.acquire_available_printer(start_index=self._printer_index)
|
||||
result = self.printers.acquire_available_printer(
|
||||
start_index=self._printer_index
|
||||
)
|
||||
if result is None:
|
||||
return None
|
||||
|
||||
@@ -81,17 +83,14 @@ class PrintWorker(threading.Thread):
|
||||
)
|
||||
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",
|
||||
"The worker got a %s printer and it's %s",
|
||||
self.printer.printer_type,
|
||||
"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.
|
||||
|
||||
Reference in New Issue
Block a user