perf: tile encoding uses all CPUs, not just --workers; robustness: async SSH sync (live log tail during transfer), per-line chunk id in log, local disk guard before SSH copy, skip .vrt, tolerate files purged mid-sync

Author: nico579

lidar2map.py

@@ -2032,6 +2032,14 @@ def __init__(self, log_path):
         # de tuiles, workers). Sans lui, la machine à états _buf/_cr_buf
         # s'entrelace entre threads → lignes de log corrompues.
         self._lock = threading.Lock()
+        self._chunk_actuel = None   # cf. definir_chunk
+
+    def definir_chunk(self, cle):
+        """Marque le chunk en cours (découpage à priori) : préfixe chaque
+        ligne de log qui suit, pour qu'un extrait du log soit auto-suffisant
+        sans avoir à remonter chercher le dernier « ── Ombrage XXX ── »."""
+        with self._lock:
+            self._chunk_actuel = cle
 
     def _log_line(self, line):
         """Écrit une ligne dans le fichier log avec horodatage."""
@@ -2041,7 +2049,8 @@ def _log_line(self, line):
         line = line.strip()
         if line:
             ts = time.strftime("%H:%M:%S")
-            self._log.write(f"[{ts}] {line}\n")
+            _cle = f"[{self._chunk_actuel}] " if self._chunk_actuel else ""
+            self._log.write(f"[{ts}] {_cle}{line}\n")
 
     def write(self, msg):
         # ── Terminal ─────────────────────────────────────────────────────────
@@ -2236,6 +2245,15 @@ def _excepthook(exc_type, exc_value, exc_tb):
 
 _activer_log()
 
+
+def _definir_chunk_log(cle):
+    """Signale à _TeeLogger le chunk en cours (découpage à priori), pour
+    préfixer chaque ligne du log qui suit (cf. _TeeLogger.definir_chunk).
+    No-op si le log n'est pas actif (tests, sys.stdout non remplacé)."""
+    _t = sys.stdout
+    if isinstance(_t, _TeeLogger):
+        _t.definir_chunk(cle)
+
 # ── Requêtes HTTP via urllib (stdlib, zéro dépendance) ──────────────────────
 _HTTP_UA = "lidar2map/1.0 (IGN WMTS/WMS)"
 
@@ -8453,6 +8471,17 @@ def _warped_3857_valide(chemin):
         return False
 
 
+def _tile_workers_defaut():
+    """Parallélisme de l'encodage de tuiles (JPEG/PNG, Pillow libère le GIL,
+    cf. le pool dans generer_mbtiles_lidar) : DÉCOUPLÉ de --workers, qui
+    plafonne les téléchargements réseau (throttle IGN ~3 en simultané, cf.
+    --laz-parallel pour la même logique côté conversion LAZ). L'encodage est
+    100% CPU local, sans lien avec ce plafond réseau : le limiter au nombre
+    de workers de download le bridait sans raison (ex. --workers 3 sur une
+    VM à 16 vCPU = 13 coeurs inutilisés pendant tout le tuilage)."""
+    return os.cpu_count() or 4
+
+
 def generer_mbtiles_lidar(tif_source, dossier_ville, nom_ville,
                     zoom_min=13, zoom_max=17, format_tuiles="auto",
                     jpeg_quality=85, bbox_natif=None, tampon_coin_max_m=0,
@@ -11323,7 +11352,7 @@ def _tuiler_tifs_ombrages(args, tifs, dossier_ville, nom_zone, bbox,
                 jpeg_quality=args.qualite_image,
                 bbox_natif=bbox, tampon_coin_max_m=tampon_coin_max_m,
                 ecraser_tuiles=args.tuiles_ecraser,
-                tile_workers=args.workers)
+                tile_workers=_tile_workers_defaut())
         else:
             print(f"  Existing MBTiles: {mbt_path.name}, direct split/conversion")
             mbt_out = mbt_path
@@ -12405,7 +12434,7 @@ def _est_du_provider(nom):
                                            bbox_natif=bbox,
                                            source_already_warped=getattr(args, "_source_already_warped", False),
                                            ecraser_tuiles=_ecraser_l,
-                                           tile_workers=args.workers)
+                                           tile_workers=_tile_workers_defaut())
             elif _mbt_path.exists():
                 print(f"  Existing MBTiles: {_mbt_path.name}, direct split/conversion")
                 _mbt_out = _mbt_path
@@ -13603,6 +13632,7 @@ def _run_split_priori(args, sous_zones, mode_desc, nom_zone, racine_pr,
 
         print(f"\n  ── Chunk {cle}  ({i_z+1}/{n_total})  {nom_z} ──")
         print(f"     {entete_chunk(coords)}")
+        _definir_chunk_log(cle)
         manifeste.debut_morceau(cle, nom_z)
         t0_z = time.time()
         try:
@@ -14001,7 +14031,7 @@ def _traiter_bbox_lidar_tuilage(args, bbox_natif, nom_z, nom_zone_base, manifest
                         jpeg_quality=args.qualite_image,
                         bbox_natif=bbox, tampon_coin_max_m=TAMPON_MAX_M,
                         ecraser_tuiles=args.tuiles_ecraser,
-                        tile_workers=args.workers)
+                        tile_workers=_tile_workers_defaut())
                 else:
                     print(f"  Existing MBTiles: {mbt_path.name}, direct split/conversion")
                     mbt_out = mbt_path
@@ -14108,6 +14138,7 @@ def _etape_ombrage(sz):
                       cle, 0, n_total)
         print(f"\n  ── Ombrage {cle}  {nom_z} ──")
         print(f"     {entete_chunk(tuple(sz[2:]))}")
+        _definir_chunk_log(cle)
         manifeste.debut_morceau(cle, nom_z)
         t0 = time.time()
         dalles_precharge = prefetch.recuperer(cle)
@@ -14124,6 +14155,7 @@ def _etape_tuilage(sz):
         if manifeste.deja_traite(cle_t) and not overwrite_actif:
             return
         nom_z = f"{nom_zone}_{cle}"
+        _definir_chunk_log(cle_t)
         manifeste.debut_morceau(cle_t, nom_z)
         t0 = time.time()
         _traiter_bbox_lidar_tuilage(args, tuple(sz[2:]), nom_z, nom_zone, manifeste, cle,

tests/_test_rlidar2map_CLI.py

@@ -562,6 +562,111 @@ def kill(self):
             self.assertTrue(ok)
             self.assertEqual(destination.read_bytes(), b"complete-new")
 
+    def test_scp_missing_remote_file_is_skipped_not_fatal(self):
+        # Le pipeline distant purge ses intermédiaires (VRT voisin, etc.)
+        # pendant que le sync tourne : un fichier de l'inventaire peut avoir
+        # disparu au moment de la copie. Vécu : FileNotFoundError distante
+        # plantait tout le lot. Le fichier manquant doit juste être ignoré,
+        # les AUTRES fichiers du même lot doivent quand même passer.
+        with tempfile.TemporaryDirectory() as tmp:
+            parsed = rlidar2map_cli.parse_options(
+                ["--local-dir", tmp, "--sync-method", "scp", "vm.example"]
+            )
+            controller = rlidar2map_cli.VmController(parsed)
+            state = remote_state("succeeded", exit_code=0)
+            local_results = Path(tmp) / state.run_id / "results"
+            local_results.mkdir(parents=True)
+
+            # PAS .vrt : exclu inconditionnellement en amont (cf.
+            # ALWAYS_EXCLUDED_EXTENSIONS) donc jamais demandé au tout, ce qui
+            # court-circuiterait le chemin de résilience testé ici. Un autre
+            # intermédiaire (warp temporaire) illustre le même risque de
+            # purge concurrente pour un fichier non filtré en amont.
+            gone_relative = "gar9_002x003/warped_3857_tmp.tif"
+            gone_encoded = rlidar2map_cli.base64.b64encode(
+                gone_relative.encode("utf-8")
+            ).decode("ascii")
+            ok_relative = "gar9_002x003/result.mbtiles"
+            ok_encoded = rlidar2map_cli.base64.b64encode(
+                ok_relative.encode("utf-8")
+            ).decode("ascii")
+            ok_content = b"complete-mbtiles"
+
+            inventory = {
+                gone_relative: (gone_encoded, (1, 2, 999, 3)),
+                ok_relative: (ok_encoded, (1, 2, len(ok_content), 3)),
+            }
+
+            def frame(payload):
+                raw = json.dumps(
+                    payload, separators=(",", ":")
+                ).encode("ascii")
+                return rlidar2map_cli.struct.pack(">Q", len(raw)) + raw
+
+            class FakeProcess:
+                def __init__(self, output, returncode):
+                    self.stdin = io.BytesIO()
+                    self.stdout = io.BytesIO(output)
+                    self._final_returncode = returncode
+                    self._returncode = None
+
+                def wait(self):
+                    if self._returncode is None:
+                        self._returncode = self._final_returncode
+                    return self._returncode
+
+                def poll(self):
+                    return self._returncode
+
+                def kill(self):
+                    self._returncode = -9
+
+            digest = rlidar2map_cli.hashlib.sha256(ok_content).hexdigest()
+            # Ordre = ordre du dict inventory (Python 3.7+ garantit l'ordre
+            # d'insertion) : "missing" pour le .vrt, puis file/trailer normal
+            # pour le .mbtiles. Code de sortie 0 (pas 74/unstable) : rien n'a
+            # été lu de façon instable, le fichier était juste absent.
+            stream = (
+                rlidar2map_cli.FILE_STREAM_MAGIC
+                + frame({"type": "missing", "path": gone_encoded})
+                + frame(
+                    {
+                        "type": "file",
+                        "path": ok_encoded,
+                        "size": len(ok_content),
+                    }
+                )
+                + ok_content
+                + frame(
+                    {
+                        "type": "trailer",
+                        "path": ok_encoded,
+                        "stable": True,
+                        "sha256": digest,
+                    }
+                )
+                + frame({"type": "end", "count": 2})
+            )
+            with mock.patch.object(
+                    controller,
+                    "_remote_results_inventory",
+                    return_value=inventory,
+            ), mock.patch.object(
+                    rlidar2map_cli.subprocess,
+                    "Popen",
+                    return_value=FakeProcess(stream, 0),
+            ):
+                ok = controller._sync_results_scp(state, local_results)
+
+            self.assertTrue(ok)
+            self.assertEqual(
+                (local_results / ok_relative).read_bytes(), ok_content
+            )
+            self.assertFalse((local_results / gone_relative).exists())
+            self.assertFalse(
+                list(local_results.parent.glob(".rlidar2map-sync-*"))
+            )
+
     def test_purge_markers_are_monotonic_across_stale_monitors(self):
         with tempfile.TemporaryDirectory() as tmp:
             parsed = rlidar2map_cli.parse_options(
@@ -663,10 +768,12 @@ def fake_run(command, **kwargs):
             self.assertNotEqual(calls[0][0], "scp")
 
     def test_sync_only_excludes_mapping(self):
+        # .vrt exclu inconditionnellement (intermédiaire jamais utile en
+        # local, cf. ALWAYS_EXCLUDED_EXTENSIONS), quel que soit --sync-only.
         for value, expected in (
-            ("tout", ()),
-            ("ombrages", (".mbtiles", ".rmap", ".sqlitedb")),
-            ("carte", (".tif",)),
+            ("tout", (".vrt",)),
+            ("ombrages", (".vrt", ".mbtiles", ".rmap", ".sqlitedb")),
+            ("carte", (".vrt", ".tif")),
         ):
             parsed = rlidar2map_cli.parse_options(
                 ["--sync-only", value, "vm.example"]
@@ -741,6 +848,95 @@ def fake_run(command, **kwargs):
             self.assertEqual(method, "ssh")
             self.assertEqual(calls, [])
 
+    def test_scp_transfer_skipped_when_local_disk_is_low(self):
+        # Garde-fou disque local avant une copie SSH : espace insuffisant ->
+        # annulation propre (retry au prochain cycle), pas de tentative de
+        # copie qui échouerait à mi-chemin ou saturerait le disque local.
+        with tempfile.TemporaryDirectory() as tmp:
+            parsed = rlidar2map_cli.parse_options(
+                ["--local-dir", tmp, "--sync-method", "auto", "vm.example"]
+            )
+            deps = rlidar2map_cli.RuntimeDeps(which=lambda _name: None)
+            controller = rlidar2map_cli.VmController(parsed, deps)
+            state = remote_state("succeeded", exit_code=0)
+            calls = []
+
+            def fake_run(command, **kwargs):
+                calls.append((list(command), kwargs))
+                return subprocess.CompletedProcess(command, 0, b"", b"")
+
+            encoded = rlidar2map_cli.base64.b64encode(
+                b"gar9_001x001_svf_flux.tif"
+            ).decode("ascii")
+            inventory = {
+                "gar9_001x001_svf_flux.tif": (encoded, (1, 2, 5_000_000_000, 4)),
+            }
+            fake_usage = mock.Mock(free=1_000)   # bien en dessous de la marge
+            local_results = controller.local_run_dir(state) / "results"
+            local_results.mkdir(parents=True, exist_ok=True)
+            with mock.patch.object(
+                rlidar2map_cli.subprocess, "run", fake_run
+            ), mock.patch.object(
+                    controller,
+                    "_remote_results_inventory",
+                    return_value=inventory,
+            ), mock.patch.object(
+                    rlidar2map_cli.shutil, "disk_usage", return_value=fake_usage,
+            ):
+                # _sync_results_scp directement (pas sync_once) : isole le
+                # garde-fou disque du transfert du log final, sans rapport.
+                ok = controller._sync_results_scp(state, local_results)
+
+            self.assertFalse(ok)
+            self.assertEqual(calls, [])
+
+    def test_print_remote_log_tail_collapses_cr_progress_across_polls(self):
+        # run.log est la capture brute (tmux/tee) : ses répétitions \r ne
+        # doivent PAS devenir des lignes [VM] distinctes (splitlines() coupe
+        # aussi sur \r) - seul l'état final d'une répétition compte, et
+        # l'état doit survivre à la frontière entre deux sondages (tail
+        # incrémental), une répétition pouvant être coupée en plein milieu.
+        with tempfile.TemporaryDirectory() as tmp:
+            parsed = rlidar2map_cli.parse_options(
+                ["--local-dir", tmp, "--sync-method", "auto", "vm.example"]
+            )
+            deps = rlidar2map_cli.RuntimeDeps(which=lambda _name: None)
+            controller = rlidar2map_cli.VmController(parsed, deps)
+            state = remote_state()
+
+            # Sondage 1 : 3 répétitions \r, la dernière coupée avant tout \r/\n
+            # (poll suivant en pleine barre de progression).
+            chunk1 = (
+                b"SVF chunked:  10% (1/10)\r"
+                b"SVF chunked:  20% (2/10)\r"
+                b"SVF chunked:  30% (3/10)"
+            )
+            # Sondage 2 : continuation -> \r finalise le 30%, PUIS la vraie
+            # ligne de fin, terminée par \n.
+            chunk2 = b"\rSVF chunked: done (10 blocks)\n"
+
+            results = iter([chunk1, chunk2])
+
+            def fake_run(_command, **_kwargs):
+                return subprocess.CompletedProcess(
+                    _command, 0, next(results), b""
+                )
+
+            printed = io.StringIO()
+            with mock.patch.object(
+                rlidar2map_cli.subprocess, "run", fake_run
+            ), mock.patch("sys.stdout", printed):
+                controller.print_remote_log_tail(state)
+                controller.print_remote_log_tail(state)
+
+            output = printed.getvalue()
+            self.assertNotIn("10%", output)
+            self.assertNotIn("20%", output)
+            self.assertNotIn("30%", output)
+            self.assertEqual(
+                output.count("SVF chunked: done (10 blocks)"), 1
+            )
+
 
 class ControllerTests(unittest.TestCase):
     def _controller_mock(self, states, sync_results):