From e5939959301d70c7acce7328c7625adbb9924ad1 Mon Sep 17 00:00:00 2001 From: belono Date: Thu, 2 Oct 2025 20:56:06 +0200 Subject: [PATCH 1/8] Improve on/offline masks consistency and its usage by the Cups connector --- src/escpos/constants.py | 3 ++- src/escpos/escpos.py | 4 ++-- src/escpos/printer/cups.py | 5 +++-- test/test_printers/test_printer_cups.py | 6 +++--- 4 files changed, 10 insertions(+), 8 deletions(-) diff --git a/src/escpos/constants.py b/src/escpos/constants.py index 34684840..976d620f 100644 --- a/src/escpos/constants.py +++ b/src/escpos/constants.py @@ -299,7 +299,8 @@ RT_STATUS: bytes = DLE + EOT RT_STATUS_ONLINE: bytes = RT_STATUS + b"\x01" RT_STATUS_PAPER: bytes = RT_STATUS + b"\x04" -RT_MASK_ONLINE: int = 8 +RT_MASK_ONLINE: int = 18 +RT_MASK_OFFLINE: int = 26 RT_MASK_PAPER: int = 18 RT_MASK_LOWPAPER: int = 30 RT_MASK_NOPAPER: int = 114 diff --git a/src/escpos/escpos.py b/src/escpos/escpos.py index fc00e713..f28bd3f9 100644 --- a/src/escpos/escpos.py +++ b/src/escpos/escpos.py @@ -72,7 +72,7 @@ QR_MODEL_2, RT_MASK_LOWPAPER, RT_MASK_NOPAPER, - RT_MASK_ONLINE, + RT_MASK_OFFLINE, RT_MASK_PAPER, RT_STATUS_ONLINE, RT_STATUS_PAPER, @@ -1442,7 +1442,7 @@ def is_online(self) -> bool: status = self.query_status(RT_STATUS_ONLINE) if len(status) == 0: return False - return not (status[0] & RT_MASK_ONLINE) + return not (status[0] & RT_MASK_OFFLINE == RT_MASK_OFFLINE) def paper_status(self) -> int: # could be IntEnum """Query the paper status of the printer. diff --git a/src/escpos/printer/cups.py b/src/escpos/printer/cups.py index bdd53c31..c9f0b1dd 100644 --- a/src/escpos/printer/cups.py +++ b/src/escpos/printer/cups.py @@ -13,6 +13,7 @@ import tempfile from typing import Literal, Optional, Type, Union +from ..constants import RT_MASK_ONLINE, RT_MASK_OFFLINE from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -206,8 +207,8 @@ def _read(self) -> bytes: 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 + return bytes([RT_MASK_OFFLINE]) + return bytes([RT_MASK_ONLINE]) def close(self) -> None: """Close CUPS connection. diff --git a/test/test_printers/test_printer_cups.py b/test/test_printers/test_printer_cups.py index 85edfe32..49ee46d6 100644 --- a/test/test_printers/test_printer_cups.py +++ b/test/test_printers/test_printer_cups.py @@ -169,8 +169,8 @@ def test_printers_no_device(cupsprinter) -> None: def test_read_no_device(cupsprinter) -> None: """ GIVEN a cups printer object - WHEN device is None - THEN check the return value is b'8' + WHEN device is None while querying for printer status + THEN check the return value is the byte '\x1a' """ cupsprinter.device = None - assert cupsprinter._read() == b"8" + assert cupsprinter._read() == bytes([26]) From 3b32f8b75e5f56e324b2db160d22a244d383ada4 Mon Sep 17 00:00:00 2001 From: belono Date: Fri, 3 Oct 2025 00:12:59 +0200 Subject: [PATCH 2/8] Fix sorting --- src/escpos/printer/cups.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/escpos/printer/cups.py b/src/escpos/printer/cups.py index c9f0b1dd..02d09306 100644 --- a/src/escpos/printer/cups.py +++ b/src/escpos/printer/cups.py @@ -13,7 +13,7 @@ import tempfile from typing import Literal, Optional, Type, Union -from ..constants import RT_MASK_ONLINE, RT_MASK_OFFLINE +from ..constants import RT_MASK_OFFLINE, RT_MASK_ONLINE from ..escpos import Escpos from ..exceptions import DeviceNotFoundError From 0b00174ae9266d6f31fc5c25aebd35a4254ab072 Mon Sep 17 00:00:00 2001 From: belono Date: Sun, 19 Oct 2025 16:37:58 +0200 Subject: [PATCH 3/8] Add test for the _read() method on the cups printer --- test/test_printers/test_printer_cups.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/test/test_printers/test_printer_cups.py b/test/test_printers/test_printer_cups.py index 49ee46d6..cf28987c 100644 --- a/test/test_printers/test_printer_cups.py +++ b/test/test_printers/test_printer_cups.py @@ -174,3 +174,28 @@ def test_read_no_device(cupsprinter) -> None: """ cupsprinter.device = None assert cupsprinter._read() == bytes([26]) + + +@pytest.mark.parametrize( + "state,expected", + [ + pytest.param(3, bytes([18])), + pytest.param(4, bytes([26])), + pytest.param(5, bytes([26])), + ], +) +def test_read(cupsprinter, mocker, state, expected): + """ + GIVEN a cups printer object and a mocked pycups device + WHEN querying for printer status + THEN check the return value is the expected + """ + mocker.patch("cups.Connection") + mocker.patch( + "escpos.printer.CupsPrinter.printers", + new={"test_printer": {"printer-state": state}}, + ) + + cupsprinter.printer_name = "test_printer" + + assert cupsprinter._read() == expected From 3a6d64517b6f2aacbbf294ea7683e338cbd99866 Mon Sep 17 00:00:00 2001 From: belono Date: Mon, 20 Oct 2025 10:29:43 +0200 Subject: [PATCH 4/8] Add missing function return value annotation --- test/test_printers/test_printer_cups.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/test_printers/test_printer_cups.py b/test/test_printers/test_printer_cups.py index cf28987c..e7d76835 100644 --- a/test/test_printers/test_printer_cups.py +++ b/test/test_printers/test_printer_cups.py @@ -184,7 +184,7 @@ def test_read_no_device(cupsprinter) -> None: pytest.param(5, bytes([26])), ], ) -def test_read(cupsprinter, mocker, state, expected): +def test_read(cupsprinter, mocker, state, expected) -> None: """ GIVEN a cups printer object and a mocked pycups device WHEN querying for printer status From 6d1b6c8117ac53ca86f8547450265c5f6a4306ba Mon Sep 17 00:00:00 2001 From: belono Date: Mon, 13 Apr 2026 21:07:02 +0200 Subject: [PATCH 5/8] Implement _read() into the LP() connector --- src/escpos/printer/lp.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/escpos/printer/lp.py b/src/escpos/printer/lp.py index fafab262..cfc7ec8c 100644 --- a/src/escpos/printer/lp.py +++ b/src/escpos/printer/lp.py @@ -14,6 +14,7 @@ import sys from typing import Literal, Optional, Union +from ..constants import RT_MASK_OFFLINE, RT_MASK_ONLINE from ..escpos import Escpos from ..exceptions import DeviceNotFoundError @@ -154,6 +155,20 @@ def open( return logging.info("LP printer enabled") + def _read(self) -> bytes: + """Return the byte corresponding to the RT status response. + + Respond on/offline given the accepting state of the print queue. + """ + p_name = subprocess.run( + ["lpstat", "-a", self.printer_name], + capture_output=True, + text=True, + ) + if p_name.returncode > 0 or "Rejecting Jobs" in p_name.stdout: + return bytes([RT_MASK_OFFLINE]) + return bytes([RT_MASK_ONLINE]) + def close(self) -> None: """Stop the subprocess.""" if not self._device: From b7abdbf4aa2d148d3e03d81ff754065e954618e0 Mon Sep 17 00:00:00 2001 From: belono Date: Mon, 13 Apr 2026 21:07:43 +0200 Subject: [PATCH 6/8] Update LP printer tests --- test/test_printers/test_printer_lp.py | 30 +++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/test/test_printers/test_printer_lp.py b/test/test_printers/test_printer_lp.py index fac00328..11da04d1 100644 --- a/test/test_printers/test_printer_lp.py +++ b/test/test_printers/test_printer_lp.py @@ -154,6 +154,36 @@ def test_auto_flush_on_close(lpprinter, mocker, caplog, capsys): assert spy.call_count == 1 +@pytest.mark.parametrize( + "stdout,returncode,expected", + [ + pytest.param("A success message...", 0, bytes([18])), + pytest.param("... Rejecting Jobs", 0, bytes([26])), + pytest.param("An error message...", 1, bytes([26])), + ], +) +def test_read(lpprinter, mocker, stdout, returncode, expected) -> None: + """ + GIVEN a lp printer object and a mocked connection + WHEN querying for printer status + THEN check the return value is the expected + """ + import subprocess + + mocker.patch("escpos.printer.LP.printers", new={"test_printer": "Test"}) + mocker.patch.object( + subprocess, + "run", + return_value=subprocess.CompletedProcess( + [], returncode=returncode, stdout=stdout + ), + ) + + lpprinter.printer_name = "test_printer" + + assert lpprinter._read() == expected + + def test_close(lpprinter, caplog, mocker): """ GIVEN a lp printer object and a mocked connection From b161585250280998d5059fab7756514452868a58 Mon Sep 17 00:00:00 2001 From: belono Date: Mon, 13 Apr 2026 22:35:08 +0200 Subject: [PATCH 7/8] Improve CupsPrinter _read() docstring --- src/escpos/printer/cups.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/escpos/printer/cups.py b/src/escpos/printer/cups.py index 02d09306..b26dfb9d 100644 --- a/src/escpos/printer/cups.py +++ b/src/escpos/printer/cups.py @@ -200,7 +200,9 @@ def _clear(self) -> None: self.pending_job = False def _read(self) -> bytes: - """Return a single-item array with the accepting state of the print queue. + """Return the byte corresponding to the RT status response. + + Respond on/offline given the accepting state of the print queue. states: idle = [3], printing a job = [4], stopped = [5] """ From 5e5bc2e7516a91786251c534c723b518f5d1d5a7 Mon Sep 17 00:00:00 2001 From: belono Date: Mon, 13 Apr 2026 23:06:41 +0200 Subject: [PATCH 8/8] Fix docs build failure caused by sphinx --- doc/conf.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/conf.py b/doc/conf.py index 89e1809f..d3b45329 100644 --- a/doc/conf.py +++ b/doc/conf.py @@ -34,7 +34,6 @@ # ones. extensions = [ "sphinx.ext.autodoc", - "sphinx_autodoc_typehints", "sphinx.ext.doctest", "sphinx.ext.todo", "sphinx.ext.coverage", @@ -44,6 +43,7 @@ "sphinx.ext.inheritance_diagram", "sphinx.ext.imgconverter", "sphinxarg.ext", + "sphinx_autodoc_typehints", "sphinxcontrib.datatemplates", "sphinxcontrib.spelling", ]