feat: détection erreurs imprimante via IPP direct (Selphy CP1300)
- script_print.sh : après lp, polling CUPS + sondage IPP direct ipp://IP:631/ipp/print Détecte input-tray-missing / media-empty en ~3-6s sans attendre le timeout CUPS - printer_service.py : ajout _get_printer_ip(), _query_ipp_direct() _get_printer_status() utilise IPP pour état/raisons/encre (source de vérité) - Timeout porté à 120s côté Python, fallback lpstat conservé
This commit is contained in:
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import re
|
||||
import subprocess
|
||||
import time
|
||||
import uuid
|
||||
@@ -258,6 +259,81 @@ class PrinterService:
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
# ── IPP direct ───────────────────────────────────────────────────────────
|
||||
|
||||
def _get_printer_ip(self, printer_name: str) -> str:
|
||||
"""Extrait l'IP de l'imprimante depuis l'URI CUPS (lpstat -v)."""
|
||||
try:
|
||||
r = subprocess.run(
|
||||
["lpstat", "-v", printer_name],
|
||||
capture_output=True, text=True, timeout=5
|
||||
)
|
||||
m = re.search(r"(\d+\.\d+\.\d+\.\d+)", r.stdout)
|
||||
return m.group(1) if m else ""
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
def _query_ipp_direct(self, printer_ip: str) -> dict:
|
||||
"""Interroge directement le serveur IPP natif de l'imprimante (sans auth).
|
||||
|
||||
Retourne un dict avec keys: state, state_reasons, accepting, markers.
|
||||
"""
|
||||
ipp_test = "/usr/share/cups/ipptool/get-printer-attributes.test"
|
||||
empty: dict = {"state": "", "state_reasons": [], "accepting": None, "markers": []}
|
||||
if not printer_ip:
|
||||
return empty
|
||||
try:
|
||||
r = subprocess.run(
|
||||
["ipptool", "-tv", f"ipp://{printer_ip}:631/ipp/print", ipp_test],
|
||||
capture_output=True, text=True, timeout=8
|
||||
)
|
||||
out = r.stdout
|
||||
except Exception as e:
|
||||
logger.debug("ipptool direct échec %s: %s", printer_ip, e)
|
||||
return empty
|
||||
|
||||
result: dict = {"state": "", "state_reasons": [], "accepting": None, "markers": []}
|
||||
|
||||
# printer-state
|
||||
m = re.search(r"printer-state \(enum\)\s*=\s*(\S+)", out)
|
||||
if m:
|
||||
result["state"] = m.group(1).lower() # idle / processing / stopped
|
||||
|
||||
# printer-state-reasons
|
||||
m = re.search(r"printer-state-reasons \([^)]+\)\s*=\s*(.+)", out)
|
||||
if m:
|
||||
reasons = [r.strip() for r in m.group(1).split(",")]
|
||||
result["state_reasons"] = [r for r in reasons if r and r != "none"]
|
||||
|
||||
# printer-is-accepting-jobs
|
||||
m = re.search(r"printer-is-accepting-jobs \(boolean\)\s*=\s*(\S+)", out)
|
||||
if m:
|
||||
result["accepting"] = m.group(1).lower() == "true"
|
||||
|
||||
# marker-levels (encre/ruban)
|
||||
names_m = re.search(r"marker-names \([^)]+\)\s*=\s*(.+)", out)
|
||||
levels_m = re.search(r"marker-levels \([^)]+\)\s*=\s*(.+)", out)
|
||||
colors_m = re.search(r"marker-colors \([^)]+\)\s*=\s*(.+)", out)
|
||||
types_m = re.search(r"marker-types \([^)]+\)\s*=\s*(.+)", out)
|
||||
if names_m and levels_m:
|
||||
names = [n.strip().strip('"') for n in names_m.group(1).split(",")]
|
||||
levels = [l.strip() for l in levels_m.group(1).split(",")]
|
||||
colors = [c.strip().strip('"') for c in colors_m.group(1).split(",")] if colors_m else []
|
||||
types = [t.strip().strip('"') for t in types_m.group(1).split(",")] if types_m else []
|
||||
for i, name in enumerate(names):
|
||||
try:
|
||||
lvl = int(levels[i]) if i < len(levels) else -1
|
||||
except ValueError:
|
||||
lvl = -1
|
||||
result["markers"].append({
|
||||
"name": name,
|
||||
"level": lvl,
|
||||
"color": colors[i] if i < len(colors) else "",
|
||||
"type": types[i] if i < len(types) else "",
|
||||
})
|
||||
|
||||
return result
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────
|
||||
|
||||
def _printer_name(self, p) -> str:
|
||||
@@ -307,18 +383,17 @@ class PrinterService:
|
||||
return {"stats_done": 0, "stats_error": 0, "stats_cancelled": 0}
|
||||
|
||||
def _get_printer_status(self, printer_name: str) -> dict:
|
||||
"""Statut CUPS complet pour une imprimante (état, jobs, accepting, uri, modèle)."""
|
||||
"""Statut complet : CUPS pour les jobs, IPP direct pour état/erreurs/encre."""
|
||||
try:
|
||||
# État de l'imprimante
|
||||
# ── Etat de base via lpstat ──
|
||||
r_state = subprocess.run(
|
||||
["lpstat", "-p", printer_name],
|
||||
capture_output=True, text=True, timeout=5
|
||||
)
|
||||
out = r_state.stdout.lower()
|
||||
|
||||
if r_state.returncode != 0 or "not found" in (r_state.stderr or "").lower():
|
||||
return {"state": "offline", "accepting": False, "jobs": [], "jobs_count": 0}
|
||||
|
||||
out = r_state.stdout.lower()
|
||||
if "idle" in out:
|
||||
state = "idle"
|
||||
elif "printing" in out or "processing" in out:
|
||||
@@ -328,18 +403,19 @@ class PrinterService:
|
||||
else:
|
||||
state = "unknown"
|
||||
|
||||
# Est-ce que l'imprimante accepte les nouveaux jobs ?
|
||||
# Est-ce que l'imprimante accepte les nouveaux jobs (CUPS) ?
|
||||
r_accept = subprocess.run(
|
||||
["lpstat", "-a", printer_name],
|
||||
capture_output=True, text=True, timeout=5
|
||||
)
|
||||
accepting = "accepting" in r_accept.stdout.lower()
|
||||
|
||||
# Liste des jobs CUPS en cours
|
||||
# Jobs CUPS en cours
|
||||
jobs = self._get_cups_jobs(printer_name)
|
||||
|
||||
# URI du périphérique (adresse IP / protocole)
|
||||
# URI du périphérique
|
||||
uri = ""
|
||||
printer_ip = ""
|
||||
try:
|
||||
r_uri = subprocess.run(
|
||||
["lpstat", "-v", printer_name],
|
||||
@@ -350,14 +426,15 @@ class PrinterService:
|
||||
parts = line.split(":", 2)
|
||||
uri = (parts[1].strip() + ":" + parts[2].strip()) if len(parts) > 2 else ""
|
||||
break
|
||||
m = re.search(r"(\d+\.\d+\.\d+\.\d+)", r_uri.stdout)
|
||||
printer_ip = m.group(1) if m else ""
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Modèle, localisation et raisons d'erreur (lpstat -l -p)
|
||||
# Modèle / location via lpstat -l -p
|
||||
model = ""
|
||||
location = ""
|
||||
reasons: list[str] = []
|
||||
state_message = ""
|
||||
try:
|
||||
r_info = subprocess.run(
|
||||
["lpstat", "-l", "-p", printer_name],
|
||||
@@ -372,24 +449,44 @@ class PrinterService:
|
||||
elif l.lower().startswith("reason:"):
|
||||
raw = l.split(":", 1)[1].strip()
|
||||
if raw and raw.lower() not in ("none", ""):
|
||||
label = self._REASON_LABELS.get(raw, raw)
|
||||
reasons.append(label)
|
||||
reasons.append(self._REASON_LABELS.get(raw, raw))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Niveaux d'encre/ruban via attributs IPP (localhost CUPS)
|
||||
markers = self._get_marker_levels(printer_name)
|
||||
# ── Sondage IPP direct sur l'imprimante (source de vérité) ──
|
||||
markers: list[dict] = []
|
||||
if printer_ip:
|
||||
ipp = self._query_ipp_direct(printer_ip)
|
||||
if ipp["state"]:
|
||||
# Mappe les états IPP → notre convention
|
||||
ipp_state_map = {
|
||||
"idle": "idle",
|
||||
"processing": "printing",
|
||||
"stopped": "disabled",
|
||||
}
|
||||
state = ipp_state_map.get(ipp["state"], ipp["state"])
|
||||
if ipp["accepting"] is not None:
|
||||
accepting = ipp["accepting"]
|
||||
if ipp["state_reasons"]:
|
||||
reasons = [
|
||||
self._REASON_LABELS.get(r, r)
|
||||
for r in ipp["state_reasons"]
|
||||
]
|
||||
if ipp["markers"]:
|
||||
markers = ipp["markers"]
|
||||
else:
|
||||
markers = self._get_marker_levels(printer_name)
|
||||
|
||||
return {
|
||||
"state": state,
|
||||
"accepting": accepting,
|
||||
"jobs": jobs,
|
||||
"state": state,
|
||||
"accepting": accepting,
|
||||
"jobs": jobs,
|
||||
"jobs_count": len(jobs),
|
||||
"uri": uri,
|
||||
"model": model,
|
||||
"location": location,
|
||||
"reasons": reasons,
|
||||
"markers": markers,
|
||||
"uri": uri,
|
||||
"model": model,
|
||||
"location": location,
|
||||
"reasons": reasons,
|
||||
"markers": markers,
|
||||
}
|
||||
except Exception as e:
|
||||
return {"state": "error", "accepting": False, "jobs": [], "jobs_count": 0, "error": str(e)}
|
||||
|
||||
Reference in New Issue
Block a user