-
Notifications
You must be signed in to change notification settings - Fork 311
Expand file tree
/
Copy pathcups.py
More file actions
222 lines (178 loc) · 6.62 KB
/
cups.py
File metadata and controls
222 lines (178 loc) · 6.62 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""This module contains the implementation of the CupsPrinter printer driver.
:author: python-escpos developers
:organization: `python-escpos <https://github.com/python-escpos>`_
:copyright: Copyright (c) 2012-2023 Bashlinux and python-escpos
:license: MIT
"""
import functools
import logging
import tempfile
from typing import Literal, Optional, Type, Union
from ..escpos import Escpos
from ..exceptions import DeviceNotFoundError
#: keeps track if the pycups dependency could be loaded (:py:class:`escpos.printer.CupsPrinter`)
_DEP_PYCUPS = False
try:
import cups
_DEP_PYCUPS = True
# Store server defaults before further configuration
DEFAULT_HOST = cups.getServer()
DEFAULT_PORT = cups.getPort()
except ImportError:
pass
# TODO: dev build mode that let's the wrapper bypass?
def is_usable() -> bool:
"""Indicate whether this component can be used due to dependencies."""
usable = False
if _DEP_PYCUPS:
usable = True
return usable
def dependency_pycups(func):
"""Indicate dependency on pycups."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
"""Throw a RuntimeError if pycups is not imported."""
if not is_usable():
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."
)
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
"""
@staticmethod
def is_usable() -> bool:
"""Indicate whether this printer class is usable.
Will return True if dependencies are available.
Will return False if not.
"""
return is_usable()
@dependency_pycups
def __init__(self, printer_name: str = "", *args, **kwargs) -> None:
"""Class constructor for CupsPrinter.
:param printer_name: CUPS printer name (Optional)
:param host: CUPS server host/ip (Optional)
:type host: str
:param port: CUPS server port (Optional)
:type port: int
"""
Escpos.__init__(self, *args, **kwargs)
self.host, self.port = args or (
kwargs.get("host", DEFAULT_HOST),
kwargs.get("port", DEFAULT_PORT),
)
self.tmpfile = tempfile.NamedTemporaryFile(delete=True)
self.printer_name = printer_name
self.job_name = ""
self.pending_job = False
self._device: Union[
Literal[False], Literal[None], Type[cups.Connection]
] = False
@property
def printers(self) -> dict:
"""Available CUPS printers."""
if self.device:
return self.device.getPrinters()
return {}
def open(
self, job_name: str = "python-escpos", raise_not_found: bool = True
) -> None:
"""Set up a new print job and target the printer.
A call to this method is required to send new jobs to
the CUPS connection after close.
Defaults to default CUPS printer.
Creates a new temporary file buffer.
By default raise an exception if device is not found.
:param raise_not_found: Default True.
False to log error but do not raise exception.
:raises: :py:exc:`~escpos.exceptions.DeviceNotFoundError`
"""
if self._device:
self.close()
cups.setServer(self.host)
cups.setPort(self.port)
self.job_name = job_name
if self.tmpfile.closed:
self.tmpfile = tempfile.NamedTemporaryFile(delete=True)
try:
# Open device
self.device: Optional[Type[cups.Connection]] = cups.Connection()
if self.device:
# Name validation, set default if no given name
self.printer_name = self.printer_name or self.device.getDefault()
assert self.printer_name in self.printers, "Incorrect printer name"
except (RuntimeError, AssertionError) as e:
# Raise exception or log error and cancel
self.device = None
if raise_not_found:
raise DeviceNotFoundError(
f"Unable to start a print job for the printer {self.printer_name}:"
+ f"\n{e}"
)
else:
logging.error(
"CupsPrinter printing %s not available", self.printer_name
)
return
logging.info("CupsPrinter printer enabled")
def _raw(self, msg: bytes) -> None:
"""Append any command sent in raw format to temporary file.
:param msg: arbitrary code to be printed
"""
self.pending_job = True
try:
self.tmpfile.write(msg)
except TypeError:
self.pending_job = False
raise TypeError("Bytes required. Printer job not opened")
def send(self) -> None:
"""Send the print job to the printer."""
assert self.device
if self.pending_job:
# Rewind tempfile
self.tmpfile.seek(0)
# Print temporary file via CUPS printer.
self.device.printFile(
self.printer_name,
self.tmpfile.name,
self.job_name,
{"document-format": cups.CUPS_FORMAT_RAW},
)
self._clear()
def _clear(self) -> None:
"""Finish the print job.
Remove temporary file.
"""
self.tmpfile.close()
self.pending_job = False
def _read(self) -> bytes:
"""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 or state in [4, 5]:
return b"8" # offline
return b"0" # online
def close(self) -> None:
"""Close CUPS connection.
Send pending job to the printer if needed.
"""
if not self._device:
return
if self.pending_job:
self.send()
logging.info("Closing CUPS connection to printer %s", self.printer_name)
self._device = False