Init photobooth - borne photo evenementielle RPi4
Backend FastAPI complet : camera gphoto2, filtres/overlays Pillow, chroma key OpenCV, multi-shot/collage, GIF, impression CUPS, email SMTP, QR code. Frontend web vanilla (HTML/CSS/JS) pour Chromium kiosk. Menu admin cache (appui long coin ecran). Toutes les fonctionnalites activables/desactivables. Scripts install + systemd pour RPi4. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
0
backend/__init__.py
Normal file
0
backend/__init__.py
Normal file
140
backend/camera.py
Normal file
140
backend/camera.py
Normal file
@@ -0,0 +1,140 @@
|
||||
import io
|
||||
import logging
|
||||
import subprocess
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from backend.config import DOSSIER_PHOTOS, charger_config
|
||||
|
||||
log = logging.getLogger("photobooth.camera")
|
||||
|
||||
# Essayer d'importer gphoto2, sinon mode simulation
|
||||
try:
|
||||
import gphoto2 as gp
|
||||
GPHOTO2_DISPONIBLE = True
|
||||
except ImportError:
|
||||
GPHOTO2_DISPONIBLE = False
|
||||
log.warning("python-gphoto2 non installe, mode simulation active")
|
||||
|
||||
|
||||
class Camera:
|
||||
"""Controle de l'appareil DSLR via gphoto2."""
|
||||
|
||||
def __init__(self):
|
||||
self.camera = None
|
||||
self.contexte = None
|
||||
self.connectee = False
|
||||
|
||||
def connecter(self) -> bool:
|
||||
"""Connecte l'appareil photo."""
|
||||
if not GPHOTO2_DISPONIBLE:
|
||||
log.info("Mode simulation : camera virtuelle connectee")
|
||||
self.connectee = True
|
||||
return True
|
||||
|
||||
try:
|
||||
self.contexte = gp.Context()
|
||||
self.camera = gp.Camera()
|
||||
self.camera.init(self.contexte)
|
||||
self.connectee = True
|
||||
log.info("Camera DSLR connectee")
|
||||
return True
|
||||
except gp.GPhoto2Error as e:
|
||||
log.error(f"Erreur connexion camera : {e}")
|
||||
self.connectee = False
|
||||
return False
|
||||
|
||||
def deconnecter(self):
|
||||
"""Deconnecte l'appareil photo."""
|
||||
if self.camera and GPHOTO2_DISPONIBLE:
|
||||
try:
|
||||
self.camera.exit(self.contexte)
|
||||
except gp.GPhoto2Error:
|
||||
pass
|
||||
self.camera = None
|
||||
self.contexte = None
|
||||
self.connectee = False
|
||||
log.info("Camera deconnectee")
|
||||
|
||||
def capturer(self) -> Path | None:
|
||||
"""Capture une photo et retourne le chemin du fichier."""
|
||||
if not self.connectee:
|
||||
log.error("Camera non connectee")
|
||||
return None
|
||||
|
||||
horodatage = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
|
||||
nom_fichier = f"photo_{horodatage}.jpg"
|
||||
chemin_dest = DOSSIER_PHOTOS / nom_fichier
|
||||
|
||||
if not GPHOTO2_DISPONIBLE:
|
||||
return self._capture_simulation(chemin_dest)
|
||||
|
||||
try:
|
||||
chemin_camera = self.camera.capture(gp.GP_CAPTURE_IMAGE, self.contexte)
|
||||
fichier_camera = self.camera.file_get(
|
||||
chemin_camera.folder, chemin_camera.name, gp.GP_FILE_TYPE_NORMAL, self.contexte
|
||||
)
|
||||
fichier_camera.save(str(chemin_dest))
|
||||
log.info(f"Photo capturee : {chemin_dest}")
|
||||
return chemin_dest
|
||||
except gp.GPhoto2Error as e:
|
||||
log.error(f"Erreur capture : {e}")
|
||||
return None
|
||||
|
||||
def preview(self) -> bytes | None:
|
||||
"""Capture un apercu live (preview) en JPEG."""
|
||||
if not self.connectee:
|
||||
return None
|
||||
|
||||
if not GPHOTO2_DISPONIBLE:
|
||||
return self._preview_simulation()
|
||||
|
||||
try:
|
||||
fichier = self.camera.capture_preview(self.contexte)
|
||||
donnees = fichier.get_data_and_size()
|
||||
return bytes(donnees)
|
||||
except gp.GPhoto2Error as e:
|
||||
log.error(f"Erreur preview : {e}")
|
||||
return None
|
||||
|
||||
def lister_appareils(self) -> list[str]:
|
||||
"""Liste les appareils photo detectes."""
|
||||
if not GPHOTO2_DISPONIBLE:
|
||||
return ["[Simulation] Camera virtuelle"]
|
||||
|
||||
try:
|
||||
appareils = gp.Camera.autodetect(self.contexte or gp.Context())
|
||||
return [f"{nom} ({port})" for nom, port in appareils]
|
||||
except gp.GPhoto2Error:
|
||||
return []
|
||||
|
||||
def _capture_simulation(self, chemin: Path) -> Path:
|
||||
"""Genere une image de test en mode simulation."""
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
img = Image.new("RGB", (1920, 1280), color=(40, 40, 40))
|
||||
draw = ImageDraw.Draw(img)
|
||||
texte = f"SIMULATION\n{datetime.now().strftime('%H:%M:%S')}"
|
||||
draw.text((960, 640), texte, fill=(255, 255, 255), anchor="mm")
|
||||
# Dessiner un cadre
|
||||
draw.rectangle([50, 50, 1870, 1230], outline=(233, 30, 99), width=4)
|
||||
img.save(chemin, "JPEG", quality=95)
|
||||
log.info(f"Photo simulation : {chemin}")
|
||||
return chemin
|
||||
|
||||
def _preview_simulation(self) -> bytes:
|
||||
"""Genere un apercu de test en mode simulation."""
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
img = Image.new("RGB", (640, 480), color=(30, 30, 30))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.text((320, 240), "PREVIEW", fill=(200, 200, 200), anchor="mm")
|
||||
draw.ellipse([300, 220, 340, 260], outline=(233, 30, 99), width=2)
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, "JPEG", quality=70)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
# Instance globale
|
||||
camera = Camera()
|
||||
71
backend/collage.py
Normal file
71
backend/collage.py
Normal file
@@ -0,0 +1,71 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
from backend.config import DOSSIER_EXPORTS, charger_config
|
||||
|
||||
log = logging.getLogger("photobooth.collage")
|
||||
|
||||
|
||||
def creer_strip(photos: list[Path], marge: int = 20, couleur_fond: str = "#ffffff") -> Path:
|
||||
"""Cree une bande verticale (strip) avec les photos."""
|
||||
images = [Image.open(p) for p in photos]
|
||||
|
||||
# Uniformiser la largeur
|
||||
largeur = min(img.width for img in images)
|
||||
images_redim = []
|
||||
for img in images:
|
||||
ratio = largeur / img.width
|
||||
hauteur = int(img.height * ratio)
|
||||
images_redim.append(img.resize((largeur, hauteur), Image.LANCZOS))
|
||||
|
||||
hauteur_totale = sum(img.height for img in images_redim) + marge * (len(images_redim) + 1)
|
||||
largeur_totale = largeur + marge * 2
|
||||
|
||||
strip = Image.new("RGB", (largeur_totale, hauteur_totale), couleur_fond)
|
||||
y = marge
|
||||
for img in images_redim:
|
||||
strip.paste(img, (marge, y))
|
||||
y += img.height + marge
|
||||
|
||||
nom = f"strip_{photos[0].stem}.jpg"
|
||||
chemin = DOSSIER_EXPORTS / nom
|
||||
strip.save(chemin, "JPEG", quality=95)
|
||||
log.info(f"Strip cree : {chemin}")
|
||||
return chemin
|
||||
|
||||
|
||||
def creer_collage(photos: list[Path], colonnes: int = 2, marge: int = 15,
|
||||
couleur_fond: str = "#ffffff") -> Path:
|
||||
"""Cree un collage en grille avec les photos."""
|
||||
images = [Image.open(p) for p in photos]
|
||||
|
||||
# Taille uniforme pour chaque cellule
|
||||
largeur_cellule = min(img.width for img in images)
|
||||
hauteur_cellule = min(img.height for img in images)
|
||||
|
||||
images_redim = []
|
||||
for img in images:
|
||||
img_crop = ImageDraw.Draw(img)
|
||||
img_redim = img.resize((largeur_cellule, hauteur_cellule), Image.LANCZOS)
|
||||
images_redim.append(img_redim)
|
||||
|
||||
lignes = (len(images_redim) + colonnes - 1) // colonnes
|
||||
largeur_totale = colonnes * largeur_cellule + (colonnes + 1) * marge
|
||||
hauteur_totale = lignes * hauteur_cellule + (lignes + 1) * marge
|
||||
|
||||
collage = Image.new("RGB", (largeur_totale, hauteur_totale), couleur_fond)
|
||||
|
||||
for i, img in enumerate(images_redim):
|
||||
col = i % colonnes
|
||||
lig = i // colonnes
|
||||
x = marge + col * (largeur_cellule + marge)
|
||||
y = marge + lig * (hauteur_cellule + marge)
|
||||
collage.paste(img, (x, y))
|
||||
|
||||
nom = f"collage_{photos[0].stem}.jpg"
|
||||
chemin = DOSSIER_EXPORTS / nom
|
||||
collage.save(chemin, "JPEG", quality=95)
|
||||
log.info(f"Collage cree : {chemin}")
|
||||
return chemin
|
||||
52
backend/config.py
Normal file
52
backend/config.py
Normal file
@@ -0,0 +1,52 @@
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
RACINE = Path(__file__).resolve().parent.parent
|
||||
CHEMIN_CONFIG = RACINE / "data" / "config.json"
|
||||
CHEMIN_CONFIG_DEFAUT = RACINE / "data" / "config.defaut.json"
|
||||
DOSSIER_PHOTOS = RACINE / "data" / "photos"
|
||||
DOSSIER_EXPORTS = RACINE / "data" / "exports"
|
||||
DOSSIER_OVERLAYS = RACINE / "frontend" / "assets" / "overlays"
|
||||
DOSSIER_FONDS = RACINE / "frontend" / "assets" / "backgrounds"
|
||||
|
||||
|
||||
def charger_config() -> dict:
|
||||
"""Charge la configuration depuis config.json."""
|
||||
if not CHEMIN_CONFIG.exists():
|
||||
if CHEMIN_CONFIG_DEFAUT.exists():
|
||||
shutil.copy(CHEMIN_CONFIG_DEFAUT, CHEMIN_CONFIG)
|
||||
else:
|
||||
return {}
|
||||
with open(CHEMIN_CONFIG, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def sauvegarder_config(config: dict):
|
||||
"""Sauvegarde la configuration dans config.json."""
|
||||
CHEMIN_CONFIG.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(CHEMIN_CONFIG, "w", encoding="utf-8") as f:
|
||||
json.dump(config, f, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
def mettre_a_jour_config(modifications: dict) -> dict:
|
||||
"""Met a jour la configuration avec les modifications fournies (merge profond)."""
|
||||
config = charger_config()
|
||||
_merge_profond(config, modifications)
|
||||
sauvegarder_config(config)
|
||||
return config
|
||||
|
||||
|
||||
def _merge_profond(base: dict, modifications: dict):
|
||||
"""Fusionne recursivement les modifications dans la base."""
|
||||
for cle, valeur in modifications.items():
|
||||
if cle in base and isinstance(base[cle], dict) and isinstance(valeur, dict):
|
||||
_merge_profond(base[cle], valeur)
|
||||
else:
|
||||
base[cle] = valeur
|
||||
|
||||
|
||||
# Initialisation des dossiers au chargement du module
|
||||
for dossier in [DOSSIER_PHOTOS, DOSSIER_EXPORTS, DOSSIER_OVERLAYS, DOSSIER_FONDS]:
|
||||
dossier.mkdir(parents=True, exist_ok=True)
|
||||
142
backend/effects.py
Normal file
142
backend/effects.py
Normal file
@@ -0,0 +1,142 @@
|
||||
import logging
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageEnhance, ImageFilter, ImageOps
|
||||
|
||||
from backend.config import DOSSIER_OVERLAYS, DOSSIER_EXPORTS, DOSSIER_FONDS
|
||||
|
||||
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",
|
||||
}
|
||||
|
||||
|
||||
def appliquer_filtre(chemin_photo: Path, filtre: str) -> Path:
|
||||
"""Applique un filtre a une photo et sauvegarde dans exports."""
|
||||
img = Image.open(chemin_photo)
|
||||
|
||||
if filtre == "nb":
|
||||
img = ImageOps.grayscale(img).convert("RGB")
|
||||
elif filtre == "sepia":
|
||||
img = ImageOps.grayscale(img)
|
||||
img = ImageOps.colorize(img, black="#704214", white="#f0e0c0")
|
||||
elif filtre == "vintage":
|
||||
img = ImageOps.grayscale(img)
|
||||
img = ImageOps.colorize(img, black="#2b1700", white="#f5deb3")
|
||||
enhancer = ImageEnhance.Contrast(img)
|
||||
img = enhancer.enhance(0.85)
|
||||
elif filtre == "contraste":
|
||||
enhancer = ImageEnhance.Contrast(img)
|
||||
img = enhancer.enhance(1.5)
|
||||
elif filtre == "lumineux":
|
||||
enhancer = ImageEnhance.Brightness(img)
|
||||
img = enhancer.enhance(1.3)
|
||||
elif filtre == "chaud":
|
||||
r, g, b = img.split()
|
||||
r = r.point(lambda x: min(255, x + 20))
|
||||
b = b.point(lambda x: max(0, x - 20))
|
||||
img = Image.merge("RGB", (r, g, b))
|
||||
elif filtre == "froid":
|
||||
r, g, b = img.split()
|
||||
r = r.point(lambda x: max(0, x - 20))
|
||||
b = b.point(lambda x: min(255, x + 20))
|
||||
img = Image.merge("RGB", (r, g, b))
|
||||
elif filtre == "flou_artistique":
|
||||
img = img.filter(ImageFilter.GaussianBlur(radius=2))
|
||||
|
||||
nom_export = f"{chemin_photo.stem}_{filtre}.jpg"
|
||||
chemin_export = DOSSIER_EXPORTS / nom_export
|
||||
img.save(chemin_export, "JPEG", quality=95)
|
||||
log.info(f"Filtre '{filtre}' applique : {chemin_export}")
|
||||
return chemin_export
|
||||
|
||||
|
||||
def appliquer_overlay(chemin_photo: Path, nom_overlay: str) -> Path:
|
||||
"""Applique un overlay PNG par-dessus la photo."""
|
||||
chemin_overlay = DOSSIER_OVERLAYS / nom_overlay
|
||||
if not chemin_overlay.exists():
|
||||
log.error(f"Overlay introuvable : {chemin_overlay}")
|
||||
return chemin_photo
|
||||
|
||||
photo = Image.open(chemin_photo).convert("RGBA")
|
||||
overlay = Image.open(chemin_overlay).convert("RGBA")
|
||||
overlay = overlay.resize(photo.size, Image.LANCZOS)
|
||||
composite = Image.alpha_composite(photo, overlay)
|
||||
resultat = composite.convert("RGB")
|
||||
|
||||
nom_export = f"{chemin_photo.stem}_overlay.jpg"
|
||||
chemin_export = DOSSIER_EXPORTS / nom_export
|
||||
resultat.save(chemin_export, "JPEG", quality=95)
|
||||
log.info(f"Overlay '{nom_overlay}' applique : {chemin_export}")
|
||||
return chemin_export
|
||||
|
||||
|
||||
def chroma_key(chemin_photo: Path, nom_fond: str | None = None,
|
||||
couleur_cle: str = "#00ff00", tolerance: int = 40) -> Path:
|
||||
"""Remplace le fond vert par un fond choisi."""
|
||||
try:
|
||||
import cv2
|
||||
import numpy as np
|
||||
except ImportError:
|
||||
log.error("OpenCV non installe, chroma key indisponible")
|
||||
return chemin_photo
|
||||
|
||||
img = cv2.imread(str(chemin_photo))
|
||||
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
|
||||
|
||||
# Plage de vert pour le chroma key
|
||||
vert_bas = np.array([35 - tolerance // 2, 50, 50])
|
||||
vert_haut = np.array([85 + tolerance // 2, 255, 255])
|
||||
masque = cv2.inRange(hsv, vert_bas, vert_haut)
|
||||
|
||||
# Adoucir le masque
|
||||
masque = cv2.GaussianBlur(masque, (5, 5), 0)
|
||||
|
||||
# Charger le fond
|
||||
if nom_fond:
|
||||
chemin_fond = DOSSIER_FONDS / nom_fond
|
||||
if chemin_fond.exists():
|
||||
fond = cv2.imread(str(chemin_fond))
|
||||
fond = cv2.resize(fond, (img.shape[1], img.shape[0]))
|
||||
else:
|
||||
fond = np.full_like(img, (200, 200, 200))
|
||||
else:
|
||||
fond = np.full_like(img, (200, 200, 200))
|
||||
|
||||
# Combiner
|
||||
masque_inv = cv2.bitwise_not(masque)
|
||||
premier_plan = cv2.bitwise_and(img, img, mask=masque_inv)
|
||||
arriere_plan = cv2.bitwise_and(fond, fond, mask=masque)
|
||||
resultat = cv2.add(premier_plan, arriere_plan)
|
||||
|
||||
nom_export = f"{chemin_photo.stem}_chroma.jpg"
|
||||
chemin_export = DOSSIER_EXPORTS / nom_export
|
||||
cv2.imwrite(str(chemin_export), resultat)
|
||||
log.info(f"Chroma key applique : {chemin_export}")
|
||||
return chemin_export
|
||||
|
||||
|
||||
def lister_overlays() -> list[str]:
|
||||
"""Liste les overlays disponibles."""
|
||||
if not DOSSIER_OVERLAYS.exists():
|
||||
return []
|
||||
return [f.name for f in DOSSIER_OVERLAYS.iterdir() if f.suffix.lower() == ".png"]
|
||||
|
||||
|
||||
def lister_fonds() -> list[str]:
|
||||
"""Liste les fonds disponibles pour le chroma key."""
|
||||
if not DOSSIER_FONDS.exists():
|
||||
return []
|
||||
extensions = {".jpg", ".jpeg", ".png"}
|
||||
return [f.name for f in DOSSIER_FONDS.iterdir() if f.suffix.lower() in extensions]
|
||||
59
backend/gallery.py
Normal file
59
backend/gallery.py
Normal file
@@ -0,0 +1,59 @@
|
||||
import logging
|
||||
import os
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from backend.config import DOSSIER_PHOTOS, DOSSIER_EXPORTS
|
||||
|
||||
log = logging.getLogger("photobooth.gallery")
|
||||
|
||||
|
||||
def lister_photos(dossier: str = "exports") -> list[dict]:
|
||||
"""Liste les photos d'un dossier avec leurs metadonnees."""
|
||||
chemin = DOSSIER_EXPORTS if dossier == "exports" else DOSSIER_PHOTOS
|
||||
photos = []
|
||||
extensions = {".jpg", ".jpeg", ".png", ".gif"}
|
||||
|
||||
if not chemin.exists():
|
||||
return photos
|
||||
|
||||
for fichier in sorted(chemin.iterdir(), key=lambda f: f.stat().st_mtime, reverse=True):
|
||||
if fichier.suffix.lower() in extensions:
|
||||
stat = fichier.stat()
|
||||
photos.append({
|
||||
"nom": fichier.name,
|
||||
"chemin": str(fichier.relative_to(chemin.parent.parent)),
|
||||
"taille": stat.st_size,
|
||||
"date": datetime.fromtimestamp(stat.st_mtime).isoformat(),
|
||||
})
|
||||
return photos
|
||||
|
||||
|
||||
def compter_photos() -> dict:
|
||||
"""Retourne le compteur de photos pour l'evenement."""
|
||||
nb_photos = len(list(DOSSIER_PHOTOS.glob("*.jpg"))) + len(list(DOSSIER_PHOTOS.glob("*.jpeg")))
|
||||
nb_exports = len(list(DOSSIER_EXPORTS.glob("*"))) - len(list(DOSSIER_EXPORTS.glob(".gitkeep")))
|
||||
return {
|
||||
"photos_prises": nb_photos,
|
||||
"exports": max(0, nb_exports),
|
||||
}
|
||||
|
||||
|
||||
def supprimer_photo(nom: str, dossier: str = "exports") -> bool:
|
||||
"""Supprime une photo par son nom."""
|
||||
chemin = DOSSIER_EXPORTS if dossier == "exports" else DOSSIER_PHOTOS
|
||||
fichier = chemin / nom
|
||||
if fichier.exists() and fichier.parent == chemin:
|
||||
fichier.unlink()
|
||||
log.info(f"Photo supprimee : {fichier}")
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def vider_galerie():
|
||||
"""Supprime toutes les photos (reset evenement)."""
|
||||
for dossier in [DOSSIER_PHOTOS, DOSSIER_EXPORTS]:
|
||||
for fichier in dossier.iterdir():
|
||||
if fichier.name != ".gitkeep":
|
||||
fichier.unlink()
|
||||
log.info("Galerie videe")
|
||||
41
backend/gif_maker.py
Normal file
41
backend/gif_maker.py
Normal file
@@ -0,0 +1,41 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image
|
||||
|
||||
from backend.config import DOSSIER_EXPORTS, charger_config
|
||||
|
||||
log = logging.getLogger("photobooth.gif")
|
||||
|
||||
|
||||
def creer_gif(photos: list[Path], duree_frame_ms: int | None = None) -> Path:
|
||||
"""Cree un GIF anime a partir d'une liste de photos."""
|
||||
config = charger_config()
|
||||
if duree_frame_ms is None:
|
||||
duree_frame_ms = config.get("gif", {}).get("duree_frame_ms", 300)
|
||||
|
||||
images = []
|
||||
for p in photos:
|
||||
img = Image.open(p).convert("RGB")
|
||||
# Redimensionner pour un GIF leger
|
||||
img.thumbnail((800, 600), Image.LANCZOS)
|
||||
images.append(img)
|
||||
|
||||
if not images:
|
||||
log.error("Aucune image pour le GIF")
|
||||
return None
|
||||
|
||||
nom = f"gif_{photos[0].stem}.gif"
|
||||
chemin = DOSSIER_EXPORTS / nom
|
||||
|
||||
images[0].save(
|
||||
chemin,
|
||||
save_all=True,
|
||||
append_images=images[1:],
|
||||
duration=duree_frame_ms,
|
||||
loop=0,
|
||||
optimize=True,
|
||||
)
|
||||
|
||||
log.info(f"GIF cree : {chemin} ({len(images)} frames, {duree_frame_ms}ms)")
|
||||
return chemin
|
||||
64
backend/mailer.py
Normal file
64
backend/mailer.py
Normal file
@@ -0,0 +1,64 @@
|
||||
import logging
|
||||
import smtplib
|
||||
from email.mime.base import MIMEBase
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
from email import encoders
|
||||
from pathlib import Path
|
||||
|
||||
from backend.config import charger_config
|
||||
|
||||
log = logging.getLogger("photobooth.mailer")
|
||||
|
||||
|
||||
def envoyer_photo(destinataire: str, chemin_photo: Path) -> bool:
|
||||
"""Envoie une photo par email."""
|
||||
config = charger_config()
|
||||
conf_email = config.get("email", {})
|
||||
conf_event = config.get("evenement", {})
|
||||
|
||||
smtp_host = conf_email.get("smtp_host", "")
|
||||
smtp_port = conf_email.get("smtp_port", 587)
|
||||
smtp_user = conf_email.get("smtp_user", "")
|
||||
smtp_password = conf_email.get("smtp_password", "")
|
||||
expediteur = conf_email.get("expediteur", smtp_user)
|
||||
|
||||
if not smtp_host or not smtp_user:
|
||||
log.error("Configuration email incomplete")
|
||||
return False
|
||||
|
||||
sujet = conf_email.get("sujet", "Votre photo - {evenement}").format(
|
||||
evenement=conf_event.get("nom", "Photobooth")
|
||||
)
|
||||
message = conf_email.get("message", "Voici votre photo !").format(
|
||||
evenement=conf_event.get("nom", "Photobooth")
|
||||
)
|
||||
|
||||
msg = MIMEMultipart()
|
||||
msg["From"] = expediteur
|
||||
msg["To"] = destinataire
|
||||
msg["Subject"] = sujet
|
||||
msg.attach(MIMEText(message, "plain", "utf-8"))
|
||||
|
||||
# Piece jointe
|
||||
if chemin_photo.exists():
|
||||
with open(chemin_photo, "rb") as f:
|
||||
piece = MIMEBase("image", "jpeg")
|
||||
piece.set_payload(f.read())
|
||||
encoders.encode_base64(piece)
|
||||
piece.add_header(
|
||||
"Content-Disposition",
|
||||
f"attachment; filename={chemin_photo.name}"
|
||||
)
|
||||
msg.attach(piece)
|
||||
|
||||
try:
|
||||
with smtplib.SMTP(smtp_host, smtp_port, timeout=10) as serveur:
|
||||
serveur.starttls()
|
||||
serveur.login(smtp_user, smtp_password)
|
||||
serveur.send_message(msg)
|
||||
log.info(f"Email envoye a {destinataire}")
|
||||
return True
|
||||
except (smtplib.SMTPException, OSError) as e:
|
||||
log.error(f"Erreur envoi email : {e}")
|
||||
return False
|
||||
364
backend/main.py
Normal file
364
backend/main.py
Normal file
@@ -0,0 +1,364 @@
|
||||
import asyncio
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File
|
||||
from fastapi.responses import FileResponse, JSONResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from backend.config import (
|
||||
RACINE, DOSSIER_PHOTOS, DOSSIER_EXPORTS, DOSSIER_OVERLAYS, DOSSIER_FONDS,
|
||||
charger_config, sauvegarder_config, mettre_a_jour_config,
|
||||
)
|
||||
from backend.camera import camera
|
||||
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.collage import creer_strip, creer_collage
|
||||
from backend.gif_maker import creer_gif
|
||||
from backend.printer import lister_imprimantes, imprimer
|
||||
from backend.mailer import envoyer_photo
|
||||
from backend.qrcode_gen import generer_qr, qr_galerie
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s")
|
||||
log = logging.getLogger("photobooth")
|
||||
|
||||
# Clients WebSocket connectes
|
||||
clients_ws: list[WebSocket] = []
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
"""Demarrage et arret de l'application."""
|
||||
log.info("Demarrage du photobooth")
|
||||
camera.connecter()
|
||||
yield
|
||||
log.info("Arret du photobooth")
|
||||
camera.deconnecter()
|
||||
|
||||
|
||||
app = FastAPI(title="Photobooth", lifespan=lifespan)
|
||||
|
||||
# Servir les fichiers statiques
|
||||
app.mount("/assets", StaticFiles(directory=str(RACINE / "frontend" / "assets")), name="assets")
|
||||
app.mount("/css", StaticFiles(directory=str(RACINE / "frontend" / "css")), name="css")
|
||||
app.mount("/js", StaticFiles(directory=str(RACINE / "frontend" / "js")), name="js")
|
||||
app.mount("/data", StaticFiles(directory=str(RACINE / "data")), name="data")
|
||||
|
||||
|
||||
# --- Pages ---
|
||||
|
||||
@app.get("/")
|
||||
async def page_principale():
|
||||
return FileResponse(str(RACINE / "frontend" / "index.html"))
|
||||
|
||||
|
||||
# --- API Config ---
|
||||
|
||||
@app.get("/api/config")
|
||||
async def api_config():
|
||||
return charger_config()
|
||||
|
||||
|
||||
@app.post("/api/config")
|
||||
async def api_config_update(modifications: dict):
|
||||
config = mettre_a_jour_config(modifications)
|
||||
await diffuser_ws({"type": "config_maj", "config": config})
|
||||
return config
|
||||
|
||||
|
||||
# --- API Camera ---
|
||||
|
||||
@app.post("/api/capturer")
|
||||
async def api_capturer():
|
||||
chemin = camera.capturer()
|
||||
if chemin is None:
|
||||
return JSONResponse({"erreur": "Echec capture"}, status_code=500)
|
||||
nom = chemin.name
|
||||
await diffuser_ws({"type": "photo_capturee", "nom": nom})
|
||||
return {"nom": nom, "chemin": f"/data/photos/{nom}"}
|
||||
|
||||
|
||||
@app.get("/api/preview")
|
||||
async def api_preview():
|
||||
donnees = camera.preview()
|
||||
if donnees is None:
|
||||
return JSONResponse({"erreur": "Preview indisponible"}, status_code=500)
|
||||
b64 = base64.b64encode(donnees).decode("ascii")
|
||||
return {"image": f"data:image/jpeg;base64,{b64}"}
|
||||
|
||||
|
||||
@app.get("/api/camera/statut")
|
||||
async def api_camera_statut():
|
||||
return {
|
||||
"connectee": camera.connectee,
|
||||
"appareils": camera.lister_appareils(),
|
||||
}
|
||||
|
||||
|
||||
@app.post("/api/camera/reconnecter")
|
||||
async def api_camera_reconnecter():
|
||||
camera.deconnecter()
|
||||
ok = camera.connecter()
|
||||
return {"connectee": ok}
|
||||
|
||||
|
||||
# --- API Effets ---
|
||||
|
||||
@app.get("/api/filtres")
|
||||
async def api_filtres():
|
||||
return FILTRES
|
||||
|
||||
|
||||
@app.post("/api/filtre")
|
||||
async def api_appliquer_filtre(donnees: dict):
|
||||
nom_photo = donnees.get("photo", "")
|
||||
filtre = donnees.get("filtre", "original")
|
||||
chemin = DOSSIER_PHOTOS / nom_photo
|
||||
if not chemin.exists():
|
||||
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
||||
if filtre == "original":
|
||||
return {"nom": nom_photo, "chemin": f"/data/photos/{nom_photo}"}
|
||||
chemin_export = appliquer_filtre(chemin, filtre)
|
||||
return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"}
|
||||
|
||||
|
||||
@app.get("/api/overlays")
|
||||
async def api_overlays():
|
||||
return lister_overlays()
|
||||
|
||||
|
||||
@app.post("/api/overlay")
|
||||
async def api_appliquer_overlay(donnees: dict):
|
||||
nom_photo = donnees.get("photo", "")
|
||||
nom_overlay = donnees.get("overlay", "")
|
||||
chemin = DOSSIER_PHOTOS / nom_photo
|
||||
if not chemin.exists():
|
||||
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
||||
chemin_export = appliquer_overlay(chemin, nom_overlay)
|
||||
return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"}
|
||||
|
||||
|
||||
@app.post("/api/chroma")
|
||||
async def api_chroma_key(donnees: dict):
|
||||
nom_photo = donnees.get("photo", "")
|
||||
nom_fond = donnees.get("fond")
|
||||
chemin = DOSSIER_PHOTOS / nom_photo
|
||||
if not chemin.exists():
|
||||
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
||||
config = charger_config()
|
||||
conf_chroma = config.get("chroma_key", {})
|
||||
chemin_export = chroma_key(
|
||||
chemin, nom_fond,
|
||||
couleur_cle=conf_chroma.get("couleur", "#00ff00"),
|
||||
tolerance=conf_chroma.get("tolerance", 40),
|
||||
)
|
||||
return {"nom": chemin_export.name, "chemin": f"/data/exports/{chemin_export.name}"}
|
||||
|
||||
|
||||
@app.get("/api/fonds")
|
||||
async def api_fonds():
|
||||
return lister_fonds()
|
||||
|
||||
|
||||
# --- API Collage / GIF ---
|
||||
|
||||
@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:
|
||||
if not c.exists():
|
||||
return JSONResponse({"erreur": f"Photo introuvable : {c.name}"}, status_code=404)
|
||||
chemin = creer_strip(chemins)
|
||||
return {"nom": chemin.name, "chemin": f"/data/exports/{chemin.name}"}
|
||||
|
||||
|
||||
@app.post("/api/collage")
|
||||
async def api_collage(donnees: dict):
|
||||
noms = donnees.get("photos", [])
|
||||
colonnes = donnees.get("colonnes", 2)
|
||||
chemins = [DOSSIER_PHOTOS / n for n in noms]
|
||||
for c in chemins:
|
||||
if not c.exists():
|
||||
return JSONResponse({"erreur": f"Photo introuvable : {c.name}"}, status_code=404)
|
||||
chemin = creer_collage(chemins, colonnes=colonnes)
|
||||
return {"nom": chemin.name, "chemin": f"/data/exports/{chemin.name}"}
|
||||
|
||||
|
||||
@app.post("/api/gif")
|
||||
async def api_gif(donnees: dict):
|
||||
noms = donnees.get("photos", [])
|
||||
chemins = [DOSSIER_PHOTOS / n for n in noms]
|
||||
for c in chemins:
|
||||
if not c.exists():
|
||||
return JSONResponse({"erreur": f"Photo introuvable : {c.name}"}, status_code=404)
|
||||
chemin = creer_gif(chemins)
|
||||
if chemin is None:
|
||||
return JSONResponse({"erreur": "Echec creation GIF"}, status_code=500)
|
||||
return {"nom": chemin.name, "chemin": f"/data/exports/{chemin.name}"}
|
||||
|
||||
|
||||
# --- API Impression ---
|
||||
|
||||
@app.get("/api/imprimantes")
|
||||
async def api_imprimantes():
|
||||
return lister_imprimantes()
|
||||
|
||||
|
||||
@app.post("/api/imprimer")
|
||||
async def api_imprimer(donnees: dict):
|
||||
nom = donnees.get("photo", "")
|
||||
# Chercher dans exports d'abord, puis photos
|
||||
chemin = DOSSIER_EXPORTS / nom
|
||||
if not chemin.exists():
|
||||
chemin = DOSSIER_PHOTOS / nom
|
||||
if not chemin.exists():
|
||||
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
||||
ok = imprimer(chemin)
|
||||
return {"succes": ok}
|
||||
|
||||
|
||||
# --- API Email ---
|
||||
|
||||
@app.post("/api/email")
|
||||
async def api_email(donnees: dict):
|
||||
email_dest = donnees.get("email", "")
|
||||
nom = donnees.get("photo", "")
|
||||
if not email_dest:
|
||||
return JSONResponse({"erreur": "Email requis"}, status_code=400)
|
||||
chemin = DOSSIER_EXPORTS / nom
|
||||
if not chemin.exists():
|
||||
chemin = DOSSIER_PHOTOS / nom
|
||||
if not chemin.exists():
|
||||
return JSONResponse({"erreur": "Photo introuvable"}, status_code=404)
|
||||
ok = envoyer_photo(email_dest, chemin)
|
||||
return {"succes": ok}
|
||||
|
||||
|
||||
# --- API QR Code ---
|
||||
|
||||
@app.get("/api/qr/galerie")
|
||||
async def api_qr_galerie():
|
||||
chemin = qr_galerie()
|
||||
if chemin is None:
|
||||
return JSONResponse({"erreur": "URL galerie non configuree"}, status_code=400)
|
||||
return {"chemin": f"/data/exports/{chemin.name}"}
|
||||
|
||||
|
||||
# --- API Galerie ---
|
||||
|
||||
@app.get("/api/galerie")
|
||||
async def api_galerie():
|
||||
return lister_photos("exports")
|
||||
|
||||
|
||||
@app.get("/api/galerie/compteur")
|
||||
async def api_compteur():
|
||||
return compter_photos()
|
||||
|
||||
|
||||
@app.delete("/api/galerie/{nom}")
|
||||
async def api_supprimer_photo(nom: str):
|
||||
ok = supprimer_photo(nom)
|
||||
return {"succes": ok}
|
||||
|
||||
|
||||
@app.post("/api/galerie/vider")
|
||||
async def api_vider_galerie():
|
||||
vider_galerie()
|
||||
return {"succes": True}
|
||||
|
||||
|
||||
# --- API Upload overlay/fond ---
|
||||
|
||||
@app.post("/api/upload/overlay")
|
||||
async def api_upload_overlay(fichier: UploadFile = File(...)):
|
||||
chemin = DOSSIER_OVERLAYS / fichier.filename
|
||||
with open(chemin, "wb") as f:
|
||||
f.write(await fichier.read())
|
||||
return {"nom": fichier.filename}
|
||||
|
||||
|
||||
@app.post("/api/upload/fond")
|
||||
async def api_upload_fond(fichier: UploadFile = File(...)):
|
||||
chemin = DOSSIER_FONDS / fichier.filename
|
||||
with open(chemin, "wb") as f:
|
||||
f.write(await fichier.read())
|
||||
return {"nom": fichier.filename}
|
||||
|
||||
|
||||
# --- API Systeme ---
|
||||
|
||||
@app.post("/api/systeme/redemarrer")
|
||||
async def api_redemarrer():
|
||||
import subprocess
|
||||
log.warning("Redemarrage systeme demande")
|
||||
subprocess.Popen(["sudo", "reboot"])
|
||||
return {"succes": True}
|
||||
|
||||
|
||||
@app.post("/api/systeme/eteindre")
|
||||
async def api_eteindre():
|
||||
import subprocess
|
||||
log.warning("Extinction systeme demandee")
|
||||
subprocess.Popen(["sudo", "shutdown", "-h", "now"])
|
||||
return {"succes": True}
|
||||
|
||||
|
||||
# --- WebSocket ---
|
||||
|
||||
@app.websocket("/ws")
|
||||
async def websocket_endpoint(ws: WebSocket):
|
||||
await ws.accept()
|
||||
clients_ws.append(ws)
|
||||
log.info(f"Client WS connecte ({len(clients_ws)} clients)")
|
||||
try:
|
||||
while True:
|
||||
data = await ws.receive_text()
|
||||
msg = json.loads(data)
|
||||
await traiter_message_ws(msg, ws)
|
||||
except WebSocketDisconnect:
|
||||
clients_ws.remove(ws)
|
||||
log.info(f"Client WS deconnecte ({len(clients_ws)} clients)")
|
||||
|
||||
|
||||
async def traiter_message_ws(msg: dict, ws: WebSocket):
|
||||
"""Traite les messages WebSocket entrants."""
|
||||
type_msg = msg.get("type", "")
|
||||
|
||||
if type_msg == "ping":
|
||||
await ws.send_json({"type": "pong"})
|
||||
elif type_msg == "preview":
|
||||
donnees = camera.preview()
|
||||
if donnees:
|
||||
b64 = base64.b64encode(donnees).decode("ascii")
|
||||
await ws.send_json({"type": "preview", "image": f"data:image/jpeg;base64,{b64}"})
|
||||
|
||||
|
||||
async def diffuser_ws(message: dict):
|
||||
"""Envoie un message a tous les clients WebSocket."""
|
||||
deconnectes = []
|
||||
for ws in clients_ws:
|
||||
try:
|
||||
await ws.send_json(message)
|
||||
except Exception:
|
||||
deconnectes.append(ws)
|
||||
for ws in deconnectes:
|
||||
clients_ws.remove(ws)
|
||||
|
||||
|
||||
# Point d'entree
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
config = charger_config()
|
||||
conf_srv = config.get("serveur", {})
|
||||
uvicorn.run(
|
||||
"backend.main:app",
|
||||
host=conf_srv.get("host", "0.0.0.0"),
|
||||
port=conf_srv.get("port", 8080),
|
||||
reload=False,
|
||||
log_level="info",
|
||||
)
|
||||
81
backend/printer.py
Normal file
81
backend/printer.py
Normal file
@@ -0,0 +1,81 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from backend.config import charger_config
|
||||
|
||||
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")
|
||||
|
||||
|
||||
def lister_imprimantes() -> list[dict]:
|
||||
"""Liste les imprimantes disponibles via CUPS."""
|
||||
if not CUPS_DISPONIBLE:
|
||||
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 []
|
||||
|
||||
|
||||
def imprimer(chemin_photo: Path, imprimante: str | None = None, copies: int | None = None) -> bool:
|
||||
"""Imprime une photo sur l'imprimante configuree."""
|
||||
config = charger_config()
|
||||
conf_imp = config.get("impression", {})
|
||||
|
||||
if imprimante is None:
|
||||
imprimante = conf_imp.get("imprimante")
|
||||
if copies is None:
|
||||
copies = conf_imp.get("copies", 1)
|
||||
|
||||
if not chemin_photo.exists():
|
||||
log.error(f"Fichier introuvable : {chemin_photo}")
|
||||
return False
|
||||
|
||||
if not CUPS_DISPONIBLE:
|
||||
log.info(f"[Simulation] Impression de {chemin_photo} x{copies}")
|
||||
return True
|
||||
|
||||
try:
|
||||
conn = cups.Connection()
|
||||
|
||||
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
|
||||
|
||||
options = {
|
||||
"copies": str(copies),
|
||||
"media": conf_imp.get("format", "10x15"),
|
||||
"fit-to-page": "true",
|
||||
}
|
||||
|
||||
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
|
||||
41
backend/qrcode_gen.py
Normal file
41
backend/qrcode_gen.py
Normal file
@@ -0,0 +1,41 @@
|
||||
import logging
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
from backend.config import DOSSIER_EXPORTS, charger_config
|
||||
|
||||
log = logging.getLogger("photobooth.qrcode")
|
||||
|
||||
try:
|
||||
import qrcode
|
||||
QRCODE_DISPONIBLE = True
|
||||
except ImportError:
|
||||
QRCODE_DISPONIBLE = False
|
||||
log.warning("qrcode non installe")
|
||||
|
||||
|
||||
def generer_qr(contenu: str, nom: str = "qr.png") -> Path | None:
|
||||
"""Genere un QR code PNG."""
|
||||
if not QRCODE_DISPONIBLE:
|
||||
log.error("Module qrcode non disponible")
|
||||
return None
|
||||
|
||||
qr = qrcode.QRCode(version=1, box_size=10, border=4)
|
||||
qr.add_data(contenu)
|
||||
qr.make(fit=True)
|
||||
img = qr.make_image(fill_color="black", back_color="white")
|
||||
|
||||
chemin = DOSSIER_EXPORTS / nom
|
||||
img.save(chemin)
|
||||
log.info(f"QR code genere : {chemin}")
|
||||
return chemin
|
||||
|
||||
|
||||
def qr_galerie() -> Path | None:
|
||||
"""Genere le QR code pointant vers la galerie en ligne."""
|
||||
config = charger_config()
|
||||
url = config.get("qr_code", {}).get("url_galerie", "")
|
||||
if not url:
|
||||
log.warning("URL galerie non configuree")
|
||||
return None
|
||||
return generer_qr(url, "qr_galerie.png")
|
||||
11
backend/requirements.txt
Normal file
11
backend/requirements.txt
Normal file
@@ -0,0 +1,11 @@
|
||||
fastapi==0.115.0
|
||||
uvicorn[standard]==0.30.0
|
||||
python-gphoto2==2.5.0
|
||||
Pillow==10.4.0
|
||||
imageio==2.35.0
|
||||
opencv-python-headless==4.10.0.84
|
||||
qrcode[pil]==7.4.2
|
||||
pycups==2.0.4
|
||||
python-multipart==0.0.9
|
||||
aiofiles==24.1.0
|
||||
websockets==12.0
|
||||
Reference in New Issue
Block a user