From 33329867d222b97066816549bab8e53a62d1e767 Mon Sep 17 00:00:00 2001 From: Patrick Kanzler Date: Thu, 17 Aug 2023 00:18:30 +0200 Subject: [PATCH] add wrapper that thros RuntimeError if not importable for pycups --- src/escpos/printer/__init__.py | 4 +- src/escpos/printer/cups.py | 240 ++++++++++++++++++--------------- tox.ini | 1 - 3 files changed, 134 insertions(+), 111 deletions(-) diff --git a/src/escpos/printer/__init__.py b/src/escpos/printer/__init__.py index 0d37832..21d07f8 100644 --- a/src/escpos/printer/__init__.py +++ b/src/escpos/printer/__init__.py @@ -2,7 +2,7 @@ """printer implementations.""" # from .win32raw import Win32Raw -# from .cups import CupsPrinter +from .cups import CupsPrinter from .dummy import Dummy from .file import File from .lp import LP @@ -17,7 +17,7 @@ __all__ = [ "Serial", "LP", "Dummy", - # "CupsPrinter", + "CupsPrinter", # "Win32Raw", ] diff --git a/src/escpos/printer/cups.py b/src/escpos/printer/cups.py index 1e5c715..472ee54 100644 --- a/src/escpos/printer/cups.py +++ b/src/escpos/printer/cups.py @@ -8,132 +8,156 @@ :license: MIT """ +import functools +import tempfile + from ..escpos import Escpos -_CUPSPRINT = False -try: - import tempfile +#: keeps track if the pycups dependency could be loaded (:py:class:`escpos.printer.CupsPrinter`) +_DEP_PYCUPS = False +try: import cups - _CUPSPRINT = True + _DEP_PYCUPS = True except ImportError: pass -if _CUPSPRINT: - class CupsPrinter(Escpos): - """Simple CUPS printer connector. +# TODO: dev build mode that let's the wrapper bypass? - .. note:: - Requires ``pycups`` which in turn needs the cups development library package: - - Ubuntu/Debian: ``libcups2-dev`` - - OpenSuse/Fedora: ``cups-devel`` +def dependency_pycups(func): + """Indicate dependency on pycups.""" - inheritance: - - .. inheritance-diagram:: escpos.printer.CupsPrinter - :parts: 1 - - """ - - def __init__(self, printer_name=None, *args, **kwargs): - """Class constructor for CupsPrinter. - - :param printer_name: CUPS printer name (Optional) - :type printer_name: str - :param host: CUPS server host/ip (Optional) - :type host: str - :param port: CUPS server port (Optional) - :type port: int - """ - Escpos.__init__(self, *args, **kwargs) - host, port = args or ( - kwargs.get("host", cups.getServer()), - kwargs.get("port", cups.getPort()), + @functools.wraps(func) + def wrapper(*args, **kwargs): + """Throw a RuntimeError if pycups is not imported.""" + if not _DEP_PYCUPS: + raise RuntimeError( + "Printing with PyCups requires the pycups library to" + "be installed. Please refer to the documentation on" + "what to install and install the dependencies for pycups." ) - cups.setServer(host) - cups.setPort(port) - self.conn = cups.Connection() - self.tmpfile = None - self.printer_name = printer_name - self.job_name = "" + return func(*args, **kwargs) + + return wrapper + + +class CupsPrinter(Escpos): + """Simple CUPS printer connector. + + .. note:: + + Requires ``pycups`` which in turn needs the cups development library package: + - Ubuntu/Debian: ``libcups2-dev`` + - OpenSuse/Fedora: ``cups-devel`` + + inheritance: + + .. inheritance-diagram:: escpos.printer.CupsPrinter + :parts: 1 + + """ + + @dependency_pycups + def __init__(self, printer_name=None, *args, **kwargs): + """Class constructor for CupsPrinter. + + :param printer_name: CUPS printer name (Optional) + :type printer_name: str + :param host: CUPS server host/ip (Optional) + :type host: str + :param port: CUPS server port (Optional) + :type port: int + """ + Escpos.__init__(self, *args, **kwargs) + host, port = args or ( + kwargs.get("host", cups.getServer()), + kwargs.get("port", cups.getPort()), + ) + cups.setServer(host) + cups.setPort(port) + self.conn = cups.Connection() + self.tmpfile = None + self.printer_name = printer_name + self.job_name = "" + self.pending_job = False + self.open() + + @property + def printers(self): + """Available CUPS printers.""" + return self.conn.getPrinters() + + def open(self, job_name="python-escpos"): + """Set up a new print job and target the printer. + + A call to this method is required to send new jobs to + the same CUPS connection. + + Defaults to default CUPS printer. + Creates a new temporary file buffer. + """ + self.job_name = job_name + if self.printer_name not in self.printers: + self.printer_name = self.conn.getDefault() + self.tmpfile = tempfile.NamedTemporaryFile(delete=True) + + def _raw(self, msg): + """Append any command sent in raw format to temporary file. + + :param msg: arbitrary code to be printed + :type msg: bytes + """ + self.pending_job = True + try: + self.tmpfile.write(msg) + except ValueError: self.pending_job = False - self.open() + raise ValueError("Printer job not opened") - @property - def printers(self): - """Available CUPS printers.""" - return self.conn.getPrinters() + @dependency_pycups + def send(self): + """Send the print job to the printer.""" + if self.pending_job: + # Rewind tempfile + self.tmpfile.seek(0) + # Print temporary file via CUPS printer. + self.conn.printFile( + self.printer_name, + self.tmpfile.name, + self.job_name, + {"document-format": cups.CUPS_FORMAT_RAW}, + ) + self._clear() - def open(self, job_name="python-escpos"): - """Set up a new print job and target the printer. + def _clear(self): + """Finish the print job. - A call to this method is required to send new jobs to - the same CUPS connection. + Remove temporary file. + """ + self.tmpfile.close() + self.pending_job = False - Defaults to default CUPS printer. - Creates a new temporary file buffer. - """ - self.job_name = job_name - if self.printer_name not in self.printers: - self.printer_name = self.conn.getDefault() - self.tmpfile = tempfile.NamedTemporaryFile(delete=True) + def _read(self): + """Return a single-item array with the accepting state of the print queue. - def _raw(self, msg): - """Append any command sent in raw format to temporary file. + states: idle = [3], printing a job = [4], stopped = [5] + """ + printer = self.printers.get(self.printer_name, {}) + state = printer.get("printer-state") + if not state: + return [] + return [state] - :param msg: arbitrary code to be printed - :type msg: bytes - """ - self.pending_job = True - try: - self.tmpfile.write(msg) - except ValueError: - self.pending_job = False - raise ValueError("Printer job not opened") + def close(self): + """Close CUPS connection. - def send(self): - """Send the print job to the printer.""" - if self.pending_job: - # Rewind tempfile - self.tmpfile.seek(0) - # Print temporary file via CUPS printer. - self.conn.printFile( - self.printer_name, - self.tmpfile.name, - self.job_name, - {"document-format": cups.CUPS_FORMAT_RAW}, - ) - self._clear() - - def _clear(self): - """Finish the print job. - - Remove temporary file. - """ - self.tmpfile.close() - self.pending_job = False - - def _read(self): - """Return a single-item array with the accepting state of the print queue. - - states: idle = [3], printing a job = [4], stopped = [5] - """ - printer = self.printers.get(self.printer_name, {}) - state = printer.get("printer-state") - if not state: - return [] - return [state] - - def close(self): - """Close CUPS connection. - - Send pending job to the printer if needed. - """ - if self.pending_job: - self.send() - if self.conn: - print("Closing CUPS connection to printer {}".format(self.printer_name)) - self.conn = None + Send pending job to the printer if needed. + """ + if self.pending_job: + self.send() + if self.conn: + print("Closing CUPS connection to printer {}".format(self.printer_name)) + self.conn = None diff --git a/tox.ini b/tox.ini index 7de5f32..9106c37 100644 --- a/tox.ini +++ b/tox.ini @@ -22,7 +22,6 @@ deps = jaconv pytest-mock hypothesis>4 python-barcode - pycups commands = pytest passenv = ESCPOS_CAPABILITIES_PICKLE_DIR, ESCPOS_CAPABILITIES_FILE, CI, TRAVIS, TRAVIS_*, APPVEYOR, APPVEYOR_*, CODECOV_* setenv = PY_IGNORE_IMPORTMISMATCH=1