396 lines
16 KiB
Python
396 lines
16 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).
|
|
|
|
Usage :
|
|
python gpx_video.py trace.gpx [-o sortie.mp4] [--duree 60] [--fps 30]
|
|
"""
|
|
|
|
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")
|
|
|
|
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)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 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 generer_video(args):
|
|
lat, lon, t, v_gpx = charger_gpx(args.gpx)
|
|
vitesse = calculer_vitesses(lat, lon, t, v_gpx)
|
|
print(f"{len(lat)} points, durée {t[-1] / 60:.1f} min, vitesse max {vitesse.max():.0f} km/h")
|
|
|
|
proj = Projection(lat, lon)
|
|
px, py = proj(lat, lon)
|
|
centre = choisir_coin_compteur(px, py)
|
|
|
|
villes = [] if args.sans_villes else trouver_villes(args.gpx, lat, lon)
|
|
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))
|
|
sortie = args.sortie or str(Path(args.gpx).with_suffix(".mp4"))
|
|
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", 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.
|
|
rampe = max(args.fps, 1)
|
|
v *= min(1.0, n / rampe, (nb_images - 1 - n) / rampe)
|
|
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)
|
|
|
|
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 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)")
|
|
generer_video(parser.parse_args())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|