Feat/conn setup down

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 name = "rapida"
 dynamic = ["version"]
 description = "python tools that facilitate the assessment of natural hazards over various domains like population, landuse, infrastructure, etc"
-requires-python = ">=3.10"
+requires-python = ">=3.12"
 authors = [
     { name = 'Ioan Ferencik'},
     { name = 'Joseph Thuha'},
@@ -63,8 +63,9 @@ dependencies = [
     "matplotlib>=3.10.9",
     "netcdf4>=1.7.3",
     "h5netcdf>=1.8.1",
-    "spacetrack>=1.4.0"
-
+    "spacetrack>=1.4.0",
+    "pyvalhalla>=3.7.0",
+    "reverse-geocoder>=1.5.1",
 ]
 
 [project.optional-dependencies]

rapida/cli/__init__.py

@@ -13,6 +13,7 @@
 from rapida.cli.publish import publish
 from rapida.cli.h3id import addh3id
 from rapida.cli.ntl import ntl
+from rapida.cli.connectivity import connectivity
 from rich.progress import Progress
 import click
 import nest_asyncio
@@ -74,6 +75,7 @@ def cli(ctx):
 cli.add_command(addh3id)
 cli.add_command(population)
 cli.add_command(ntl)
+cli.add_command(connectivity)
 
 if __name__ == '__main__':
     cli()

rapida/cli/aclick.py

@@ -2,8 +2,17 @@
 import click
 from functools import wraps
 import asyncio
+import logging
 
-
+def setup_debug_logging(ctx, param, value):
+    """Callback that switches the log level to DEBUG instantly."""
+    if value:
+        logger = logging.getLogger('rapida')
+        logger.setLevel(logging.DEBUG)
+        for handler in logger.handlers:
+            handler.setLevel(logging.DEBUG)
+        logger.debug("Debug logging enabled globally.")
+    return value
 class AsyncCommand(click.Command):
     """
     Async wrapper designed to work alongside nest_asyncio in Jupyter
@@ -34,16 +43,32 @@ def wrapped_callback(*c_args, **c_kwargs):
             self.callback = wrapped_callback
 class RapidaCommandGroup(click.Group):
     """
-    Combined group that handles async subcommands and lists keys.
+    Combined group that handles async subcommands, forces help on no arguments,
+    and lists keys accurately.
     """
 
     def list_commands(self, ctx):
         return self.commands.keys()
 
+    def add_command(self, cmd: click.Command, name: str = None) -> None:
+        # Catch-all: forces help display if a subcommand is run empty
+        cmd.no_args_is_help = True
+        # 2. Automatically inject --debug into every registered subcommand
+        cmd.params.append(
+            click.Option(
+                ['--debug'],
+                is_flag=True,
+                help='Enable debug logging.',
+                expose_value=False,
+                callback=setup_debug_logging
+            )
+        )
+        super().add_command(cmd, name)
+
     def command(self, *args, **kwargs):
-        # Automatically wrap all @group.command() calls in AsyncCommand
+        # Automatically wrap all inline @group.command() calls in AsyncCommand
         kwargs.setdefault('cls', AsyncCommand)
-        return super().command(*args, no_args_is_help=True, **kwargs)
+        return super().command(*args, **kwargs)
 
     def group(self, *args, **kwargs):
         # Ensure nested groups inherit this behavior

rapida/cli/connectivity.py

@@ -0,0 +1,94 @@
+from typing import Union
+
+import click
+import logging
+import tempfile
+from rapida.util.bbox_param_type import BboxParamType
+from rapida.connectivity import run_connectivity_analysis
+from rapida.cli.aclick import AsyncCommand
+from rapida.connectivity.isochrone import MODE_MAP
+
+logger = logging.getLogger(__name__)
+
+
+def parse_intervals(ctx, param, value):
+    """Parses a comma-separated string of numbers into a list of integers."""
+    if not value:
+        return [5, 15, 30, 60]  # Fallback default
+
+    try:
+        # Split by comma, strip spaces, and cast to int
+        return [int(x.strip()) for x in value.split(",")]
+    except ValueError:
+        raise click.BadParameter("Time intervals must be a comma-separated list of integers (e.g., 5,15,30,60).")
+
+@click.command(short_help='run connectivity analysis', cls=AsyncCommand)
+
+@click.option('-b', '--bbox',
+              required=True,
+              type=BboxParamType(),
+              help='Bounding box xmin/west, ymin/south, xmax/east, ymax/north'
+              )
+@click.option(
+        '-m', "--mode", "travel_mode",
+        type=click.Choice(MODE_MAP, case_sensitive=False),
+        default='walk',
+        required=True,
+        help=f"The means of travel when delineating isochrones"
+
+    )
+
+@click.option(
+    '-ti', '--time-intervals',
+    type=str,
+    default="5,15,30,60",
+    callback=parse_intervals,
+    help="Comma-separated time intervals in minutes for the catchment areas."
+)
+
+@click.option(
+    '-bd', '--barriers-dataset',
+    type=click.Path(exists=True, file_okay=True, dir_okay=False, readable=True),
+    default=None,
+    help="Path to an OGR-supported vector data source (e.g., GPKG, Shapefile) containing exclusion zones."
+)
+@click.option(
+    '-bl', '--barriers-layer',
+    type=str,
+    default="0",
+    help="Name or index of the layer to use from barriers dataset. Defaults to the first layer (layer 0)."
+)
+
+@click.option('-bb', "--barriers-buffer",
+    type=int,
+    default=5,
+    required=False,
+    help="The value in meters to used to buffer the geometries in barriers/dataset/layer in case the barriers are lines"
+)
+@click.option(
+    "--dst-dir",
+    "-d",           # Short option
+    "dst_dir",     # Function argument name
+    type=click.Path(
+        exists=False,      # Set to True if you want Click to fail if the dir doesn't exist yet
+        file_okay=False,   # Strictly enforce that this is a directory, not a file
+        dir_okay=True,
+        resolve_path=True  # Resolves relative paths (like '.') to absolute paths automatically
+    ),
+    default=tempfile.gettempdir(),   # Defaults to the current working directory
+    show_default=True,     # Tells the user what the default is in the --help menu
+    help="Destination directory to save the downloaded OSM pbf files."
+)
+
+
+@click.pass_context
+async def connectivity(ctx, bbox:tuple[float, float, float, float]=None, travel_mode:str=None,
+                       time_intervals:list[int] =None, dst_dir:str=None,
+                       barriers_dataset:str=None, barriers_layer:str=None, barriers_buffer:int=None
+    ):
+    logger.info(f'Running connectivity analysis ')
+    progress = ctx.obj.get('progress')
+    return await run_connectivity_analysis(
+        bbox=bbox, dst_dir=dst_dir, travel_mode=travel_mode, time_intervals=time_intervals,
+        barriers_dataset=barriers_dataset, barriers_layer=barriers_layer, barriers_buffer=barriers_buffer, progress=progress
+    )

rapida/cli/ntl.py

@@ -1,12 +1,10 @@
 import logging
-import numbers
 import os.path
 from datetime import datetime
 from typing import Iterable
 import click
 import tempfile
 
-
 from rapida.cli import RapidaCommandGroup
 from rapida.ntl.nasa.const import ARCHIVE, OPERATIONAL, PROCESSING_LEVEL_NAMES, PRODUCT_NAMES, NTL_FILENAME_PATTERN, ROUTES, COLLECTIONS
 from rapida.ntl.nasa.search import search as nasa_search
@@ -113,7 +111,7 @@ def search():
 
 
 @click.pass_context
-async def search_noaa(ctx, bbox:tuple[numbers.Number]=None, nominal_date:datetime=None, satellites:list[str] = [], cmask:bool=None):
+async def search_noaa(ctx, bbox:tuple[float, float, float, float]=None, nominal_date:datetime=None, satellites:list[str] = [], cmask:bool=None):
 
     progress = ctx.obj.get('progress')
     table = Table(title=f"VIIRS satellites granules for the night of  {nominal_date.date()} covering {bbox}",
@@ -186,7 +184,7 @@ async def search_noaa(ctx, bbox:tuple[numbers.Number]=None, nominal_date:datetim
     )
 
 @click.pass_context
-def search_nasa(ctx, bbox:tuple[numbers.Number, numbers.Number, numbers.Number, numbers.Number]=None, nominal_date:datetime=None, stream:str = None, processing_level:str=None, route:str=None):
+def search_nasa(ctx, bbox:tuple[tuple[float, float, float, float]]=None, nominal_date:datetime=None, stream:str = None, processing_level:str=None, route:str=None):
 
     progress = ctx.obj.get('progress')
 
@@ -390,7 +388,7 @@ async def download_noaa(ctx, satellite:str=None, timestamp:str=None, products:It
 
 
 @click.pass_context
-async def bulk_download(ctx, bbox:tuple[numbers.Number]=None, start_date:datetime=None, end_date:datetime=None,
+async def bulk_download(ctx, bbox:tuple[float, float, float, float]=None, start_date:datetime=None, end_date:datetime=None,
                             products:str=None, dst_dir:str=None):
     progress = ctx.obj.get('progress')
 
@@ -455,7 +453,7 @@ async def bulk_download(ctx, bbox:tuple[numbers.Number]=None, start_date:datetim
 
 
 @click.pass_context
-async def fetch(ctx, bbox:tuple[numbers.Number]=None, nominal_date:datetime=None, deliverable:str=None, dst_dir:str=None):
+async def fetch(ctx, bbox:tuple[float, float, float, float]=None, nominal_date:datetime=None, deliverable:str=None, dst_dir:str=None):
 
     progress = ctx.obj.get('progress')
     return await fetch_ntl(bbox=bbox,nominal_date=nominal_date, deliverable=deliverable, progress=progress, dst_dir=dst_dir )
@@ -525,7 +523,7 @@ async def fetch(ctx, bbox:tuple[numbers.Number]=None, nominal_date:datetime=None
 )
 
 @click.pass_context
-async def detect(ctx, bbox:tuple[numbers.Number]=None, nominal_date:datetime=None, deliverable:str=None,
+async def detect(ctx, bbox:tuple[float, float, float, float]=None, nominal_date:datetime=None, deliverable:str=None,
                  mask_clouds:bool=True, dst_dir:str=None, percentage_drop:int=None, display:bool=False):
     progress = ctx.obj.get('progress')
     return await detect_outage(

rapida/connectivity/__init__.py

@@ -0,0 +1,28 @@
+import json
+import os.path
+from rapida.util.bbox_param_type import get_best_semantic_label
+from rich.progress import Progress
+from rapida.connectivity.io import prepare_osm_pbf,extract_health_sites, extract_origins_from_geojson
+from rapida.connectivity.graph import compile_valhalla_graph
+from rapida.connectivity.isochrone import connectivity_areas
+
+
+
+async def run_connectivity_analysis(
+        bbox:tuple[float, float, float, float]=None, travel_mode:str=None, time_intervals:list[int] =None,
+        dst_dir:str=None, barriers_dataset:str=None, barriers_layer:str=None, barriers_buffer:int=None, progress:Progress=None
+    ):
+    bbox_label = get_best_semantic_label(bbox=bbox)
+    dest_dir = os.path.join(dst_dir, bbox_label)
+    bbox_pbf = await prepare_osm_pbf(bbox=bbox, dst_dir=dest_dir, progress=progress)
+    health_sites = await extract_health_sites(pbf_path=bbox_pbf, dst_dir=dest_dir, progress=progress)
+    dag_tar_path = await compile_valhalla_graph(pbf_path=bbox_pbf,dst_dir=dest_dir, progress=progress)
+    origins = extract_origins_from_geojson(geojson_path=health_sites)
+    results = await connectivity_areas(
+        tar_path=dag_tar_path, origins=origins, travel_mode=travel_mode, intervals_minutes=time_intervals,
+        barriers_dataset=barriers_dataset, barriers_layer=barriers_layer, barriers_buffer=barriers_buffer
+                                       )
+    with open(os.path.join(dest_dir, 'isochrones.geojson'), "w") as f:
+        json.dump(results, f, indent=2)
+
+    return

rapida/connectivity/graph.py

@@ -0,0 +1,120 @@
+import os
+import json
+import asyncio
+from rapida.connectivity.runcli import run_cli
+from valhalla import get_config
+from valhalla.config import _sanitize_config, default_config
+from typing import Union
+from pathlib import Path
+
+
+def get_config_fixed(
+    tile_extract: Union[str, Path] = "valhalla_tiles.tar",
+    tile_dir: Union[str, Path] = "valhalla_tiles",
+    verbose: bool = False,
+) -> dict:
+    """
+    Returns a default Valhalla configuration.
+
+    :param tile_extract: The file path (with .tar extension) of the tile extract (mjolnir.tile_extract), if present. Preferred over tile_dir.
+    :param tile_dir: The directory path where the graph tiles are stored (mjolnir.tile_dir), if present.
+    :param verbose: Whether you want to see Valhalla's logs on stdout (mjolnir.logging). Default False.
+    """
+
+    config = _sanitize_config(default_config.copy())
+
+    config["mjolnir"]["tile_dir"] = (
+        ""
+        if isinstance(tile_dir, str) and not str(tile_dir)
+        else str(Path(tile_dir).resolve(strict=True))
+    )
+    config["mjolnir"]["tile_extract"] = (
+        ""
+        if isinstance(tile_extract, str) and not str(tile_extract)
+        else str(Path(tile_extract).resolve())
+    )
+
+    config["logging"]["type"] = "std_out" if verbose else ""
+
+    return config
+
+
+async def compile_valhalla_graph(pbf_path: str, dst_dir: str, progress=None) -> str:
+    """
+    Compiles the raw OSM PBF into a highly optimized Valhalla routing DAG.
+    Offloads the C++ execution to a background thread to keep uvloop unblocked.
+    """
+    os.makedirs(dst_dir, exist_ok=True)
+    tile_dir = os.path.join(dst_dir, "valhalla_tiles")
+    os.makedirs(tile_dir, exist_ok=True)
+    tar_path = os.path.join(dst_dir, "valhalla_tiles.tar")
+    #os.makedirs(tar_path, exist_ok=True)
+    config_path = os.path.join(dst_dir, "valhalla.json")
+
+    # 1. Generate the Valhalla JSON configuration natively
+    if progress:
+        progress.console.print("[cyan]Generating Valhalla engine configuration...[/cyan]")
+
+    try:
+        valhalla_conf = get_config(
+            tile_dir=tile_dir,
+            tile_extract=tar_path,
+            verbose=False
+        )
+    except Exception:
+        valhalla_conf = get_config_fixed(
+            tile_dir=tile_dir,
+            tile_extract=tar_path,
+            verbose=False
+        )
+        # ---------------------------------------------------------
+        # INJECT CUSTOM LIMITS BEFORE SAVING THE BUILD CONFIG
+        # ---------------------------------------------------------
+        if "service_limits" not in valhalla_conf:
+            valhalla_conf["service_limits"] = {}
+        if "isochrone" not in valhalla_conf["service_limits"]:
+            valhalla_conf["service_limits"]["isochrone"] = {}
+
+        # Expand the max_locations limit to allow system-wide bulk routing
+        valhalla_conf["service_limits"]["isochrone"]["max_locations"] = 5000
+        # ---------------------------------------------------------
+
+    with open(config_path, "w") as f:
+        json.dump(valhalla_conf, f, indent=4)
+
+    # 2. Define the blocking C++ execution function
+    def run_compiler():
+        # 1. Build the Admin Database
+        # This parses country borders, timezones, and local driving rules (e.g., right vs left side of road)
+        if progress:
+            progress.console.print("[cyan]Building admin rules database...[/cyan]")
+        run_cli([
+            "valhalla_build_admins",  "-c",  config_path, pbf_path
+        ])
+
+        # 2. Build the Routing Tiles
+        # This generates the actual mathematical DAG and writes it to the 'valhalla_tiles' folder
+        if progress:
+            progress.console.print("[cyan]Building routing graph ...[/cyan]")
+        run_cli([
+            "valhalla_build_tiles", "-c", config_path, pbf_path
+        ])
+
+        # 3. Compress into the Extract
+        # This reads the generated folder and packs it into the high-performance memory-mapped .tar file
+        if progress:
+            progress.console.print("[cyan]Compressing graph into a tarball...[/cyan]")
+        run_cli([
+            "valhalla_build_extract", "-c", config_path, "-v", "--overwrite"
+        ])
+
+    # 3. Offload compilation to a worker thread
+    if progress:
+        progress.console.print("[cyan]Compiling binary DAG ...[/cyan]")
+
+    await asyncio.to_thread(run_compiler)
+
+    if progress:
+        progress.console.print(f"[bold green]✓ Valhalla DAG compiled successfully: {tar_path}[/bold green]")
+
+    return tar_path