[WIP] point-clouds: generate virtual point cloud (.vpc)

kart/point_cloud/mosaic_util.py

@@ -0,0 +1,38 @@
+import logging
+
+from tempfile import NamedTemporaryFile
+from kart.subprocess_util import check_output
+
+
+L = logging.getLogger("kart.point_cloud")
+
+
+def write_vpc_for_directory(directory_path):
+    """
+    Given a folder containing some LAZ/LAS files, write a mosaic file that
+    combines them all into a single Virtual Point Cloud (VPC). The VPC will
+    contain references to the tiles, rather than replicating their contents.
+    """
+    vrt_path = directory_path / f"{directory_path.name}.vpc"
+
+    tiles = [str(p) for p in directory_path.glob("*.copc.laz")]
+    if not tiles:
+        vrt_path.unlink(missing_ok=True)
+        return
+
+    with NamedTemporaryFile("w+t", suffix=".kart_tiles", encoding="utf-8") as tile_list:
+        tile_list.write("\n".join(tiles))
+        tile_list.flush()
+
+        try:
+            check_output(
+                [
+                    "pdal_wrench",
+                    "build_vpc",
+                    f"--output={vrt_path}",
+                    f"--input-file-list={tile_list.name}",
+                ]
+            )
+        except FileNotFoundError:
+            L.warning("pdal_wrench not found. Skipping VPC generation.")
+            return

kart/point_cloud/v1.py

@@ -1,3 +1,7 @@
+import logging
+import os
+import time
+
 from kart.tile.tile_dataset import TileDataset
 from kart.list_of_conflicts import ListOfConflicts, InvalidNewValue
 from kart.point_cloud.metadata_util import (
@@ -16,6 +20,9 @@
 )
 
 
+L = logging.getLogger("kart.point_cloud")
+
+
 class PointCloudV1(TileDataset):
     """A V1 point-cloud (LIDAR) dataset."""
 
@@ -66,8 +73,12 @@ def get_tile_path_pattern(
 
     @classmethod
     def write_mosaic_for_directory(cls, directory_path):
-        # TODO: Not yet implemented
-        pass
+        from kart.point_cloud.mosaic_util import write_vpc_for_directory
+
+        if os.environ.get("KART_POINT_CLOUD_VPCS"):
+            t0 = time.time()
+            write_vpc_for_directory(directory_path)
+            L.info("Generated VPC in %ss", (time.time() - t0))
 
     def get_dirty_dataset_paths(self, workdir_diff_cache):
         # TODO - improve finding and handling of non-standard tile filenames.