""" ============================================================================= SERVEUR WEB — gérer les animations depuis un navigateur Workshop « Vitrine pixels » ============================================================================= Ce programme transforme le Raspberry Pi en petit SITE WEB. Depuis n'importe quel ordinateur ou téléphone connecté au même réseau, on ouvre l'adresse du Pi dans un navigateur (par exemple http://172.16.50.27 en Wi-Fi ; voir INTERFACE_WEB.md) et on peut : - DÉPOSER des animations (GIF, PNG animés, séquences de PNG…) ; - voir un APERÇU animé de chacune ; - choisir celles qui sont JOUÉES (activer / désactiver) et leur ORDRE ; - en jouer une TOUT DE SUITE, ou passer à la suivante ; - les SUPPRIMER (elles passent d'abord par une corbeille). Il ne parle pas lui-même à la matrice : c'est toujours lecteur.py qui joue les animations. Le serveur se contente de ranger les fichiers dans le dossier des animations, et de laisser des « post-it » au lecteur (voir lecteur.py, partie 5). COMMENT LE LANCER ----------------- Sur le Raspberry Pi (normalement, c'est le service « vitrine-web » qui le lance tout seul, voir MODE_VITRINE.md) : sudo ~/vitrine-env/bin/python serveur_web.py Sur votre ordinateur, pour essayer (avec le lecteur en simulateur à côté) : python3 serveur_web.py --port 8080 puis ouvrir http://localhost:8080 dans le navigateur. LES OUTILS UTILISÉS ------------------- - Flask : une bibliothèque Python pour fabriquer des sites web. Chaque « route » (@app.route) associe une adresse (/, /ajouter…) à une fonction Python qui fabrique la réponse. - Jinja : le système de « gabarits » de Flask. La page web est écrite en HTML dans web/page.html, avec des trous {{ … }} que Python remplit. - Pillow : pour vérifier les images reçues et fabriquer les aperçus. ORGANISATION DU PROGRAMME ------------------------- 1. Réglages. 2. Les noms de fichiers. 3. Lister les animations et leurs informations. 4. Ranger : ordre, activer / désactiver, corbeille. 5. Les aperçus animés. 6. Les pages et les actions (les « routes »). 7. Le programme principal. """ import argparse import io import os import re import shutil import socket import subprocess import tempfile import unicodedata from flask import (Flask, Response, abort, flash, jsonify, redirect, render_template, request, send_file, url_for) from PIL import Image import lecteur from matrice import LARGEUR, HAUTEUR # ============================================================================= # 1. RÉGLAGES # ============================================================================= # Port du serveur. 80 est le port « normal » du web : on tape juste # http://172.16.50.27 dans le navigateur. (Il faut être administrateur.) PORT = 80 # Mot de passe pour accéder à la page. None = pas de mot de passe : # toute personne sur le même réseau peut ajouter ou supprimer des animations. # Exemple : MOT_DE_PASSE = "pixels" (le navigateur demandera un nom # d'utilisateur, n'importe lequel, et ce mot de passe). MOT_DE_PASSE = None # Taille maximum d'un envoi, en mégaoctets (tous les fichiers ensemble). TAILLE_MAX_MO = 200 # Nombre maximum d'images dans une SÉQUENCE de PNG (120 images = 10 secondes # à 12 images/s, 5 secondes à 24 images/s). MAX_IMAGES_SEQUENCE = 120 # Au-delà de ce nombre de PNG envoyés d'un coup dans le cadre « Un ou # plusieurs fichiers », c'est sûrement une séquence envoyée au mauvais # endroit (chaque PNG deviendrait une animation) : on refuse et on explique. MAX_PNG_SEPARES = 10 # Où ranger les envois le temps de les vérifier : le dossier temporaire du # système (sur la carte SD), pas /run qui est en mémoire vive et trop petit. DOSSIER_ENVOIS = tempfile.gettempdir() # Sous-dossiers spéciaux du dossier des animations. Leur nom commence par # un point : le lecteur les ignore (voir lecteur.est_cache). DESACTIVEES = ".desactivees" CORBEILLE = ".corbeille" ICI = os.path.dirname(os.path.abspath(__file__)) app = Flask(__name__, template_folder=os.path.join(ICI, "web"), # où est page.html static_folder=os.path.join(ICI, "web"), # où est style.css static_url_path="/static") app.secret_key = os.urandom(16) # nécessaire pour les messages « flash » app.config["MAX_CONTENT_LENGTH"] = TAILLE_MAX_MO * 1024 * 1024 DOSSIER = lecteur.trouver_dossier() # peut être changé par --dossier # ============================================================================= # 2. LES NOMS DE FICHIERS # ============================================================================= def nettoyer_nom(nom): """« Pluie de cœurs (v2)!! » → « pluie_de_coeurs_v2 ». Sans accent, sans espace, sans caractère spécial : des noms qui ne posent de problème à aucun ordinateur (et pas de « ../ » malicieux).""" nom = unicodedata.normalize("NFKD", nom) nom = nom.replace("œ", "oe").replace("Œ", "oe").replace("æ", "ae") nom = nom.encode("ascii", "ignore").decode("ascii").lower() # enlève les accents nom = re.sub(r"[^a-z0-9]+", "_", nom).strip("_") return nom or "animation" def sans_numero(nom): """« 03_vague » → « vague » : on retire le numéro d'ordre du début.""" return re.sub(r"^\d+[_-]?", "", nom) def decouper(nom): """« 03_vague.gif » → (« 03_vague », « .gif ») ; un dossier n'a pas d'extension.""" base, extension = os.path.splitext(nom) if extension.lower() not in lecteur.EXTENSIONS: return nom, "" return base, extension.lower() def chemin_sur(relatif): """Transforme un nom reçu du navigateur en chemin, en vérifiant qu'il reste DANS le dossier des animations. Sinon, quelqu'un pourrait demander « ../../etc/passwd »… On refuse tout ce qui sort du dossier.""" chemin = os.path.realpath(os.path.join(DOSSIER, relatif)) racine = os.path.realpath(DOSSIER) if not chemin.startswith(racine + os.sep) or not os.path.exists(chemin): abort(404) return chemin def nom_libre(dossier, nom): """Si « vague.gif » existe déjà dans le dossier, propose « vague_2.gif »…""" base, extension = decouper(nom) candidat, n = nom, 2 while os.path.exists(os.path.join(dossier, candidat)): candidat = f"{base}_{n}{extension}" n += 1 return candidat # ============================================================================= # 3. LISTER LES ANIMATIONS ET LEURS INFORMATIONS # ============================================================================= def actives(): """Noms des animations jouées, dans l'ordre de lecture.""" return [os.path.basename(c) for c in lecteur.lister_animations(DOSSIER)] def desactivees(): return [os.path.basename(c) for c in lecteur.lister_animations(os.path.join(DOSSIER, DESACTIVEES))] def date_de_modif(chemin): """Date de dernière modification (pour savoir si un aperçu est à refaire).""" if os.path.isdir(chemin): return max([os.path.getmtime(chemin)] + [os.path.getmtime(f) for f in lecteur.images_du_dossier(chemin)]) return os.path.getmtime(chemin) # Clé du cache : le nom SANS numéro + la date, pour qu'un simple changement # d'ordre (01_vague → 02_vague) ne force pas à tout recalculer. _cache = {} # clé → (informations, aperçu GIF) def analyser(chemin): """Charge une animation comme le lecteur, et en tire les informations affichées sur la page + l'aperçu. Le résultat est gardé en mémoire (« cache ») tant que le fichier ne change pas : le Pi 3 n'est pas rapide.""" date = date_de_modif(chemin) cle = (sans_numero(os.path.basename(chemin)), date) if cle in _cache: return _cache[cle] # Taille d'origine (avant adaptation à 32 × 8). premier = lecteur.images_du_dossier(chemin)[0] if os.path.isdir(chemin) else chemin with Image.open(premier) as image: taille = image.size etapes = lecteur.charger(chemin) fixe = len(etapes) == 1 infos = { "images": len(etapes), "duree": sum(d for _, d in etapes), "fixe": fixe, "sequence": os.path.isdir(chemin), "taille": taille, "taille_ok": taille == (LARGEUR, HAUTEUR), "version": int(date), } apercu = fabriquer_apercu(etapes, fixe) _cache[cle] = (infos, apercu) return infos, apercu # ============================================================================= # 4. RANGER : ORDRE, ACTIVER / DÉSACTIVER, CORBEILLE # ============================================================================= # # L'ordre de lecture, c'est l'ordre ALPHABÉTIQUE des noms (voir lecteur.py). # Pour changer l'ordre, on renomme donc les fichiers : 01_…, 02_…, 03_… def renumeroter(noms): """Renomme les animations actives pour qu'elles portent les numéros 01, 02, 03… dans l'ordre de la liste « noms ».""" # 1re étape : noms temporaires (commençant par un point, donc ignorés par # le lecteur), pour ne jamais écraser un fichier qui a déjà le nom visé. temporaires = [] for i, nom in enumerate(noms): temporaire = f".renommage_{i}_{nom}" os.rename(os.path.join(DOSSIER, nom), os.path.join(DOSSIER, temporaire)) temporaires.append(temporaire) # 2e étape : les noms définitifs. # Tous les numéros ont le même nombre de chiffres (01…99, ou 001…999 au-delà # de 99 animations), sinon l'ordre alphabétique mettrait « 100 » avant « 11 ». chiffres = max(2, len(str(len(noms)))) for i, (nom, temporaire) in enumerate(zip(noms, temporaires)): definitif = f"{i + 1:0{chiffres}d}_{sans_numero(nom)}" os.rename(os.path.join(DOSSIER, temporaire), os.path.join(DOSSIER, definitif)) def ranger_dans(sous_dossier, nom, depuis=None): """Déplace une animation dans un sous-dossier (désactivées ou corbeille).""" depuis = depuis or DOSSIER destination = os.path.join(DOSSIER, sous_dossier) os.makedirs(destination, exist_ok=True) nouveau = nom_libre(destination, sans_numero(nom)) shutil.move(os.path.join(depuis, nom), os.path.join(destination, nouveau)) def ajouter_a_la_fin(chemin_source, nom): """Place une nouvelle animation à la fin de la liste de lecture.""" noms = actives() nom = nom_libre(DOSSIER, f"{len(noms) + 1:02d}_{nom}") shutil.move(chemin_source, os.path.join(DOSSIER, nom)) renumeroter(noms + [nom]) # ============================================================================= # 5. LES APERÇUS ANIMÉS # ============================================================================= # Agrandissement des aperçus : le grand côté fait 192 points # (32 × 8 → 192 × 48 si la matrice est couchée, 8 × 32 → 48 × 192 si elle est debout). ZOOM = 192 // max(LARGEUR, HAUTEUR) def fabriquer_apercu(etapes, fixe): """Fabrique un GIF agrandi, exactement comme la matrice va l'afficher (après adaptation de la taille). Renvoie les octets du fichier.""" images, durees = [], [] for image, duree in etapes: images.append(image.resize((LARGEUR * ZOOM, HAUTEUR * ZOOM), lecteur.PLUS_PROCHE_VOISIN)) durees.append(max(20, round(duree * 1000))) if fixe: durees = [1000] sortie = io.BytesIO() # un « fichier » en mémoire, sans toucher au disque images[0].save(sortie, format="GIF", save_all=True, append_images=images[1:], duration=durees, loop=0) return sortie.getvalue() # ============================================================================= # 6. LES PAGES ET LES ACTIONS (LES « ROUTES ») # ============================================================================= @app.before_request def verifier_mot_de_passe(): """Avant chaque page : si un mot de passe est réglé, on le demande.""" if MOT_DE_PASSE is None: return None auth = request.authorization if auth is None or auth.password != MOT_DE_PASSE: return Response("Mot de passe nécessaire.", 401, {"WWW-Authenticate": 'Basic realm="Vitrine pixels"'}) return None def lire_en_cours(): try: with open(lecteur.FICHIER_EN_COURS) as f: return f.read().strip() except OSError: return "" def adresses_du_pi(): """Les adresses du Pi sur ses réseaux (câble, Wi-Fi), pour les afficher en bas de la page : elles peuvent changer d'un jour à l'autre.""" try: sortie = subprocess.run(["hostname", "-I"], capture_output=True, text=True, timeout=2).stdout adresses = [a for a in sortie.split() if "." in a] # IPv4 seulement except (OSError, subprocess.SubprocessError): adresses = [] return adresses or [socket.gethostname()] def fiche(nom, sous_dossier=""): """Tout ce que la page doit savoir sur une animation.""" relatif = os.path.join(sous_dossier, nom) if sous_dossier else nom chemin = os.path.join(DOSSIER, relatif) try: infos, _ = analyser(chemin) erreur = None except Exception as e: # fichier abîmé : on l'affiche quand même, avec l'erreur infos, erreur = {"version": 0}, str(e) return {"nom": nom, "titre": sans_numero(decouper(nom)[0]).replace("_", " "), "relatif": relatif, "infos": infos, "erreur": erreur} @app.route("/") def accueil(): """La page principale.""" corbeille = os.path.join(DOSSIER, CORBEILLE) nb_corbeille = len(os.listdir(corbeille)) if os.path.isdir(corbeille) else 0 return render_template( "page.html", actives=[fiche(n) for n in actives()], desactivees=[fiche(n, DESACTIVEES) for n in desactivees()], en_cours=lire_en_cours(), nb_corbeille=nb_corbeille, dossier=DOSSIER, adresses=adresses_du_pi(), port="" if request.host.endswith(":80") or ":" not in request.host else ":" + request.host.rsplit(":", 1)[1], extensions=",".join(lecteur.EXTENSIONS), taille_max=TAILLE_MAX_MO, max_images=MAX_IMAGES_SEQUENCE, largeur=LARGEUR, hauteur=HAUTEUR, ) @app.route("/apercu/") def apercu(relatif): """L'aperçu GIF d'une animation (affiché dans la page par une balise ).""" try: _, octets = analyser(chemin_sur(relatif)) except Exception: abort(404) reponse = send_file(io.BytesIO(octets), mimetype="image/gif") reponse.headers["Cache-Control"] = "max-age=86400" # l'URL change avec la version return reponse @app.route("/etat") def etat(): """Ce que joue le lecteur en ce moment (la page le redemande toutes les 2 s).""" return jsonify(en_cours=lire_en_cours()) def verifier_animation(chemin): """Vérifie qu'une animation se charge vraiment, comme le ferait le lecteur. Renvoie un message d'erreur, ou None si tout va bien.""" try: if not lecteur.charger(chemin): return "aucune image trouvée" except Exception as erreur: return f"fichier illisible ({erreur})" return None @app.route("/ajouter", methods=["POST"]) def ajouter(): """Reçoit un ou plusieurs fichiers : chacun devient une animation.""" fichiers = [f for f in request.files.getlist("fichiers") if f.filename] if not fichiers: flash("Aucun fichier choisi.", "erreur") nb_png = sum(f.filename.lower().endswith(".png") for f in fichiers) if nb_png > MAX_PNG_SEPARES: flash(f"{nb_png} PNG envoyés d'un coup : est-ce une séquence d'images ? " "Utilisez plutôt le cadre « Une séquence de PNG », sinon chaque image " "deviendrait une animation séparée.", "erreur") return redirect(url_for("accueil")) for fichier in fichiers: base, extension = os.path.splitext(fichier.filename) extension = extension.lower() if extension not in lecteur.EXTENSIONS: flash(f"« {fichier.filename} » : format non accepté " f"(acceptés : {', '.join(lecteur.EXTENSIONS)}).", "erreur") continue # On enregistre d'abord dans un dossier temporaire, on vérifie, puis on range. descripteur, temporaire = tempfile.mkstemp(suffix=extension, dir=DOSSIER_ENVOIS) os.close(descripteur) fichier.save(temporaire) probleme = verifier_animation(temporaire) if probleme: os.remove(temporaire) flash(f"« {fichier.filename} » refusé : {probleme}.", "erreur") continue ajouter_a_la_fin(temporaire, nettoyer_nom(sans_numero(base)) + extension) flash(f"« {fichier.filename} » ajouté à la fin de la liste.", "ok") return redirect(url_for("accueil")) @app.route("/ajouter_sequence", methods=["POST"]) def ajouter_sequence(): """Reçoit plusieurs PNG qui forment UNE animation (export Krita en séquence).""" fichiers = sorted((f for f in request.files.getlist("images") if f.filename), key=lambda f: f.filename) # ordre des noms = ordre des images nom = nettoyer_nom(request.form.get("nom", "") or "sequence") try: ips = float(request.form.get("ips", "12").replace(",", ".")) assert 0.5 <= ips <= 60 except (ValueError, AssertionError): flash("Images par seconde : un nombre entre 0,5 et 60.", "erreur") return redirect(url_for("accueil")) if len(fichiers) < 2: flash("Une séquence demande au moins 2 images PNG.", "erreur") return redirect(url_for("accueil")) if len(fichiers) > MAX_IMAGES_SEQUENCE: flash(f"{len(fichiers)} images : c'est trop, une séquence peut en compter " f"{MAX_IMAGES_SEQUENCE} au maximum. Découpez-la en plusieurs " "animations, ou exportez-la en GIF.", "erreur") return redirect(url_for("accueil")) temporaire = tempfile.mkdtemp(dir=DOSSIER_ENVOIS) for i, fichier in enumerate(fichiers): if not fichier.filename.lower().endswith(".png"): shutil.rmtree(temporaire) flash(f"« {fichier.filename} » n'est pas un PNG.", "erreur") return redirect(url_for("accueil")) fichier.save(os.path.join(temporaire, f"{nom}_{i:04d}.png")) with open(os.path.join(temporaire, "images_par_seconde.txt"), "w") as f: f.write(f"{ips:g}\n") probleme = verifier_animation(temporaire) if probleme: shutil.rmtree(temporaire) flash(f"Séquence refusée : {probleme}.", "erreur") else: ajouter_a_la_fin(temporaire, nom) flash(f"Séquence « {nom} » ajoutée ({len(fichiers)} images, {ips:g} i/s).", "ok") return redirect(url_for("accueil")) @app.route("/action", methods=["POST"]) def action(): """Tous les boutons des fiches : monter, descendre, jouer, désactiver…""" quoi = request.form.get("action") nom = os.path.basename(request.form.get("nom", "")) # basename : pas de « ../ » noms = actives() if quoi == "suivante": lecteur.ecrire_post_it(lecteur.FICHIER_DEMANDE, lecteur.SUIVANTE) flash("On passe à l'animation suivante.", "ok") elif quoi in ("monter", "descendre") and nom in noms: i = noms.index(nom) j = i - 1 if quoi == "monter" else i + 1 if 0 <= j < len(noms): noms[i], noms[j] = noms[j], noms[i] # on échange les deux places renumeroter(noms) elif quoi == "jouer": relatif = request.form.get("relatif", "") chemin_sur(relatif) # vérifie que l'animation existe bien lecteur.ecrire_post_it(lecteur.FICHIER_DEMANDE, relatif) flash(f"« {nom} » va être jouée tout de suite.", "ok") elif quoi == "desactiver" and nom in noms: ranger_dans(DESACTIVEES, nom) noms.remove(nom) renumeroter(noms) flash(f"« {nom} » ne sera plus jouée.", "ok") elif quoi == "activer" and nom in desactivees(): ajouter_a_la_fin(os.path.join(DOSSIER, DESACTIVEES, nom), nom) flash(f"« {nom} » est ajoutée à la fin de la liste.", "ok") elif quoi == "supprimer": if nom in noms: ranger_dans(CORBEILLE, nom) noms.remove(nom) renumeroter(noms) elif nom in desactivees(): ranger_dans(CORBEILLE, nom, depuis=os.path.join(DOSSIER, DESACTIVEES)) flash(f"« {nom} » est dans la corbeille.", "ok") elif quoi == "vider_corbeille": shutil.rmtree(os.path.join(DOSSIER, CORBEILLE), ignore_errors=True) flash("Corbeille vidée.", "ok") return redirect(url_for("accueil") + request.form.get("ancre", "")) @app.errorhandler(413) def trop_gros(erreur): flash(f"Envoi trop gros (maximum {TAILLE_MAX_MO} Mo).", "erreur") return redirect(url_for("accueil")) @app.errorhandler(OSError) def erreur_disque(erreur): # Cas typique : partition protégée en écriture (overlay, voir MODE_VITRINE.md). flash(f"Impossible d'écrire dans le dossier des animations ({erreur}). " "La carte SD est-elle protégée en écriture ?", "erreur") return redirect(url_for("accueil")) # ============================================================================= # 7. LE PROGRAMME PRINCIPAL # ============================================================================= def main(): global DOSSIER options = argparse.ArgumentParser(description="Interface web de la vitrine pixels.") options.add_argument("--port", type=int, default=PORT) options.add_argument("--dossier", help="dossier d'animations à gérer") args = options.parse_args() if args.dossier: DOSSIER = os.path.abspath(args.dossier) os.makedirs(DOSSIER, exist_ok=True) print("Dossier des animations :", DOSSIER) print(f"Ouvrez http://:{args.port} dans un navigateur.") # host="0.0.0.0" : accepter les visites venant des autres ordinateurs du réseau. app.run(host="0.0.0.0", port=args.port, threaded=True) if __name__ == "__main__": main()