Merge branch 'master' into master

.github/workflows/black.yml

@@ -1,10 +1,13 @@
-name: Lint
+name: Lint (Black code style)
 
 on: [push, pull_request]
 
 jobs:
-  lint:
+  black-code-style:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
       - uses: psf/black@stable
+        with:
+          version: "23.3.0"
+

.github/workflows/documentation.yml

@@ -25,7 +25,8 @@ jobs:
       - name: Install packages
         run:
           sudo apt-get update -y &&
-          sudo apt-get install -y git python3-sphinx graphviz libenchant1c2a &&
+          sudo apt-get install -y git python3-sphinx graphviz libenchant-2-2 &&
-          sudo pip install tox
+          sudo apt-get install -y gcc libcups2-dev python3-dev python3-setuptools &&
+          sudo pip install tox pycups
       - name: Test doc build
         run: tox -e docs

.github/workflows/pythonpackage.yml

@@ -15,14 +15,14 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ['3.6', '3.7', '3.8', '3.9', '3.10']
+        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
 
     steps:
     - uses: actions/checkout@v3
       with:
         submodules: 'recursive'
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v4.5.0
+      uses: actions/setup-python@v4.6.0
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies

.gitignore

@@ -6,6 +6,7 @@ $~
 .idea/
 .directory
 .cache/
+settings.json
 
 # temporary data
 temp
@@ -22,6 +23,9 @@ src/escpos/version.py
 .hypothesis
 .pytest_cache/
 
+# pyenv
+.python-version
+
 # testing temporary directories
 test/test-cli-output/
 

.mailmap

@@ -14,3 +14,5 @@ Sergio Pulgarin <sergio.pulgarin@gmail.com>
 reck31 <rakesh.gunduka@gmail.com>
 Alex Debiasio <alex.debiasio@thinkin.io>             <alex.debiasio@studenti.unitn.it>
 Maximilian Wagenbach <maximilian.wagenbach@native-instruments.de>
+<belono@users.noreply.github.com>                    <tiotil.lindeman@gmail.com>
+belono <belono@users.noreply.github.com>             Benito López <belono@users.noreply.github.com>

.vscode/settings.json

@@ -1,5 +1,5 @@
 {
-  "restructuredtext.confPath": "${workspaceFolder}/doc",
+  "esbonio.sphinx.confDir": "${workspaceFolder}/doc",
   "files.watcherExclude": {
     "**/.git/objects/**": true,
     "**/.git/subtree-cache/**": true,

doc/conf.py

@@ -66,8 +66,8 @@ source_suffix = ".rst"
 master_doc = "index"
 
 # General information about the project.
-project = u"python-escpos"
+project = "python-escpos"
-copyright = u"2016, Manuel F Martinez and others"
+copyright = "2016, Manuel F Martinez and others"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the
@@ -229,8 +229,8 @@ latex_documents = [
     (
         "index",
         "python-escpos.tex",
-        u"python-escpos Documentation",
+        "python-escpos Documentation",
-        u"Manuel F Martinez and others",
+        "Manuel F Martinez and others",
         "manual",
     ),
 ]
@@ -264,8 +264,8 @@ man_pages = [
     (
         "index",
         "python-escpos",
-        u"python-escpos Documentation",
+        "python-escpos Documentation",
-        [u"Manuel F Martinez and others"],
+        ["Manuel F Martinez and others"],
         1,
     )
 ]
@@ -283,8 +283,8 @@ texinfo_documents = [
     (
         "index",
         "python-escpos",
-        u"python-escpos Documentation",
+        "python-escpos Documentation",
-        u"Manuel F Martinez and others",
+        "Manuel F Martinez and others",
         "python-escpos",
         "One line description of project.",
         "Miscellaneous",

doc/user/printers.rst

@@ -1,9 +1,9 @@
 ********
 Printers
 ********
-:Last Reviewed: 2017-01-25
+:Last Reviewed: 2022-11-25
 
-As of now there are 5 different type of printer implementations.
+As of now there are 7 different type of printer implementations.
 
 USB
 ---
@@ -75,3 +75,26 @@ all of the "output" as raw ESC/POS in a string and returns that.
       :member-order: bysource
       :noindex:
 
+CUPS
+----
+This driver uses `pycups` in order to communicate with a CUPS server.
+Supports both local and remote CUPS printers and servers.
+The printer must be properly configured in CUPS administration.
+The connector generates a print job that is added to the CUPS queue.
+
+.. todo:: fix import in documentation
+
+LP
+----
+This driver uses the UNIX command `lp` in order to communicate with a CUPS server.
+Supports local and remote CUPS printers.
+The printer must be properly configured in CUPS administration.
+The connector spawns a new sub-process where the command lp is executed.
+
+No dependencies required, but somehow the print queue will affect some print job such as barcode.
+
+.. autoclass:: escpos.printer.LP
+      :members:
+      :special-members:
+      :member-order: bysource
+      :noindex:

examples/weather.py

@@ -68,7 +68,7 @@ def forecast(idx):
     printer.text(deg)
     printer.text("\n")
     # take care of pesky unicode dash
-    printer.text(cond.replace(u"\u2013", "-").encode("utf-8"))
+    printer.text(cond.replace("\u2013", "-").encode("utf-8"))
     printer.text("\n \n")
 
 

setup.cfg

@@ -18,11 +18,11 @@ classifiers =
     Operating System :: OS Independent
     Programming Language :: Python
     Programming Language :: Python :: 3
-    Programming Language :: Python :: 3.6
     Programming Language :: Python :: 3.7
     Programming Language :: Python :: 3.8
     Programming Language :: Python :: 3.9
     Programming Language :: Python :: 3.10
+    Programming Language :: Python :: 3.11
     Programming Language :: Python :: Implementation :: CPython
     Topic :: Software Development :: Libraries :: Python Modules
     Topic :: Office/Business :: Financial :: Point-Of-Sale

src/escpos/escpos.py

@@ -592,7 +592,6 @@ class Escpos(object):
         text_distance=1,
         center=True,
     ):
-
         image_writer = ImageWriter()
 
         # Check if barcode type exists
@@ -1075,7 +1074,7 @@ class EscposIO(object):
         for line in lines:
             self.printer.set(**params)
             if isinstance(text, six.text_type):
-                self.printer.text(u"{0}\n".format(line))
+                self.printer.text("{0}\n".format(line))
             else:
                 self.printer.text("{0}\n".format(line))
 

src/escpos/magicencode.py

@@ -77,7 +77,7 @@ class Encoder(object):
             assert len(encodable_chars) == 128
             return encodable_chars
         elif "python_encode" in codepage:
-            encodable_chars = [u" "] * 128
+            encodable_chars = [" "] * 128
             for i in range(0, 128):
                 codepoint = i + 128
                 try:

src/escpos/printer.py

@@ -9,14 +9,35 @@
 """
 
 
-import serial
+import os
 import socket
+import subprocess
+import sys
+
+import serial
 import usb.core
 import usb.util
 
 from .escpos import Escpos
 from .exceptions import USBNotFoundError
 
+_WIN32PRINT = False
+try:
+    import win32print
+
+    _WIN32PRINT = True
+except ImportError:
+    pass
+
+_CUPSPRINT = False
+try:
+    import cups
+    import tempfile
+
+    _CUPSPRINT = True
+except ImportError:
+    pass
+
 
 class Usb(Escpos):
     """USB printer
@@ -371,14 +392,6 @@ class Dummy(Escpos):
         pass
 
 
-_WIN32PRINT = False
-try:
-    import win32print
-
-    _WIN32PRINT = True
-except ImportError:
-    pass
-
 if _WIN32PRINT:
 
     class Win32Raw(Escpos):
@@ -419,3 +432,171 @@ if _WIN32PRINT:
             if self.hPrinter is None:
                 raise Exception("Printer job not opened")
             win32print.WritePrinter(self.hPrinter, msg)
+
+
+if _CUPSPRINT:
+
+    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_
+        """
+
+        def __init__(self, printer_name=None, *args, **kwargs):
+            """CupsPrinter class constructor.
+
+            :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"):
+            """Setup a new print job and target 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
+                raise ValueError("Printer job not opened")
+
+        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
+
+
+if not sys.platform.startswith("win"):
+
+    class LP(Escpos):
+        """Simple UNIX lp command raw printing.
+
+        Thanks to `Oyami-Srk comment <https://github.com/python-escpos/python-escpos/pull/348#issuecomment-549558316>`_.
+        """
+
+        def __init__(self, printer_name: str, *args, **kwargs):
+            """LP class constructor.
+
+            :param printer_name: CUPS printer name (Optional)
+            :type printer_name: str
+            :param auto_flush: Automatic flush after every _raw() (Optional)
+            :type auto_flush: bool
+            """
+            Escpos.__init__(self, *args, **kwargs)
+            self.printer_name = printer_name
+            self.auto_flush = kwargs.get("auto_flush", True)
+            self.open()
+
+        def open(self):
+            """Invoke _lp_ in a new subprocess and wait for commands."""
+            self.lp = subprocess.Popen(
+                ["lp", "-d", self.printer_name, "-o", "raw"],
+                stdin=subprocess.PIPE,
+                stdout=open(os.devnull, "w"),
+            )
+
+        def close(self):
+            """Stop the subprocess."""
+            self.lp.terminate()
+
+        def flush(self):
+            """End line and wait for new commands"""
+            if self.lp.stdin.writable():
+                self.lp.stdin.write(b"\n")
+            if self.lp.stdin.closed is False:
+                self.lp.stdin.close()
+            self.lp.wait()
+            self.open()
+
+        def _raw(self, msg):
+            """Write raw command(s) to the printer.
+
+            :param msg: arbitrary code to be printed
+            :type msg: bytes
+            """
+            if self.lp.stdin.writable():
+                self.lp.stdin.write(msg)
+            else:
+                raise Exception("Not a valid pipe for lp process")
+            if self.auto_flush:
+                self.flush()

test/test_function_set.py

@@ -255,7 +255,6 @@ def test_align_right():
 
 
 def test_densities():
-
     for density in range(8):
         instance = printer.Dummy()
         instance.set(density=density)

test/test_magicencode.py

@@ -24,13 +24,13 @@ class TestEncoder:
     """
 
     def test_can_encode(self):
-        assert not Encoder({"CP437": 1}).can_encode("CP437", u"€")
+        assert not Encoder({"CP437": 1}).can_encode("CP437", "€")
-        assert Encoder({"CP437": 1}).can_encode("CP437", u"á")
+        assert Encoder({"CP437": 1}).can_encode("CP437", "á")
         assert not Encoder({"foobar": 1}).can_encode("foobar", "a")
 
     def test_find_suitable_encoding(self):
-        assert not Encoder({"CP437": 1}).find_suitable_encoding(u"€")
+        assert not Encoder({"CP437": 1}).find_suitable_encoding("€")
-        assert Encoder({"CP858": 1}).find_suitable_encoding(u"€") == "CP858"
+        assert Encoder({"CP858": 1}).find_suitable_encoding("€") == "CP858"
 
     @raises(ValueError)
     def test_get_encoding(self):
@@ -90,7 +90,7 @@ class TestMagicEncode:
                 encoder=Encoder({"CP437": 1}),
                 encoding="CP437",
             )
-            encode.write(u"€ ist teuro.")
+            encode.write("€ ist teuro.")
             assert driver.output == b"_ ist teuro."
 
     class TestForceEncoding:

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = py35, py36, py37, py38, py39, py310, docs, flake8
+envlist = py37, py38, py39, py310, py311, docs, flake8
 
 [gh-actions]
 python =
@@ -9,6 +9,7 @@ python =
     3.8: py38
     3.9: py39
     3.10: py310
+    3.11: py311
 
 [testenv]
 deps = nose
@@ -22,7 +23,7 @@ deps = nose
        hypothesis>4
        python-barcode
 commands = pytest --cov escpos
-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_*
 
 [testenv:docs]
 basepython = python