Fix type checking - per device type annotations
This commit is contained in:
parent
67c0ab03ba
commit
3f775fbaa8
@ -11,7 +11,7 @@
|
|||||||
import functools
|
import functools
|
||||||
import logging
|
import logging
|
||||||
import tempfile
|
import tempfile
|
||||||
from typing import Optional, Type
|
from typing import Literal, Optional, Type, Union
|
||||||
|
|
||||||
from ..escpos import Escpos
|
from ..escpos import Escpos
|
||||||
from ..exceptions import DeviceNotFoundError
|
from ..exceptions import DeviceNotFoundError
|
||||||
@ -74,6 +74,8 @@ class CupsPrinter(Escpos):
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
_device: Union[Literal[False], Literal[None], cups.Connection] = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_usable() -> bool:
|
def is_usable() -> bool:
|
||||||
"""Indicate whether this printer class is usable.
|
"""Indicate whether this printer class is usable.
|
||||||
|
@ -9,6 +9,7 @@
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
from typing import IO, Literal, Optional, Union
|
||||||
|
|
||||||
from ..escpos import Escpos
|
from ..escpos import Escpos
|
||||||
from ..exceptions import DeviceNotFoundError
|
from ..exceptions import DeviceNotFoundError
|
||||||
@ -33,6 +34,8 @@ class File(Escpos):
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
_device: Union[Literal[False], Literal[None], IO[bytes]] = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_usable() -> bool:
|
def is_usable() -> bool:
|
||||||
"""Indicate whether this printer class is usable.
|
"""Indicate whether this printer class is usable.
|
||||||
@ -67,7 +70,7 @@ class File(Escpos):
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
# Open device
|
# Open device
|
||||||
self.device = open(self.devfile, "wb")
|
self.device: Optional[IO[bytes]] = open(self.devfile, "wb")
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
# Raise exception or log error and cancel
|
# Raise exception or log error and cancel
|
||||||
self.device = None
|
self.device = None
|
||||||
@ -82,7 +85,8 @@ class File(Escpos):
|
|||||||
|
|
||||||
def flush(self) -> None:
|
def flush(self) -> None:
|
||||||
"""Flush printing content."""
|
"""Flush printing content."""
|
||||||
self.device.flush()
|
if self.device:
|
||||||
|
self.device.flush()
|
||||||
|
|
||||||
def _raw(self, msg):
|
def _raw(self, msg):
|
||||||
"""Print any command sent in raw format.
|
"""Print any command sent in raw format.
|
||||||
@ -101,5 +105,5 @@ class File(Escpos):
|
|||||||
logging.info("Closing File connection to printer %s", self.devfile)
|
logging.info("Closing File connection to printer %s", self.devfile)
|
||||||
if not self.auto_flush:
|
if not self.auto_flush:
|
||||||
self.flush()
|
self.flush()
|
||||||
self.device.close()
|
self._device.close()
|
||||||
self._device = False
|
self._device = False
|
||||||
|
@ -12,6 +12,7 @@ import functools
|
|||||||
import logging
|
import logging
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
|
from typing import Literal, Optional, Union
|
||||||
|
|
||||||
from ..escpos import Escpos
|
from ..escpos import Escpos
|
||||||
from ..exceptions import DeviceNotFoundError
|
from ..exceptions import DeviceNotFoundError
|
||||||
@ -53,6 +54,8 @@ class LP(Escpos):
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
_device: Union[Literal[False], Literal[None], subprocess.Popen] = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_usable() -> bool:
|
def is_usable() -> bool:
|
||||||
"""Indicate whether this printer class is usable.
|
"""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()
|
self.printer_name = self.printer_name or self._get_system_default_printer()
|
||||||
assert self.printer_name in self.printers, "Incorrect printer name"
|
assert self.printer_name in self.printers, "Incorrect printer name"
|
||||||
# Open device
|
# Open device
|
||||||
self.device = subprocess.Popen(
|
self.device: Optional[subprocess.Popen] = subprocess.Popen(
|
||||||
["lp", "-d", self.printer_name, "-t", self.job_name, "-o", "raw"],
|
["lp", "-d", self.printer_name, "-t", self.job_name, "-o", "raw"],
|
||||||
stdin=subprocess.PIPE,
|
stdin=subprocess.PIPE,
|
||||||
stdout=subprocess.DEVNULL,
|
stdout=subprocess.DEVNULL,
|
||||||
@ -159,11 +162,14 @@ class LP(Escpos):
|
|||||||
self._is_closing = True
|
self._is_closing = True
|
||||||
if not self.auto_flush:
|
if not self.auto_flush:
|
||||||
self.flush()
|
self.flush()
|
||||||
self.device.terminate()
|
self._device.terminate()
|
||||||
self._device = False
|
self._device = False
|
||||||
|
|
||||||
def flush(self) -> None:
|
def flush(self) -> None:
|
||||||
"""End line and wait for new commands."""
|
"""End line and wait for new commands."""
|
||||||
|
if not self.device or not self.device.stdin:
|
||||||
|
return
|
||||||
|
|
||||||
if self._flushed:
|
if self._flushed:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
@ -10,7 +10,7 @@
|
|||||||
|
|
||||||
import logging
|
import logging
|
||||||
import socket
|
import socket
|
||||||
from typing import Union
|
from typing import Literal, Optional, Union
|
||||||
|
|
||||||
from ..escpos import Escpos
|
from ..escpos import Escpos
|
||||||
from ..exceptions import DeviceNotFoundError
|
from ..exceptions import DeviceNotFoundError
|
||||||
@ -48,6 +48,8 @@ class Network(Escpos):
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
_device: Union[Literal[False], Literal[None], socket.socket] = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_usable() -> bool:
|
def is_usable() -> bool:
|
||||||
"""Indicate whether this printer class is usable.
|
"""Indicate whether this printer class is usable.
|
||||||
@ -91,7 +93,9 @@ class Network(Escpos):
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
# Open device
|
# 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.settimeout(self.timeout)
|
||||||
self.device.connect((self.host, self.port))
|
self.device.connect((self.host, self.port))
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
@ -124,8 +128,8 @@ class Network(Escpos):
|
|||||||
return
|
return
|
||||||
logging.info("Closing Network connection to printer %s", self.host)
|
logging.info("Closing Network connection to printer %s", self.host)
|
||||||
try:
|
try:
|
||||||
self.device.shutdown(socket.SHUT_RDWR)
|
self._device.shutdown(socket.SHUT_RDWR)
|
||||||
except socket.error:
|
except socket.error:
|
||||||
pass
|
pass
|
||||||
self.device.close()
|
self._device.close()
|
||||||
self._device = False
|
self._device = False
|
||||||
|
@ -11,7 +11,7 @@
|
|||||||
|
|
||||||
import functools
|
import functools
|
||||||
import logging
|
import logging
|
||||||
from typing import Optional, Union
|
from typing import Literal, Optional, Union
|
||||||
|
|
||||||
from ..escpos import Escpos
|
from ..escpos import Escpos
|
||||||
from ..exceptions import DeviceNotFoundError
|
from ..exceptions import DeviceNotFoundError
|
||||||
@ -64,6 +64,8 @@ class Serial(Escpos):
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
_device: Union[Literal[False], Literal[None], serial.Serial] = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_usable() -> bool:
|
def is_usable() -> bool:
|
||||||
"""Indicate whether this printer class is usable.
|
"""Indicate whether this printer class is usable.
|
||||||
@ -170,7 +172,7 @@ class Serial(Escpos):
|
|||||||
if not self._device:
|
if not self._device:
|
||||||
return
|
return
|
||||||
logging.info("Closing Serial connection to printer %s", self.devfile)
|
logging.info("Closing Serial connection to printer %s", self.devfile)
|
||||||
if self.device.is_open:
|
if self._device and self._device.is_open:
|
||||||
self.device.flush()
|
self._device.flush()
|
||||||
self.device.close()
|
self._device.close()
|
||||||
self._device = False
|
self._device = False
|
||||||
|
@ -9,7 +9,7 @@
|
|||||||
"""
|
"""
|
||||||
import functools
|
import functools
|
||||||
import logging
|
import logging
|
||||||
from typing import Dict, Optional, Type, Union
|
from typing import Dict, Literal, Optional, Type, Union
|
||||||
|
|
||||||
from ..escpos import Escpos
|
from ..escpos import Escpos
|
||||||
from ..exceptions import DeviceNotFoundError, USBNotFoundError
|
from ..exceptions import DeviceNotFoundError, USBNotFoundError
|
||||||
@ -63,6 +63,8 @@ class Usb(Escpos):
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
_device: Union[Literal[False], Literal[None], Type[usb.core.Device]] = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_usable() -> bool:
|
def is_usable() -> bool:
|
||||||
"""Indicate whether this printer class is usable.
|
"""Indicate whether this printer class is usable.
|
||||||
@ -197,5 +199,5 @@ class Usb(Escpos):
|
|||||||
logging.info(
|
logging.info(
|
||||||
"Closing Usb connection to printer %s", tuple(self.usb_args.values())
|
"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
|
self._device = False
|
||||||
|
@ -10,6 +10,7 @@
|
|||||||
|
|
||||||
import functools
|
import functools
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Literal, Optional, Union
|
||||||
|
|
||||||
from ..escpos import Escpos
|
from ..escpos import Escpos
|
||||||
from ..exceptions import DeviceNotFoundError
|
from ..exceptions import DeviceNotFoundError
|
||||||
@ -18,6 +19,7 @@ from ..exceptions import DeviceNotFoundError
|
|||||||
_DEP_WIN32PRINT = False
|
_DEP_WIN32PRINT = False
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
import _win32typing
|
||||||
import pywintypes
|
import pywintypes
|
||||||
import win32print
|
import win32print
|
||||||
|
|
||||||
@ -63,6 +65,8 @@ class Win32Raw(Escpos):
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
_device: Union[Literal[False], Literal[None], _win32typing.PyPrinterHANDLE] = False
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_usable() -> bool:
|
def is_usable() -> bool:
|
||||||
"""Indicate whether this printer class is usable.
|
"""Indicate whether this printer class is usable.
|
||||||
@ -111,7 +115,9 @@ class Win32Raw(Escpos):
|
|||||||
self.printer_name = self.printer_name or win32print.GetDefaultPrinter()
|
self.printer_name = self.printer_name or win32print.GetDefaultPrinter()
|
||||||
assert self.printer_name in self.printers, "Incorrect printer name"
|
assert self.printer_name in self.printers, "Incorrect printer name"
|
||||||
# Open device
|
# Open device
|
||||||
self.device = win32print.OpenPrinter(self.printer_name)
|
self.device: Optional[
|
||||||
|
_win32typing.PyPrinterHANDLE
|
||||||
|
] = win32print.OpenPrinter(self.printer_name)
|
||||||
if self.device:
|
if self.device:
|
||||||
self.current_job = win32print.StartDocPrinter(
|
self.current_job = win32print.StartDocPrinter(
|
||||||
hprinter=self.device, level=1, _tuple=(job_name, None, "RAW")
|
hprinter=self.device, level=1, _tuple=(job_name, None, "RAW")
|
||||||
@ -136,9 +142,9 @@ class Win32Raw(Escpos):
|
|||||||
if not self._device:
|
if not self._device:
|
||||||
return
|
return
|
||||||
logging.info("Closing Win32Raw connection to printer %s", self.printer_name)
|
logging.info("Closing Win32Raw connection to printer %s", self.printer_name)
|
||||||
win32print.EndPagePrinter(self.device)
|
win32print.EndPagePrinter(self._device)
|
||||||
win32print.EndDocPrinter(self.device)
|
win32print.EndDocPrinter(self._device)
|
||||||
win32print.ClosePrinter(self.device)
|
win32print.ClosePrinter(self._device)
|
||||||
self._device = False
|
self._device = False
|
||||||
|
|
||||||
@dependency_win32print
|
@dependency_win32print
|
||||||
|
Loading…
x
Reference in New Issue
Block a user