Fix LP printer stall

This commit is contained in:
belono 2023-09-21 18:57:41 +02:00
parent a8e1c583e0
commit f8deb69f43

View File

@ -12,7 +12,6 @@ import functools
import logging import logging
import subprocess import subprocess
import sys import sys
from typing import ByteString
from ..escpos import Escpos from ..escpos import Escpos
from ..exceptions import DeviceNotFoundError from ..exceptions import DeviceNotFoundError
@ -74,8 +73,41 @@ class LP(Escpos):
self.printer_name = printer_name self.printer_name = printer_name
self.auto_flush = kwargs.get("auto_flush", True) self.auto_flush = kwargs.get("auto_flush", True)
@property
def printers(self) -> dict:
"""Available CUPS printers."""
p_names = subprocess.run(
["lpstat", "-e"], # Get printer names
capture_output=True,
text=True,
)
p_devs = subprocess.run(
["lpstat", "-v"], # Get attached devices
capture_output=True,
text=True,
)
# List and trim output lines
names = [name for name in p_names.stdout.split("\n") if name]
devs = [dev for dev in p_devs.stdout.split("\n") if dev]
# return a dict of {printer name: attached device} pairs
return {name: dev.split()[-1] for name in names for dev in devs if name in dev}
def _get_system_default_printer(self) -> str:
"""Return the system's default printer name."""
p_name = subprocess.run(
["lpstat", "-d"],
capture_output=True,
text=True,
)
name = p_name.stdout.split()[-1]
if name not in self.printers:
return ""
return name
@dependency_linux_lp @dependency_linux_lp
def open(self, raise_not_found: bool = True) -> None: def open(
self, job_name: str = "python-escpos", raise_not_found: bool = True
) -> None:
"""Invoke _lp_ in a new subprocess and wait for commands. """Invoke _lp_ in a new subprocess and wait for commands.
By default raise an exception if device is not found. By default raise an exception if device is not found.
@ -88,24 +120,25 @@ class LP(Escpos):
if self._device: if self._device:
self.close() self.close()
self.job_name = job_name
try:
# Name validation, set default if no given name
self.printer_name = self.printer_name or self._get_system_default_printer()
assert self.printer_name in self.printers, "Incorrect printer name"
# Open device # Open device
self.device = subprocess.Popen( self.device = subprocess.Popen(
["lp", "-d", self.printer_name, "-o", "raw"], ["lp", "-d", self.printer_name, "-t", self.job_name, "-o", "raw"],
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE, stderr=subprocess.DEVNULL,
) )
except (AssertionError, subprocess.SubprocessError) as e:
error: ByteString = b""
if self.device and self.device.stderr:
error = self.device.stderr.read()
if bool(error):
# Raise exception or log error and cancel # Raise exception or log error and cancel
self.device = None self.device = None
if raise_not_found: if raise_not_found:
raise DeviceNotFoundError( raise DeviceNotFoundError(
f"Unable to start a print job for the printer {self.printer_name}:" f"Unable to start a print job for the printer {self.printer_name}:"
+ f"\n{error!r}" + f"\n{e}"
) )
else: else:
logging.error("LP printing %s not available", self.printer_name) logging.error("LP printing %s not available", self.printer_name)
@ -138,6 +171,6 @@ class LP(Escpos):
if self.device.stdin.writable(): if self.device.stdin.writable():
self.device.stdin.write(msg) self.device.stdin.write(msg)
else: else:
raise Exception("Not a valid pipe for lp process") raise subprocess.SubprocessError("Not a valid pipe for lp process")
if self.auto_flush: if self.auto_flush:
self.flush() self.flush()