Detach kernel driver on backend initialization

brother_label_printer/backends/__init__.py

@@ -14,6 +14,13 @@ def is_usb_printer(dev):
 class PyUSBBackend():
     def __init__(self, dev):
         self.dev = dev
+        try:
+            if self.dev.is_kernel_driver_active(0):
+                self.dev.detach_kernel_driver(0)
+        except NotImplementedError:
+            pass
+        except usb.core.USBError:
+            pass
         self.lock = threading.Lock()
 
     @classmethod
@@ -29,6 +36,7 @@ def write(self, data: bytes):
     def read(self, count: int) -> bytes:
         return self.dev.read(0x81, count)
 
+
 class BTSerialBackend():
     def __init__(self, dev):
         self.dev = dev

brother_label_printer/printers/__init__.py

@@ -54,7 +54,7 @@ def estimate_label_size(self, label: Label) -> Tuple[float, float]:
         xpixels, ypixels = label.size
         return (xpixels / xdpi) * 2.54, (ypixels / ydpi) * 2.54
 
-    def print_label(self, label: Label) -> BaseStatus:
+    def print_label(self, label: Label, copies=1) -> BaseStatus:
         """Print the label"""
 
     @abstractmethod

brother_label_printer/printers/brother_pt700.py

@@ -2,13 +2,12 @@
 Brother P-Touch P700 Driver
 """
 import io
-import random
 import struct
+import threading
 import time
 from collections import namedtuple
 from enum import Enum, IntEnum
-from itertools import chain, islice
-from pprint import pprint
+from itertools import islice
 from typing import Iterable, Sequence
 from math import ceil
 import logging
@@ -28,6 +27,14 @@ def batch_iter_bytes(b, size):
     return iter(lambda: bytes(tuple(islice(i, size))), b"")
 
 
+def create_copies(b, size, copies):
+    result = list()
+    for i in range(copies):
+        result.append(batch_iter_bytes(b, size))
+
+    return result
+
+
 class INFO_OFFSETS(IntEnum):
     PRINTHEAD_MARK = 0
     MODEL_CODE = 4
@@ -171,6 +178,11 @@ class P700(BasePrinter):
     """Printer Class for the Brother P-Touch P700/PT-700 Printer"""
     DPI = (180, 180)
 
+    def __init__(self, io_obj: io.BufferedIOBase):
+        super().__init__(io_obj)
+        self.status = self.get_status()
+        self._check_print_status = False
+
     def connect(self) -> None:
         """Connect to Printer"""
         self.io.write(b'\x00' * 100)
@@ -201,62 +213,135 @@ def _debug_status(self):
     def get_label_width(self):
         return self.get_status().tape_info.width
 
-    def print_label(self, label: Label) -> Status:
-        status = self.get_status()
-        if not status.ready():
+    def print_label(self, label: Label, copies=1) -> Status:
+        self.status = self.get_status()
+        if not self.status.ready():
             raise IOError("Printer is not ready")
 
-        img = label.render(height=status.tape_info.printarea)
-        logger.debug("printarea is %s dots", status.tape_info.printarea)
+        img = label.render(height=self.status.tape_info.printarea)
+        logger.debug("printarea is %s dots", self.status.tape_info.printarea)
         if not img.mode == "1":
             raise ValueError("render output has invalid mode '1'")
         img = img.transpose(Image.ROTATE_270).transpose(
             Image.FLIP_TOP_BOTTOM)
         img = ImageChops.invert(img)
 
         logger.info("label output size: %s", img.size)
-        logger.info("tape info: %s", status.tape_info)
+        logger.info("tape info: %s", self.status.tape_info)
 
         img_bytes = img.tobytes()
 
         with self.io.lock:
             self._raw_print(
-                status, batch_iter_bytes(img_bytes, ceil(img.size[0] / 8)))
-
-        # wait for label to finish printing
-        time.sleep(len(img_bytes)/1000 if len(img_bytes)/1000 > 5 else 5)
+                self.status, create_copies(img_bytes, ceil(img.size[0] / 8), copies))
 
         return self.get_status()
 
     def _dummy_print(self, status: Status, document: Iterable[bytes]) -> None:
         for line in document:
-            encode_line(line, status.tape_info)
+            print(b'G' + encode_line(line, status.tape_info))
+        print('------')
+        for line in document:
+            print(b'G' + encode_line(line, status.tape_info))
+
+    def _print_status_check(self):
+        while self._check_print_status:
+            data = self.io.read(32)
+            if len(data) == 32:
+                self.status = Status(data)
+            time.sleep(0.1)
 
-    def _raw_print(self, status: Status, document: Iterable[bytes]) -> None:
+    def _raw_print(self, status: Status, documents: list[Iterable[bytes]]) -> None:
         logger.info("starting print")
 
         self.connect()
+        self._check_print_status = True
+
+        status_thread = threading.Thread(target=self._print_status_check)
+        status_thread.start()
+
+        try:
+            self.set_raster_mode()
+            self.set_various_mode()
+            self.set_advanced_mode()
+            self.set_margin(14)
+            self.set_compression_mode()
+
+            for i in range(len(documents)):
+                for line in documents[i]:
+                    self.io.write(b'G' + encode_line(line, status.tape_info))
+
+                self.print_empty_row()
+
+                if i+1 < len(documents):
+                    self.next_page()
+                    end = time.time() + 20
+                    while True:
+                        if self.status.data.get('status_type') == 6 and self.status.data.get('phase_type') == 0:
+                            break
+                        time.sleep(0.1)
+                        if time.time() > end:
+                            raise TimeoutError("The printer did not reach receiving state for the next page")
+
+            # end page
+            self.last_page_end()
+            logger.info("end of page")
+
+            end = time.time() + 20
+            while True:
+                if self.status.data.get('status_type') == 1:
+                    break
+                time.sleep(0.1)
+                if time.time() > end:
+                    raise TimeoutError("The printer did not reach printing complete state")
+
+        finally:
+            self._check_print_status = False
+            status_thread.join()
+
+    def last_page_end(self):
+        self.io.write(b'\x1A')
 
-        # raster mode
-        self.io.write(b'\x1B\x69\x69\x01')
-
-        # Various mode
-        self.io.write(b'\x1B\x69\x4D\x40')
-
-        # Advanced mode
-        self.io.write(b'\x1B\x69\x4B\x08')
-
-        # margin
-        self.io.write(b'\x1B\x69\x64\x0E\x00')
-
-        # Compression mode
-        self.io.write(b'\x4D\x02')
-
-        for line in document:
-            self.io.write(b'G' + encode_line(line, status.tape_info))
+    def next_page(self):
+        self.io.write(b'\x0C')
 
+    def print_empty_row(self):
         self.io.write(b'\x5A')
 
-        # end page
-        self.io.write(b'\x1A')
-        logger.info("end of page")
+    def set_compression_mode(self, tiff=True):
+        data = b'\x4D' + self.build_byte({1: tiff})
+        self.io.write(data)
+
+    def set_margin(self, margin: int):
+        data = b'\x1B\x69\x64' + margin.to_bytes(2, 'little')
+        self.io.write(data)
+
+    def set_advanced_mode(self, no_chain_printing=True, special_tape=False, no_buffer_clearing=False):
+        """
+        No chain printing
+        When printing multiple copies, the labels are fed after the last one is printed.
+        1:No chain printing(Feeding and cutting are performed after the last one is printed.)
+        0:Chain printing(Feeding and cutting are not performed after the last one is printed.)
+        Special tape (no cutting)
+        Labels are not cut when special tape is installed.
+        1.Special tape (no cutting) ON 0:Special tape (no cutting) OFF
+        No buffer clearing when printing
+        The expansion buffer of the machine is not cleared with the “no buffer clearing when printing”
+        """
+        data = b'\x1B\x69\x4B' + self.build_byte({3: no_chain_printing, 4: special_tape, 7: no_buffer_clearing})
+        self.io.write(data)
+
+    def set_various_mode(self, cut=True, mirror=False):
+        """
+        Autocut 1.Automatically cuts 0.Does not automatically cut
+        Mirror printing 1. Mirror printing 0. No mirror printing
+        """
+        data = b'\x1B\x69\x4D' + self.build_byte({6: cut, 7: mirror})
+        self.io.write(data)
+
+    def set_raster_mode(self):
+        self.io.write(b'\x1B\x69\x61\x01')
+
+    @staticmethod
+    def build_byte(bits: dict):
+        return bytes([int(''.join([str(int(bits.get(i, 0))) for i in reversed(range(8))]), 2)])

setup.py

@@ -4,7 +4,7 @@
     README_TEXT = readme.read()
 
 setup(
-    name="labelprinterkit-avis",
+    name="brother-label-printer",
     version="0.0.5",
     description="A library for creating and printing labels",
     use_scm_version=True,
@@ -28,5 +28,5 @@
         'pyserial',
         'qrcode'
     ],
-    python_requires='>=3.4'
+    python_requires='>=3.9'
 )