Building centroid PMTiles (#325)

* adds a pipeline step to create a building centroid pmtile layer

* snapshot

* fix subprocess spacing issue

* remove extra vars

* in building centroids, round two decimals

* change rounding from 2 to 3

* cast to double

* select only risk>0

* lint

* update rounding and remove gen-ids

* retrigger ci

* change tippecanoe tile size syntax

* cli order

* switch to zg

* change building tippecanoe settings

Author: norlandrhagen

ocr/config.py

@@ -778,12 +778,14 @@ def wipe(self):
                 f'Wiping vector data storage at these locations:\n'
                 f'- {self.building_geoparquet_uri.parent}\n'
                 f'- {self.buildings_pmtiles_uri.parent}\n'
+                f'- {self.building_centroids_pmtiles_uri.parent}\n'
                 f'- {self.region_geoparquet_uri}\n'
                 f'- {self.aggregated_region_analysis_uri}\n'
                 f'- {self.tracts_summary_stats_uri.parent}\n'
             )
         self.upath_delete(self.building_geoparquet_uri.parent)
         self.upath_delete(self.buildings_pmtiles_uri.parent)
+        self.upath_delete(self.building_centroids_pmtiles_uri.parent)
         self.upath_delete(self.region_geoparquet_uri)
         self.upath_delete(self.aggregated_region_analysis_uri)
         self.upath_delete(self.tracts_summary_stats_uri.parent)
@@ -802,6 +804,12 @@ def buildings_pmtiles_uri(self) -> UPath:
         path.parent.mkdir(parents=True, exist_ok=True)
         return path
 
+    @functools.cached_property
+    def building_centroids_pmtiles_uri(self) -> UPath:
+        path = UPath(f'{self.storage_root}/{self.pmtiles_prefix}/building_centroids.pmtiles')
+        path.parent.mkdir(parents=True, exist_ok=True)
+        return path
+
     @functools.cached_property
     def region_pmtiles_uri(self) -> UPath:
         path = UPath(f'{self.storage_root}/{self.pmtiles_prefix}/regions.pmtiles')
@@ -929,6 +937,7 @@ def nv(name: str, value: str | None):
                 nv('Tracts summary stats', str(self.tracts_summary_stats_uri)),
                 nv('Counties summary stats', str(self.counties_summary_stats_uri)),
                 nv('Buildings PMTiles', str(self.buildings_pmtiles_uri)),
+                nv('Buildings PMTiles', str(self.building_centroids_pmtiles_uri)),
                 nv('Region PMTiles', str(self.region_pmtiles_uri)),
             ]
         )

ocr/deploy/cli.py

@@ -299,9 +299,10 @@ def run(
                 'software': COILED_SOFTWARE,
             },
         )
-        manager.wait_for_completion(exit_on_failure=True)
 
         if write_regional_stats:
+            manager.wait_for_completion(exit_on_failure=True)
+
             manager = _get_manager(Platform.COILED, config.debug)
 
             manager.submit_job(
@@ -330,6 +331,19 @@ def run(
 
         # ------------- 03  Tiles ---------------
 
+        manager = _get_manager(Platform.COILED, config.debug)
+        manager.submit_job(
+            command='ocr create-building-centroid-pmtiles',
+            name=f'create-building-centroid-pmtiles-{config.environment.value}',
+            kwargs={
+                **_coiled_kwargs(config, env_file),
+                'vm_type': 'c8g.12xlarge',
+                'scheduler_vm_type': 'c8g.12xlarge',
+                'disk_size': 250,
+                'software': COILED_SOFTWARE,
+            },  # PMTiles creation needs more disk space
+        )
+
         manager = _get_manager(Platform.COILED, config.debug)
         manager.submit_job(
             command='ocr create-building-pmtiles',
@@ -412,6 +426,15 @@ def run(
         )
         manager.wait_for_completion(exit_on_failure=True)
 
+        # Create building centroid PMTiles from the consolidated geoparquet file
+        manager = _get_manager(Platform.LOCAL, config.debug)
+        manager.submit_job(
+            command='ocr create-building-centroid-pmtiles',
+            name=f'create-building-centroid-pmtiles-{config.environment.value}',
+            kwargs={
+                **_local_kwargs(),
+            },
+        )
         # Create PMTiles from the consolidated geoparquet file
         manager = _get_manager(Platform.LOCAL, config.debug)
         manager.submit_job(
@@ -423,18 +446,6 @@ def run(
         )
         manager.wait_for_completion(exit_on_failure=True)
 
-        # TODO: Should we run this locally? It will most likely break due to lack of resources
-        # # Create pyramid
-        # manager = _get_manager(Platform.LOCAL, config.debug)
-        # manager.submit_job(
-        #     command='ocr create-pyramid',
-        #     name=f'create-pyramid-{config.environment.value}',
-        #     kwargs={
-        #         **_local_kwargs(),
-        #     },
-        # )
-        # manager.wait_for_completion(exit_on_failure=True)
-
     if config.debug:
         # Print out the pretty paths
         console.log('Run complete. Current configuration paths:')
@@ -814,6 +825,67 @@ def create_building_pmtiles(
     create_building_pmtiles(config=config)
 
 
+@app.command()
+def create_building_centroid_pmtiles(
+    env_file: Path | None = typer.Option(
+        None,
+        '-e',
+        '--env-file',
+        help='Path to the environment variables file. These will be used to set up the OCRConfiguration',
+        show_default=True,
+        exists=True,
+        file_okay=True,
+        resolve_path=True,
+    ),
+    platform: Platform | None = typer.Option(
+        None,
+        '-p',
+        '--platform',
+        help='If set, schedule this command on the specified platform instead of running inline.',
+        show_default=True,
+    ),
+    vm_type: str | None = typer.Option(
+        'c8g.8xlarge', '--vm-type', help='Coiled VM type override (Coiled only).'
+    ),
+    disk_size: int | None = typer.Option(250, '--disk-size', help='Disk size in GB (Coiled only).'),
+):
+    """
+    Create building centroid PMTiles from the consolidated geoparquet file.
+    """
+
+    # Schedule if requested and not already inside a batch task
+    if platform is not None and not _in_batch():
+        config = load_config(env_file)
+        manager = _get_manager(platform, config.debug)
+        command = 'ocr create-building-pmtiles'
+        name = f'create-building-pmtiles-{config.environment.value}'
+
+        if platform == Platform.COILED:
+            COILED_SOFTWARE = os.environ.get('COILED_SOFTWARE_ENV_NAME')
+            if COILED_SOFTWARE is None or not COILED_SOFTWARE.strip():
+                console.log(
+                    '[red]Error: COILED_SOFTWARE_ENV_NAME environment variable is not set. '
+                    'This must be set to the name of a Coiled software environment with OCR installed. Proceeding with package sync...[/red]'
+                )
+            kwargs = {**_coiled_kwargs(config, env_file)}
+            kwargs['vm_type'] = vm_type
+            kwargs['scheduler_vm_type'] = vm_type
+            kwargs['disk_size'] = disk_size
+            kwargs['software'] = COILED_SOFTWARE
+        else:
+            kwargs = {**_local_kwargs()}
+
+        manager.submit_job(command=command, name=name, kwargs=kwargs)
+        manager.wait_for_completion(exit_on_failure=True)
+        return
+
+    from ocr.pipeline.create_building_centroid_pmtiles import create_building_centroid_pmtiles
+
+    config = load_config(env_file)
+
+    create_building_centroid_pmtiles(config=config)
+
+
 @app.command()
 def create_pyramid(
     env_file: Path | None = typer.Option(

ocr/pipeline/create_building_centroid_pmtiles.py

@@ -0,0 +1,92 @@
+import subprocess
+import tempfile
+from pathlib import Path
+
+import duckdb
+from upath import UPath
+
+from ocr.config import OCRConfig
+from ocr.console import console
+from ocr.utils import apply_s3_creds, copy_or_upload, get_temp_dir, install_load_extensions
+
+
+def create_building_centroid_pmtiles(
+    config: OCRConfig,
+):
+    """Nearly identical to create_building_pmtiles.py, but creates centroid only layer for higher zoom levels."""
+
+    input_path = f'{config.vector.region_geoparquet_uri}/*.parquet'  # type: ignore[attr-defined]
+
+    output_path = config.vector.building_centroids_pmtiles_uri  # type: ignore[attr-defined]
+
+    needs_s3 = any(str(p).startswith('s3://') for p in [input_path, output_path])
+
+    connection = duckdb.connect(database=':memory:')
+
+    try:
+        install_load_extensions(aws=needs_s3, spatial=True, httpfs=True, con=connection)
+        if needs_s3:
+            apply_s3_creds(region='us-west-2', con=connection)
+
+        with tempfile.TemporaryDirectory(dir=get_temp_dir()) as tmpdir:
+            tmp_path = UPath(tmpdir)
+            local_pmtiles = tmp_path / 'aggregated.pmtiles'
+            ndjson_path = Path(tmpdir) / 'buildings.ndjson'
+
+            if config.debug:
+                console.log(f'Exporting features from {input_path} to NDJSON in {ndjson_path}')
+
+            copy_sql = f"""
+            COPY (
+                SELECT
+                    'Feature' AS type,
+                    json_object(
+                    '0', ROUND(CAST(wind_risk_2011 AS DOUBLE), 3),
+                    '1', ROUND(CAST(wind_risk_2047 AS DOUBLE), 3)
+                    ) AS properties,
+                    json(ST_AsGeoJson(ST_Centroid(geometry))) AS geometry
+                FROM read_parquet('{input_path}')
+                WHERE
+                    wind_risk_2011 > 0
+                    AND
+                    wind_risk_2047 > 0
+            ) TO '{ndjson_path.as_posix()}' (FORMAT json);
+            """
+            connection.execute(copy_sql)
+
+            if config.debug:
+                console.log('NDJSON export complete')
+                console.log(f'Generating PMTiles at {local_pmtiles}')
+            # import ipdb; ipdb.set_trace()
+            tippecanoe_cmd = [
+                'tippecanoe',
+                '-o',
+                str(local_pmtiles),
+                '-l',
+                'risk',
+                '-n',
+                'centroid',
+                '-f',
+                '-P',
+                '--drop-fraction-as-needed',
+                '--no-feature-limit',
+                '--extend-zooms-if-still-dropping',
+                '-zg',
+                '-q',
+                str(ndjson_path),
+            ]
+            subprocess.run(tippecanoe_cmd, check=True)
+
+            if config.debug:
+                console.log('Tippecanoe tiles generation complete')
+                console.log(f'Uploading PMTiles to {output_path}')
+
+            copy_or_upload(local_pmtiles, output_path)
+
+            if config.debug:
+                console.log('PMTiles upload completed successfully')
+    finally:
+        try:
+            connection.close()
+        except Exception:
+            pass

ocr/pipeline/create_building_pmtiles.py

@@ -79,6 +79,7 @@ def create_building_pmtiles(
                 '-q',
                 '--extend-zooms-if-still-dropping',
                 '-zg',
+                '-Z 6',
                 '--generate-ids',
                 str(ndjson_path),
             ]

tests/test_config.py

@@ -438,6 +438,11 @@ def test_cached_properties(self, temp_dir):
             str(config.buildings_pmtiles_uri)
             == f'{temp_dir}/output/fire-risk/vector/qa/pmtiles/buildings.pmtiles'
         )
+        # Test building centroid pmtiles uri
+        assert (
+            str(config.building_centroids_pmtiles_uri)
+            == f'{temp_dir}/output/fire-risk/vector/qa/pmtiles/building_centroids.pmtiles'
+        )
 
     def test_summary_stats_uris(self, temp_dir):
         """Test summary statistics URI properties."""