add wrapper that thros RuntimeError if not importable for pycups

This commit is contained in:
Patrick Kanzler 2023-08-17 00:18:30 +02:00
parent 1a2273a5b3
commit 33329867d2
3 changed files with 134 additions and 111 deletions

View File

@ -2,7 +2,7 @@
"""printer implementations.""" """printer implementations."""
# from .win32raw import Win32Raw # from .win32raw import Win32Raw
# from .cups import CupsPrinter from .cups import CupsPrinter
from .dummy import Dummy from .dummy import Dummy
from .file import File from .file import File
from .lp import LP from .lp import LP
@ -17,7 +17,7 @@ __all__ = [
"Serial", "Serial",
"LP", "LP",
"Dummy", "Dummy",
# "CupsPrinter", "CupsPrinter",
# "Win32Raw", # "Win32Raw",
] ]

View File

@ -8,132 +8,156 @@
:license: MIT :license: MIT
""" """
import functools
import tempfile
from ..escpos import Escpos from ..escpos import Escpos
_CUPSPRINT = False #: keeps track if the pycups dependency could be loaded (:py:class:`escpos.printer.CupsPrinter`)
try: _DEP_PYCUPS = False
import tempfile
try:
import cups import cups
_CUPSPRINT = True _DEP_PYCUPS = True
except ImportError: except ImportError:
pass pass
if _CUPSPRINT:
class CupsPrinter(Escpos): # TODO: dev build mode that let's the wrapper bypass?
"""Simple CUPS printer connector.
.. note::
Requires ``pycups`` which in turn needs the cups development library package: def dependency_pycups(func):
- Ubuntu/Debian: ``libcups2-dev`` """Indicate dependency on pycups."""
- OpenSuse/Fedora: ``cups-devel``
inheritance: @functools.wraps(func)
def wrapper(*args, **kwargs):
.. inheritance-diagram:: escpos.printer.CupsPrinter """Throw a RuntimeError if pycups is not imported."""
:parts: 1 if not _DEP_PYCUPS:
raise RuntimeError(
""" "Printing with PyCups requires the pycups library to"
"be installed. Please refer to the documentation on"
def __init__(self, printer_name=None, *args, **kwargs): "what to install and install the dependencies for pycups."
"""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) return func(*args, **kwargs)
cups.setPort(port)
self.conn = cups.Connection() return wrapper
self.tmpfile = None
self.printer_name = printer_name
self.job_name = "" 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.pending_job = False
self.open() raise ValueError("Printer job not opened")
@property @dependency_pycups
def printers(self): def send(self):
"""Available CUPS printers.""" """Send the print job to the printer."""
return self.conn.getPrinters() 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"): def _clear(self):
"""Set up a new print job and target the printer. """Finish the print job.
A call to this method is required to send new jobs to Remove temporary file.
the same CUPS connection. """
self.tmpfile.close()
self.pending_job = False
Defaults to default CUPS printer. def _read(self):
Creates a new temporary file buffer. """Return a single-item array with the accepting state of the print queue.
"""
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): states: idle = [3], printing a job = [4], stopped = [5]
"""Append any command sent in raw format to temporary file. """
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 def close(self):
:type msg: bytes """Close CUPS connection.
"""
self.pending_job = True
try:
self.tmpfile.write(msg)
except ValueError:
self.pending_job = False
raise ValueError("Printer job not opened")
def send(self): Send pending job to the printer if needed.
"""Send the print job to the printer.""" """
if self.pending_job: if self.pending_job:
# Rewind tempfile self.send()
self.tmpfile.seek(0) if self.conn:
# Print temporary file via CUPS printer. print("Closing CUPS connection to printer {}".format(self.printer_name))
self.conn.printFile( self.conn = None
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

View File

@ -22,7 +22,6 @@ deps = jaconv
pytest-mock pytest-mock
hypothesis>4 hypothesis>4
python-barcode python-barcode
pycups
commands = pytest commands = pytest
passenv = ESCPOS_CAPABILITIES_PICKLE_DIR, ESCPOS_CAPABILITIES_FILE, CI, TRAVIS, TRAVIS_*, APPVEYOR, APPVEYOR_*, CODECOV_* passenv = ESCPOS_CAPABILITIES_PICKLE_DIR, ESCPOS_CAPABILITIES_FILE, CI, TRAVIS, TRAVIS_*, APPVEYOR, APPVEYOR_*, CODECOV_*
setenv = PY_IGNORE_IMPORTMISMATCH=1 setenv = PY_IGNORE_IMPORTMISMATCH=1