Impression Mitsubishi K60 + cadres + interface admin distante
- Intégration selphy_print (backend CUPS dyesub Mitsubishi CP-K60DW-S) - Formats impression : 15x20, 10x15, 10x15 2-strips avec cadre et orientation portrait/paysage - Cadres d'impression par format (strip/10x15/15x20) : upload, sélection, aperçu live - Cadres démo générés : pellicule noir/vintage/couleurs + bordures classique/doré/rose gold - Écran de choix de cadre avant capture (clic direct, aperçu grand format) - Filtres réduits à 3 (Couleur, N&B, Sépia) et appliqués sur tous les strips - Port 80 via authbind, route /admin avec redirection auto depuis poste distant - Autologin LightDM corrigé (pam-autologin-service) - Bouton évacuer bourrage papier dans admin - Fix : _gp_lock manquant dans Camera.__init__ Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
import io
|
||||
import logging
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
@@ -66,6 +67,7 @@ class Camera:
|
||||
self.webcam = None
|
||||
self.webcam_index = -1
|
||||
self.preview_dslr_ok = True # True si le dernier preview DSLR a reussi
|
||||
self._gp_lock = threading.Lock()
|
||||
|
||||
def connecter(self, source=None) -> bool:
|
||||
"""Connecte la camera. source: 'gphoto2', 'webcam:0', 'webcam:2', etc."""
|
||||
|
||||
@@ -4,6 +4,7 @@ from pathlib import Path
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
from backend.config import DOSSIER_EXPORTS, charger_config
|
||||
from backend.effects import appliquer_cadre_impression
|
||||
|
||||
log = logging.getLogger("photobooth.collage")
|
||||
|
||||
@@ -28,7 +29,8 @@ def _crop_center(img: Image.Image, ratio_cible: float) -> Image.Image:
|
||||
return img.crop((0, offset, img.width, offset + nouvelle_hauteur))
|
||||
|
||||
|
||||
def creer_strip(photos: list[Path], marge: int = 10, couleur_fond: str = "#000000") -> Path:
|
||||
def creer_strip(photos: list[Path], marge: int = 10, couleur_fond: str = "#000000",
|
||||
cadre: str | None = None) -> Path:
|
||||
"""Cree une bande portrait style pellicule (4 photos de gauche a droite).
|
||||
|
||||
Pellicule noire avec perforations, photos separees par des bandes noires.
|
||||
@@ -83,6 +85,9 @@ def creer_strip(photos: list[Path], marge: int = 10, couleur_fond: str = "#00000
|
||||
strip.paste(img, (x, perf_hauteur))
|
||||
x += largeur_cellule + sep
|
||||
|
||||
if cadre:
|
||||
strip = appliquer_cadre_impression(strip, "strip", cadre)
|
||||
|
||||
nom = f"strip_{photos[0].stem}.jpg"
|
||||
chemin = DOSSIER_EXPORTS / nom
|
||||
strip.save(chemin, "JPEG", quality=95, dpi=(300, 300))
|
||||
@@ -115,7 +120,7 @@ def creer_impression_strip(chemin_strip: Path, marge: int = 20,
|
||||
|
||||
|
||||
def creer_collage(photos: list[Path], colonnes: int = 2, marge: int = 25,
|
||||
couleur_fond: str = "#ffffff") -> Path:
|
||||
couleur_fond: str = "#ffffff", cadre: str | None = None) -> Path:
|
||||
"""Cree un collage en grille au format 10x15cm."""
|
||||
nb = len(photos)
|
||||
images = [Image.open(p) for p in photos]
|
||||
@@ -141,6 +146,9 @@ def creer_collage(photos: list[Path], colonnes: int = 2, marge: int = 25,
|
||||
y = marge + lig * (hauteur_cellule + marge)
|
||||
collage.paste(img, (x, y))
|
||||
|
||||
if cadre:
|
||||
collage = appliquer_cadre_impression(collage, "10x15", cadre)
|
||||
|
||||
nom = f"collage_{photos[0].stem}.jpg"
|
||||
chemin = DOSSIER_EXPORTS / nom
|
||||
collage.save(chemin, "JPEG", quality=95, dpi=(300, 300))
|
||||
|
||||
@@ -11,6 +11,8 @@ DOSSIER_EXPORTS = RACINE / "data" / "exports"
|
||||
DOSSIER_OVERLAYS = RACINE / "frontend" / "assets" / "overlays"
|
||||
DOSSIER_FONDS = RACINE / "frontend" / "assets" / "backgrounds"
|
||||
DOSSIER_ANIMATIONS = RACINE / "frontend" / "assets" / "animations"
|
||||
DOSSIER_CADRES = RACINE / "frontend" / "assets" / "cadres"
|
||||
FORMATS_CADRES = ["strip", "10x15", "15x20"]
|
||||
|
||||
|
||||
def charger_config() -> dict:
|
||||
@@ -51,3 +53,5 @@ def _merge_profond(base: dict, modifications: dict):
|
||||
# Initialisation des dossiers au chargement du module
|
||||
for dossier in [DOSSIER_PHOTOS, DOSSIER_EXPORTS, DOSSIER_OVERLAYS, DOSSIER_FONDS, DOSSIER_ANIMATIONS]:
|
||||
dossier.mkdir(parents=True, exist_ok=True)
|
||||
for fmt in FORMATS_CADRES:
|
||||
(DOSSIER_CADRES / fmt).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@@ -4,21 +4,15 @@ from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageEnhance, ImageFilter, ImageOps
|
||||
|
||||
from backend.config import DOSSIER_OVERLAYS, DOSSIER_EXPORTS, DOSSIER_FONDS
|
||||
from backend.config import DOSSIER_OVERLAYS, DOSSIER_EXPORTS, DOSSIER_FONDS, DOSSIER_CADRES
|
||||
|
||||
log = logging.getLogger("photobooth.effects")
|
||||
|
||||
# Filtres disponibles
|
||||
FILTRES = {
|
||||
"original": "Original",
|
||||
"nb": "Noir & Blanc",
|
||||
"sepia": "Sepia",
|
||||
"vintage": "Vintage",
|
||||
"contraste": "Contraste fort",
|
||||
"lumineux": "Lumineux",
|
||||
"chaud": "Tons chauds",
|
||||
"froid": "Tons froids",
|
||||
"flou_artistique": "Flou artistique",
|
||||
"original": "Couleur",
|
||||
"nb": "Noir & Blanc",
|
||||
"sepia": "Sepia",
|
||||
}
|
||||
|
||||
|
||||
@@ -127,6 +121,28 @@ def chroma_key(chemin_photo: Path, nom_fond: str | None = None,
|
||||
return chemin_export
|
||||
|
||||
|
||||
def appliquer_cadre_impression(img: Image.Image, format_papier: str, nom_cadre: str) -> Image.Image:
|
||||
"""Composite un cadre PNG par-dessus l'image assemblée."""
|
||||
chemin = DOSSIER_CADRES / format_papier / nom_cadre
|
||||
if not chemin.exists():
|
||||
log.warning(f"Cadre introuvable : {chemin}")
|
||||
return img
|
||||
cadre = Image.open(chemin).convert("RGBA")
|
||||
if cadre.size != img.size:
|
||||
cadre = cadre.resize(img.size, Image.LANCZOS)
|
||||
base = img.convert("RGBA")
|
||||
composite = Image.alpha_composite(base, cadre)
|
||||
return composite.convert("RGB")
|
||||
|
||||
|
||||
def lister_cadres(format_papier: str) -> list[str]:
|
||||
"""Liste les cadres disponibles pour un format donné."""
|
||||
dossier = DOSSIER_CADRES / format_papier
|
||||
if not dossier.exists():
|
||||
return []
|
||||
return sorted(f.name for f in dossier.iterdir() if f.suffix.lower() == ".png")
|
||||
|
||||
|
||||
def lister_overlays() -> list[str]:
|
||||
"""Liste les overlays disponibles."""
|
||||
if not DOSSIER_OVERLAYS.exists():
|
||||
|
||||
104
backend/main.py
104
backend/main.py
@@ -9,12 +9,13 @@ from contextlib import asynccontextmanager
|
||||
|
||||
import cv2
|
||||
|
||||
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File
|
||||
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect, UploadFile, File
|
||||
from fastapi.responses import FileResponse, JSONResponse, HTMLResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from backend.config import (
|
||||
RACINE, DOSSIER_PHOTOS, DOSSIER_EXPORTS, DOSSIER_OVERLAYS, DOSSIER_FONDS, DOSSIER_ANIMATIONS,
|
||||
DOSSIER_CADRES, FORMATS_CADRES,
|
||||
charger_config, sauvegarder_config, mettre_a_jour_config,
|
||||
)
|
||||
from backend.camera import camera, GPHOTO2_DISPONIBLE
|
||||
@@ -23,7 +24,7 @@ try:
|
||||
except ImportError:
|
||||
gp = None
|
||||
from backend.gallery import lister_photos, compter_photos, supprimer_photo, vider_galerie
|
||||
from backend.effects import appliquer_filtre, appliquer_overlay, chroma_key, lister_overlays, lister_fonds, FILTRES
|
||||
from backend.effects import appliquer_filtre, appliquer_overlay, chroma_key, lister_overlays, lister_fonds, lister_cadres, FILTRES
|
||||
from backend.collage import creer_strip, creer_collage, creer_impression_strip
|
||||
from backend.destinations import distribuer_photo, detecter_usb, compteur_restant, reset_compteur, recuperer_booth_password
|
||||
from backend.printer import lister_imprimantes, imprimer
|
||||
@@ -150,8 +151,17 @@ app.mount("/data", StaticFiles(directory=str(RACINE / "data")), name="data")
|
||||
|
||||
# --- Pages ---
|
||||
|
||||
@app.get("/admin", include_in_schema=False)
|
||||
async def page_admin():
|
||||
return FileResponse(str(RACINE / "frontend" / "index.html"))
|
||||
|
||||
|
||||
@app.get("/")
|
||||
async def page_principale():
|
||||
async def page_principale(request: Request):
|
||||
from fastapi.responses import RedirectResponse
|
||||
host = request.client.host if request.client else ""
|
||||
if host not in ("127.0.0.1", "::1", "localhost"):
|
||||
return RedirectResponse(url="/admin", status_code=302)
|
||||
return FileResponse(str(RACINE / "frontend" / "index.html"))
|
||||
|
||||
|
||||
@@ -454,10 +464,16 @@ async def api_supprimer_fond(nom: str):
|
||||
@app.post("/api/strip")
|
||||
async def api_strip(donnees: dict):
|
||||
noms = donnees.get("photos", [])
|
||||
chemins = [DOSSIER_PHOTOS / n for n in noms]
|
||||
for c in chemins:
|
||||
source = donnees.get("source", "photos")
|
||||
dossier_src = DOSSIER_EXPORTS if source == "exports" else DOSSIER_PHOTOS
|
||||
chemins = []
|
||||
for n in noms:
|
||||
c = dossier_src / n
|
||||
if not c.exists():
|
||||
return JSONResponse({"erreur": f"Photo introuvable : {c.name}"}, status_code=404)
|
||||
c = DOSSIER_EXPORTS / n # fallback
|
||||
if not c.exists():
|
||||
return JSONResponse({"erreur": f"Photo introuvable : {n}"}, status_code=404)
|
||||
chemins.append(c)
|
||||
chemin_strip = creer_strip(chemins)
|
||||
# Creer aussi la page d'impression (2 bandes sur 10x15 paysage)
|
||||
chemin_print = creer_impression_strip(chemin_strip)
|
||||
@@ -489,6 +505,17 @@ async def api_imprimantes():
|
||||
return lister_imprimantes()
|
||||
|
||||
|
||||
@app.post("/api/imprimante/evacuer")
|
||||
async def api_evacuer_bourrage():
|
||||
"""Annule tous les jobs et réactive l'imprimante pour débloquer un bourrage."""
|
||||
import subprocess
|
||||
config = charger_config()
|
||||
nom = config.get("impression", {}).get("imprimante", "Mitsubishi")
|
||||
subprocess.run(["sudo", "cancel", "-a", nom], capture_output=True)
|
||||
subprocess.run(["sudo", "cupsenable", nom], capture_output=True)
|
||||
return {"succes": True, "message": f"Jobs annulés, {nom} réactivée"}
|
||||
|
||||
|
||||
@app.post("/api/imprimer")
|
||||
async def api_imprimer(donnees: dict):
|
||||
nom = donnees.get("photo", "")
|
||||
@@ -503,10 +530,11 @@ async def api_imprimer(donnees: dict):
|
||||
chemin = DOSSIER_PHOTOS / nom
|
||||
if not chemin.exists():
|
||||
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
||||
ok = imprimer(chemin, copies=copies)
|
||||
if ok:
|
||||
cadre_override = donnees.get("cadre") # cadre choisi en session, prioritaire sur le config
|
||||
resultat = imprimer(chemin, copies=copies, cadre_override=cadre_override)
|
||||
if resultat.get("succes"):
|
||||
distribuer_photo(chemin, imprimee=True)
|
||||
return {"succes": ok}
|
||||
return resultat
|
||||
|
||||
|
||||
# --- API Booth (galerie live) ---
|
||||
@@ -620,7 +648,7 @@ async def api_detecter_usb():
|
||||
return detecter_usb()
|
||||
|
||||
|
||||
# --- API Cadres actifs ---
|
||||
# --- API Cadres actifs (overlays photo) ---
|
||||
|
||||
@app.get("/api/cadres")
|
||||
async def api_cadres():
|
||||
@@ -637,6 +665,60 @@ async def api_cadres_update(donnees: dict):
|
||||
return {"actifs": actifs}
|
||||
|
||||
|
||||
# --- API Cadres impression (par format) ---
|
||||
|
||||
@app.get("/api/cadres-impression/{format_papier}")
|
||||
async def api_cadres_impression_list(format_papier: str):
|
||||
if format_papier not in FORMATS_CADRES:
|
||||
return JSONResponse({"erreur": "Format inconnu"}, status_code=400)
|
||||
config = charger_config()
|
||||
actif = config.get("impression", {}).get("cadres_actifs", {}).get(format_papier)
|
||||
return {"format": format_papier, "disponibles": lister_cadres(format_papier), "actif": actif}
|
||||
|
||||
|
||||
@app.post("/api/cadres-impression/{format_papier}/actif")
|
||||
async def api_cadres_impression_set(format_papier: str, donnees: dict):
|
||||
if format_papier not in FORMATS_CADRES:
|
||||
return JSONResponse({"erreur": "Format inconnu"}, status_code=400)
|
||||
nom = donnees.get("nom") # None = désactiver
|
||||
config = charger_config()
|
||||
cadres_actifs = config.get("impression", {}).get("cadres_actifs", {})
|
||||
cadres_actifs[format_papier] = nom
|
||||
mettre_a_jour_config({"impression": {"cadres_actifs": cadres_actifs}})
|
||||
return {"format": format_papier, "actif": nom}
|
||||
|
||||
|
||||
@app.post("/api/upload/cadre/{format_papier}")
|
||||
async def api_upload_cadre(format_papier: str, fichier: UploadFile = File(...)):
|
||||
if format_papier not in FORMATS_CADRES:
|
||||
return JSONResponse({"erreur": "Format inconnu"}, status_code=400)
|
||||
if not fichier.filename.lower().endswith(".png"):
|
||||
return JSONResponse({"erreur": "Seuls les PNG sont acceptés"}, status_code=400)
|
||||
dossier = DOSSIER_CADRES / format_papier
|
||||
dossier.mkdir(parents=True, exist_ok=True)
|
||||
chemin = dossier / fichier.filename
|
||||
with open(chemin, "wb") as f:
|
||||
f.write(await fichier.read())
|
||||
return {"nom": fichier.filename, "format": format_papier}
|
||||
|
||||
|
||||
@app.delete("/api/cadres-impression/{format_papier}/{nom}")
|
||||
async def api_supprimer_cadre(format_papier: str, nom: str):
|
||||
if format_papier not in FORMATS_CADRES:
|
||||
return JSONResponse({"erreur": "Format inconnu"}, status_code=400)
|
||||
chemin = DOSSIER_CADRES / format_papier / nom
|
||||
if not chemin.exists() or chemin.parent != DOSSIER_CADRES / format_papier:
|
||||
return JSONResponse({"erreur": "Fichier introuvable"}, status_code=404)
|
||||
chemin.unlink()
|
||||
# Désactiver si c'était le cadre actif
|
||||
config = charger_config()
|
||||
cadres_actifs = config.get("impression", {}).get("cadres_actifs", {})
|
||||
if cadres_actifs.get(format_papier) == nom:
|
||||
cadres_actifs[format_papier] = None
|
||||
mettre_a_jour_config({"impression": {"cadres_actifs": cadres_actifs}})
|
||||
return {"succes": True}
|
||||
|
||||
|
||||
# --- API Animations compte a rebours ---
|
||||
|
||||
ANIMATIONS_CAR = {
|
||||
@@ -769,7 +851,7 @@ if __name__ == "__main__":
|
||||
uvicorn.run(
|
||||
"backend.main:app",
|
||||
host=conf_srv.get("host", "0.0.0.0"),
|
||||
port=conf_srv.get("port", 8080),
|
||||
port=conf_srv.get("port", 80),
|
||||
reload=False,
|
||||
log_level="info",
|
||||
)
|
||||
|
||||
@@ -1,81 +1,265 @@
|
||||
import logging
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from backend.config import charger_config
|
||||
from backend.effects import appliquer_cadre_impression
|
||||
|
||||
log = logging.getLogger("photobooth.printer")
|
||||
|
||||
# Essayer d'importer cups
|
||||
try:
|
||||
import cups
|
||||
CUPS_DISPONIBLE = True
|
||||
except ImportError:
|
||||
CUPS_DISPONIBLE = False
|
||||
log.warning("pycups non installe, impression indisponible")
|
||||
# PageSize CUPS + dimensions exactes à 300 DPI pour la Mitsubishi CP-K60DW-S
|
||||
# Le resize vers ces dimensions exactes évite toute bordure blanche
|
||||
FORMATS = {
|
||||
"10x15": {"cups": "w288h432", "px": (1200, 1800)},
|
||||
"15x15": {"cups": "w432h432", "px": (1800, 1800)},
|
||||
"15x20": {"cups": "w432h576", "px": (1800, 2400)},
|
||||
"10x15-2up": {"cups": "w288h432-div2", "px": (600, 1800)}, # 1 strip 5x15cm, printer duplique
|
||||
"15x20-2up": {"cups": "w432h576-div2", "px": (900, 2400)}, # 1 strip 7.5x20cm, printer duplique
|
||||
}
|
||||
|
||||
ERREUR_FICHIER = "fichier_introuvable"
|
||||
ERREUR_IMPRIMANTE = "imprimante_indisponible"
|
||||
ERREUR_BOURRAGE = "bourrage_papier"
|
||||
ERREUR_PAPIER = "plus_de_papier"
|
||||
ERREUR_IMPRESSION = "erreur_impression"
|
||||
|
||||
|
||||
def _run(cmd: list[str], timeout: int = 10) -> tuple[int, str, str]:
|
||||
try:
|
||||
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
||||
return r.returncode, r.stdout, r.stderr
|
||||
except subprocess.TimeoutExpired:
|
||||
return -1, "", "timeout"
|
||||
except FileNotFoundError:
|
||||
return -1, "", f"commande introuvable : {cmd[0]}"
|
||||
|
||||
|
||||
def preparer_apercu(chemin: Path, largeur: int, hauteur: int,
|
||||
orientation: str = "paysage") -> Path:
|
||||
"""
|
||||
Génère un aperçu JPEG recadré aux bonnes dimensions (pour le frontend).
|
||||
orientation : 'portrait' pivote le DSLR paysage de 90° avant recadrage.
|
||||
"""
|
||||
from PIL import Image, ImageOps
|
||||
img = Image.open(chemin).convert("RGB")
|
||||
img = ImageOps.exif_transpose(img)
|
||||
if orientation == "portrait" and img.width > img.height:
|
||||
img = img.rotate(90, expand=True)
|
||||
img = _crop_to(img, largeur, hauteur)
|
||||
tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
|
||||
img.save(tmp.name, "JPEG", quality=85)
|
||||
return Path(tmp.name)
|
||||
|
||||
|
||||
def _crop_to(img, largeur: int, hauteur: int):
|
||||
"""Cover crop centre vers les dimensions exactes."""
|
||||
from PIL import Image
|
||||
iw, ih = img.size
|
||||
ratio_cible = largeur / hauteur
|
||||
ratio_src = iw / ih
|
||||
if ratio_src > ratio_cible:
|
||||
new_w = int(ih * ratio_cible)
|
||||
left = (iw - new_w) // 2
|
||||
img = img.crop((left, 0, left + new_w, ih))
|
||||
elif ratio_src < ratio_cible:
|
||||
new_h = int(iw / ratio_cible)
|
||||
top = (ih - new_h) // 2
|
||||
img = img.crop((0, top, iw, top + new_h))
|
||||
return img.resize((largeur, hauteur), Image.LANCZOS)
|
||||
|
||||
|
||||
def _preparer_image(chemin: Path, largeur: int, hauteur: int,
|
||||
format_papier: str | None = None, cadre: str | None = None,
|
||||
orientation: str = "paysage") -> Path:
|
||||
"""
|
||||
Pivote si nécessaire, recadre aux dimensions exactes (cover crop),
|
||||
applique le cadre, sauvegarde en temp.
|
||||
"""
|
||||
try:
|
||||
from PIL import Image, ImageOps
|
||||
except ImportError:
|
||||
log.warning("PIL absent — image envoyée telle quelle, risque de bordures")
|
||||
return chemin
|
||||
|
||||
img = Image.open(chemin).convert("RGB")
|
||||
img = ImageOps.exif_transpose(img) # respecte l'orientation EXIF
|
||||
|
||||
# Portrait demandé mais DSLR paysage → rotation 90°
|
||||
if orientation == "portrait" and img.width > img.height:
|
||||
img = img.rotate(90, expand=True)
|
||||
# Paysage demandé mais image portrait (ex. prise avec DSLR tourné) → rotation -90°
|
||||
elif orientation == "paysage" and img.height > img.width:
|
||||
img = img.rotate(-90, expand=True)
|
||||
|
||||
img = _crop_to(img, largeur, hauteur)
|
||||
|
||||
if cadre and format_papier:
|
||||
img = appliquer_cadre_impression(img, format_papier, cadre)
|
||||
|
||||
tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
|
||||
img.save(tmp.name, "JPEG", quality=95, dpi=(300, 300))
|
||||
log.debug(f"Image préparée : {orientation} {largeur}x{hauteur}px → {tmp.name}")
|
||||
return Path(tmp.name)
|
||||
|
||||
|
||||
def _statut(nom: str) -> str:
|
||||
"""Retourne 'idle', 'printing', 'stopped' ou 'inconnue'."""
|
||||
code, out, _ = _run(["lpstat", "-p", nom])
|
||||
if code != 0:
|
||||
return "inconnue"
|
||||
out = out.lower()
|
||||
if "disabled" in out or "stopped" in out:
|
||||
return "stopped"
|
||||
if "printing" in out or "processing" in out:
|
||||
return "printing"
|
||||
if "idle" in out or "inactive" in out:
|
||||
return "idle"
|
||||
return "inconnue"
|
||||
|
||||
|
||||
def _reactiver(nom: str) -> bool:
|
||||
"""Réactive l'imprimante via cupsenable (sudoers sans mot de passe requis)."""
|
||||
code, _, err = _run(["sudo", "cupsenable", nom], timeout=5)
|
||||
if code == 0:
|
||||
log.info(f"Imprimante {nom} réactivée")
|
||||
return True
|
||||
log.error(f"Impossible de réactiver {nom} : {err}")
|
||||
return False
|
||||
|
||||
|
||||
def _detecter_erreur_physique(nom: str) -> str | None:
|
||||
"""Interroge l'état physique via le backend dyesub."""
|
||||
try:
|
||||
r = subprocess.run(
|
||||
["/usr/lib/cups/backend/gutenprint53+usb", "-s"],
|
||||
capture_output=True, text=True, timeout=10,
|
||||
)
|
||||
combined = (r.stdout + r.stderr).lower()
|
||||
except Exception:
|
||||
return None
|
||||
if "jam" in combined:
|
||||
return ERREUR_BOURRAGE
|
||||
if "media out" in combined or "no media" in combined or "remaining: 0/" in combined:
|
||||
return ERREUR_PAPIER
|
||||
return None
|
||||
|
||||
|
||||
def lister_imprimantes() -> list[dict]:
|
||||
"""Liste les imprimantes disponibles via CUPS."""
|
||||
if not CUPS_DISPONIBLE:
|
||||
"""Liste les imprimantes CUPS disponibles."""
|
||||
code, out, _ = _run(["lpstat", "-p"])
|
||||
if code != 0 or not out.strip():
|
||||
return [{"nom": "[Simulation] Imprimante virtuelle", "statut": "prete"}]
|
||||
|
||||
try:
|
||||
conn = cups.Connection()
|
||||
imprimantes = conn.getPrinters()
|
||||
return [
|
||||
{
|
||||
"nom": nom,
|
||||
"statut": "prete" if info.get("printer-state") == 3 else "occupee",
|
||||
"info": info.get("printer-info", ""),
|
||||
}
|
||||
for nom, info in imprimantes.items()
|
||||
]
|
||||
except cups.IPPError as e:
|
||||
log.error(f"Erreur CUPS : {e}")
|
||||
return []
|
||||
imprimantes = []
|
||||
for ligne in out.splitlines():
|
||||
if not ligne.startswith("printer "):
|
||||
continue
|
||||
parties = ligne.split()
|
||||
if len(parties) < 2:
|
||||
continue
|
||||
nom = parties[1]
|
||||
arretee = "disabled" in ligne.lower() or "stopped" in ligne.lower()
|
||||
imprimantes.append({
|
||||
"nom": nom,
|
||||
"statut": "arretee" if arretee else "prete",
|
||||
})
|
||||
return imprimantes or [{"nom": "[Simulation] Imprimante virtuelle", "statut": "prete"}]
|
||||
|
||||
|
||||
def imprimer(chemin_photo: Path, imprimante: str | None = None, copies: int | None = None) -> bool:
|
||||
"""Imprime une photo sur l'imprimante configuree."""
|
||||
def imprimer(
|
||||
chemin_photo: Path,
|
||||
imprimante: str | None = None,
|
||||
copies: int | None = None,
|
||||
format_papier: str | None = None,
|
||||
cadre_override: str | None = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Imprime une photo sans bordure blanche.
|
||||
L'image est recadrée et redimensionnée aux dimensions exactes (300 DPI)
|
||||
avant envoi à CUPS.
|
||||
|
||||
Retourne {"succes": True} ou {"succes": False, "erreur": CODE, "message": str}
|
||||
"""
|
||||
config = charger_config()
|
||||
conf_imp = config.get("impression", {})
|
||||
|
||||
if imprimante is None:
|
||||
imprimante = conf_imp.get("imprimante")
|
||||
imprimante = conf_imp.get("imprimante", "Mitsubishi")
|
||||
if copies is None:
|
||||
copies = conf_imp.get("copies", 1)
|
||||
if format_papier is None:
|
||||
format_papier = conf_imp.get("format", "15x20")
|
||||
|
||||
fmt = FORMATS.get(format_papier, FORMATS["15x20"])
|
||||
page_size = fmt["cups"]
|
||||
largeur, hauteur = fmt["px"]
|
||||
|
||||
fmt_base = format_papier.replace("-2up", "")
|
||||
cadre = cadre_override if cadre_override is not None else conf_imp.get("cadres_actifs", {}).get(fmt_base)
|
||||
orientation = conf_imp.get("orientations", {}).get(fmt_base, "portrait")
|
||||
|
||||
if not chemin_photo.exists():
|
||||
log.error(f"Fichier introuvable : {chemin_photo}")
|
||||
return False
|
||||
return {"succes": False, "erreur": ERREUR_FICHIER, "message": "Fichier photo introuvable"}
|
||||
|
||||
if not CUPS_DISPONIBLE:
|
||||
log.info(f"[Simulation] Impression de {chemin_photo} x{copies}")
|
||||
return True
|
||||
chemin_print = _preparer_image(chemin_photo, largeur, hauteur, fmt_base, cadre, orientation)
|
||||
tmp_cree = chemin_print != chemin_photo
|
||||
|
||||
if _statut(imprimante) == "stopped":
|
||||
log.warning(f"Imprimante {imprimante} arrêtée — réactivation")
|
||||
if not _reactiver(imprimante):
|
||||
return {
|
||||
"succes": False,
|
||||
"erreur": ERREUR_IMPRIMANTE,
|
||||
"message": "Imprimante arrêtée, réactivation impossible (vérifiez sudo cupsenable)",
|
||||
}
|
||||
time.sleep(1)
|
||||
|
||||
try:
|
||||
conn = cups.Connection()
|
||||
for tentative in range(1, 4):
|
||||
cmd = [
|
||||
"lp",
|
||||
"-d", imprimante,
|
||||
"-n", str(copies),
|
||||
"-o", f"PageSize={page_size}",
|
||||
"-o", "StpiShrinkOutput=Crop",
|
||||
str(chemin_print),
|
||||
]
|
||||
|
||||
if imprimante is None:
|
||||
imprimante = conn.getDefault()
|
||||
if imprimante is None:
|
||||
imprimantes = conn.getPrinters()
|
||||
if imprimantes:
|
||||
imprimante = list(imprimantes.keys())[0]
|
||||
else:
|
||||
log.error("Aucune imprimante trouvee")
|
||||
return False
|
||||
code, out, err = _run(cmd, timeout=30)
|
||||
|
||||
options = {
|
||||
"copies": str(copies),
|
||||
"media": conf_imp.get("format", "10x15"),
|
||||
"fit-to-page": "true",
|
||||
if code == 0:
|
||||
job = out.strip()
|
||||
log.info(f"Impression lancée : {job} ({format_papier} {largeur}x{hauteur}px, x{copies})")
|
||||
return {"succes": True, "job": job}
|
||||
|
||||
log.warning(f"Impression échouée (tentative {tentative}/3) : {err.strip()}")
|
||||
|
||||
if tentative < 3:
|
||||
if _statut(imprimante) == "stopped":
|
||||
erreur = _detecter_erreur_physique(imprimante)
|
||||
if erreur == ERREUR_BOURRAGE:
|
||||
return {
|
||||
"succes": False,
|
||||
"erreur": ERREUR_BOURRAGE,
|
||||
"message": "Bourrage papier — retirez le papier coincé puis réessayez",
|
||||
}
|
||||
if erreur == ERREUR_PAPIER:
|
||||
return {
|
||||
"succes": False,
|
||||
"erreur": ERREUR_PAPIER,
|
||||
"message": "Plus de papier — rechargez le rouleau",
|
||||
}
|
||||
_reactiver(imprimante)
|
||||
time.sleep(2)
|
||||
|
||||
return {
|
||||
"succes": False,
|
||||
"erreur": ERREUR_IMPRESSION,
|
||||
"message": "Impression échouée après 3 tentatives",
|
||||
}
|
||||
|
||||
job_id = conn.printFile(imprimante, str(chemin_photo), "Photobooth", options)
|
||||
log.info(f"Impression lancee : job #{job_id} sur {imprimante}")
|
||||
return True
|
||||
|
||||
except cups.IPPError as e:
|
||||
log.error(f"Erreur impression : {e}")
|
||||
return False
|
||||
finally:
|
||||
if tmp_cree:
|
||||
chemin_print.unlink(missing_ok=True)
|
||||
|
||||
Reference in New Issue
Block a user