Tuiles visibles en direct après régénération et déploiement webapp distante
- URLs images versionnées (?v=mtime) : une tuile recalculée change d'URL et force le rechargement navigateur (cache heuristique contourné), y compris en plein run via /api/tiles ; veille permanente 15 s sur la carte - simulation locale du mode deux machines (docker-compose.local-2m.yml) : worker GPU :8974 + webapp légère :8973 au cache séparé output-webapp/ - override webapp pour le Pi 5 (192.168.3.3) : volume /srv/lidar/output, rsync vers le worker, labels Traefik (proxy/websecure/myresolver) - purge 0,5 m : worker/process et politique générale passés à 0,2 m seul - intègre le travail parallèle non commité : export mosaïque multi-dalles (export.py + /api/export), sous-tuilage intégral des couches, légendes VIZ_LEGENDS, docs et tests associés (213 tests verts)
This commit is contained in:
Antoine Jacquin
429
lidar_pipeline/export.py
Normal file
429
lidar_pipeline/export.py
Normal file
@ -0,0 +1,429 @@
|
||||
"""Export multi-dalles : mosaïque d'images adjacentes pour consultation terrain.
|
||||
|
||||
Assemble les visualisations de plusieurs dalles LiDAR contiguës (1 km chacune,
|
||||
Lambert 93) en une seule image (PNG/JPEG/WebP) ou un PDF à une page par
|
||||
visualisation, avec habillage (bandeau titre, échelle, nord) et légende de la
|
||||
couche exportée : titre, dégradé de la colormap avec bornes, lecture du rendu
|
||||
couleurs ↔ terrain et méthode de calcul (textes de VIZ_LEGENDS, index.py).
|
||||
Les dalles sont collées bord à bord, sans aucun trait ni espace entre elles :
|
||||
les images sources (tif_to_crop) sont des carrés purs de 1 km qui se
|
||||
joignent parfaitement.
|
||||
|
||||
Pensé pour un usage sur téléphone depuis la carte (webapp.py /api/export) :
|
||||
Pillow seul, aucune dépendance au pipeline — fonctionne sur la webapp légère
|
||||
(Raspberry Pi) depuis le cache local des tuiles.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger("lidar")
|
||||
|
||||
# Formats de sortie acceptés (clé API → format Pillow).
|
||||
EXPORT_FORMATS = {"png": "PNG", "jpeg": "JPEG", "webp": "WEBP", "pdf": "PDF"}
|
||||
# Extension de fichier par format.
|
||||
EXPORT_EXTS = {"png": "png", "jpeg": "jpg", "webp": "webp", "pdf": "pdf"}
|
||||
|
||||
# Garde-fou mémoire : mosaïque RGB au-delà de ~1,6 Go (Raspberry Pi) refusée
|
||||
# en pleine résolution — l'utilisateur choisit alors une taille réduite.
|
||||
_MAX_CANVAS_BYTES = 1_600_000_000
|
||||
# PDF multi-pages : toutes les pages vivent en mémoire au moment de
|
||||
# l'enregistrement (save_all) — côté maximal ramené à cette valeur.
|
||||
_PDF_PAGE_MAX_PX = 6000
|
||||
|
||||
|
||||
def _load_font(size, bold=False):
|
||||
"""Police TrueType si disponible, sinon police embarquée de Pillow."""
|
||||
from PIL import ImageFont
|
||||
names = ("DejaVuSans-Bold.ttf", "DejaVuSans.ttf") if bold else ("DejaVuSans.ttf",)
|
||||
for name in names:
|
||||
for directory in ("/usr/share/fonts/truetype/dejavu/",
|
||||
"/usr/share/fonts/dejavu/",
|
||||
"/usr/share/fonts/TTF/"):
|
||||
path = directory + name
|
||||
if Path(path).exists():
|
||||
try:
|
||||
return ImageFont.truetype(path, size)
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
return ImageFont.load_default(size=size) # Pillow >= 10.1
|
||||
except TypeError:
|
||||
return ImageFont.load_default()
|
||||
|
||||
|
||||
def _hex_rgb(hx):
|
||||
"""'#rrggbb' → (r, g, b)."""
|
||||
hx = hx.lstrip('#')
|
||||
return tuple(int(hx[i:i + 2], 16) for i in (0, 2, 4))
|
||||
|
||||
|
||||
def _gradient_bar(gradient, width, height):
|
||||
"""Barre de dégradé interpolée linéairement entre les arrêts de colormap."""
|
||||
from PIL import Image
|
||||
stops = [_hex_rgb(c) for c in gradient]
|
||||
bar = Image.new("RGB", (width, height))
|
||||
n = len(stops) - 1
|
||||
for x in range(width):
|
||||
t = x / max(width - 1, 1) * n
|
||||
i = min(int(t), n - 1)
|
||||
f = t - i
|
||||
color = tuple(int(round(stops[i][k] + (stops[i + 1][k] - stops[i][k]) * f))
|
||||
for k in range(3))
|
||||
for y in range(height):
|
||||
bar.putpixel((x, y), color)
|
||||
return bar
|
||||
|
||||
|
||||
def _wrap_text(measure, text, font, max_w):
|
||||
"""Découpe un texte en lignes ≤ max_w px (les \\n explicites sont conservés)."""
|
||||
lines = []
|
||||
for para in text.split("\n"):
|
||||
words = para.split()
|
||||
if not words:
|
||||
lines.append("")
|
||||
continue
|
||||
cur = words[0]
|
||||
for word in words[1:]:
|
||||
if measure.textlength(cur + " " + word, font=font) <= max_w:
|
||||
cur += " " + word
|
||||
else:
|
||||
lines.append(cur)
|
||||
cur = word
|
||||
lines.append(cur)
|
||||
return lines
|
||||
|
||||
|
||||
def _find_tile_images(vis_dir, cells, viz_keys, resolution):
|
||||
"""Chemins d'image par dalle : {(col, row): {viz_key: Path}}.
|
||||
|
||||
Raises:
|
||||
ValueError: dalles absentes à la résolution demandée, ou visualisations
|
||||
manquantes sur certaines dalles.
|
||||
"""
|
||||
from .index import scan_tiles
|
||||
by_cell = {(t["col"], t["row"]): t for t in scan_tiles(vis_dir)
|
||||
if t["resolution"] == resolution}
|
||||
missing = [f"{c},{r}" for (c, r) in cells if (c, r) not in by_cell]
|
||||
if missing:
|
||||
raise ValueError(
|
||||
f"dale(s) sans visualisation à {str(resolution).replace('.', ',')} m/px : "
|
||||
f"{', '.join(missing)}")
|
||||
images, missing_viz = {}, []
|
||||
for cell in cells:
|
||||
t = by_cell[cell]
|
||||
per_viz = {}
|
||||
for key in viz_keys:
|
||||
info = t["viz"].get(key)
|
||||
if not info:
|
||||
missing_viz.append(f"{cell[0]},{cell[1]} ({key})")
|
||||
continue
|
||||
per_viz[key] = Path(t["dir_path"]) / info["filename"]
|
||||
images[cell] = per_viz
|
||||
if missing_viz:
|
||||
raise ValueError("visualisation manquante sur : " + ", ".join(missing_viz))
|
||||
return images
|
||||
|
||||
|
||||
def _nice_scale_m(width_m):
|
||||
"""Longueur d'échelle ronde (m) tenant dans ~25 % de la largeur."""
|
||||
for cand in (25, 50, 100, 200, 250, 500, 1000, 2000, 5000, 10000):
|
||||
if cand <= 0.25 * width_m:
|
||||
return cand
|
||||
return 10000
|
||||
|
||||
|
||||
def _fmt_scale(v):
|
||||
return f"{v / 1000:g} km" if v >= 1000 else f"{v} m"
|
||||
|
||||
|
||||
def _build_mosaic(images, viz_key, tile_px, n_cols, n_rows, col_min, row_max):
|
||||
"""Colle les dalles bord à bord — aucun trait, aucun espace entre elles."""
|
||||
from PIL import Image
|
||||
canvas = Image.new("RGB", (n_cols * tile_px, n_rows * tile_px), "white")
|
||||
for (col, row), per_viz in images.items():
|
||||
img = Image.open(str(per_viz[viz_key]))
|
||||
if img.mode != "RGB":
|
||||
img = img.convert("RGB")
|
||||
if img.size != (tile_px, tile_px):
|
||||
img = img.resize((tile_px, tile_px))
|
||||
# Dalle (col, row) : X ∈ [col, col+1] km, Y ∈ [row-1, row] km.
|
||||
# Ligne 0 de l'image = nord → les rows décroissantes descendent.
|
||||
x = (col - col_min) * tile_px
|
||||
y = (row_max - row) * tile_px
|
||||
canvas.paste(img, (x, y))
|
||||
img.close()
|
||||
return canvas
|
||||
|
||||
|
||||
def _legend_layout(legend, mosaic, w):
|
||||
"""Mise en page du bandeau légende (polices, lignes, hauteur), ou None.
|
||||
|
||||
Mesure le texte sur la mosaïque (sans la modifier) pour dimensionner le
|
||||
bandeau avant la création du canevas final.
|
||||
"""
|
||||
from PIL import ImageDraw
|
||||
measure = ImageDraw.Draw(mosaic)
|
||||
# Police corps du bandeau : généreuse (lecture de terrain sur téléphone),
|
||||
# bornée pour rester proportionnée aux petits exports réduits.
|
||||
body = max(16, min(30, w // 100))
|
||||
gap = max(8, int(body * 0.55))
|
||||
line_h = int(body * 1.45)
|
||||
x0 = max(18, w // 60)
|
||||
max_w = max(40, w - 2 * x0)
|
||||
text_font = _load_font(body)
|
||||
lines = _wrap_text(measure, legend["legend"], text_font, max_w)
|
||||
desc_lines = _wrap_text(measure, "Calcul : " + legend["description"],
|
||||
text_font, max_w)
|
||||
has_bar = bool(legend.get("gradient") and legend.get("ticks"))
|
||||
bar_w = max(100, min(w // 3, 640))
|
||||
bar_h = max(10, int(body * 0.9))
|
||||
height = (gap + int(body * 1.75) + gap
|
||||
+ (bar_h + line_h if has_bar else 0) + gap
|
||||
+ len(lines) * line_h + gap + len(desc_lines) * line_h + gap)
|
||||
return {"body": body, "gap": gap, "line_h": line_h, "x0": x0,
|
||||
"title_font": _load_font(int(body * 1.35), bold=True),
|
||||
"text_font": text_font, "lines": lines, "desc_lines": desc_lines,
|
||||
"has_bar": has_bar, "bar_w": bar_w, "bar_h": bar_h, "height": height}
|
||||
|
||||
|
||||
def _draw_legend(out, draw, legend, layout, y_base):
|
||||
"""Dessine le bandeau légende sous le pied : titre, dégradé + bornes,
|
||||
lecture du rendu (lignes de la légende) et méthode de calcul."""
|
||||
y = y_base + layout["gap"]
|
||||
draw.text((layout["x0"], y), legend["title"], fill=(20, 20, 20),
|
||||
font=layout["title_font"], anchor="la")
|
||||
y += int(layout["body"] * 1.75) + layout["gap"]
|
||||
if layout["has_bar"]:
|
||||
bar = _gradient_bar(legend["gradient"], layout["bar_w"], layout["bar_h"])
|
||||
out.paste(bar, (layout["x0"], y))
|
||||
# Cadre dessiné juste à l'extérieur de la barre : les pixels
|
||||
# extrêmes du dégradé (bornes de l'échelle) restent visibles.
|
||||
draw.rectangle((layout["x0"] - 1, y - 1,
|
||||
layout["x0"] + layout["bar_w"], y + layout["bar_h"]),
|
||||
outline=(90, 90, 90), width=1)
|
||||
tick_y = y + layout["bar_h"] + 2
|
||||
draw.text((layout["x0"], tick_y), legend["ticks"][0], fill=(20, 20, 20),
|
||||
font=layout["text_font"], anchor="la")
|
||||
draw.text((layout["x0"] + layout["bar_w"], tick_y), legend["ticks"][1],
|
||||
fill=(20, 20, 20), font=layout["text_font"], anchor="ra")
|
||||
y = tick_y + layout["line_h"]
|
||||
y += layout["gap"]
|
||||
for line in layout["lines"]:
|
||||
draw.text((layout["x0"], y), line, fill=(20, 20, 20),
|
||||
font=layout["text_font"], anchor="la")
|
||||
y += layout["line_h"]
|
||||
y += layout["gap"]
|
||||
for line in layout["desc_lines"]:
|
||||
draw.text((layout["x0"], y), line, fill=(90, 90, 90),
|
||||
font=layout["text_font"], anchor="la")
|
||||
y += layout["line_h"]
|
||||
|
||||
|
||||
def _decorate(mosaic, m_per_px, title, subtitle, legend=None):
|
||||
"""Ajoute un bandeau de titre au-dessus et un pied (échelle + nord) dessous.
|
||||
|
||||
Avec `legend` (entrée de VIZ_LEGENDS), un bandeau explicatif de la couche
|
||||
est ajouté sous le pied : titre, dégradé de la colormap avec bornes,
|
||||
lecture du rendu et méthode de calcul.
|
||||
|
||||
L'habillage ne touche jamais la mosaïque : les dalles restent jointives.
|
||||
"""
|
||||
from PIL import Image, ImageDraw
|
||||
w, h = mosaic.size
|
||||
top_h = max(90, h // 16)
|
||||
foot_h = max(84, h // 20)
|
||||
layout = _legend_layout(legend, mosaic, w) if legend else None
|
||||
legend_h = layout["height"] if layout else 0
|
||||
out = Image.new("RGB", (w, h + top_h + foot_h + legend_h), "white")
|
||||
out.paste(mosaic, (0, top_h))
|
||||
d = ImageDraw.Draw(out)
|
||||
|
||||
title_font = _load_font(max(22, int(top_h * 0.34)))
|
||||
sub_font = _load_font(max(14, int(top_h * 0.20)))
|
||||
foot_font = _load_font(max(14, int(foot_h * 0.26)))
|
||||
for text, font, cy in ((title, title_font, top_h * 0.40),
|
||||
(subtitle, sub_font, top_h * 0.76)):
|
||||
tw = d.textlength(text, font=font)
|
||||
d.text(((w - tw) / 2, cy), text, fill=(20, 20, 20), font=font,
|
||||
anchor="lm")
|
||||
|
||||
# Pied : échelle à gauche, nord à droite.
|
||||
width_m = w * m_per_px
|
||||
scale_m = _nice_scale_m(width_m)
|
||||
bar_px = max(40, int(scale_m / m_per_px))
|
||||
bar_y = h + top_h + foot_h * 0.42
|
||||
x0, y0 = max(18, w // 60), bar_y - foot_h * 0.16
|
||||
seg = bar_px / 4
|
||||
for i in range(4):
|
||||
cx = x0 + i * seg
|
||||
d.rectangle((cx, y0, cx + seg, y0 + foot_h * 0.20),
|
||||
fill=(15, 15, 15) if i % 2 == 0 else "white",
|
||||
outline=(15, 15, 15), width=2)
|
||||
d.text((x0 + bar_px / 2, y0 - foot_h * 0.22), _fmt_scale(scale_m),
|
||||
fill=(20, 20, 20), font=foot_font, anchor="ms")
|
||||
|
||||
nx = w - max(30, w // 24)
|
||||
ny = h + top_h + foot_h * 0.55
|
||||
s = foot_h * 0.30
|
||||
d.polygon([(nx, ny - s * 1.5), (nx - s, ny + s), (nx, ny + s * 0.4),
|
||||
(nx + s, ny + s)], fill=(15, 15, 15))
|
||||
d.text((nx, ny - s * 2.2), "N", fill=(20, 20, 20), font=foot_font,
|
||||
anchor="ms")
|
||||
if layout:
|
||||
y_base = h + top_h + foot_h
|
||||
d.line((0, y_base, w, y_base), fill=(210, 210, 210), width=1)
|
||||
_draw_legend(out, d, legend, layout, y_base)
|
||||
return out
|
||||
|
||||
|
||||
def export_filename(cells, resolution, viz_keys, fmt):
|
||||
"""Nom de fichier déterministe + horodatage (évite les collisions)."""
|
||||
cols = [c for c, _ in cells]
|
||||
rows = [r for _, r in cells]
|
||||
res = str(resolution).replace(".", "p")
|
||||
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
viz = "-".join(viz_keys)[:60]
|
||||
return (f"export_{min(cols)}-{max(cols)}_{min(rows)}-{max(rows)}"
|
||||
f"_r{res}_{viz}_{stamp}.{EXPORT_EXTS[fmt]}")
|
||||
|
||||
|
||||
def build_export(vis_dir, cells, viz_keys, resolution, fmt, out_dir,
|
||||
max_side=0):
|
||||
"""Assemble des dalles adjacentes en une image ou un PDF.
|
||||
|
||||
Args:
|
||||
vis_dir: dossier output/visualisations (dalles scannées via index).
|
||||
cells: liste (col, row) des dalles à assembler (km L93).
|
||||
viz_keys: mots-clés de visualisation (ex: slope, hillshade_multi).
|
||||
Plusieurs clés → un export par couche : PDF multi-pages (une page
|
||||
par visualisation) ou une image par visualisation.
|
||||
resolution: résolution des dalles en m/px (0.5 ou 0.2).
|
||||
fmt: png, jpeg, webp ou pdf.
|
||||
out_dir: dossier de sortie (output/exports, créé si besoin).
|
||||
max_side: réduire le plus grand côté de la mosaïque à N pixels
|
||||
(0 = pleine résolution).
|
||||
|
||||
Returns:
|
||||
{file: Path du premier fichier, width, height, pages,
|
||||
files: [{file, width, height, pages}, …] — 1 par couche en image,
|
||||
1 entrée pour le PDF multi-pages}
|
||||
|
||||
Raises:
|
||||
ValueError: dalles ou visualisations manquantes, format invalide,
|
||||
zone trop grande en mémoire.
|
||||
"""
|
||||
from PIL import Image
|
||||
resample = getattr(Image, 'LANCZOS', 1)
|
||||
if fmt not in EXPORT_FORMATS:
|
||||
raise ValueError(f"format invalide : {fmt!r} "
|
||||
f"(attendus : {', '.join(EXPORT_FORMATS)})")
|
||||
viz_keys = [str(k) for k in viz_keys if k]
|
||||
if not viz_keys:
|
||||
raise ValueError("aucune visualisation demandée")
|
||||
|
||||
images = _find_tile_images(vis_dir, cells, viz_keys, resolution)
|
||||
cols = [c for c, _ in cells]
|
||||
rows = [r for _, r in cells]
|
||||
col_min, col_max = min(cols), max(cols)
|
||||
row_min, row_max = min(rows), max(rows)
|
||||
n_cols = col_max - col_min + 1
|
||||
n_rows = row_max - row_min + 1
|
||||
# Côté réel d'une dalle (2000 px à 0,5 m, 5000 à 0,2 m) : lu sur la
|
||||
# première image plutôt que recalculé, pour coller aux fichiers réels.
|
||||
first_path = next(iter(images.values()))[viz_keys[0]]
|
||||
with Image.open(str(first_path)) as _im:
|
||||
tile_px = max(_im.size)
|
||||
tile_px = max(1, tile_px)
|
||||
canvas_bytes = n_cols * tile_px * n_rows * tile_px * 3
|
||||
if canvas_bytes > _MAX_CANVAS_BYTES:
|
||||
raise ValueError(
|
||||
f"zone trop grande en pleine résolution "
|
||||
f"({n_cols}×{n_rows} dalles à {resolution} m/px ≈ "
|
||||
f"{canvas_bytes / 1e9:.1f} Go) — choisissez une taille réduite")
|
||||
|
||||
from .index import VIZ_LABELS, VIZ_LEGENDS
|
||||
res_txt = str(resolution).replace(".", ",")
|
||||
date_txt = datetime.now().strftime("%d/%m/%Y")
|
||||
subtitle = (f"Lambert 93 · X {col_min}–{col_max + 1} km · "
|
||||
f"Y {row_min - 1}–{row_max} km · {res_txt} m/px · "
|
||||
f"{len(cells)} dalle{'s' if len(cells) > 1 else ''} · {date_txt}")
|
||||
|
||||
out_dir = Path(out_dir)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
logger.info(f"Export : {n_cols}×{n_rows} dalles à {res_txt} m/px, "
|
||||
f"{' × '.join(viz_keys)} ({fmt})")
|
||||
|
||||
# PDF multi-pages : toutes les pages sont retenues pour save_all —
|
||||
# côté plafonné pour rester en mémoire raisonnable.
|
||||
page_cap = _PDF_PAGE_MAX_PX if fmt == "pdf" and len(viz_keys) > 1 else None
|
||||
|
||||
# Options d'enregistrement hétérogènes (str/int/float/bool)
|
||||
save_kw: dict = {"format": EXPORT_FORMATS[fmt]}
|
||||
if fmt == "jpeg":
|
||||
save_kw["quality"] = 90
|
||||
elif fmt == "webp":
|
||||
save_kw["quality"] = 90
|
||||
elif fmt == "pdf":
|
||||
save_kw["resolution"] = 150.0
|
||||
|
||||
# Une couche = un export : page du PDF multi-pages, ou fichier image
|
||||
# dédié (chacun porte la légende de SA couche).
|
||||
pages = [] # pages PDF décorées (save_all)
|
||||
files = [] # fichiers écrits (1 par couche en image, le PDF sinon)
|
||||
try:
|
||||
for key in viz_keys:
|
||||
mosaic = _build_mosaic(images, key, tile_px, n_cols, n_rows,
|
||||
col_min, row_max)
|
||||
m_per_px = resolution
|
||||
caps = [c for c in (int(max_side), page_cap) if c]
|
||||
cap = min(caps) if caps else 0
|
||||
if cap and max(mosaic.size) > cap:
|
||||
ratio = cap / max(mosaic.size)
|
||||
size = (max(1, round(mosaic.size[0] * ratio)),
|
||||
max(1, round(mosaic.size[1] * ratio)))
|
||||
mosaic = mosaic.resize(size, resample)
|
||||
m_per_px = resolution / ratio
|
||||
decorated = _decorate(mosaic, m_per_px, VIZ_LABELS.get(key, key),
|
||||
subtitle, legend=VIZ_LEGENDS.get(key))
|
||||
mosaic.close()
|
||||
if fmt != "pdf":
|
||||
key_file = out_dir / export_filename(cells, resolution, [key], fmt)
|
||||
decorated.save(str(key_file), **save_kw)
|
||||
decorated.close()
|
||||
files.append(key_file)
|
||||
else:
|
||||
pages.append(decorated)
|
||||
if fmt == "pdf":
|
||||
out_file = out_dir / export_filename(cells, resolution, viz_keys, fmt)
|
||||
if len(pages) > 1:
|
||||
save_kw["save_all"] = True
|
||||
save_kw["append_images"] = pages[1:]
|
||||
pages[0].save(str(out_file), **save_kw)
|
||||
files.append(out_file)
|
||||
finally:
|
||||
for p in pages:
|
||||
p.close()
|
||||
|
||||
result_files = []
|
||||
for f in files:
|
||||
entry = {"file": f, "width": 0, "height": 0, "pages": 1}
|
||||
if fmt == "pdf":
|
||||
# Pages de même dimension que la première mosaïque décorée.
|
||||
first = pages[0] if pages else None
|
||||
entry["width"], entry["height"] = first.size if first else (0, 0)
|
||||
entry["pages"] = len(pages)
|
||||
else:
|
||||
with Image.open(str(f)) as img:
|
||||
entry["width"], entry["height"] = img.size
|
||||
result_files.append(entry)
|
||||
head = result_files[0] if result_files else {
|
||||
"file": None, "width": 0, "height": 0, "pages": 0}
|
||||
result = {"file": head["file"], "width": head["width"],
|
||||
"height": head["height"], "pages": head["pages"],
|
||||
"files": result_files}
|
||||
total = sum(f.stat().st_size for f in files)
|
||||
logger.info(f"✓ Export terminé : {len(files)} fichier"
|
||||
f"{'s' if len(files) > 1 else ''} ({total / 1e6:.1f} Mo)")
|
||||
return result
|
||||
Reference in New Issue
Block a user