From 3f775fbaa82cc594341934555a9b04760320973c Mon Sep 17 00:00:00 2001 From: belono Date: Sun, 8 Oct 2023 21:16:23 +0200 Subject: [PATCH] Fix type checking - per device type annotations --- src/escpos/printer/cups.py | 4 +++- src/escpos/printer/file.py | 10 +++++++--- src/escpos/printer/lp.py | 10 ++++++++-- src/escpos/printer/network.py | 12 ++++++++---- src/escpos/printer/serial.py | 10 ++++++---- src/escpos/printer/usb.py | 6 ++++-- src/escpos/printer/win32raw.py | 14 ++++++++++---- 7 files changed, 46 insertions(+), 20 deletions(-) diff --git a/src/escpos/printer/cups.py b/src/escpos/printer/cups.py index a3168a5..98885e2 100644 --- a/src/escpos/printer/cups.py +++ b/src/escpos/printer/cups.py @@ -11,7 +11,7 @@ import functools import logging import tempfile -from typing import Optional, Type +from typing import Literal, Optional, Type, Union from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -74,6 +74,8 @@ class CupsPrinter(Escpos): """ + _device: Union[Literal[False], Literal[None], cups.Connection] = False + @staticmethod def is_usable() -> bool: """Indicate whether this printer class is usable. diff --git a/src/escpos/printer/file.py b/src/escpos/printer/file.py index 852d758..2eda7d1 100644 --- a/src/escpos/printer/file.py +++ b/src/escpos/printer/file.py @@ -9,6 +9,7 @@ """ import logging +from typing import IO, Literal, Optional, Union from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -33,6 +34,8 @@ class File(Escpos): """ + _device: Union[Literal[False], Literal[None], IO[bytes]] = False + @staticmethod def is_usable() -> bool: """Indicate whether this printer class is usable. @@ -67,7 +70,7 @@ class File(Escpos): try: # Open device - self.device = open(self.devfile, "wb") + self.device: Optional[IO[bytes]] = open(self.devfile, "wb") except OSError as e: # Raise exception or log error and cancel self.device = None @@ -82,7 +85,8 @@ class File(Escpos): def flush(self) -> None: """Flush printing content.""" - self.device.flush() + if self.device: + self.device.flush() def _raw(self, msg): """Print any command sent in raw format. @@ -101,5 +105,5 @@ class File(Escpos): logging.info("Closing File connection to printer %s", self.devfile) if not self.auto_flush: self.flush() - self.device.close() + self._device.close() self._device = False diff --git a/src/escpos/printer/lp.py b/src/escpos/printer/lp.py index 498b3b5..55c1b68 100644 --- a/src/escpos/printer/lp.py +++ b/src/escpos/printer/lp.py @@ -12,6 +12,7 @@ import functools import logging import subprocess import sys +from typing import Literal, Optional, Union from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -53,6 +54,8 @@ class LP(Escpos): """ + _device: Union[Literal[False], Literal[None], subprocess.Popen] = False + @staticmethod def is_usable() -> bool: """Indicate whether this printer class is usable. @@ -132,7 +135,7 @@ class LP(Escpos): self.printer_name = self.printer_name or self._get_system_default_printer() assert self.printer_name in self.printers, "Incorrect printer name" # Open device - self.device = subprocess.Popen( + self.device: Optional[subprocess.Popen] = subprocess.Popen( ["lp", "-d", self.printer_name, "-t", self.job_name, "-o", "raw"], stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, @@ -159,11 +162,14 @@ class LP(Escpos): self._is_closing = True if not self.auto_flush: self.flush() - self.device.terminate() + self._device.terminate() self._device = False def flush(self) -> None: """End line and wait for new commands.""" + if not self.device or not self.device.stdin: + return + if self._flushed: return diff --git a/src/escpos/printer/network.py b/src/escpos/printer/network.py index 16323e2..67313fd 100644 --- a/src/escpos/printer/network.py +++ b/src/escpos/printer/network.py @@ -10,7 +10,7 @@ import logging import socket -from typing import Union +from typing import Literal, Optional, Union from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -48,6 +48,8 @@ class Network(Escpos): """ + _device: Union[Literal[False], Literal[None], socket.socket] = False + @staticmethod def is_usable() -> bool: """Indicate whether this printer class is usable. @@ -91,7 +93,9 @@ class Network(Escpos): try: # Open device - self.device = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.device: Optional[socket.socket] = socket.socket( + socket.AF_INET, socket.SOCK_STREAM + ) self.device.settimeout(self.timeout) self.device.connect((self.host, self.port)) except OSError as e: @@ -124,8 +128,8 @@ class Network(Escpos): return logging.info("Closing Network connection to printer %s", self.host) try: - self.device.shutdown(socket.SHUT_RDWR) + self._device.shutdown(socket.SHUT_RDWR) except socket.error: pass - self.device.close() + self._device.close() self._device = False diff --git a/src/escpos/printer/serial.py b/src/escpos/printer/serial.py index 5ba10ce..bfae5fc 100644 --- a/src/escpos/printer/serial.py +++ b/src/escpos/printer/serial.py @@ -11,7 +11,7 @@ import functools import logging -from typing import Optional, Union +from typing import Literal, Optional, Union from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -64,6 +64,8 @@ class Serial(Escpos): """ + _device: Union[Literal[False], Literal[None], serial.Serial] = False + @staticmethod def is_usable() -> bool: """Indicate whether this printer class is usable. @@ -170,7 +172,7 @@ class Serial(Escpos): if not self._device: return logging.info("Closing Serial connection to printer %s", self.devfile) - if self.device.is_open: - self.device.flush() - self.device.close() + if self._device and self._device.is_open: + self._device.flush() + self._device.close() self._device = False diff --git a/src/escpos/printer/usb.py b/src/escpos/printer/usb.py index 3aa9646..3f45fc4 100644 --- a/src/escpos/printer/usb.py +++ b/src/escpos/printer/usb.py @@ -9,7 +9,7 @@ """ import functools import logging -from typing import Dict, Optional, Type, Union +from typing import Dict, Literal, Optional, Type, Union from ..escpos import Escpos from ..exceptions import DeviceNotFoundError, USBNotFoundError @@ -63,6 +63,8 @@ class Usb(Escpos): """ + _device: Union[Literal[False], Literal[None], Type[usb.core.Device]] = False + @staticmethod def is_usable() -> bool: """Indicate whether this printer class is usable. @@ -197,5 +199,5 @@ class Usb(Escpos): logging.info( "Closing Usb connection to printer %s", tuple(self.usb_args.values()) ) - usb.util.dispose_resources(self.device) + usb.util.dispose_resources(self._device) self._device = False diff --git a/src/escpos/printer/win32raw.py b/src/escpos/printer/win32raw.py index e367460..9c3e17e 100644 --- a/src/escpos/printer/win32raw.py +++ b/src/escpos/printer/win32raw.py @@ -10,6 +10,7 @@ import functools import logging +from typing import Literal, Optional, Union from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -18,6 +19,7 @@ from ..exceptions import DeviceNotFoundError _DEP_WIN32PRINT = False try: + import _win32typing import pywintypes import win32print @@ -63,6 +65,8 @@ class Win32Raw(Escpos): """ + _device: Union[Literal[False], Literal[None], _win32typing.PyPrinterHANDLE] = False + @staticmethod def is_usable() -> bool: """Indicate whether this printer class is usable. @@ -111,7 +115,9 @@ class Win32Raw(Escpos): self.printer_name = self.printer_name or win32print.GetDefaultPrinter() assert self.printer_name in self.printers, "Incorrect printer name" # Open device - self.device = win32print.OpenPrinter(self.printer_name) + self.device: Optional[ + _win32typing.PyPrinterHANDLE + ] = win32print.OpenPrinter(self.printer_name) if self.device: self.current_job = win32print.StartDocPrinter( hprinter=self.device, level=1, _tuple=(job_name, None, "RAW") @@ -136,9 +142,9 @@ class Win32Raw(Escpos): if not self._device: return logging.info("Closing Win32Raw connection to printer %s", self.printer_name) - win32print.EndPagePrinter(self.device) - win32print.EndDocPrinter(self.device) - win32print.ClosePrinter(self.device) + win32print.EndPagePrinter(self._device) + win32print.EndDocPrinter(self._device) + win32print.ClosePrinter(self._device) self._device = False @dependency_win32print