575 lines
26 KiB
Python
575 lines
26 KiB
Python
"""Génère une vidéo MP4 carrée à partir d'une trace GPX.
|
|
|
|
Contenu de la vidéo : trace blanche sur fond noir, cercle blanc figurant le
|
|
véhicule, compteur de vitesse analogique à aiguille et noms des villes
|
|
traversées (géocodage inverse Nominatim / OpenStreetMap, mis en cache).
|
|
|
|
Une trace longue peut être découpée en plusieurs vidéos, chacune cadrée sur sa
|
|
portion : de durée de trajet identique (--morceaux, --duree-morceau) ou avec un
|
|
nombre de villes donné (--villes-par-morceau).
|
|
|
|
Usage :
|
|
python gpx_video.py trace.gpx [-o sortie.mp4] [--duree 60] [--fps 30]
|
|
[--morceaux N | --duree-morceau MIN |
|
|
--villes-par-morceau [N]]
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
TAILLE = 1024
|
|
SS = 2 # facteur de suréchantillonnage pour l'anticrénelage
|
|
MARGE = 60
|
|
RAYON_COMPTEUR = 150
|
|
RAYON_VEHICULE = 11
|
|
EPAISSEUR_TRACE = 3
|
|
PAS_VILLES_M = 1000
|
|
NOMINATIM_URL = "https://nominatim.openstreetmap.org/reverse"
|
|
USER_AGENT = "gpx-video-compteur/1.0 (script personnel)"
|
|
CLES_VILLE = ("city", "town", "village", "municipality", "hamlet")
|
|
LARGEUR_ZOOM = TAILLE // 2 # largeur du nom de ville zoomé
|
|
HAUTEUR_ZOOM_MAX = TAILLE // 4 # limite pour les noms très courts
|
|
FONDU_ZOOM = 0.3 # durée des fondus d'apparition/disparition (s de vidéo)
|
|
SEUIL_ARRET_KMH = 3 # sous cette vitesse moyenne, un segment est un arrêt
|
|
TRANSITION_ARRET = 1.0 # durée (s de trajet) du retour de l'aiguille à 0
|
|
|
|
BLANC = (255, 255, 255)
|
|
GRIS = (170, 170, 170)
|
|
NOIR = (0, 0, 0)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Lecture GPX et calculs
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _local(tag):
|
|
return tag.rsplit("}", 1)[-1]
|
|
|
|
|
|
def charger_gpx(chemin):
|
|
"""Retourne des tableaux numpy lat, lon, t (secondes), vitesse (km/h ou None)."""
|
|
racine = ET.parse(chemin).getroot()
|
|
lats, lons, temps, vitesses = [], [], [], []
|
|
for el in racine.iter():
|
|
if _local(el.tag) != "trkpt":
|
|
continue
|
|
t_txt, v_txt = None, None
|
|
for enfant in el.iter():
|
|
nom = _local(enfant.tag)
|
|
if nom == "time":
|
|
t_txt = enfant.text
|
|
elif nom == "speed":
|
|
v_txt = enfant.text
|
|
if t_txt is None:
|
|
continue
|
|
lats.append(float(el.get("lat")))
|
|
lons.append(float(el.get("lon")))
|
|
temps.append(datetime.fromisoformat(t_txt.strip().replace("Z", "+00:00")).timestamp())
|
|
vitesses.append(float(v_txt) * 3.6 if v_txt else math.nan)
|
|
|
|
if len(lats) < 2:
|
|
sys.exit("Erreur : la trace doit contenir au moins 2 points horodatés (<time>).")
|
|
|
|
ordre = np.argsort(temps, kind="stable")
|
|
lat = np.array(lats)[ordre]
|
|
lon = np.array(lons)[ordre]
|
|
t = np.array(temps)[ordre]
|
|
v = np.array(vitesses)[ordre]
|
|
return lat, lon, t - t[0], v
|
|
|
|
|
|
def distances_haversine(lat, lon):
|
|
"""Distance (m) entre points successifs."""
|
|
r = 6371000.0
|
|
p1, p2 = np.radians(lat[:-1]), np.radians(lat[1:])
|
|
dp = p2 - p1
|
|
dl = np.radians(lon[1:] - lon[:-1])
|
|
a = np.sin(dp / 2) ** 2 + np.cos(p1) * np.cos(p2) * np.sin(dl / 2) ** 2
|
|
return 2 * r * np.arcsin(np.sqrt(a))
|
|
|
|
|
|
def calculer_vitesses(lat, lon, t, v_gpx):
|
|
"""Vitesse en km/h : extension GPX si disponible, sinon distance/temps."""
|
|
if np.isnan(v_gpx).mean() < 0.5:
|
|
idx = np.arange(len(v_gpx))
|
|
ok = ~np.isnan(v_gpx)
|
|
v = np.interp(idx, idx[ok], v_gpx[ok])
|
|
else:
|
|
d = distances_haversine(lat, lon)
|
|
dt = np.diff(t)
|
|
seg = np.where(dt > 0, d / np.where(dt > 0, dt, 1), 0.0) * 3.6
|
|
v = np.concatenate([[seg[0]], (seg[:-1] + seg[1:]) / 2, [seg[-1]]])
|
|
noyau = np.ones(3) / 3
|
|
lisse = np.convolve(np.pad(v, 1, mode="edge"), noyau, mode="valid")
|
|
return np.clip(lisse, 0, None)
|
|
|
|
|
|
def annuler_arrets(lat, lon, t, v):
|
|
"""Met la vitesse à 0 pendant les arrêts (trace immobile).
|
|
|
|
Un segment est immobile si sa vitesse moyenne (distance/durée) est sous
|
|
SEUIL_ARRET_KMH. Le GPS enregistre souvent peu ou pas de points à
|
|
l'arrêt : sans correction, l'aiguille resterait figée sur la vitesse des
|
|
points encadrant la pause. Des points à vitesse nulle sont donc insérés
|
|
au début et à la fin de chaque segment immobile long.
|
|
"""
|
|
d = distances_haversine(lat, lon)
|
|
dt = np.diff(t)
|
|
immobile = d < SEUIL_ARRET_KMH / 3.6 * dt
|
|
v = v.copy()
|
|
# Point entouré de segments immobiles (ou en bout de trace) : à l'arrêt.
|
|
avant = np.concatenate([[True], immobile])
|
|
apres = np.concatenate([immobile, [True]])
|
|
v[avant & apres] = 0.0
|
|
|
|
longs = np.flatnonzero(immobile & (dt > 2 * TRANSITION_ARRET))
|
|
if len(longs) == 0:
|
|
return lat, lon, t, v
|
|
t_ajout = np.concatenate([t[longs] + TRANSITION_ARRET, t[longs + 1] - TRANSITION_ARRET])
|
|
t2 = np.concatenate([t, t_ajout])
|
|
lat2 = np.concatenate([lat, np.interp(t_ajout, t, lat)])
|
|
lon2 = np.concatenate([lon, np.interp(t_ajout, t, lon)])
|
|
v2 = np.concatenate([v, np.zeros(len(t_ajout))])
|
|
ordre = np.argsort(t2, kind="stable")
|
|
return lat2[ordre], lon2[ordre], t2[ordre], v2[ordre]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Villes (Nominatim)
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _nominatim(lat, lon):
|
|
params = urllib.parse.urlencode({
|
|
"format": "jsonv2", "lat": f"{lat:.6f}", "lon": f"{lon:.6f}",
|
|
"zoom": 13, "addressdetails": 1, "accept-language": "fr",
|
|
})
|
|
req = urllib.request.Request(f"{NOMINATIM_URL}?{params}", headers={"User-Agent": USER_AGENT})
|
|
# Certificats toujours vérifiés, mais sans le mode strict de Python 3.13
|
|
# que certains CA locaux (antivirus, proxy) ne respectent pas.
|
|
contexte = ssl.create_default_context()
|
|
contexte.verify_flags &= ~getattr(ssl, "VERIFY_X509_STRICT", 0)
|
|
with urllib.request.urlopen(req, timeout=20, context=contexte) as rep:
|
|
adresse = json.load(rep).get("address", {})
|
|
for cle in CLES_VILLE:
|
|
if cle in adresse:
|
|
return adresse[cle]
|
|
return None
|
|
|
|
|
|
def trouver_villes(chemin_gpx, lat, lon):
|
|
"""Liste de (nom, lat, lon) : villes traversées, dans l'ordre de passage."""
|
|
cache_path = Path(chemin_gpx).with_suffix(".villes.json")
|
|
if cache_path.exists():
|
|
print(f"Villes lues depuis le cache {cache_path.name}")
|
|
return [tuple(v) for v in json.loads(cache_path.read_text(encoding="utf-8"))]
|
|
|
|
cumul = np.concatenate([[0.0], np.cumsum(distances_haversine(lat, lon))])
|
|
cibles = np.arange(0, cumul[-1] + 1, PAS_VILLES_M)
|
|
indices = sorted(set(np.searchsorted(cumul, cibles).clip(0, len(lat) - 1).tolist()) | {len(lat) - 1})
|
|
|
|
points_par_ville = {}
|
|
erreurs = echecs_consecutifs = 0
|
|
for n, i in enumerate(indices, 1):
|
|
print(f"\rGéocodage inverse {n}/{len(indices)}", end="", flush=True)
|
|
try:
|
|
nom = _nominatim(lat[i], lon[i])
|
|
echecs_consecutifs = 0
|
|
except Exception as exc: # réseau indisponible, quota...
|
|
print(f"\n point {i} ignoré : {exc}")
|
|
nom = None
|
|
erreurs += 1
|
|
echecs_consecutifs += 1
|
|
if echecs_consecutifs >= 3:
|
|
print("Géocodage abandonné : vidéo générée sans villes.")
|
|
return []
|
|
if nom:
|
|
points_par_ville.setdefault(nom, []).append(i)
|
|
time.sleep(1.05) # politique d'usage Nominatim : 1 requête/s max
|
|
print()
|
|
|
|
villes = []
|
|
for nom, idx in points_par_ville.items():
|
|
# étiquette au point de trace le plus proche du centre des échantillons
|
|
cy, cx = lat[idx].mean(), lon[idx].mean()
|
|
proche = int(np.argmin((lat - cy) ** 2 + ((lon - cx) * math.cos(math.radians(cy))) ** 2))
|
|
villes.append((nom, float(lat[proche]), float(lon[proche])))
|
|
|
|
if erreurs == 0:
|
|
cache_path.write_text(json.dumps(villes, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"{len(villes)} villes : {', '.join(v[0] for v in villes)}")
|
|
return villes
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Mise en page
|
|
# --------------------------------------------------------------------------
|
|
|
|
class Projection:
|
|
"""Projection équirectangulaire locale vers pixels (image 1x)."""
|
|
|
|
def __init__(self, lat, lon):
|
|
self.k = math.cos(math.radians((lat.min() + lat.max()) / 2))
|
|
x, y = lon * self.k, lat
|
|
self.x0, self.y1 = x.min(), y.max()
|
|
largeur, hauteur = x.max() - x.min(), y.max() - y.min()
|
|
zone = TAILLE - 2 * MARGE
|
|
self.echelle = zone / max(largeur, hauteur, 1e-9)
|
|
self.dx = MARGE + (zone - largeur * self.echelle) / 2
|
|
self.dy = MARGE + (zone - hauteur * self.echelle) / 2
|
|
|
|
def __call__(self, lat, lon):
|
|
px = self.dx + (np.asarray(lon) * self.k - self.x0) * self.echelle
|
|
py = self.dy + (self.y1 - np.asarray(lat)) * self.echelle
|
|
return px, py
|
|
|
|
|
|
def choisir_coin_compteur(px, py):
|
|
"""Coin le moins encombré par la trace pour placer le compteur."""
|
|
d = RAYON_COMPTEUR + 20
|
|
coins = [(TAILLE - d, TAILLE - d), (d, TAILLE - d), (TAILLE - d, d), (d, d)]
|
|
def encombrement(c):
|
|
return int(np.sum((px - c[0]) ** 2 + (py - c[1]) ** 2 < (RAYON_COMPTEUR + 25) ** 2))
|
|
return min(coins, key=encombrement)
|
|
|
|
|
|
def charger_police(taille, gras=False):
|
|
noms = ["arialbd.ttf", "DejaVuSans-Bold.ttf"] if gras else ["arial.ttf", "DejaVuSans.ttf"]
|
|
for nom in noms:
|
|
for chemin in (nom, f"C:/Windows/Fonts/{nom}"):
|
|
try:
|
|
return ImageFont.truetype(chemin, taille)
|
|
except OSError:
|
|
pass
|
|
return ImageFont.load_default(taille)
|
|
|
|
|
|
def angle_vitesse(v, vmax):
|
|
"""Angle (radians, sens trigo) de l'aiguille : 0 à 210°, vmax à -30°."""
|
|
frac = min(max(v / vmax, 0.0), 1.0)
|
|
return math.radians(210 - 240 * frac)
|
|
|
|
|
|
def dessiner_fond(px, py, villes_px, centre, vmax):
|
|
"""Calque statique (résolution SS) : trace, villes, cadran du compteur."""
|
|
s = SS
|
|
img = Image.new("RGB", (TAILLE * s, TAILLE * s), NOIR)
|
|
dr = ImageDraw.Draw(img)
|
|
|
|
# Trace
|
|
points = list(zip((px * s).tolist(), (py * s).tolist()))
|
|
dr.line(points, fill=BLANC, width=EPAISSEUR_TRACE * s, joint="curve")
|
|
|
|
# Compteur : fond et cadran
|
|
cx, cy = centre[0] * s, centre[1] * s
|
|
r = RAYON_COMPTEUR * s
|
|
dr.ellipse([cx - r - 12 * s, cy - r - 12 * s, cx + r + 12 * s, cy + r + 12 * s], fill=NOIR)
|
|
dr.ellipse([cx - r, cy - r, cx + r, cy + r], outline=BLANC, width=3 * s)
|
|
police_graduation = charger_police(int(19 * s), gras=True)
|
|
for vit in range(0, int(vmax) + 1, 10):
|
|
a = angle_vitesse(vit, vmax)
|
|
majeure = vit % 20 == 0
|
|
r_ext = r - 10 * s
|
|
r_int = r_ext - (22 if majeure else 11) * s
|
|
dr.line([cx + r_int * math.cos(a), cy - r_int * math.sin(a),
|
|
cx + r_ext * math.cos(a), cy - r_ext * math.sin(a)],
|
|
fill=BLANC, width=(4 if majeure else 2) * s)
|
|
if majeure:
|
|
r_txt = r_int - 20 * s
|
|
dr.text((cx + r_txt * math.cos(a), cy - r_txt * math.sin(a)), str(vit),
|
|
font=police_graduation, fill=BLANC, anchor="mm")
|
|
dr.text((cx, cy + r * 0.72), "km/h", font=charger_police(int(18 * s)), fill=GRIS, anchor="mm")
|
|
|
|
# Villes : point + étiquette, en évitant chevauchements et compteur
|
|
police_ville = charger_police(int(20 * s), gras=True)
|
|
occupes = [(cx - r - 12 * s, cy - r - 12 * s, cx + r + 12 * s, cy + r + 12 * s)]
|
|
decalages = [(1, 0), (-1, 0), (0, -1), (0, 1), (1, -1), (-1, -1), (1, 1), (-1, 1)]
|
|
ancres = {(1, 0): "lm", (-1, 0): "rm", (0, -1): "mb", (0, 1): "mt",
|
|
(1, -1): "lb", (-1, -1): "rb", (1, 1): "lt", (-1, 1): "rt"}
|
|
for nom, x, y in villes_px:
|
|
x, y = x * s, y * s
|
|
dr.ellipse([x - 5 * s, y - 5 * s, x + 5 * s, y + 5 * s], fill=NOIR, outline=BLANC, width=2 * s)
|
|
meilleur = None
|
|
for ddx, ddy in decalages:
|
|
pos = (x + ddx * 12 * s, y + ddy * 12 * s)
|
|
boite = dr.textbbox(pos, nom, font=police_ville, anchor=ancres[(ddx, ddy)])
|
|
dans_image = boite[0] >= 0 and boite[1] >= 0 and boite[2] <= TAILLE * s and boite[3] <= TAILLE * s
|
|
conflits = sum(_intersecte(boite, o) for o in occupes)
|
|
# nombre de pixels de trace recouverts par l'étiquette
|
|
sur_trace = int(np.sum((px * s >= boite[0]) & (px * s <= boite[2]) &
|
|
(py * s >= boite[1]) & (py * s <= boite[3])))
|
|
score = (not dans_image) * 1000 + conflits * 100 + sur_trace
|
|
if meilleur is None or score < meilleur[0]:
|
|
meilleur = (score, pos, ancres[(ddx, ddy)], boite)
|
|
_, pos, ancre, boite = meilleur
|
|
dr.text(pos, nom, font=police_ville, fill=BLANC, anchor=ancre,
|
|
stroke_width=4 * s, stroke_fill=NOIR)
|
|
occupes.append(boite)
|
|
return img
|
|
|
|
|
|
def _intersecte(a, b):
|
|
return not (a[2] < b[0] or b[2] < a[0] or a[3] < b[1] or b[3] < a[1])
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Rendu des images et encodage
|
|
# --------------------------------------------------------------------------
|
|
|
|
def rendu_patch(fond_ss, boite, dessin):
|
|
"""Dessine sur une zone suréchantillonnée puis la réduit en taille 1x."""
|
|
x0, y0, x1, y1 = boite
|
|
patch = fond_ss.crop((x0 * SS, y0 * SS, x1 * SS, y1 * SS))
|
|
dessin(ImageDraw.Draw(patch), x0 * SS, y0 * SS)
|
|
return patch.resize((x1 - x0, y1 - y0), Image.LANCZOS)
|
|
|
|
|
|
def rendu_zoom(nom):
|
|
"""Masques 1x (contour, texte) du nom de ville zoomé et position du coin haut-gauche."""
|
|
l_max, h_max = LARGEUR_ZOOM * SS, HAUTEUR_ZOOM_MAX * SS
|
|
|
|
def mesure(taille):
|
|
police = charger_police(taille, gras=True)
|
|
contour = max(taille // 14, SS)
|
|
b = police.getbbox(nom, anchor="mm", stroke_width=contour)
|
|
return police, contour, b[2] - b[0], b[3] - b[1]
|
|
|
|
# Taille de police la plus grande dont le texte, contour compris, tient
|
|
# dans la zone : un nom long est réduit, jamais tronqué.
|
|
ref = 100 * SS
|
|
_, _, largeur, hauteur = mesure(ref)
|
|
taille = max(int(ref * min(l_max / max(largeur, 1), h_max / max(hauteur, 1))), 1)
|
|
police, contour, largeur, hauteur = mesure(taille)
|
|
while taille > 1 and (largeur > l_max or hauteur > h_max):
|
|
taille = max(taille - max(taille // 50, 1), 1)
|
|
police, contour, largeur, hauteur = mesure(taille)
|
|
|
|
# Dessin sur un canevas largement dimensionné puis recadrage sur l'encre
|
|
# réelle : aucun pixel (anticrénelage, accents, contour) n'est coupé.
|
|
marge = 4 * contour + 4 * SS
|
|
canevas = (largeur + 2 * marge, hauteur + 2 * marge)
|
|
centre_ss = (canevas[0] // 2, canevas[1] // 2)
|
|
masques_ss = []
|
|
for largeur_trait in (contour, 0):
|
|
m = Image.new("L", canevas, 0)
|
|
ImageDraw.Draw(m).text(centre_ss, nom, font=police, fill=255, anchor="mm",
|
|
stroke_width=largeur_trait, stroke_fill=255)
|
|
masques_ss.append(m)
|
|
x0, y0, x1, y1 = masques_ss[0].getbbox() or (0, 0, SS, SS)
|
|
# boîte alignée sur la grille 1x, avec 1 px de marge pour le filtre LANCZOS
|
|
x0, y0 = (x0 // SS - 1) * SS, (y0 // SS - 1) * SS
|
|
x1, y1 = (math.ceil(x1 / SS) + 1) * SS, (math.ceil(y1 / SS) + 1) * SS
|
|
l1, h1 = (x1 - x0) // SS, (y1 - y0) // SS
|
|
masques = [m.crop((x0, y0, x1, y1)).resize((l1, h1), Image.LANCZOS) for m in masques_ss]
|
|
return masques[0], masques[1], (TAILLE // 2 - l1 // 2, TAILLE // 2 - h1 // 2)
|
|
|
|
|
|
def bornes_morceaux(args, duree_trace, t_villes):
|
|
"""Temps (s) de début/fin des morceaux, selon l'option de découpage."""
|
|
if args.villes_par_morceau:
|
|
# Groupes consécutifs de villes (au plus N, tailles équilibrées),
|
|
# coupés à mi-temps entre la dernière ville d'un groupe et la suivante.
|
|
tv = sorted(t_villes)
|
|
nb = max(math.ceil(len(tv) / args.villes_par_morceau), 1)
|
|
debuts = [round(k * len(tv) / nb) for k in range(1, nb)]
|
|
coupes = [(tv[i - 1] + tv[i]) / 2 for i in debuts]
|
|
return np.array([0.0, *coupes, duree_trace])
|
|
nb = 1
|
|
if args.morceaux:
|
|
nb = args.morceaux
|
|
elif args.duree_morceau:
|
|
nb = max(math.ceil(duree_trace / 60 / args.duree_morceau), 1)
|
|
return np.linspace(0, duree_trace, nb + 1)
|
|
|
|
|
|
def extraire_morceau(lat, lon, t, vitesse, t0, t1):
|
|
"""Portion de trace entre t0 et t1 (s), bornes interpolées, temps ramené à 0."""
|
|
dedans = (t > t0) & (t < t1)
|
|
tt = np.concatenate([[t0], t[dedans], [t1]])
|
|
return (np.interp(tt, t, lat), np.interp(tt, t, lon), tt - t0,
|
|
np.interp(tt, t, vitesse))
|
|
|
|
|
|
def generer_video(args):
|
|
lat, lon, t, v_gpx = charger_gpx(args.gpx)
|
|
vitesse = calculer_vitesses(lat, lon, t, v_gpx)
|
|
lat, lon, t, vitesse = annuler_arrets(lat, lon, t, vitesse)
|
|
print(f"{len(lat)} points, durée {t[-1] / 60:.1f} min, vitesse max {vitesse.max():.0f} km/h")
|
|
|
|
villes = [] if args.sans_villes else trouver_villes(args.gpx, lat, lon)
|
|
sortie = Path(args.sortie or Path(args.gpx).with_suffix(".mp4"))
|
|
|
|
if args.villes_par_morceau and not villes:
|
|
print("Attention : aucune ville (géocodage indisponible ou --sans-villes), "
|
|
"--villes-par-morceau ignoré : vidéo unique.")
|
|
args.villes_par_morceau = None
|
|
|
|
# Chaque ville va dans le morceau où le véhicule passe à son point d'étiquette.
|
|
t_villes = [t[int(np.argmin((lat - la) ** 2 + ((lon - lo) * math.cos(math.radians(la))) ** 2))]
|
|
for _, la, lo in villes]
|
|
bornes = bornes_morceaux(args, t[-1], t_villes)
|
|
nb = len(bornes) - 1
|
|
if nb == 1:
|
|
rendre_video(lat, lon, t, vitesse, villes, sortie, args)
|
|
return
|
|
|
|
chiffres = len(str(nb))
|
|
for k in range(nb):
|
|
t0, t1 = bornes[k], bornes[k + 1]
|
|
dernier = k == nb - 1
|
|
villes_k = [v for v, tv in zip(villes, t_villes) if t0 <= tv < t1 or (dernier and tv == t1)]
|
|
sortie_k = sortie.with_name(f"{sortie.stem}_{k + 1:0{chiffres}d}{sortie.suffix}")
|
|
print(f"Morceau {k + 1}/{nb} : {t0 / 60:.1f}-{t1 / 60:.1f} min, {len(villes_k)} villes")
|
|
rendre_video(*extraire_morceau(lat, lon, t, vitesse, t0, t1), villes_k, sortie_k, args,
|
|
rampe_debut=k == 0, rampe_fin=dernier)
|
|
|
|
|
|
def rendre_video(lat, lon, t, vitesse, villes, sortie, args, rampe_debut=True, rampe_fin=True):
|
|
"""Rend une vidéo complète (cadrage, fond, images) pour la trace donnée."""
|
|
proj = Projection(lat, lon)
|
|
px, py = proj(lat, lon)
|
|
centre = choisir_coin_compteur(px, py)
|
|
|
|
villes_px = [(nom, *map(float, proj(la, lo))) for nom, la, lo in villes]
|
|
|
|
fond_ss = dessiner_fond(px, py, villes_px, centre, args.vmax)
|
|
fond = fond_ss.resize((TAILLE, TAILLE), Image.LANCZOS)
|
|
|
|
marge_c = RAYON_COMPTEUR + 12
|
|
boite_compteur = (centre[0] - marge_c, centre[1] - marge_c, centre[0] + marge_c, centre[1] + marge_c)
|
|
# Masque circulaire : seul le disque du compteur est recollé, pas les
|
|
# coins du carré (sinon le véhicule y disparaît).
|
|
masque_compteur = Image.new("L", (2 * marge_c * SS, 2 * marge_c * SS), 0)
|
|
ImageDraw.Draw(masque_compteur).ellipse([0, 0, 2 * marge_c * SS - 1, 2 * marge_c * SS - 1], fill=255)
|
|
masque_compteur = masque_compteur.resize((2 * marge_c, 2 * marge_c), Image.LANCZOS)
|
|
police_vitesse = charger_police(int(46 * SS), gras=True)
|
|
|
|
nb_images = int(round(args.duree * args.fps))
|
|
|
|
# Zoom sur les villes : image d'arrivée du véhicule et masques précalculés
|
|
zooms = []
|
|
if args.zoom_villes:
|
|
for nom, la, lo in villes:
|
|
proche = int(np.argmin((lat - la) ** 2 + ((lon - lo) * math.cos(math.radians(la))) ** 2))
|
|
n_arrivee = t[proche] / t[-1] * (nb_images - 1)
|
|
zooms.append((n_arrivee, *rendu_zoom(nom)))
|
|
zooms.sort(key=lambda z: z[0])
|
|
duree_zoom = args.duree_zoom * args.fps
|
|
fondu = max(FONDU_ZOOM * args.fps, 1)
|
|
cmd = ["ffmpeg", "-y", "-loglevel", "error",
|
|
"-f", "rawvideo", "-pix_fmt", "rgb24", "-s", f"{TAILLE}x{TAILLE}", "-r", str(args.fps), "-i", "-",
|
|
"-c:v", "libx264", "-preset", "medium", "-crf", "18", "-pix_fmt", "yuv420p",
|
|
"-movflags", "+faststart", str(sortie)]
|
|
ffmpeg = subprocess.Popen(cmd, stdin=subprocess.PIPE)
|
|
|
|
debut = time.time()
|
|
for n in range(nb_images):
|
|
tt = t[-1] * n / max(nb_images - 1, 1)
|
|
la, lo = np.interp(tt, t, lat), np.interp(tt, t, lon)
|
|
v = float(np.interp(tt, t, vitesse))
|
|
# Le GPS démarre/s'arrête souvent sans point à vitesse nulle :
|
|
# l'aiguille part de 0 et y revient sur la première/dernière seconde
|
|
# (seulement aux extrémités du trajet, pas aux coupures entre morceaux).
|
|
rampe = max(args.fps, 1)
|
|
v *= min(1.0, n / rampe if rampe_debut else 1.0,
|
|
(nb_images - 1 - n) / rampe if rampe_fin else 1.0)
|
|
x, y = (float(c) for c in proj(la, lo))
|
|
|
|
image = fond.copy()
|
|
|
|
# Véhicule
|
|
rv = RAYON_VEHICULE + 3
|
|
bx0, by0 = int(math.floor(x)) - rv, int(math.floor(y)) - rv
|
|
boite_v = (bx0, by0, bx0 + 2 * rv + 2, by0 + 2 * rv + 2)
|
|
def dessin_vehicule(dr, ox, oy):
|
|
cxs, cys, r = x * SS - ox, y * SS - oy, RAYON_VEHICULE * SS
|
|
dr.ellipse([cxs - r, cys - r, cxs + r, cys + r], fill=BLANC, outline=NOIR, width=SS * 2)
|
|
image.paste(rendu_patch(fond_ss, boite_v, dessin_vehicule), boite_v[:2])
|
|
|
|
# Aiguille et vitesse numérique
|
|
def dessin_aiguille(dr, ox, oy):
|
|
cxs, cys = centre[0] * SS - ox, centre[1] * SS - oy
|
|
a = angle_vitesse(v, args.vmax)
|
|
longueur, arriere, demi = (RAYON_COMPTEUR - 30) * SS, 22 * SS, 6 * SS
|
|
ux, uy = math.cos(a), -math.sin(a)
|
|
nx, ny = -uy, ux
|
|
dr.polygon([(cxs + ux * longueur, cys + uy * longueur),
|
|
(cxs + nx * demi - ux * arriere, cys + ny * demi - uy * arriere),
|
|
(cxs - nx * demi - ux * arriere, cys - ny * demi - uy * arriere)], fill=BLANC)
|
|
rm = 13 * SS
|
|
dr.ellipse([cxs - rm, cys - rm, cxs + rm, cys + rm], fill=BLANC)
|
|
dr.ellipse([cxs - rm / 2.5, cys - rm / 2.5, cxs + rm / 2.5, cys + rm / 2.5], fill=NOIR)
|
|
dr.text((cxs, cys + RAYON_COMPTEUR * SS * 0.46), f"{v:.0f}",
|
|
font=police_vitesse, fill=BLANC, anchor="mm")
|
|
image.paste(rendu_patch(fond_ss, boite_compteur, dessin_aiguille), boite_compteur[:2], masque_compteur)
|
|
|
|
# Nom de la ville en grand (la plus récemment atteinte)
|
|
actives = [z for z in zooms if z[0] <= n < z[0] + duree_zoom]
|
|
if actives:
|
|
n_arrivee, masque_contour, masque_texte, pos = actives[-1]
|
|
f = min(1.0, (n - n_arrivee) / fondu, (n_arrivee + duree_zoom - n) / fondu)
|
|
if f < 1.0:
|
|
masque_contour = masque_contour.point(lambda p: int(p * f))
|
|
masque_texte = masque_texte.point(lambda p: int(p * f))
|
|
image.paste(NOIR, pos, masque_contour)
|
|
image.paste(BLANC, pos, masque_texte)
|
|
|
|
ffmpeg.stdin.write(image.tobytes())
|
|
if n % 10 == 0 or n == nb_images - 1:
|
|
ecoule = time.time() - debut
|
|
reste = ecoule / (n + 1) * (nb_images - n - 1)
|
|
print(f"\rImage {n + 1}/{nb_images} (reste ~{reste:.0f} s)", end="", flush=True)
|
|
|
|
ffmpeg.stdin.close()
|
|
code = ffmpeg.wait()
|
|
print()
|
|
if code != 0:
|
|
sys.exit(f"Erreur ffmpeg (code {code})")
|
|
print(f"Vidéo écrite : {sortie}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Génère une vidéo 1024x1024 (trace, véhicule, compteur, villes) depuis un GPX.")
|
|
parser.add_argument("gpx", help="fichier GPX d'entrée")
|
|
parser.add_argument("-o", "--sortie", help="fichier MP4 de sortie (défaut : même nom que le GPX)")
|
|
parser.add_argument("--duree", type=float, default=60,
|
|
help="durée de la vidéo (de chaque morceau si découpage) en secondes (défaut 60)")
|
|
parser.add_argument("--fps", type=int, default=30, help="images par seconde (défaut 30)")
|
|
parser.add_argument("--vmax", type=float, default=200, help="graduation max du compteur en km/h (défaut 200)")
|
|
parser.add_argument("--sans-villes", action="store_true", help="ne pas afficher les villes (pas d'accès réseau)")
|
|
parser.add_argument("--zoom-villes", action="store_true",
|
|
help="afficher le nom de chaque ville en grand au centre quand le véhicule y arrive")
|
|
parser.add_argument("--duree-zoom", type=float, default=2.0,
|
|
help="durée d'affichage du nom zoomé en secondes de vidéo (défaut 2)")
|
|
decoupage = parser.add_mutually_exclusive_group()
|
|
decoupage.add_argument("--morceaux", type=int,
|
|
help="découper le trajet en N vidéos de même durée de trajet (sortie_1.mp4, ...)")
|
|
decoupage.add_argument("--duree-morceau", type=float,
|
|
help="découper le trajet en morceaux égaux d'au plus MIN minutes de trajet chacun")
|
|
decoupage.add_argument("--villes-par-morceau", type=int, nargs="?", const=20, metavar="N",
|
|
help="découper le trajet en morceaux d'au plus N villes chacun (défaut 20)")
|
|
args = parser.parse_args()
|
|
if args.morceaux is not None and args.morceaux < 1:
|
|
parser.error("--morceaux doit être >= 1")
|
|
if args.duree_morceau is not None and args.duree_morceau <= 0:
|
|
parser.error("--duree-morceau doit être > 0")
|
|
if args.villes_par_morceau is not None and args.villes_par_morceau < 1:
|
|
parser.error("--villes-par-morceau doit être >= 1")
|
|
generer_video(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|