vitrine-pixels/8x32 PI/serveur_web.py

"""
=============================================================================
  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.
TAILLE_MAX_MO = 50

# 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,
        largeur=LARGEUR, hauteur=HAUTEUR,
    )


@app.route("/apercu/<path:relatif>")
def apercu(relatif):
    """L'aperçu GIF d'une animation (affiché dans la page par une balise <img>)."""
    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")
    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=lecteur.DOSSIER_ETAT)
        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"))

    temporaire = tempfile.mkdtemp(dir=lecteur.DOSSIER_ETAT)
    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://<adresse du Pi>:{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()