Compare commits
1 Commits
adb5e9422d
...
multi-prin
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
070f037449 |
@@ -10,9 +10,9 @@ import time
|
|||||||
from enum import Enum
|
from enum import Enum
|
||||||
import uuid
|
import uuid
|
||||||
import threading
|
import threading
|
||||||
import usb.core
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from uuid import uuid4
|
from uuid import uuid4
|
||||||
|
import usb.core
|
||||||
from PIL import Image, ImageEnhance, ImageDraw, ImageFont
|
from PIL import Image, ImageEnhance, ImageDraw, ImageFont
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
|
||||||
@@ -124,7 +124,9 @@ class EscPosPrinter(Printer):
|
|||||||
try:
|
try:
|
||||||
# This also calls open(), which we need to close()
|
# This also calls open(), which we need to close()
|
||||||
# or else the device will appear as busy.
|
# 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:
|
except escpos.exceptions.DeviceNotFoundError as e:
|
||||||
self.app.logger.error(
|
self.app.logger.error(
|
||||||
"The USB device is not plugged in : %s",
|
"The USB device is not plugged in : %s",
|
||||||
@@ -221,7 +223,8 @@ class EscPosPrinter(Printer):
|
|||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
"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 %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"]):
|
if len(signature) > int(self.configuration_file["signature_max_length"]):
|
||||||
self.app.logger.warning(
|
self.app.logger.warning(
|
||||||
@@ -230,7 +233,8 @@ class EscPosPrinter(Printer):
|
|||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
"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 %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
|
# Do the actual printing
|
||||||
@@ -262,10 +266,10 @@ class EscPosPrinter(Printer):
|
|||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
"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 %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)):
|
if not os.path.isfile(str(path)):
|
||||||
self.app.logger.warning("File does not exist : " + str(path))
|
self.app.logger.warning("File does not exist : " + str(path))
|
||||||
raise OSError(
|
raise OSError(
|
||||||
@@ -310,9 +314,7 @@ class EscPosPrinter(Printer):
|
|||||||
try:
|
try:
|
||||||
self.printer.close()
|
self.printer.close()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.app.logger.error(
|
self.app.logger.error("Could not close the printer connexion %s", str(e))
|
||||||
"Could not close the printer connexion %s", str(e)
|
|
||||||
)
|
|
||||||
raise RuntimeError("Could not close the printer connexion. ") from e
|
raise RuntimeError("Could not close the printer connexion. ") from e
|
||||||
|
|
||||||
self.app.logger.info("Printed a picture")
|
self.app.logger.info("Printed a picture")
|
||||||
@@ -456,7 +458,9 @@ class BrotherPrinter(Printer):
|
|||||||
vendor_id=vendor_id,
|
vendor_id=vendor_id,
|
||||||
product_id=self.device_id,
|
product_id=self.device_id,
|
||||||
serial_number=serial_number,
|
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.app.logger.debug("Created a Brother printer !")
|
||||||
self.ready = True
|
self.ready = True
|
||||||
@@ -472,7 +476,12 @@ class BrotherPrinter(Printer):
|
|||||||
self._print_img(text_to_image(msg))
|
self._print_img(text_to_image(msg))
|
||||||
self._print_img(text_to_image(signature))
|
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
|
Print a raster image via a Brother QL printer
|
||||||
"""
|
"""
|
||||||
@@ -512,20 +521,20 @@ class BrotherPrinter(Printer):
|
|||||||
# del be
|
# del be
|
||||||
|
|
||||||
instructions = convert(
|
instructions = convert(
|
||||||
qlr=qlr,
|
qlr=qlr,
|
||||||
images=[im],
|
images=[im],
|
||||||
label=self.printer_info.label_type,
|
label=self.printer_info.label_type,
|
||||||
dither=True,
|
dither=True,
|
||||||
compress=False,
|
compress=False,
|
||||||
dpi_600=False,
|
dpi_600=False,
|
||||||
cut=False,
|
cut=False,
|
||||||
)
|
)
|
||||||
|
|
||||||
success = send(
|
success = send(
|
||||||
instructions=instructions,
|
instructions=instructions,
|
||||||
printer_identifier=BACKEND_STRING_DESCR,
|
printer_identifier=BACKEND_STRING_DESCR,
|
||||||
backend_identifier="pyusb",
|
backend_identifier="pyusb",
|
||||||
blocking=False
|
blocking=False,
|
||||||
)
|
)
|
||||||
|
|
||||||
self.app.logger.debug(str(success))
|
self.app.logger.debug(str(success))
|
||||||
@@ -534,7 +543,7 @@ class BrotherPrinter(Printer):
|
|||||||
|
|
||||||
except usb.core.USBError as e:
|
except usb.core.USBError as e:
|
||||||
if "timeout error" in str(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:
|
except Exception as e:
|
||||||
raise e
|
raise e
|
||||||
# finally:
|
# finally:
|
||||||
@@ -560,27 +569,25 @@ class BrotherPrinter(Printer):
|
|||||||
# be.dispose()
|
# be.dispose()
|
||||||
# del be
|
# del be
|
||||||
instructions = convert(
|
instructions = convert(
|
||||||
qlr=qlr,
|
qlr=qlr,
|
||||||
images=[signature_file],
|
images=[signature_file],
|
||||||
label=self.printer_info.label_type,
|
label=self.printer_info.label_type,
|
||||||
dither=True,
|
dither=True,
|
||||||
compress=False,
|
compress=False,
|
||||||
dpi_600=False,
|
dpi_600=False,
|
||||||
cut=False,
|
cut=False,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
success = send(
|
success = send(
|
||||||
instructions=instructions,
|
instructions=instructions,
|
||||||
printer_identifier=BACKEND_STRING_DESCR,
|
printer_identifier=BACKEND_STRING_DESCR,
|
||||||
backend_identifier="pyusb",
|
backend_identifier="pyusb",
|
||||||
blocking=False
|
blocking=False,
|
||||||
)
|
)
|
||||||
self.app.logger.debug(str(success))
|
self.app.logger.debug(str(success))
|
||||||
self.app.logger.debug("Printed signature")
|
self.app.logger.debug("Printed signature")
|
||||||
time.sleep(2)
|
time.sleep(2)
|
||||||
|
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise e
|
raise e
|
||||||
finally:
|
finally:
|
||||||
@@ -619,7 +626,6 @@ class BrotherPrinter(Printer):
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise e
|
raise e
|
||||||
|
|
||||||
|
|
||||||
def _qr(self):
|
def _qr(self):
|
||||||
# This should not be reached by a print on a BrotherQL
|
# This should not be reached by a print on a BrotherQL
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|||||||
@@ -91,25 +91,35 @@ class Printers:
|
|||||||
)
|
)
|
||||||
|
|
||||||
match (manufacturer):
|
match (manufacturer):
|
||||||
case ("EPSON"):
|
case "EPSON":
|
||||||
try:
|
try:
|
||||||
# We create a new EscPosPrinter()
|
# We create a new EscPosPrinter()
|
||||||
self.app.logger.debug("Trying to create a new EPSON printer")
|
self.app.logger.debug("Trying to create a new EPSON printer")
|
||||||
prid = dev.idProduct
|
prid = dev.idProduct
|
||||||
vendir = dev.idVendor
|
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:
|
except Exception as e:
|
||||||
raise e
|
raise e
|
||||||
|
|
||||||
# If the object creation is successfull, we add it to the list of Printers
|
# If the object creation is successfull, we add it to the list of Printers
|
||||||
printers.add(escpos_printer)
|
printers.add(escpos_printer)
|
||||||
self.app.logger.debug("Found a %s printer", manufacturer)
|
self.app.logger.debug("Found a %s printer", manufacturer)
|
||||||
case ("Brother"):
|
case "Brother":
|
||||||
try:
|
try:
|
||||||
# We create a new BrotherPrinter()
|
# We create a new BrotherPrinter()
|
||||||
self.app.logger.debug("Trying to create a new BROTHER printer")
|
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:
|
except Exception as e:
|
||||||
self.app.logger.error(
|
self.app.logger.error(
|
||||||
"Could not create a %s printer class with %s:%s",
|
"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
|
# If the object creation is successfull, we add it to the list of Printers
|
||||||
printers.add(brother_printer)
|
printers.add(brother_printer)
|
||||||
self.app.logger.debug("Found a %s printer", manufacturer)
|
self.app.logger.debug("Found a %s printer", manufacturer)
|
||||||
case (_):
|
case _:
|
||||||
self.app.logger.debug("Found a printer of unknown manufacturer : %s", manufacturer)
|
self.app.logger.debug(
|
||||||
|
"Found a printer of unknown manufacturer : %s", manufacturer
|
||||||
|
)
|
||||||
# end match
|
# end match
|
||||||
|
|
||||||
self.app.logger.debug("Found %s printers", len(printers))
|
self.app.logger.debug("Found %s printers", len(printers))
|
||||||
@@ -160,7 +172,14 @@ class Printers:
|
|||||||
"""Return the current list of printers"""
|
"""Return the current list of printers"""
|
||||||
printers_info_list = []
|
printers_info_list = []
|
||||||
for i in self.printers:
|
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
|
return printers_info_list
|
||||||
|
|
||||||
|
|||||||
@@ -41,7 +41,9 @@ class Web:
|
|||||||
|
|
||||||
return True
|
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.
|
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...")
|
self.app.logger.debug("Ho great, I'm alive... I'm ready to work another day...")
|
||||||
|
|
||||||
def _acquire_available_printer(self):
|
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:
|
if result is None:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@@ -81,17 +83,14 @@ class PrintWorker(threading.Thread):
|
|||||||
)
|
)
|
||||||
self.printer = self._acquire_available_printer()
|
self.printer = self._acquire_available_printer()
|
||||||
self.app.logger.debug(
|
self.app.logger.debug(
|
||||||
|
"The worker got a %s printer and it's %s",
|
||||||
"The worker got a %s printer and it's %s",
|
self.printer.printer_type,
|
||||||
self.printer.printer_type,
|
"Ready" if self.printer.ready else "Not ready",
|
||||||
"Ready" if self.printer.ready else "Not ready",
|
|
||||||
)
|
)
|
||||||
|
|
||||||
if self.printer is None:
|
if self.printer is None:
|
||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
self.printer.use_printer()
|
self.printer.use_printer()
|
||||||
|
|
||||||
# If the printer is not ready, we wait for it to become ready.
|
# If the printer is not ready, we wait for it to become ready.
|
||||||
|
|||||||
Reference in New Issue
Block a user