python-escpos · belono · Oct 2, 2025 · Oct 2, 2025 · Oct 19, 2025 · Oct 20, 2025
diff --git 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",
 ]

diff --git 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
@@ -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
@@ -13,6 +13,7 @@
 import tempfile
 from typing import Literal, Optional, Type, Union
 
+from ..constants import RT_MASK_OFFLINE, RT_MASK_ONLINE
 from ..escpos import Escpos
 from ..exceptions import DeviceNotFoundError
 
@@ -199,15 +200,17 @@ 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]
         """
         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/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:

diff --git a/test/test_printers/test_printer_cups.py b/test/test_printers/test_printer_cups.py
@@ -169,8 +169,33 @@ 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])
+
+
+@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) -> None:
+    """
+    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
diff --git 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