From 44470cc66deeeb9c13aeb0dd856bca7af30e02cf Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Wed, 27 May 2026 02:25:06 +0200
Subject: [PATCH 1/9] refine nasa search and create downlaod structure

---
 rapida/cli/aclick.py      |   2 +-
 rapida/cli/ntl.py         | 107 +++++++++++++++++++++++++++++++-------
 rapida/ntl/cache.py       |  49 +++++++++++++++++
 rapida/ntl/nasa/const.py  |  92 ++++++++++++++++++--------------
 rapida/ntl/nasa/io.py     |  23 ++++++++
 rapida/ntl/nasa/search.py | 101 ++++++++++++++++++++---------------
 rapida/ntl/nasa/util.py   |  37 +++++++++++++
 rapida/ntl/noaa/cmask.py  |  13 ++---
 rapida/ntl/noaa/io.py     |  10 ++--
 9 files changed, 322 insertions(+), 112 deletions(-)
 create mode 100644 rapida/ntl/cache.py
 create mode 100644 rapida/ntl/nasa/io.py
 create mode 100644 rapida/ntl/nasa/util.py

diff --git a/rapida/cli/aclick.py b/rapida/cli/aclick.py
index fbdb1d6..a16a0cf 100644
--- a/rapida/cli/aclick.py
+++ b/rapida/cli/aclick.py
@@ -43,7 +43,7 @@ def list_commands(self, ctx):
     def command(self, *args, **kwargs):
         # Automatically wrap all @group.command() calls in AsyncCommand
         kwargs.setdefault('cls', AsyncCommand)
-        return super().command(*args, **kwargs)
+        return super().command(*args, no_args_is_help=True, **kwargs)
 
     def group(self, *args, **kwargs):
         # Ensure nested groups inherit this behavior
diff --git a/rapida/cli/ntl.py b/rapida/cli/ntl.py
index d19a815..3b596f3 100644
--- a/rapida/cli/ntl.py
+++ b/rapida/cli/ntl.py
@@ -3,10 +3,12 @@
 from datetime import date
 import click
 from rapida.cli import RapidaCommandGroup
-from rapida.ntl.nasa.const import ARCHIVE, OPERATIONAL, PROCESSING_LEVEL_NAMES
+from rapida.ntl.nasa.const import ARCHIVE, OPERATIONAL, PROCESSING_LEVEL_NAMES, PRODUCT_NAMES
 from rapida.ntl.nasa.search import search as nasa_search
 from rapida.ntl.noaa.search import async_search_granules, VIIRSNavigator
 from rapida.util.bbox_param_type import BboxParamType
+from rapida.ntl.nasa.io import download as download_from_nasa
+
 from rich.table import Table
 logger = logging.getLogger(__name__)
 
@@ -37,12 +39,13 @@ def handle_parse_result(self, ctx, opts, args):
 def ntl():
     """Nighttime Lights VIIRS data and impact detection"""
     pass
-@ntl.group(short_help=f'Search for available NTL data products across tiers and streams')
+
+@ntl.group(short_help=f'Search for available NTL data')
 def search():
     """Search for available NTL data products across distinct data streams."""
     pass
 
-@search.command(name='noaa', short_help=f'Search for available NTL data from operational NOAA stream')
+@search.command(name='noaa', short_help=f'Search for available  NTL operational data from NOAA source')
 
 @click.option('-b', '--bbox',
               required=True,
@@ -114,17 +117,17 @@ async def search_noaa(ctx, bbox:tuple[numbers.Number]=None, target_date:date=Non
         progress.console.print(f"\n[dim]Note: Each granule represents {1025 / 12:.2f}s of instrument data.[/dim]")
 
 
-@search.command(name='nasa', short_help=f'Search for available NTL data from NASA science archive stream')
+@search.command(name='nasa', short_help=f'Search for available NTL science data from NASA source')
 
 @click.option('-b', '--bbox',
               required=True,
               type=BboxParamType(),
               help='Bounding box xmin/west, ymin/south, xmax/east, ymax/north'
               )
-@click.option("--date", "target_date",
+@click.option("--date", "nominal_date",
               type=click.DateTime(formats=["%Y-%m-%d"]),
               required=True,
-              help=''
+              help='The human experience of a specific night, local time zone matched to the center of bbox'
               )
 @click.option(
         '-s', '--stream',
@@ -144,32 +147,96 @@ async def search_noaa(ctx, bbox:tuple[numbers.Number]=None, target_date:date=Non
     )
 
 @click.pass_context
-def search_nasa(ctx, bbox:tuple[numbers.Number]=None, target_date:date=None, stream:str = None, processing_level:str=None):
+def search_nasa(ctx, bbox:tuple[numbers.Number]=None, nominal_date:date=None, stream:str = None, processing_level:str=None):
 
     progress = ctx.obj.get('progress')
 
-    urls = nasa_search(processing_level=processing_level, target_date=target_date,
+    urls = nasa_search(processing_level=processing_level, nominal_date=nominal_date,
                        bbox=bbox, stream=stream, progress=progress)
 
     if urls:
-        table = Table(title=f" {processing_level} VIIRS satellites tiles for the night of  {target_date.date()} covering {bbox}",
+        table = Table(title=f" {processing_level} VIIRS satellites tiles for the night of  {nominal_date.date()}-{nominal_date.strftime('%Y%j')} covering {bbox}",
                       title_style="bold yellow")
         table.add_column("Product", style="red", justify='center')
+        table.add_column("Timestamp", style="red", justify='center')
+        table.add_column("Tile", style="red", justify='center')
         table.add_column("URI", style="green", justify='center')
-        for url in urls:
-            table.add_row(*url)
+        for e in urls:
+            table.add_row(*e)
         progress.console.print(table)
-@ntl.command(short_help=f'Download selected NTL data')
-async def download():
-    logger.info('Downloading NTL')
 
-@ntl.command(short_help=f'Execute crisis impact detection (48h Alerts / 72h Assessments)')
-async def detect():
-    logger.info('Detecting impact on the ground')
+
+@ntl.group(short_help=f'Download NTL data ')
+def download():
+    pass
+
+
+
+@download.command(name='nasa', short_help=f'Download NTL products from NASA')
+#
+# @click.option('-b', '--bbox',
+#               required=True,
+#               type=BboxParamType(),
+#               help='Bounding box xmin/west, ymin/south, xmax/east, ymax/north'
+#               )
+@click.option( "-t", "--timestamp", "timestamp",
+               type=str,
+               required=True,
+               help='Granule timestamp string as date and time. Ex: 202604152232 '
+               )
+@click.option(
+    "-p",
+                "product",
+                type=click.Choice(PRODUCT_NAMES, case_sensitive=True),
+                required=True,
+                help=f'The product to download.'
+
+    )
+
+@click.option(
+    "--dest-dir",
+    "-d",           # Short option
+    "dest_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="/tmp",           # 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 the images."
+)
+
+
+@click.pass_context
+async def download_nasa(ctx, bbox:tuple[numbers.Number]=None, timestamp:str = None, product:str=None, dest_dir:str=None):
+    progress = ctx.obj.get('progress')
+
+    downloaded_files = await download_from_nasa(timestamp=timestamp, product=product, dst_dir=dest_dir,progress=progress)
+
+
+
+
+@download.command(name='noaa', short_help=f'Download operational NTL data from NOAA')
+
+
+@click.pass_context
+async def download_noaa(ctx, ):
+    logger.info('Downloading from NOAA')
+
+
+
 
 
-@ntl.command(short_help=f'Track long-term resilience and recovery curves (2-3 Week horizon)')
-async def monitor():
-    logger.info('Monitoring recovery')
+# @ntl.command(short_help=f'Execute crisis impact detection (48h Alerts / 72h Assessments)')
+# @click.pass_context
+# async def detect(ctx):
+#     logger.info('Detecting impact on the ground')
+#
+#
+# @ntl.command(short_help=f'Track long-term resilience and recovery curves (2-3 Week horizon)')
+# async def monitor():
+#     logger.info('Monitoring recovery')
 
 
diff --git a/rapida/ntl/cache.py b/rapida/ntl/cache.py
new file mode 100644
index 0000000..a45f44b
--- /dev/null
+++ b/rapida/ntl/cache.py
@@ -0,0 +1,49 @@
+import shelve
+import time
+import os
+import tempfile
+import hashlib
+import json
+
+
+
+MAX_AGE_SECONDS = 6 * 3600  # 6 hours
+CACHE_PATH = os.path.join(tempfile.gettempdir(), "ntl_search_cache")
+
+
+
+def search_id(search_params: dict) -> str:
+    """Generates a deterministic unique ID based on the STAC search parameters."""
+    # Sort the dictionary keys to ensure the same parameters always produce the exact same hash
+    param_string = json.dumps(search_params, sort_keys=True)
+    return hashlib.md5(param_string.encode('utf-8')).hexdigest()
+
+
+
+def store(key:str=None, url:str=None, tile:str=None, cache_path=CACHE_PATH):
+    with shelve.open(cache_path) as cache:
+        record = cache.get(key, None)
+        if record is None:
+            record = {tile:url}, time.time()
+        else:
+            record[0].update({tile:url})
+        cache[key] = record
+
+
+
+
+
+def get_urls(key:str=None, tile:str=None, cache_path=CACHE_PATH):
+    with shelve.open(cache_path) as cache:
+        record = cache.get(key, None)
+        if record is None:
+            return
+        for tiles, creation_time in record:
+            # Invalidate purely on read
+            if time.time() - creation_time > MAX_AGE_SECONDS:
+                del cache[key]
+                return  # Expired
+            if tile and tile in tiles:
+                return tiles[tile]
+            return tiles
+
diff --git a/rapida/ntl/nasa/const.py b/rapida/ntl/nasa/const.py
index 9c7262e..63a71d9 100644
--- a/rapida/ntl/nasa/const.py
+++ b/rapida/ntl/nasa/const.py
@@ -1,46 +1,47 @@
 import json
 from pystac_client import Client
-
+import re
 PRODUCT = 46
 
 COLLECTIONS_STRING = \
 '''
 {
-  "LANCEMODIS": {
-    "A1": [
-      "VJ146A1_NRT_2",
-      "VNP46A1_NRT_1",
-      "VNP46A1_NRT_2"
-    ],
-    "A1G": [
-      "VJ146A1G_NRT_2",
-      "VNP46A1G_NRT_2",
-      "VNP46A1G_NRT_1"
-    ],
-    "A2": [
-      "VNP46A2_NRT_2"
-    ]
-  },
-  "LAADS": {
-    "A1": [
-      "VJ146A1_2",
-      "VNP46A1_2"
-    ],
-    "A2": [
-      "VJ146A2_2",
-      "VNP46A2_2"
-    ],
-    "A3": [
-      "VJ146A3_2",
-      "VNP46A3_2",
-      "VNP46A3_1"
-    ],
-    "A4": [
-      "VJ146A4_2",
-      "VNP46A4_2",
-      "VNP46A4_1"
-    ]
-  }
+    "LANCEMODIS": {
+        "A1": [
+            "VJ146A1_NRT_2",
+            "VNP46A1_NRT_1",
+            "VNP46A1_NRT_2"
+        ],
+        "A2": [
+            "VJ146A2_NRT_2",
+            "VNP46A2_NRT_2"
+        ],
+        "A1G": [
+            "VJ146A1G_NRT_2",
+            "VNP46A1G_NRT_2",
+            "VNP46A1G_NRT_1"
+        ]
+    },
+    "LAADS": {
+        "A1": [
+            "VJ146A1_2",
+            "VNP46A1_2"
+        ],
+        "A2": [
+            "VJ146A2_2",
+            "VNP46A2_2"
+        ],
+        "A3": [
+            "VJ146A3_2",
+            "VNP46A3_2",
+            "VNP46A3_1"
+        ],
+        "A4": [
+            "VJ146A4_2",
+            "VNP46A4_2",
+            "VNP46A4_1"
+        ]
+    }
 }
 '''
 COLLECTIONS = json.loads(COLLECTIONS_STRING)
@@ -63,6 +64,19 @@
 PROCESSING_LEVELS = {stream_name: list(stream_data.keys()) for stream_name, stream_data in COLLECTIONS.items()}
 PROCESSING_LEVEL_NAMES = {CATALOG2STREAM[stream_name]: list(stream_data.keys()) for stream_name, stream_data in COLLECTIONS.items()}
 
+NTL_FILENAME_PATTERN = re.compile(
+    r"^(?P<product>V[A-Z0-9_]+)\."
+    r"A(?P<year>\d{4})(?P<doy>\d{3})\."
+    r"(?:(?P<time>\d{4})\.)?"              # NEW: Optional HHMM overpass time
+    r"(?P<tile>h(?P<h>\d{2})v(?P<v>\d{2}))\."
+    r"(?P<version>\d{3})"
+    r"(?:\.(?P<production_time>\d{13}))?"
+    r"\.h5$"
+)
+PRODUCTS = set([item for stream in COLLECTIONS.values() for level, prod_list in stream.items() for item in prod_list])
+PRODUCT_NAMES = [p.rsplit('_', 1)[0] if p.count('_') == 2 else p.split('_')[0] for p in PRODUCTS]
+
+
 def generate_collections(catalogs=CATALOGS, product_filter=PRODUCT ):
     collections = {}
     for catalog_name in catalogs:
@@ -92,7 +106,7 @@ def processing_levels(collections=COLLECTIONS):
 
 if __name__ == '__main__':
 
-    #COLLECTIONS = generate_collections()
+    COLLECTIONS = generate_collections()
     print(json.dumps(COLLECTIONS, indent=4))
 
     # SOURCES = tuple(COLLECTIONS)
@@ -100,5 +114,5 @@ def processing_levels(collections=COLLECTIONS):
     #levels = sorted({level for stream in COLLECTIONS.values() for level in stream})
     # print(PROCESSING_LEVELS)
     # print(levels)
-    # print(PROCESSING_LEVELS)
-    # print(PROCESSING_LEVEL_NAMES)
\ No newline at end of file
+
+    #print(PRODUCTS)
\ No newline at end of file
diff --git a/rapida/ntl/nasa/io.py b/rapida/ntl/nasa/io.py
new file mode 100644
index 0000000..89ef9e0
--- /dev/null
+++ b/rapida/ntl/nasa/io.py
@@ -0,0 +1,23 @@
+from pathlib import Path
+import httpx
+from rapida.ntl.nasa.util import get_intersecting_tiles
+import os
+from rich.progress import Progress
+from rapida.ntl import cache
+
+
+async def download(timestamp: str = None, product: str = None, dst_dir:str=None, progress:Progress=None):
+
+
+    key = f'{product.upper()}_{timestamp}'
+
+
+    # EarthAccess token
+    ea_token = os.environ.get('EARTHDATA_TOKEN')
+    headers = {"Authorization": f"Bearer {ea_token}"}
+
+
+    # tasks = []
+    # semaphore = asyncio.Semaphore(5)
+    # async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:
+    #     pass
\ No newline at end of file
diff --git a/rapida/ntl/nasa/search.py b/rapida/ntl/nasa/search.py
index 3459b54..2c2b555 100644
--- a/rapida/ntl/nasa/search.py
+++ b/rapida/ntl/nasa/search.py
@@ -1,9 +1,14 @@
+import json
+import os.path
+from rapida.ntl import cache
 from rapida.ntl.nasa.const import (
     STREAM2CATALOG,
     COLLECTIONS,
     CMR_STAC_ROOT,
     OPERATIONAL, ARCHIVE,
+    NTL_FILENAME_PATTERN
 )
+from rapida.ntl.nasa.util import timestamp_format
 import math
 from datetime import datetime, timedelta, date
 import logging
@@ -13,6 +18,8 @@
 logger = logging.getLogger(__name__)
 
 
+
+
 def calculate_night_hours(midlat: float, day_of_year: int) -> int:
     """
     Calculates the average hours of nighttime for a given latitude and Julian day.
@@ -55,31 +62,42 @@ def stac_search(stream:str, processing_level:str, dt:datetime, bbox:tuple[float]
     stac_url = f'{CMR_STAC_ROOT}{catalog_name}'
     urls = []
     catalog = Client.open(url=stac_url)
+
     search_result = catalog.search(
-        collections=[available_collections],
+        collections=available_collections,
         datetime=dt,
         bbox=bbox
     )
 
-    logger.info(f"Found {search_result.matched()} granule(s) at {stac_url}")
     if search_result.matched():
-
+        logger.info(f"Found {search_result.matched()} granule(s) at {stac_url}")
         items = search_result.item_collection()
 
         for itm in items:
+            #print(json.dumps(itm.to_dict(), indent=4))
             for asset_key, asset in itm.assets.items():
                 # Look for the .h5 file, but specifically grab the HTTPS link
                 if asset.href.endswith('.h5') and asset.href.startswith('https'):
-                    urls.append((itm.collection_id, asset.href))
+                    url = asset.href
+                    _, file_name = os.path.split(url)
+                    match = NTL_FILENAME_PATTERN.match(file_name)
+                    meta = match.groupdict()
+                    product = meta['product']
+                    dt = datetime.strptime(f'{meta["year"]}{meta["doy"]}', '%Y%j')
+                    tile = meta['tile']
+                    timestamp = dt.strftime(timestamp_format(product_id=product))
+                    key = f'product_{asset_key}'
+                    cache.store(key=key, url=url, tile=tile)
+                    urls.append((product,timestamp, tile, url))
 
         return urls
 
 
 def search(
         processing_level: str,
-        target_date: date,
+        nominal_date: date,
         bbox: tuple[float, float, float, float],
-        stream: str = OPERATIONAL,
+        stream: str = None,
         max_concurrency: int = 5,
 
         progress: Progress = None
@@ -94,47 +112,44 @@ def search(
     plevel = processing_level.upper()
     # --- 1. Harmonized Temporal Logic ---
     if stream == OPERATIONAL:
-        days_difference = abs((now-target_date).days)
+        days_difference = abs((now - nominal_date).days)
         if days_difference > 7:
-            raise ValueError(f'Invalid target_date={target_date}.{stream} stream holds max 7 days of data. ')
-        if 'A1' in plevel:
-            # NRT A1 uses a calculated night window across the UTC midnight boundary
-            midlat = (minlat + maxlat) * 0.5
-            night_hrs = calculate_night_hours(midlat, day_of_year=int(target_date.strftime('%j')))
-            start_dt = target_date - timedelta(hours=night_hrs / 2)
-            end_dt = target_date + timedelta(hours=night_hrs / 2)
-            dt = [start_dt, end_dt]
-        if 'A2' in plevel:
-            dt = target_date.replace(hour=12, minute=0, second=0)
-    elif stream == ARCHIVE:
-        if 'A1' in plevel or 'A2' in plevel:
-            # A2 and historical/standard A1 target noon dead-center
-            dt = target_date.replace(hour=12, minute=0, second=0)
-        elif 'A3' in plevel:
-            # A3 Monthly composites target mid-month. If current month, step back one month.
-            if now.month == target_date.month and now.year == target_date.year:
-                prev_month = target_date.replace(day=1) - timedelta(days=1)
-                dt = prev_month.replace(day=15)
-            else:
-                dt = target_date.replace(day=15)
-        elif 'A4' in plevel:
-            if now.year > target_date.year:
-                raise ValueError(f'Can not search in future! Please adjust target date')
-            if now.year == target_date.year:
-                # A4 Annual composites target July 1st
-                dt = target_date.replace(year=now.year-1,month=7, day=1)
-            else:
-                # A4 Annual composites target July 1st
-                dt = target_date.replace(month=7, day=1)
+            raise ValueError(f'Invalid target_date={nominal_date}.{stream} stream holds max 7 days of data. ')
+
+    if 'A1' in plevel or 'A2' in plevel:
+        # A2 and historical/standard A1 target noon dead-center
+        midlat = (minlat + maxlat) * 0.5
+        midlon = (minlon + maxlon) * 0.5
+        utc_offset_hours = midlon / 15.0
+        local_overpass_utc = nominal_date + timedelta(hours=1.5) - timedelta(hours=utc_offset_hours)
+        local_overpass_doy = local_overpass_utc.strftime('%j')
+        start_dt = local_overpass_utc - timedelta(minutes=45)
+        end_dt = local_overpass_utc + timedelta(minutes=45)
+        dt = [start_dt, end_dt]
+    elif 'A3' in plevel:
+        # A3 Monthly composites target mid-month. If current month, step back one month.
+        if now.month == nominal_date.month and now.year == nominal_date.year:
+            prev_month = nominal_date.replace(day=1) - timedelta(days=1)
+            dt = prev_month.replace(day=15)
+        else:
+            dt = nominal_date.replace(day=15)
+    elif 'A4' in plevel:
+        if now.year > nominal_date.year:
+            raise ValueError(f'Can not search in future! Please adjust target date')
+        if now.year == nominal_date.year:
+            # A4 Annual composites target July 1st
+            dt = nominal_date.replace(year=now.year - 1, month=7, day=1)
+        else:
+            # A4 Annual composites target July 1st
+            dt = nominal_date.replace(month=7, day=1)
     else:
-        raise ValueError(f'Invalid stream {stream} for ')
-
+        raise ValueError(f'Invalid stream {stream} for NASA NTL data')
 
 
     # --- 2. Catalog Search ---
     if progress:
         progress_task = progress.add_task(
-            description=f'[red]Searching {processing_level} ({stream}) imagery for {target_date.date()}',
+            description=f'[red]Searching {processing_level} ({stream}) imagery for {nominal_date.date()}',
             total=None
         )
 
@@ -146,8 +161,8 @@ def search(
     )
 
     if not urls:
-        logger.info(f"No imagery found for {processing_level} on {target_date.date()}")
-
+        logger.info(f"No imagery found for {processing_level} on {nominal_date.date()}")
+        return
     else:
         if progress and 'progress_task' in locals():
             progress.update(progress_task,
@@ -155,4 +170,4 @@ def search(
                             completed=len(urls),
                             total=len(urls))
 
-    return urls
\ No newline at end of file
+        return urls
\ No newline at end of file
diff --git a/rapida/ntl/nasa/util.py b/rapida/ntl/nasa/util.py
new file mode 100644
index 0000000..ef76f33
--- /dev/null
+++ b/rapida/ntl/nasa/util.py
@@ -0,0 +1,37 @@
+import math
+def get_intersecting_tiles(bbox: tuple[float, float, float, float]) -> list[tuple[int, int]]:
+    """
+    Identifies VIIRS Sinusoidal tiles (h, v) intersecting a geographic bounding box.
+    bbox format: (min_lon, min_lat, max_lon, max_lat)
+    :return tuple of ints representing pairs of tile coordinates (horizontal, vertical)
+    """
+    min_lon, min_lat, max_lon, max_lat = bbox
+
+    # VIIRS standard sinusoidal grid is approx 10x10 degrees at the equator
+    # h runs 0 to 35 (180W to 180E)
+    # v runs 0 to 17 (90N to 90S)
+    h_min = math.floor((min_lon + 180) / 10)
+    h_max = math.floor((max_lon + 180) / 10)
+    v_min = math.floor((90 - max_lat) / 10)
+    v_max = math.floor((90 - min_lat) / 10)
+
+    tiles = []
+    for v in range(max(0, v_min), min(18, v_max + 1)):
+        for h in range(max(0, h_min), min(36, h_max + 1)):
+            tiles.append((h, v))
+
+    return tiles
+
+TIMESTAMP_FORMATS = {
+    "A1": "%Y%j",  # Daily: Year + Julian Day (e.g., 2026134)
+    "A2": "%Y%j",  # Daily: Year + Julian Day (e.g., 2026134)
+    "A3": "%Y%m",  # Monthly: Year + Month (e.g., 202605)
+    "A4": "%Y"     # Yearly: Year only (e.g., 2026)
+}
+
+def timestamp_format(product_id: str) -> str:
+    """Determine the correct temporal format string based on product name."""
+    # Check if A1, A2, A3, or A4 is in the product string (e.g., 'VNP46A1')
+    for identifier, time_format in TIMESTAMP_FORMATS.items():
+        if identifier in product_id:
+            return time_format
diff --git a/rapida/ntl/noaa/cmask.py b/rapida/ntl/noaa/cmask.py
index a6c263f..5ec2b2b 100644
--- a/rapida/ntl/noaa/cmask.py
+++ b/rapida/ntl/noaa/cmask.py
@@ -31,13 +31,14 @@ def bbox_in_hdf(hdf_url: str, bbox: Iterable[float]):
             if not bbox_poly.intersects(bounds_poly):
                 return False
             intersection_poly = bbox_poly.intersection(bounds_poly)
-            perc_intersection = round(intersection_poly.area/bbox_poly.area * 100)
+            perc_intersection = round(intersection_poly.area/bbox_poly.area * 100
+                                      )
 
             # with open("/tmp/bbox.geojson", "w") as ff:
             #     ff.write(to_geojson(bbox_poly))
-            n = filename.split('_')[3]
-            with open(f"/tmp/granule_{n}.geojson", "w") as f:
-                f.write(to_geojson(bounds_poly))
+            # n = filename.split('_')[3]
+            # with open(f"/tmp/granule_{n}.geojson", "w") as f:
+            #     f.write(to_geojson(bounds_poly))
             return True, perc_intersection
 
 
@@ -156,7 +157,7 @@ def cloud_coverage(hdf_url: str, bbox: list) -> int:
         '',  # Output to RAM
         subdataset_str,
         format='MEM',
-        dstSRS='EPSG:4326',
+        dstSRS='EPSG:4326', # works here because we are counting pixels not planar metrics
         outputBounds=[lon_min, lat_min, lon_max, lat_max],
         xRes=0.00675, yRes=0.00675,  # 750 meters in decimal degrees
         dstNodata=-128,
@@ -203,5 +204,5 @@ def cloud_coverage_batch(urls: list[str], bbox: Iterable[float], max_threads: in
                 if progress and master_task is not None:
                     progress.update(master_task, advance=1)
     now = datetime.datetime.now()
-    logger.idebug(f'Computed cloud coverage for {len(urls)} granules in {(now-then).total_seconds()} secs')
+    logger.debug(f'Computed cloud coverage for {len(urls)} granules in {(now-then).total_seconds()} secs')
     return results
diff --git a/rapida/ntl/noaa/io.py b/rapida/ntl/noaa/io.py
index 687a1a0..e16795d 100644
--- a/rapida/ntl/noaa/io.py
+++ b/rapida/ntl/noaa/io.py
@@ -1,6 +1,6 @@
 import obstore
-from sympy.physics.units import percent
-
+import os
+import asyncio
 from rapida.ntl.noaa.const import PRODUCTS, PRODUCT_NAMES, VIIRS_URLS,PUBLIC_CONFIG,SOURCE_NAMES
 from datetime import datetime, timedelta
 from typing import Iterable
@@ -9,6 +9,8 @@
 from rapida.ntl.noaa.cmask import bbox_in_hdf
 from urllib.parse import urlparse
 
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -126,4 +128,6 @@ async def find_ntl(satellite: str = None, bbox: Iterable[float] = None, dt: date
                 f'No valid/intersecting data for product {product} and satellite {satellite} for the night {dt}')
 
 
-    return found
\ No newline at end of file
+    return found
+
+

From 10e535ff058f052b1c07a00f41ab77e81f1ef1d5 Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Wed, 27 May 2026 12:36:34 +0200
Subject: [PATCH 2/9] NTL download from NASA

---
 rapida/cli/__init__.py              |  1 -
 rapida/cli/ntl.py                   | 40 ++++++++++++++++++++---------
 rapida/ntl/cache.py                 | 26 ++++++++++++-------
 rapida/ntl/nasa/io.py               | 26 +++++++++++++------
 rapida/ntl/nasa/search.py           |  2 +-
 rapida/ntl/nasa/util.py             |  4 +--
 rapida/util/download_remote_file.py | 10 +++++---
 7 files changed, 73 insertions(+), 36 deletions(-)

diff --git a/rapida/cli/__init__.py b/rapida/cli/__init__.py
index c010c8e..77ccbf3 100644
--- a/rapida/cli/__init__.py
+++ b/rapida/cli/__init__.py
@@ -14,7 +14,6 @@
 from rapida.cli.h3id import addh3id
 from rapida.cli.ntl import ntl
 from rich.progress import Progress
-
 import click
 import nest_asyncio
 nest_asyncio.apply()
diff --git a/rapida/cli/ntl.py b/rapida/cli/ntl.py
index 3b596f3..1be4ba2 100644
--- a/rapida/cli/ntl.py
+++ b/rapida/cli/ntl.py
@@ -1,9 +1,10 @@
 import logging
 import numbers
+import os.path
 from datetime import date
 import click
 from rapida.cli import RapidaCommandGroup
-from rapida.ntl.nasa.const import ARCHIVE, OPERATIONAL, PROCESSING_LEVEL_NAMES, PRODUCT_NAMES
+from rapida.ntl.nasa.const import ARCHIVE, OPERATIONAL, PROCESSING_LEVEL_NAMES, PRODUCT_NAMES, NTL_FILENAME_PATTERN
 from rapida.ntl.nasa.search import search as nasa_search
 from rapida.ntl.noaa.search import async_search_granules, VIIRSNavigator
 from rapida.util.bbox_param_type import BboxParamType
@@ -173,12 +174,8 @@ def download():
 
 
 @download.command(name='nasa', short_help=f'Download NTL products from NASA')
-#
-# @click.option('-b', '--bbox',
-#               required=True,
-#               type=BboxParamType(),
-#               help='Bounding box xmin/west, ymin/south, xmax/east, ymax/north'
-#               )
+
+
 @click.option( "-t", "--timestamp", "timestamp",
                type=str,
                required=True,
@@ -192,11 +189,15 @@ def download():
                 help=f'The product to download.'
 
     )
+@click.option('--tile',
+              required=False,
+              type=str,
+              help='A specific tile number conforming to NASA BalckMarble 10x10 degrres tile numbering. Ex: h21v03 '
+              )
 
 @click.option(
-    "--dest-dir",
-    "-d",           # Short option
-    "dest_dir",     # Function argument name
+    "--dst-dir",
+    "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
@@ -210,12 +211,27 @@ def download():
 
 
 @click.pass_context
-async def download_nasa(ctx, bbox:tuple[numbers.Number]=None, timestamp:str = None, product:str=None, dest_dir:str=None):
+async def download_nasa(ctx, timestamp:str = None, product:str=None, tile:str=None, dst_dir:str=None):
     progress = ctx.obj.get('progress')
 
-    downloaded_files = await download_from_nasa(timestamp=timestamp, product=product, dst_dir=dest_dir,progress=progress)
+    downloaded_files = await download_from_nasa(timestamp=timestamp, product=product, tile=tile, dst_dir=dst_dir,progress=progress)
 
+    if downloaded_files:
+        table = Table(title=f"Downloaded files for {product.upper()} {timestamp} ", title_style="bold yellow")
 
+        table.add_column("Path", style="red", justify='center')
+        table.add_column("Timestamp", style="red", justify='center')
+        table.add_column("Tile", style="red", justify='center')
+        table.add_column("Size", style="green", justify='center')
+        for local_file_path in downloaded_files:
+            _, file_name = os.path.split(local_file_path)
+            file_size = os.path.getsize(local_file_path)
+            m = NTL_FILENAME_PATTERN.match(file_name)
+            meta = m.groupdict()
+            tile = meta['tile']
+            table.add_row(local_file_path, timestamp, tile, f'{file_size}')
+
+        progress.console.print(table)
 
 
 @download.command(name='noaa', short_help=f'Download operational NTL data from NOAA')
diff --git a/rapida/ntl/cache.py b/rapida/ntl/cache.py
index a45f44b..6c114e3 100644
--- a/rapida/ntl/cache.py
+++ b/rapida/ntl/cache.py
@@ -26,7 +26,9 @@ def store(key:str=None, url:str=None, tile:str=None, cache_path=CACHE_PATH):
         if record is None:
             record = {tile:url}, time.time()
         else:
-            record[0].update({tile:url})
+            tiles, creation_time = record
+            if not tile in tiles:
+                record[0].update({tile:url})
         cache[key] = record
 
 
@@ -36,14 +38,20 @@ def store(key:str=None, url:str=None, tile:str=None, cache_path=CACHE_PATH):
 def get_urls(key:str=None, tile:str=None, cache_path=CACHE_PATH):
     with shelve.open(cache_path) as cache:
         record = cache.get(key, None)
+
         if record is None:
             return
-        for tiles, creation_time in record:
-            # Invalidate purely on read
-            if time.time() - creation_time > MAX_AGE_SECONDS:
-                del cache[key]
-                return  # Expired
-            if tile and tile in tiles:
-                return tiles[tile]
-            return tiles
+        # 1. Directly unpack the tuple
+        tiles, creation_time = record
+
+        # 2. Check for expiration
+        if time.time() - creation_time > MAX_AGE_SECONDS:
+            del cache[key]
+            return  # Expired
+
+        # 3. Handle the tile request
+        if tile and tile in tiles:
+            return tiles[tile],
+
+        return tuple(tiles.values())
 
diff --git a/rapida/ntl/nasa/io.py b/rapida/ntl/nasa/io.py
index 89ef9e0..503c658 100644
--- a/rapida/ntl/nasa/io.py
+++ b/rapida/ntl/nasa/io.py
@@ -1,23 +1,33 @@
 from pathlib import Path
 import httpx
 from rapida.ntl.nasa.util import get_intersecting_tiles
+import asyncio
 import os
 from rich.progress import Progress
 from rapida.ntl import cache
-
-
-async def download(timestamp: str = None, product: str = None, dst_dir:str=None, progress:Progress=None):
+from rapida.util.download_remote_file import download_remote_files
+import logging
+logger = logging.getLogger(__name__)
+async def download(timestamp: str = None, product: str = None, tile:str=None, dst_dir:str=None, progress:Progress=None):
 
 
     key = f'{product.upper()}_{timestamp}'
-
+    urls = cache.get_urls(key=key, tile=tile)
+    if not urls:
+        logger.info(f'Failed to locate information in {cache.CACHE_PATH} for {product}-{timestamp}-{tile or ""} \n' \
+                       f'Consider searching first.')
+        return 
 
     # EarthAccess token
     ea_token = os.environ.get('EARTHDATA_TOKEN')
     headers = {"Authorization": f"Bearer {ea_token}"}
 
 
-    # tasks = []
-    # semaphore = asyncio.Semaphore(5)
-    # async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:
-    #     pass
\ No newline at end of file
+
+    async with asyncio.Semaphore(5) as semaphore:
+        # async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:
+        #     pass
+
+        return await download_remote_files(
+            file_urls=urls,dst_folder=dst_dir, progress=progress, headers=headers
+        )
diff --git a/rapida/ntl/nasa/search.py b/rapida/ntl/nasa/search.py
index 2c2b555..e377e31 100644
--- a/rapida/ntl/nasa/search.py
+++ b/rapida/ntl/nasa/search.py
@@ -86,7 +86,7 @@ def stac_search(stream:str, processing_level:str, dt:datetime, bbox:tuple[float]
                     dt = datetime.strptime(f'{meta["year"]}{meta["doy"]}', '%Y%j')
                     tile = meta['tile']
                     timestamp = dt.strftime(timestamp_format(product_id=product))
-                    key = f'product_{asset_key}'
+                    key = f'{product}_{timestamp}'
                     cache.store(key=key, url=url, tile=tile)
                     urls.append((product,timestamp, tile, url))
 
diff --git a/rapida/ntl/nasa/util.py b/rapida/ntl/nasa/util.py
index ef76f33..3137737 100644
--- a/rapida/ntl/nasa/util.py
+++ b/rapida/ntl/nasa/util.py
@@ -23,8 +23,8 @@ def get_intersecting_tiles(bbox: tuple[float, float, float, float]) -> list[tupl
     return tiles
 
 TIMESTAMP_FORMATS = {
-    "A1": "%Y%j",  # Daily: Year + Julian Day (e.g., 2026134)
-    "A2": "%Y%j",  # Daily: Year + Julian Day (e.g., 2026134)
+    "A1": "%Y%m%d",  # Daily: Year + Julian Day (e.g., 2026134)
+    "A2": "%Y%m%d",  # Daily: Year + Julian Day (e.g., 2026134)
     "A3": "%Y%m",  # Monthly: Year + Month (e.g., 202605)
     "A4": "%Y"     # Yearly: Year only (e.g., 2026)
 }
diff --git a/rapida/util/download_remote_file.py b/rapida/util/download_remote_file.py
index 785ebad..291222a 100644
--- a/rapida/util/download_remote_file.py
+++ b/rapida/util/download_remote_file.py
@@ -230,7 +230,10 @@ async def download_remote_files(
         dst_folder: str,
         progress=None,
         target_path_func=None,
-        force=False, connect_timeout=250, data_read_timeout=9000
+        headers:dict[str:str] = None,
+        force=False,
+        connect_timeout=250,
+        data_read_timeout=9000
 ):
     """
     Download remote files from a list of URLs.
@@ -241,8 +244,9 @@ async def download_remote_files(
     :param target_path_func: A function that takes a URL as an argument and returns a path to save the file to.
     """
     try:
+
         timeout = aiohttp.ClientTimeout(connect=connect_timeout, sock_connect=data_read_timeout)
-        async with aiohttp.ClientSession(timeout=timeout) as client_session:
+        async with aiohttp.ClientSession(timeout=timeout, headers=headers) as client_session:
             tasks = []
             os.makedirs(dst_folder, exist_ok=True)
             for file_url in file_urls:
@@ -254,7 +258,7 @@ async def download_remote_files(
                     file_name = os.path.basename(target_path)
                 else:
                     file_name = os.path.basename(file_url)
-                    target_path = os.path.join(dst_folder, file_name)
+                target_path = os.path.join(dst_folder, file_name)
 
                 tasks.append(
                     asyncio.create_task(

From 3351ffdcb9431ccfeafeb64900b04526cd9fcbfd Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Wed, 27 May 2026 20:48:23 +0200
Subject: [PATCH 3/9] search caching for nasa

---
 rapida/cli/ntl.py         |  22 +++-
 rapida/ntl/cache.py       |   2 +-
 rapida/ntl/nasa/const.py  |  24 ++++-
 rapida/ntl/nasa/io.py     |  13 +--
 rapida/ntl/nasa/search.py | 209 +++++++++++++++++++++++++-------------
 rapida/ntl/nasa/util.py   |   2 +-
 rapida/ntl/noaa/cmask.py  |  26 +++--
 7 files changed, 202 insertions(+), 96 deletions(-)

diff --git a/rapida/cli/ntl.py b/rapida/cli/ntl.py
index 1be4ba2..e17912f 100644
--- a/rapida/cli/ntl.py
+++ b/rapida/cli/ntl.py
@@ -1,10 +1,13 @@
 import logging
 import numbers
 import os.path
-from datetime import date
+from datetime import datetime
+
 import click
+
+
 from rapida.cli import RapidaCommandGroup
-from rapida.ntl.nasa.const import ARCHIVE, OPERATIONAL, PROCESSING_LEVEL_NAMES, PRODUCT_NAMES, NTL_FILENAME_PATTERN
+from rapida.ntl.nasa.const import ARCHIVE, OPERATIONAL, PROCESSING_LEVEL_NAMES, PRODUCT_NAMES, NTL_FILENAME_PATTERN, ROUTES
 from rapida.ntl.nasa.search import search as nasa_search
 from rapida.ntl.noaa.search import async_search_granules, VIIRSNavigator
 from rapida.util.bbox_param_type import BboxParamType
@@ -84,7 +87,7 @@ def search():
 
 
 @click.pass_context
-async def search_noaa(ctx, bbox:tuple[numbers.Number]=None, target_date:date=None, satellites:list[str] = [], cmask:bool=None  ):
+async def search_noaa(ctx, bbox:tuple[numbers.Number]=None, target_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  {target_date.date()} covering {bbox}",
@@ -147,13 +150,22 @@ async def search_noaa(ctx, bbox:tuple[numbers.Number]=None, target_date:date=Non
             )
     )
 
+@click.option(
+        '-r', '--route',
+        type=click.Choice(ROUTES, case_sensitive=False),
+        default='API',
+        required=True,
+        help=f"Route to use when searching for data. Options are STAC for CMR STAC or API for NASA blackmarble API"
+
+    )
+
 @click.pass_context
-def search_nasa(ctx, bbox:tuple[numbers.Number]=None, nominal_date:date=None, stream:str = None, processing_level:str=None):
+def search_nasa(ctx, bbox:tuple[numbers.Number]=None, nominal_date:datetime=None, stream:str = None, processing_level:str=None, route:str=None):
 
     progress = ctx.obj.get('progress')
 
     urls = nasa_search(processing_level=processing_level, nominal_date=nominal_date,
-                       bbox=bbox, stream=stream, progress=progress)
+                       bbox=bbox, stream=stream, route=route, progress=progress)
 
     if urls:
         table = Table(title=f" {processing_level} VIIRS satellites tiles for the night of  {nominal_date.date()}-{nominal_date.strftime('%Y%j')} covering {bbox}",
diff --git a/rapida/ntl/cache.py b/rapida/ntl/cache.py
index 6c114e3..0377e36 100644
--- a/rapida/ntl/cache.py
+++ b/rapida/ntl/cache.py
@@ -35,7 +35,7 @@ def store(key:str=None, url:str=None, tile:str=None, cache_path=CACHE_PATH):
 
 
 
-def get_urls(key:str=None, tile:str=None, cache_path=CACHE_PATH):
+def fetch(key:str=None, tile:str=None, cache_path=CACHE_PATH):
     with shelve.open(cache_path) as cache:
         record = cache.get(key, None)
 
diff --git a/rapida/ntl/nasa/const.py b/rapida/ntl/nasa/const.py
index 63a71d9..c4999f1 100644
--- a/rapida/ntl/nasa/const.py
+++ b/rapida/ntl/nasa/const.py
@@ -77,6 +77,20 @@
 PRODUCT_NAMES = [p.rsplit('_', 1)[0] if p.count('_') == 2 else p.split('_')[0] for p in PRODUCTS]
 
 
+API_SOURCES = {
+    OPERATIONAL: 'https://nrt3.modaps.eosdis.nasa.gov/archive/allData/5200', #NRT lance
+    ARCHIVE: 'https://ladsweb.modaps.eosdis.nasa.gov/archive/allData/5200' #LADDS
+}
+API_CONTENT = {
+    OPERATIONAL: 'https://nrt3.modaps.eosdis.nasa.gov/api/v2/content/details/allData/5200',
+    ARCHIVE: 'https://ladsweb.modaps.eosdis.nasa.gov/api/v2/content/details/allData/5200'
+}
+API_PRODUCTS = {catalog: {level: sorted({prod.rsplit('_', 1)[0] for prod in products}) for level, products in levels.items()} for catalog, levels in COLLECTIONS.items()}
+
+ROUTES = 'STAC', 'API'
+
+
+
 def generate_collections(catalogs=CATALOGS, product_filter=PRODUCT ):
     collections = {}
     for catalog_name in catalogs:
@@ -106,13 +120,15 @@ def processing_levels(collections=COLLECTIONS):
 
 if __name__ == '__main__':
 
-    COLLECTIONS = generate_collections()
-    print(json.dumps(COLLECTIONS, indent=4))
+    #COLLECTIONS = generate_collections()
+    #print(json.dumps(COLLECTIONS, indent=4))
 
     # SOURCES = tuple(COLLECTIONS)
     #
-    #levels = sorted({level for stream in COLLECTIONS.values() for level in stream})
+    levels = {level for stream, prods in COLLECTIONS.items() for level in stream}
+
+
     # print(PROCESSING_LEVELS)
-    # print(levels)
+    print(cleaned_dict)
 
     #print(PRODUCTS)
\ No newline at end of file
diff --git a/rapida/ntl/nasa/io.py b/rapida/ntl/nasa/io.py
index 503c658..81e5115 100644
--- a/rapida/ntl/nasa/io.py
+++ b/rapida/ntl/nasa/io.py
@@ -12,7 +12,7 @@ async def download(timestamp: str = None, product: str = None, tile:str=None, ds
 
 
     key = f'{product.upper()}_{timestamp}'
-    urls = cache.get_urls(key=key, tile=tile)
+    urls = cache.fetch(key=key, tile=tile)
     if not urls:
         logger.info(f'Failed to locate information in {cache.CACHE_PATH} for {product}-{timestamp}-{tile or ""} \n' \
                        f'Consider searching first.')
@@ -23,11 +23,8 @@ async def download(timestamp: str = None, product: str = None, tile:str=None, ds
     headers = {"Authorization": f"Bearer {ea_token}"}
 
 
+    semaphore = asyncio.Semaphore(5)
 
-    async with asyncio.Semaphore(5) as semaphore:
-        # async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:
-        #     pass
-
-        return await download_remote_files(
-            file_urls=urls,dst_folder=dst_dir, progress=progress, headers=headers
-        )
+    return await download_remote_files(
+        file_urls=urls,dst_folder=dst_dir, progress=progress, headers=headers
+    )
diff --git a/rapida/ntl/nasa/search.py b/rapida/ntl/nasa/search.py
index e377e31..81c6958 100644
--- a/rapida/ntl/nasa/search.py
+++ b/rapida/ntl/nasa/search.py
@@ -1,25 +1,22 @@
 import json
 import os.path
 from rapida.ntl import cache
-from rapida.ntl.nasa.const import (
-    STREAM2CATALOG,
-    COLLECTIONS,
-    CMR_STAC_ROOT,
-    OPERATIONAL, ARCHIVE,
-    NTL_FILENAME_PATTERN
-)
+from rapida.ntl.nasa import const
 from rapida.ntl.nasa.util import timestamp_format
 import math
 from datetime import datetime, timedelta, date
 import logging
 from pystac_client import Client
 from rich.progress import Progress
-
+from rapida.ntl.nasa.util import get_intersecting_tiles
+import httpx
 logger = logging.getLogger(__name__)
 
 
 
 
+
+
 def calculate_night_hours(midlat: float, day_of_year: int) -> int:
     """
     Calculates the average hours of nighttime for a given latitude and Julian day.
@@ -44,22 +41,105 @@ def calculate_night_hours(midlat: float, day_of_year: int) -> int:
     return int(round(night_hrs))
 
 
+def url2result(url:str=None, store=True):
+
+    _, file_name = os.path.split(url)
+    match = const.NTL_FILENAME_PATTERN.match(file_name)
+    meta = match.groupdict()
+    product = meta['product']
+    dt = datetime.strptime(f'{meta["year"]}{meta["doy"]}', '%Y%j')
+    tile = meta['tile']
+    timestamp = dt.strftime(timestamp_format(product_id=product))
+    key = f'{product}_{timestamp}'
+    if store:
+        cache.store(key=key, url=url, tile=tile)
+    return product, timestamp, tile, url
+
+
+
+def calculate_local_utc(stream:str, processing_level:str, nominal_date: datetime, bbox:tuple[float]):
+    """
+    Calculate VIIRS satellites local overpass time in UTC TZ
+    :param stream:
+    :param processing_level:
+    :param nominal_date:
+    :param bbox:
+    :return:
+    """
+    minlon, minlat, maxlon, maxlat = bbox
+    now = datetime.now()
+    plevel = processing_level.upper()
+    # --- 1. Harmonized Temporal Logic ---
+    if stream == const.OPERATIONAL:
+        days_difference = abs((now - nominal_date).days)
+        if days_difference > 7:
+            raise ValueError(f'Invalid target_date={nominal_date}.{stream} stream holds max 7 days of data. ')
+
+    if 'A1' in plevel or 'A2' in plevel:
+        # A2 and historical/standard A1 target noon dead-center
+
+        midlon = (minlon + maxlon) * 0.5
+        utc_offset_hours = midlon / 15.0
+        local_overpass_utc = nominal_date + timedelta(hours=1.5) - timedelta(hours=utc_offset_hours)
+
+        dt = local_overpass_utc
+    elif 'A3' in plevel:
+        # A3 Monthly composites target mid-month. If current month, step back one month.
+        if now.month == nominal_date.month and now.year == nominal_date.year:
+            prev_month = nominal_date.replace(day=1) - timedelta(days=1)
+            dt = prev_month.replace(day=15)
+        else:
+            dt = nominal_date.replace(day=15)
+    elif 'A4' in plevel:
+        if now.year > nominal_date.year:
+            raise ValueError(f'Can not search in future! Please adjust target date')
+        if now.year == nominal_date.year:
+            # A4 Annual composites target July 1st
+            dt = nominal_date.replace(year=now.year - 1, month=7, day=1)
+        else:
+            # A4 Annual composites target July 1st
+            dt = nominal_date.replace(month=7, day=1)
+    else:
+        raise ValueError(f'Invalid stream {stream} for NASA NTL data')
+    return dt
+
+
+def api_search(stream:str, products:str, dt:datetime, bbox:tuple[float])-> list[str]:
+    tiles = get_intersecting_tiles(bbox=bbox)
+    urls = []
+    for product in products:
+        content_url = f'{const.API_CONTENT[stream]}/{product}/{dt.strftime("%Y/%j")}'
+        with httpx.Client() as client:
+            # Fetch the JSON directory listing
+            response = client.get(content_url, timeout=10.0)
+            response.raise_for_status()
+            # MODAPS returns a flat JSON array of file objects
+            payload = response.json()['content']
+            for item in payload:
+                file_name = item.get("name", "")
+                for tile in tiles:
+                    if tile in file_name:
+                        url = item.get('downloadsLink')
+                        result = url2result(url=url, store=True)
+                        urls.append(result)
+
+
+    return urls
+
 def stac_search(stream:str, processing_level:str, dt:datetime, bbox:tuple[float]):
 
-    catalog_name = STREAM2CATALOG[stream]
-    catalog_collections = COLLECTIONS[catalog_name]
+    catalog_name = const.STREAM2CATALOG[stream]
+    catalog_collections = const.COLLECTIONS[catalog_name]
     catalog_processing_levels = catalog_collections.keys()
     assert processing_level in catalog_processing_levels, (f'Invalid processing level {processing_level} for {catalog_name}. \''
                                                            f'Valid processing levels {catalog_processing_levels}')
 
     available_collections = sorted(catalog_collections[processing_level], reverse=True) #we preffe
 
-
-
     logger.info(f'Searching for {processing_level} imagery in catalog "{catalog_name}" collections: {available_collections}')
     logger.debug(f'Searching for {processing_level} imagery in catalog "{catalog_name}" collections: {available_collections} ' \
                  f'for {dt} and {bbox} geaographic area')
-    stac_url = f'{CMR_STAC_ROOT}{catalog_name}'
+    stac_url = f'{const.CMR_STAC_ROOT}{catalog_name}'
     urls = []
     catalog = Client.open(url=stac_url)
 
@@ -79,27 +159,19 @@ def stac_search(stream:str, processing_level:str, dt:datetime, bbox:tuple[float]
                 # Look for the .h5 file, but specifically grab the HTTPS link
                 if asset.href.endswith('.h5') and asset.href.startswith('https'):
                     url = asset.href
-                    _, file_name = os.path.split(url)
-                    match = NTL_FILENAME_PATTERN.match(file_name)
-                    meta = match.groupdict()
-                    product = meta['product']
-                    dt = datetime.strptime(f'{meta["year"]}{meta["doy"]}', '%Y%j')
-                    tile = meta['tile']
-                    timestamp = dt.strftime(timestamp_format(product_id=product))
-                    key = f'{product}_{timestamp}'
-                    cache.store(key=key, url=url, tile=tile)
-                    urls.append((product,timestamp, tile, url))
+                    result = url2result(url=url, store=True)
+                    urls.append(result)
 
         return urls
 
 
 def search(
         processing_level: str,
-        nominal_date: date,
+        nominal_date: datetime,
         bbox: tuple[float, float, float, float],
         stream: str = None,
+        route:str = None,
         max_concurrency: int = 5,
-
         progress: Progress = None
 ):
     """
@@ -107,58 +179,53 @@ def search(
     Forces both NRT and STD streams to download files locally first for maximum
     processing speed, bypassing slow network/vsicurl VRT parsing.
     """
-    minlon, minlat, maxlon, maxlat = bbox
-    now = datetime.now()
-    plevel = processing_level.upper()
-    # --- 1. Harmonized Temporal Logic ---
-    if stream == OPERATIONAL:
-        days_difference = abs((now - nominal_date).days)
-        if days_difference > 7:
-            raise ValueError(f'Invalid target_date={nominal_date}.{stream} stream holds max 7 days of data. ')
-
-    if 'A1' in plevel or 'A2' in plevel:
-        # A2 and historical/standard A1 target noon dead-center
-        midlat = (minlat + maxlat) * 0.5
-        midlon = (minlon + maxlon) * 0.5
-        utc_offset_hours = midlon / 15.0
-        local_overpass_utc = nominal_date + timedelta(hours=1.5) - timedelta(hours=utc_offset_hours)
-        local_overpass_doy = local_overpass_utc.strftime('%j')
-        start_dt = local_overpass_utc - timedelta(minutes=45)
-        end_dt = local_overpass_utc + timedelta(minutes=45)
-        dt = [start_dt, end_dt]
-    elif 'A3' in plevel:
-        # A3 Monthly composites target mid-month. If current month, step back one month.
-        if now.month == nominal_date.month and now.year == nominal_date.year:
-            prev_month = nominal_date.replace(day=1) - timedelta(days=1)
-            dt = prev_month.replace(day=15)
-        else:
-            dt = nominal_date.replace(day=15)
-    elif 'A4' in plevel:
-        if now.year > nominal_date.year:
-            raise ValueError(f'Can not search in future! Please adjust target date')
-        if now.year == nominal_date.year:
-            # A4 Annual composites target July 1st
-            dt = nominal_date.replace(year=now.year - 1, month=7, day=1)
-        else:
-            # A4 Annual composites target July 1st
-            dt = nominal_date.replace(month=7, day=1)
-    else:
-        raise ValueError(f'Invalid stream {stream} for NASA NTL data')
-
-
+    stream_name = const.STREAM2CATALOG[stream]
+    stream_products = const.API_PRODUCTS[stream_name]
+    stream_processing_levels = stream_products.keys()
+    assert processing_level.upper() in stream_processing_levels, (
+        f'Invalid processing level {processing_level} for {stream_name}. \''
+        f'Valid processing levels {stream_processing_levels}')
+
+    dt = calculate_local_utc(stream=stream,processing_level=processing_level,
+                             nominal_date=nominal_date, bbox=bbox)
+    products = stream_products[processing_level]
+    cached_results = []
+    expected_products_count = len(products)
+    found_products_count = 0
+
+    for product in products:
+        timestamp = dt.strftime(timestamp_format(product_id=product))
+        key = f'{product}_{timestamp}'
+        urls = cache.fetch(key=key)
+        if urls:
+            found_products_count += 1
+            for url in urls:
+                cached_results.append(url2result(url=url, store=False))
+
+    # Only short-circuit if the cache successfully returned data for EVERY product requested
+    if found_products_count == expected_products_count:
+        logger.info("Full cache hit. Bypassing network search.")
+        return cached_results
     # --- 2. Catalog Search ---
     if progress:
         progress_task = progress.add_task(
             description=f'[red]Searching {processing_level} ({stream}) imagery for {nominal_date.date()}',
             total=None
         )
-
-    urls = stac_search(
-        stream=stream,
-        processing_level=processing_level,
-        dt=dt,
-        bbox=bbox,
-    )
+    if route == 'API':
+        urls = api_search(
+            stream=stream,
+            products=products,
+            dt=dt,
+            bbox=bbox,
+        )
+    else:
+        urls = stac_search(
+            stream=stream,
+            processing_level=processing_level,
+            dt=dt,
+            bbox=bbox,
+        )
 
     if not urls:
         logger.info(f"No imagery found for {processing_level} on {nominal_date.date()}")
diff --git a/rapida/ntl/nasa/util.py b/rapida/ntl/nasa/util.py
index 3137737..3209d9d 100644
--- a/rapida/ntl/nasa/util.py
+++ b/rapida/ntl/nasa/util.py
@@ -18,7 +18,7 @@ def get_intersecting_tiles(bbox: tuple[float, float, float, float]) -> list[tupl
     tiles = []
     for v in range(max(0, v_min), min(18, v_max + 1)):
         for h in range(max(0, h_min), min(36, h_max + 1)):
-            tiles.append((h, v))
+            tiles.append(f'h{h:02d}v{v:02d}')
 
     return tiles
 
diff --git a/rapida/ntl/noaa/cmask.py b/rapida/ntl/noaa/cmask.py
index 5ec2b2b..0789bbf 100644
--- a/rapida/ntl/noaa/cmask.py
+++ b/rapida/ntl/noaa/cmask.py
@@ -11,11 +11,15 @@
 import numpy as np
 from rich.progress import Progress
 from osgeo import gdal
-
+from shapely.ops import transform
 
 gdal.UseExceptions()
 logger = logging.getLogger(__name__)
 
+def shift_to_360(lon, lat, z=None):
+    """Transforms standard -180/180 coordinates to a 0-360 system."""
+    shifted_lon = lon + 360.0 if lon < 0 else lon
+    return (shifted_lon, lat) if z is None else (shifted_lon, lat, z)
 
 def bbox_in_hdf(hdf_url: str, bbox: Iterable[float]):
     fs = fsspec.filesystem("http")
@@ -27,13 +31,23 @@ def bbox_in_hdf(hdf_url: str, bbox: Iterable[float]):
             # Now it reads at HTTP speeds without the boto3 overhead
             bounds_poly = wkt.loads(hfile.attrs['geospatial_bounds'].decode('utf-8'))
             bbox_poly = box(*bbox, ccw=True)
+            # 2. The Kiribati Ghost Detection
+            minx, miny, maxx, maxy = bounds_poly.bounds
+            is_idl_crosser = (maxx - minx) > 300
+
+            if is_idl_crosser:
+                # Transform both geometries to 0-360 space to fix the tear
+                working_bounds = transform(shift_to_360, bounds_poly)
+                working_bbox = transform(shift_to_360, bbox_poly)
+            else:
+                # Use standard geometries
+                working_bounds = bounds_poly
+                working_bbox = bbox_poly
             #if not bbox_poly.within(bounds_poly):
-            if not bbox_poly.intersects(bounds_poly):
+            if not working_bbox.intersects(working_bounds):
                 return False
-            intersection_poly = bbox_poly.intersection(bounds_poly)
-            perc_intersection = round(intersection_poly.area/bbox_poly.area * 100
-                                      )
-
+            intersection_poly = working_bbox.intersection(working_bounds)
+            perc_intersection = round(intersection_poly.area/working_bbox.area * 100)
             # with open("/tmp/bbox.geojson", "w") as ff:
             #     ff.write(to_geojson(bbox_poly))
             # n = filename.split('_')[3]

From a3239c15dda74664b9f394e18f05c737c5828e70 Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Thu, 28 May 2026 08:58:09 +0200
Subject: [PATCH 4/9] search caching for noaa

---
 rapida/ntl/cache.py       | 22 +++++++++------
 rapida/ntl/nasa/io.py     |  4 ---
 rapida/ntl/nasa/search.py |  2 +-
 rapida/ntl/noaa/cmask.py  | 22 +++++++++++----
 rapida/ntl/noaa/io.py     | 57 +++++++++++++++++++++++++++++++++++++++
 rapida/ntl/noaa/search.py | 10 +++++--
 6 files changed, 97 insertions(+), 20 deletions(-)

diff --git a/rapida/ntl/cache.py b/rapida/ntl/cache.py
index 0377e36..9429c1d 100644
--- a/rapida/ntl/cache.py
+++ b/rapida/ntl/cache.py
@@ -20,15 +20,21 @@ def search_id(search_params: dict) -> str:
 
 
 
-def store(key:str=None, url:str=None, tile:str=None, cache_path=CACHE_PATH):
+def store(key:str=None, value:str=None, tile:str=None, cache_path=CACHE_PATH):
     with shelve.open(cache_path) as cache:
         record = cache.get(key, None)
         if record is None:
-            record = {tile:url}, time.time()
+            if tile:
+                record = {tile:value}, time.time()
+            else:
+                record = value, time.time()
         else:
             tiles, creation_time = record
-            if not tile in tiles:
-                record[0].update({tile:url})
+            if tile:
+                if not tile in tiles:
+                    record[0].update({tile:value})
+            else:
+                record[0] = value
         cache[key] = record
 
 
@@ -38,7 +44,6 @@ def store(key:str=None, url:str=None, tile:str=None, cache_path=CACHE_PATH):
 def fetch(key:str=None, tile:str=None, cache_path=CACHE_PATH):
     with shelve.open(cache_path) as cache:
         record = cache.get(key, None)
-
         if record is None:
             return
         # 1. Directly unpack the tuple
@@ -48,10 +53,11 @@ def fetch(key:str=None, tile:str=None, cache_path=CACHE_PATH):
         if time.time() - creation_time > MAX_AGE_SECONDS:
             del cache[key]
             return  # Expired
-
         # 3. Handle the tile request
         if tile and tile in tiles:
             return tiles[tile],
-
-        return tuple(tiles.values())
+        if isinstance(tiles, dict):
+            return tuple(tiles.values())
+        else:
+            return tiles
 
diff --git a/rapida/ntl/nasa/io.py b/rapida/ntl/nasa/io.py
index 81e5115..15ab913 100644
--- a/rapida/ntl/nasa/io.py
+++ b/rapida/ntl/nasa/io.py
@@ -21,10 +21,6 @@ async def download(timestamp: str = None, product: str = None, tile:str=None, ds
     # EarthAccess token
     ea_token = os.environ.get('EARTHDATA_TOKEN')
     headers = {"Authorization": f"Bearer {ea_token}"}
-
-
-    semaphore = asyncio.Semaphore(5)
-
     return await download_remote_files(
         file_urls=urls,dst_folder=dst_dir, progress=progress, headers=headers
     )
diff --git a/rapida/ntl/nasa/search.py b/rapida/ntl/nasa/search.py
index 81c6958..ea8905c 100644
--- a/rapida/ntl/nasa/search.py
+++ b/rapida/ntl/nasa/search.py
@@ -52,7 +52,7 @@ def url2result(url:str=None, store=True):
     timestamp = dt.strftime(timestamp_format(product_id=product))
     key = f'{product}_{timestamp}'
     if store:
-        cache.store(key=key, url=url, tile=tile)
+        cache.store(key=key, value=url, tile=tile)
     return product, timestamp, tile, url
 
 
diff --git a/rapida/ntl/noaa/cmask.py b/rapida/ntl/noaa/cmask.py
index 0789bbf..36087e3 100644
--- a/rapida/ntl/noaa/cmask.py
+++ b/rapida/ntl/noaa/cmask.py
@@ -12,8 +12,10 @@
 from rich.progress import Progress
 from osgeo import gdal
 from shapely.ops import transform
-
+from rapida.ntl import cache
 gdal.UseExceptions()
+
+
 logger = logging.getLogger(__name__)
 
 def shift_to_360(lon, lat, z=None):
@@ -160,6 +162,10 @@ def cloud_coverage_fast(hdf_url: str, bbox: Iterable[float],
 
 def cloud_coverage(hdf_url: str, bbox: list) -> int:
 
+
+    _, file_name = os.path.split(hdf_url)
+    cc = cache.fetch(key=file_name)
+    if cc is not None:return cc
     lon_min, lat_min, lon_max, lat_max = bbox
     subdataset_str = f'NETCDF:"/vsicurl/{hdf_url}":CloudMaskBinary'
 
@@ -185,8 +191,10 @@ def cloud_coverage(hdf_url: str, bbox: list) -> int:
 
     if valid_data.size == 0:
         raise Exception(f'Failed to compute cloud coverage for {hdf_url}. No valid data.')
-
-    return int((np.count_nonzero(valid_data == 1) / valid_data.size) * 100)
+    cc = int((np.count_nonzero(valid_data == 1) / valid_data.size) * 100)
+    print(cc)
+    cache.store(key=file_name, value=cc)
+    return cc
 
 
 def cloud_coverage_batch(urls: list[str], bbox: Iterable[float], max_threads: int = 5, progress: Progress = None):
@@ -195,7 +203,7 @@ def cloud_coverage_batch(urls: list[str], bbox: Iterable[float], max_threads: in
     if progress:
         master_task = progress.add_task(
             description=f"[cyan]Computing cloud coverage .... ",
-            total=len(urls)
+            total=None
         )
     then = datetime.datetime.now()
 
@@ -207,7 +215,7 @@ def cloud_coverage_batch(urls: list[str], bbox: Iterable[float], max_threads: in
             for url in urls
         }
 
-        for future in concurrent.futures.as_completed(future_to_url):
+        for future in concurrent.futures.as_completed(future_to_url, timeout=60):
             url = future_to_url[future]
             try:
                 results[url] = future.result()
@@ -219,4 +227,8 @@ def cloud_coverage_batch(urls: list[str], bbox: Iterable[float], max_threads: in
                     progress.update(master_task, advance=1)
     now = datetime.datetime.now()
     logger.debug(f'Computed cloud coverage for {len(urls)} granules in {(now-then).total_seconds()} secs')
+    if progress and 'master_task' in locals():
+        progress.remove_task(master_task)
+
+
     return results
diff --git a/rapida/ntl/noaa/io.py b/rapida/ntl/noaa/io.py
index e16795d..efe7426 100644
--- a/rapida/ntl/noaa/io.py
+++ b/rapida/ntl/noaa/io.py
@@ -130,4 +130,61 @@ async def find_ntl(satellite: str = None, bbox: Iterable[float] = None, dt: date
 
     return found
 
+async def locate_file(satellite:str=None, dt=None, source:str=None, products: Iterable[str] = PRODUCT_NAMES):
+    found = {}
+    # The "Solid" way: Generate stores using from_url
+    viirs_stores = {
+        sat: {
+            source: obstore.store.from_url(url, config=PUBLIC_CONFIG)
+            for source, url in sources.items()
+        }
+        for sat, sources in VIIRS_URLS.items()
+    }
+    stores = viirs_stores[satellite]
 
+    # 1. Safely determine the primary and alternate sources
+    primary_source = source if source else random.choice(SOURCE_NAMES)
+    alt_source = SOURCE_NAMES[0] if primary_source == SOURCE_NAMES[1] else SOURCE_NAMES[1]
+    entries_cache = {}
+    for product_name in products:
+        #match_found = False
+        product = PRODUCTS[product_name]
+        sources_to_try = [primary_source, alt_source]
+        time_pattern = dt.strftime('s%Y%m%d%H%M' if 'cloud' in product.lower() else 'd%Y%m%d_t%H%M')
+        for current_source in sources_to_try:
+            store = stores[current_source]
+
+            date_path = dt.strftime('/%Y/%m/%d/')
+            prefix = f"{product}{date_path}"
+            cache_key = (current_source, prefix)
+            if cache_key not in entries_cache:
+                try:
+                    entries_cache[cache_key] = await obstore.list(store, prefix=prefix).collect_async()
+                except Exception as e:
+                    logger.warning(f"Failed to list {prefix} from {current_source}: {e}")
+                    entries_cache[cache_key] = []
+
+            entries = entries_cache[cache_key]
+            if not entries:
+                continue
+            match_gen = (
+                e for e in entries
+                if time_pattern in e['path'] and e['path'].lower().endswith(('.nc', '.h5'))
+            )
+
+            # next() takes the first match, or returns None if the generator is empty
+            selected_entry = next(match_gen, None)
+
+            if selected_entry:
+                file_path = selected_entry['path']
+                file_size = selected_entry.get('size', 0)  # Safe get
+                if current_source not in found:  # reset
+                    found[current_source] = []
+                found[current_source].append((file_path, file_size))
+                break  # Found it! Stop looking in fallback sources for this product
+
+            else:
+                logger.debug(f"Pattern {time_pattern} not found in {current_source} for {product_name}")
+
+
+    return found
diff --git a/rapida/ntl/noaa/search.py b/rapida/ntl/noaa/search.py
index b396bc3..2e1efc5 100644
--- a/rapida/ntl/noaa/search.py
+++ b/rapida/ntl/noaa/search.py
@@ -15,12 +15,12 @@
 import logging
 from typing import Iterable, Optional
 from rapida.ntl.noaa.io import (
-find_ntl, public_url, parse_noaa_timestamp
+find_ntl, public_url, parse_noaa_timestamp, locate_file
 )
 
 from rapida.ntl.noaa.cmask import cloud_coverage_batch
 from rapida.ntl.noaa.const import  PRODUCTS_RE
-
+from rapida.ntl import cache
 logger = logging.getLogger(__name__)
 
 
@@ -183,6 +183,7 @@ async def track_task(agranule):
 
         if progress and progress_task is not None:
             progress.update(progress_task, description = f'Selected {len(results)} granule(s) for satellite {satellite}')
+            progress.remove_task(progress_task)
         return results
 
     except ExceptionGroup as eg:
@@ -537,6 +538,9 @@ def night_granules(self, bbox:Iterable[float]=None, target_date:date=None, cmask
         return selected_granules
 
 
+
+
+
 def search_granules(satellites:Optional[Iterable[str]]=None,
                     target_date:date=None, bbox:Iterable[float] = None,
                    cmask:bool=False, progress=None):
@@ -625,6 +629,8 @@ async def async_search_granules(
     else:
         selected_granules = list(found_granules.values())
         selected_granules.sort(key=lambda g: g.rank, reverse=True)
+
+
     return selected_granules
 
 # if __name__ == '__main__':

From f4356c1a5ce613e4e9a6eac5ca6ac2f67d7df63f Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Thu, 28 May 2026 10:32:09 +0200
Subject: [PATCH 5/9] download for operational/noaa

---
 rapida/cli/ntl.py     |  60 ++++++++++++++++++++--
 rapida/ntl/noaa/io.py | 113 ++++++++++++++++++++++++++++++++++--------
 2 files changed, 149 insertions(+), 24 deletions(-)

diff --git a/rapida/cli/ntl.py b/rapida/cli/ntl.py
index e17912f..395c2f5 100644
--- a/rapida/cli/ntl.py
+++ b/rapida/cli/ntl.py
@@ -2,7 +2,7 @@
 import numbers
 import os.path
 from datetime import datetime
-
+from typing import Iterable
 import click
 
 
@@ -12,7 +12,8 @@
 from rapida.ntl.noaa.search import async_search_granules, VIIRSNavigator
 from rapida.util.bbox_param_type import BboxParamType
 from rapida.ntl.nasa.io import download as download_from_nasa
-
+from rapida.ntl.noaa.const import SOURCE_NAMES, PRODUCT_NAMES as OPER_PRODUCT_NAMES
+from rapida.ntl.noaa.io import download as download_from_noaa, bytesto
 from rich.table import Table
 logger = logging.getLogger(__name__)
 
@@ -247,15 +248,66 @@ async def download_nasa(ctx, timestamp:str = None, product:str=None, tile:str=No
 
 
 @download.command(name='noaa', short_help=f'Download operational NTL data from NOAA')
+@click.option(
+    "--sat", "-s",
+    "satellite", # This will be the name of the argument in your function
+    type=click.Choice(VIIRSNavigator.SATELLITES, case_sensitive=False),
+    multiple=False,
+    help=f"Target satellite(s). One of ({','.join(VIIRSNavigator.SATELLITES)}) that produced the granule."
+)
 
+@click.option("--timestamp", "-t", "timestamp", type=str, required=True, help='Granule timestamp string as date and time. Ex: 202604152232 ')
+@click.option(
+    "--products",
+                "-p",
+                "products",
+                type=click.Choice(OPER_PRODUCT_NAMES, case_sensitive=False),
+                default=OPER_PRODUCT_NAMES,
+                multiple=True,
+                required=False,
+                help=f'One or more of the operational products: {",".join(OPER_PRODUCT_NAMES)} to download.'
+    )
+@click.option("-src", '--src', "source",
+              type=click.Choice(SOURCE_NAMES, case_sensitive=False),
+              required=False,
+              help='The source {AMAZON/GOOGLE} where to search for the granules.'
+              )
+@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="/tmp",           # 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 the images."
+)
 
 @click.pass_context
-async def download_noaa(ctx, ):
-    logger.info('Downloading from NOAA')
+async def download_noaa(ctx, satellite:str=None, timestamp:str=None, products:Iterable[str]=None, source:str=None, dst_dir:str=None ):
+    progress = ctx.obj.get('progress')
 
 
+    downloaded_files = await download_from_noaa(satellite=satellite,timestamp=timestamp,
+                           source=source, products=products,dest_dir=dst_dir, progress=progress)
 
+    if downloaded_files:
+        table = Table(title=f"VIIRS satellites images for the night of  {timestamp} ",
+                      title_style="bold yellow")
 
+        table.add_column("Satellite", style="green", justify='center')
+        table.add_column("Timestamp (UTC)", style="cyan", justify='center')
+        table.add_column("Downloaded file", justify="left", style="red")
+        table.add_column("File size", justify="center", style="white")
+
+        for _, local_file_path, file_size in downloaded_files:
+            values = satellite, timestamp, f'{local_file_path}', f'{bytesto(file_size, "m"):.2f} MB'
+            table.add_row(*values)
+        progress.console.print(table)
 
 # @ntl.command(short_help=f'Execute crisis impact detection (48h Alerts / 72h Assessments)')
 # @click.pass_context
diff --git a/rapida/ntl/noaa/io.py b/rapida/ntl/noaa/io.py
index efe7426..b13aefb 100644
--- a/rapida/ntl/noaa/io.py
+++ b/rapida/ntl/noaa/io.py
@@ -8,12 +8,21 @@
 import logging
 from rapida.ntl.noaa.cmask import bbox_in_hdf
 from urllib.parse import urlparse
-
-
+from rapida.ntl.noaa.const import PRODUCT2NAME
+import aiofiles
 
 logger = logging.getLogger(__name__)
 
 
+# The "Solid" way: Generate stores using from_url
+VIIRS_STORES = {
+    sat: {
+        source: obstore.store.from_url(url, config=PUBLIC_CONFIG)
+        for source, url in sources.items()
+    }
+    for sat, sources in VIIRS_URLS.items()
+}
+
 def parse_noaa_timestamp(time_str: str) -> datetime:
     """
     Converts a NOAA VIIRS string (e.g., '202604010001018') into a timezone-naive UTC datetime.
@@ -42,17 +51,10 @@ async def find_ntl(satellite: str = None, bbox: Iterable[float] = None, dt: date
                    products: Iterable[str] = PRODUCT_NAMES, source: str = None):
     found = {}
 
-    # The "Solid" way: Generate stores using from_url
-    viirs_stores = {
-        sat: {
-            source: obstore.store.from_url(url, config=PUBLIC_CONFIG)
-            for source, url in sources.items()
-        }
-        for sat, sources in VIIRS_URLS.items()
-    }
 
 
-    stores = viirs_stores[satellite]
+
+    stores = VIIRS_STORES[satellite]
 
     # 1. Safely determine the primary and alternate sources
     primary_source = source if source else random.choice(SOURCE_NAMES)
@@ -132,15 +134,8 @@ async def find_ntl(satellite: str = None, bbox: Iterable[float] = None, dt: date
 
 async def locate_file(satellite:str=None, dt=None, source:str=None, products: Iterable[str] = PRODUCT_NAMES):
     found = {}
-    # The "Solid" way: Generate stores using from_url
-    viirs_stores = {
-        sat: {
-            source: obstore.store.from_url(url, config=PUBLIC_CONFIG)
-            for source, url in sources.items()
-        }
-        for sat, sources in VIIRS_URLS.items()
-    }
-    stores = viirs_stores[satellite]
+
+    stores = VIIRS_STORES[satellite]
 
     # 1. Safely determine the primary and alternate sources
     primary_source = source if source else random.choice(SOURCE_NAMES)
@@ -188,3 +183,81 @@ async def locate_file(satellite:str=None, dt=None, source:str=None, products: It
 
 
     return found
+
+
+
+async def fetch_file(satellite:str=None, provider:str=None, path:str=None, size:int=None, dst_dir:str=None,
+                     progress=None, progress_task = None):
+    try:
+        adir = os.path.abspath(dst_dir)
+        if not os.path.exists(adir):
+            os.mkdir(adir)
+        down_task = None
+        store = VIIRS_STORES[satellite][provider]
+        rel_path, file_name = os.path.split(path)
+        product = rel_path.split('/')[0]
+        product_name = PRODUCT2NAME[product]
+        if progress:
+            down_task = progress.add_task(f'[red]Downloading  {file_name} from {provider}', total=size)
+        dst_file_path = os.path.join(adir, file_name)
+        response = await obstore.get_async(store, path)
+        async with aiofiles.open(dst_file_path, 'wb') as local_file:
+            # The 'get' call is the async request
+            async for chunk in  response.stream():
+                await local_file.write(chunk)
+                if progress and down_task is not None:
+                    progress.update(down_task, advance=len(chunk))
+
+        if os.stat(dst_file_path).st_size == size:
+            if progress and progress_task is not None:
+                progress.update(progress_task, description=f'[green]Downloaded {file_name} from {provider}', advance=1)
+            return product_name, dst_file_path, size
+    except Exception:
+
+        raise
+    finally:
+        if progress:
+            if down_task:progress.remove_task(down_task)
+
+
+async def fetch_ntl(found_paths:dict[str, list]=None, satellite:str=None, dst_dir='/tmp', progress=None):
+
+    # Download logic (Surgical io to local SSD)
+    tasks = []
+    progress_task = None
+    try:
+        async with asyncio.TaskGroup() as tg:
+            for provider, files in found_paths.items():
+                if progress:
+                    progress_task = progress.add_task(description=f'Downloading VIIRS images...', total=len(files))
+                for path, size in files:
+                    tasks.append(tg.create_task(fetch_file(
+                        satellite=satellite, provider=provider,
+                        path=path, size=size, progress=progress,
+                        dst_dir=dst_dir, progress_task = progress_task
+                    )))
+    except ExceptionGroup as eg:
+        for e in eg.exceptions:
+            logger.error(f"❌ Sub-task failed: {e}")
+    finally:
+
+        results = [t.result() for t in tasks]
+
+        return results
+
+
+
+
+
+async def download(satellite:str=None, timestamp:str=None, source:str=None,
+        products:Iterable[str]=PRODUCT_NAMES, dest_dir='/tmp', progress=None):
+    dt = datetime.strptime(timestamp, '%Y%m%d%H%M')
+    logger.info(f'Locating files for satellite {satellite} timestamp {timestamp} ')
+    found_files = await locate_file(satellite=satellite, dt=dt, source=source, products=products)
+    return  await fetch_ntl(found_paths=found_files, dst_dir=dest_dir, satellite=satellite, progress=progress)
+
+
+def bytesto(bytes, to, bsize=1024):
+    a = {'k' : 1, 'm': 2, 'g' : 3, 't' : 4, 'p' : 5, 'e' : 6 }
+    r = float(bytes)
+    return bytes / (bsize ** a[to])
\ No newline at end of file

From abefbbe4f11ba2a445b058ee8754a00be489c666 Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Thu, 28 May 2026 12:10:14 +0200
Subject: [PATCH 6/9] add token guard

---
 rapida/cli/ntl.py                   |  5 ++++-
 rapida/ntl/nasa/io.py               |  7 +++++++
 rapida/ntl/noaa/cmask.py            |  1 -
 rapida/util/download_remote_file.py | 20 ++++++++++++--------
 4 files changed, 23 insertions(+), 10 deletions(-)

diff --git a/rapida/cli/ntl.py b/rapida/cli/ntl.py
index 395c2f5..72e337d 100644
--- a/rapida/cli/ntl.py
+++ b/rapida/cli/ntl.py
@@ -15,6 +15,9 @@
 from rapida.ntl.noaa.const import SOURCE_NAMES, PRODUCT_NAMES as OPER_PRODUCT_NAMES
 from rapida.ntl.noaa.io import download as download_from_noaa, bytesto
 from rich.table import Table
+from rapida.project.project import Project
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -291,7 +294,7 @@ async def download_nasa(ctx, timestamp:str = None, product:str=None, tile:str=No
 async def download_noaa(ctx, satellite:str=None, timestamp:str=None, products:Iterable[str]=None, source:str=None, dst_dir:str=None ):
     progress = ctx.obj.get('progress')
 
-
+    project = Project(os.getcwd())
     downloaded_files = await download_from_noaa(satellite=satellite,timestamp=timestamp,
                            source=source, products=products,dest_dir=dst_dir, progress=progress)
 
diff --git a/rapida/ntl/nasa/io.py b/rapida/ntl/nasa/io.py
index 15ab913..5541eb7 100644
--- a/rapida/ntl/nasa/io.py
+++ b/rapida/ntl/nasa/io.py
@@ -13,6 +13,8 @@ async def download(timestamp: str = None, product: str = None, tile:str=None, ds
 
     key = f'{product.upper()}_{timestamp}'
     urls = cache.fetch(key=key, tile=tile)
+
+
     if not urls:
         logger.info(f'Failed to locate information in {cache.CACHE_PATH} for {product}-{timestamp}-{tile or ""} \n' \
                        f'Consider searching first.')
@@ -20,6 +22,11 @@ async def download(timestamp: str = None, product: str = None, tile:str=None, ds
 
     # EarthAccess token
     ea_token = os.environ.get('EARTHDATA_TOKEN')
+
+    # Add this sanity check
+    if not ea_token:
+        raise ValueError("CRITICAL: EARTHDATA_TOKEN environment variable is not set or is empty!")
+
     headers = {"Authorization": f"Bearer {ea_token}"}
     return await download_remote_files(
         file_urls=urls,dst_folder=dst_dir, progress=progress, headers=headers
diff --git a/rapida/ntl/noaa/cmask.py b/rapida/ntl/noaa/cmask.py
index 36087e3..63443c9 100644
--- a/rapida/ntl/noaa/cmask.py
+++ b/rapida/ntl/noaa/cmask.py
@@ -192,7 +192,6 @@ def cloud_coverage(hdf_url: str, bbox: list) -> int:
     if valid_data.size == 0:
         raise Exception(f'Failed to compute cloud coverage for {hdf_url}. No valid data.')
     cc = int((np.count_nonzero(valid_data == 1) / valid_data.size) * 100)
-    print(cc)
     cache.store(key=file_name, value=cc)
     return cc
 
diff --git a/rapida/util/download_remote_file.py b/rapida/util/download_remote_file.py
index 291222a..423fd6e 100644
--- a/rapida/util/download_remote_file.py
+++ b/rapida/util/download_remote_file.py
@@ -99,19 +99,24 @@ async def download_file(file_url=None, dst_file_path=None,
     for attempt in range(no_attempts):
         try:
             down_task = None
-            async with session.get(file_url, timeout=data_read_timeout) as response:
+            async with session.get(file_url, timeout=data_read_timeout, allow_redirects=True) as response:
                 if response.status == 200:
+                    # Use .get() to avoid KeyError if the header is missing
+                    remote_size_str = response.headers.get('Content-Length')
+                    remote_size = int(remote_size_str) if remote_size_str else None
 
-                    remote_size = int(response.headers['Content-Length'])
                     if os.path.exists(dst_file_path):
-                        if not force and os.path.getsize(dst_file_path) == remote_size:
+                        # Only compare sizes if remote_size is known
+                        if not force and remote_size is not None and os.path.getsize(dst_file_path) == remote_size:
                             logger.debug(f'Returning local file {dst_file_path}')
                             return dst_file_path
                         else:
                             os.remove(dst_file_path)
 
                     if progress:
+                        # total=None allows rich to show a spinner instead of a % bar for unknown sizes
                         down_task = progress.add_task(f'[cyan]Downloading {file_url}', total=remote_size)
+
                     async with aiofiles.open(dst_file_path, 'wb') as local_file:
                         while True:
                             chunk = await response.content.read(chunk_size)
@@ -122,8 +127,9 @@ async def download_file(file_url=None, dst_file_path=None,
                                 progress.update(down_task, advance=len(chunk))
 
                     size = os.path.getsize(dst_file_path)
-                    if size != remote_size:
-                        raise Exception(f'{file_url} is was not downloaded correctly!')
+                    # Only validate size if the server provided a Content-Length
+                    if remote_size is not None and size != remote_size:
+                        raise Exception(f'{file_url} was not downloaded correctly!')
 
                     logger.debug(f'File {dst_file_path} was successfully downloaded')
 
@@ -138,7 +144,7 @@ async def download_file(file_url=None, dst_file_path=None,
                 os.remove(dst_file_path)
             raise ce
         except Exception as e:
-            logger.error(f'Exception "{e}" was encountered in while downloading {file_url}')
+            logger.error(f'Exception "{e}" was encountered while downloading {file_url}')
             if os.path.exists(dst_file_path):
                 os.remove(dst_file_path)
             if attempt == no_attempts - 1:
@@ -148,8 +154,6 @@ async def download_file(file_url=None, dst_file_path=None,
             if down_task and progress:
                 progress.remove_task(down_task)
 
-
-
 async def download_s3_object(
     url,
     filename,

From 72f6acf63f7395ce0be58cc1d2a39d60831e6573 Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Thu, 28 May 2026 12:39:12 +0200
Subject: [PATCH 7/9] add token guard

---
 rapida/ntl/nasa/search.py | 17 +++++++++--------
 1 file changed, 9 insertions(+), 8 deletions(-)

diff --git a/rapida/ntl/nasa/search.py b/rapida/ntl/nasa/search.py
index ea8905c..0ad844f 100644
--- a/rapida/ntl/nasa/search.py
+++ b/rapida/ntl/nasa/search.py
@@ -57,7 +57,7 @@ def url2result(url:str=None, store=True):
 
 
 
-def calculate_local_utc(stream:str, processing_level:str, nominal_date: datetime, bbox:tuple[float]):
+def calculate_local_utc(stream:str, processing_level:str, nominal_date: datetime, bbox:tuple[float], route:str=None):
     """
     Calculate VIIRS satellites local overpass time in UTC TZ
     :param stream:
@@ -84,21 +84,21 @@ def calculate_local_utc(stream:str, processing_level:str, nominal_date: datetime
 
         dt = local_overpass_utc
     elif 'A3' in plevel:
+        day = 15 if route == 'STAC' else 1
         # A3 Monthly composites target mid-month. If current month, step back one month.
         if now.month == nominal_date.month and now.year == nominal_date.year:
             prev_month = nominal_date.replace(day=1) - timedelta(days=1)
-            dt = prev_month.replace(day=15)
+            dt = prev_month.replace(day=day)
         else:
-            dt = nominal_date.replace(day=15)
+            dt = nominal_date.replace(day=day)
     elif 'A4' in plevel:
         if now.year > nominal_date.year:
             raise ValueError(f'Can not search in future! Please adjust target date')
+        month = 7 if route == 'STAC' else 1 # A4 Annual composites target July 1st on stac and jan 1st on api
         if now.year == nominal_date.year:
-            # A4 Annual composites target July 1st
-            dt = nominal_date.replace(year=now.year - 1, month=7, day=1)
+            dt = nominal_date.replace(year=now.year - 1, month=month, day=1)
         else:
-            # A4 Annual composites target July 1st
-            dt = nominal_date.replace(month=7, day=1)
+            dt = nominal_date.replace(month=month, day=1)
     else:
         raise ValueError(f'Invalid stream {stream} for NASA NTL data')
     return dt
@@ -108,6 +108,7 @@ def api_search(stream:str, products:str, dt:datetime, bbox:tuple[float])-> list[
     tiles = get_intersecting_tiles(bbox=bbox)
     urls = []
     for product in products:
+
         content_url = f'{const.API_CONTENT[stream]}/{product}/{dt.strftime("%Y/%j")}'
         with httpx.Client() as client:
             # Fetch the JSON directory listing
@@ -187,7 +188,7 @@ def search(
         f'Valid processing levels {stream_processing_levels}')
 
     dt = calculate_local_utc(stream=stream,processing_level=processing_level,
-                             nominal_date=nominal_date, bbox=bbox)
+                             nominal_date=nominal_date, bbox=bbox, route=route)
     products = stream_products[processing_level]
     cached_results = []
     expected_products_count = len(products)

From 8332187e0875ecbcf4075b651ba2f7255a81634a Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Fri, 29 May 2026 01:19:15 +0200
Subject: [PATCH 8/9] bulk download

---
 rapida/cli/ntl.py         | 135 +++++++++++++++++++++++++++++++++++---
 rapida/ntl/nasa/const.py  |   4 +-
 rapida/ntl/nasa/io.py     | 113 +++++++++++++++++++++++++++++++
 rapida/ntl/nasa/search.py |  29 +++++---
 4 files changed, 260 insertions(+), 21 deletions(-)

diff --git a/rapida/cli/ntl.py b/rapida/cli/ntl.py
index 72e337d..661bfa4 100644
--- a/rapida/cli/ntl.py
+++ b/rapida/cli/ntl.py
@@ -4,18 +4,18 @@
 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
+from rapida.ntl.nasa.const import ARCHIVE, OPERATIONAL, PROCESSING_LEVEL_NAMES, PRODUCT_NAMES, PRODUCTS, \
+    NTL_FILENAME_PATTERN, ROUTES, COLLECTIONS
 from rapida.ntl.nasa.search import search as nasa_search
 from rapida.ntl.noaa.search import async_search_granules, VIIRSNavigator
 from rapida.util.bbox_param_type import BboxParamType
-from rapida.ntl.nasa.io import download as download_from_nasa
+from rapida.ntl.nasa.io import download as download_from_nasa, bulk_download
 from rapida.ntl.noaa.const import SOURCE_NAMES, PRODUCT_NAMES as OPER_PRODUCT_NAMES
 from rapida.ntl.noaa.io import download as download_from_noaa, bytesto
 from rich.table import Table
-from rapida.project.project import Project
+from rapida.ntl.nasa.io import bulk_download as bdownload
 
 
 logger = logging.getLogger(__name__)
@@ -41,6 +41,47 @@ def handle_parse_result(self, ctx, opts, args):
         return super().handle_parse_result(ctx, opts, args)
 
 
+def validate_products_strict(ctx, param, value):
+    if not value:
+        return value
+
+    # Check for mixed catalogs
+    has_nrt = any('nrt' in p.lower() for p in value)
+    has_std = any('nrt' not in p.lower() for p in value)
+    nrt_choices  = [item for p in COLLECTIONS['LANCEMODIS'].values() for item in p]
+    std_choices = [item for p in COLLECTIONS['LAADS'].values() for item in p]
+    if has_nrt and has_std:
+        raise click.BadParameter(
+            f"Cannot mix NRT and Standard products in the same command. "
+            f"They belong to different catalogs:\n"
+            f"LANCEMODIS - NOAA operational: {', '.join(nrt_choices)}\n" 
+            f"LAADS - NASA archive : {', '.join(std_choices)}"
+        )
+
+    if has_nrt:
+        return tuple(nrt_choices)
+    else:
+        return tuple(std_choices)
+
+class NASAProductsChoiceOption(click.Option):
+    """
+    Custom Click option that dynamically validates choices
+    """
+
+    def handle_parse_result(self, ctx, opts, args):
+        # Retrieve the value of 'stream' that click has already processed
+        products = opts.get('products')
+        has_nrt = 'nrt' in products[0].lower()
+        if has_nrt:
+            valid_choices = [item for p in COLLECTIONS['LANCEMODIS'].values() for item in p]
+
+        else:
+            valid_choices = [item for p  in COLLECTIONS['LAADS'].values() for item in p]
+
+        self.type = click.Choice([c.upper() for c in valid_choices], case_sensitive=False)
+        return super().handle_parse_result(ctx, opts, args)
+
+
 @click.group(cls=RapidaCommandGroup)
 
 
@@ -169,7 +210,7 @@ def search_nasa(ctx, bbox:tuple[numbers.Number]=None, nominal_date:datetime=None
     progress = ctx.obj.get('progress')
 
     urls = nasa_search(processing_level=processing_level, nominal_date=nominal_date,
-                       bbox=bbox, stream=stream, route=route, progress=progress)
+                       bbox=bbox, stream=stream, route=route, progress=progress, push_to_cache=True)
 
     if urls:
         table = Table(title=f" {processing_level} VIIRS satellites tiles for the night of  {nominal_date.date()}-{nominal_date.strftime('%Y%j')} covering {bbox}",
@@ -220,7 +261,7 @@ def download():
         dir_okay=True,
         resolve_path=True  # Resolves relative paths (like '.') to absolute paths automatically
     ),
-    default="/tmp",           # Defaults to the current working directory
+    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 the images."
 )
@@ -285,7 +326,7 @@ async def download_nasa(ctx, timestamp:str = None, product:str=None, tile:str=No
         dir_okay=True,
         resolve_path=True  # Resolves relative paths (like '.') to absolute paths automatically
     ),
-    default="/tmp",           # Defaults to the current working directory
+    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 the images."
 )
@@ -293,8 +334,6 @@ async def download_nasa(ctx, timestamp:str = None, product:str=None, tile:str=No
 @click.pass_context
 async def download_noaa(ctx, satellite:str=None, timestamp:str=None, products:Iterable[str]=None, source:str=None, dst_dir:str=None ):
     progress = ctx.obj.get('progress')
-
-    project = Project(os.getcwd())
     downloaded_files = await download_from_noaa(satellite=satellite,timestamp=timestamp,
                            source=source, products=products,dest_dir=dst_dir, progress=progress)
 
@@ -312,6 +351,82 @@ async def download_noaa(ctx, satellite:str=None, timestamp:str=None, products:It
             table.add_row(*values)
         progress.console.print(table)
 
+
+
+
+@download.command(name='bulk', short_help=f'Download daily  NTL products from NASA BlackMarble using STAC ')
+
+@click.option('-b', '--bbox',
+              required=True,
+              type=BboxParamType(),
+              help='Bounding box xmin/west, ymin/south, xmax/east, ymax/north'
+              )
+@click.option("--from", "start_date",
+              type=click.DateTime(formats=["%Y-%m-%d"]),
+              required=True,
+              help='The start date of required period'
+              )
+@click.option("--to", "end_date",
+              type=click.DateTime(formats=["%Y-%m-%d"]),
+              required=True,
+              help='The end date of required period'
+              )
+
+@click.option(
+        '-p','--products',
+        # type=click.Choice(PRODUCTS, case_sensitive=False),
+        cls=NASAProductsChoiceOption,
+        #callback=validate_products_strict,
+        required=True,
+        multiple=True,
+        help=f"One or more STAC collections hosting different processing level or products limited to one stream. "
+    )
+@click.option(
+    "--dst-dir",
+    "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 the images."
+)
+
+
+@click.pass_context
+async def bulk_download(ctx, bbox:tuple[numbers.Number]=None, start_date:datetime=None, end_date:datetime=None,
+                            products:str=None, dst_dir:str=None):
+    progress = ctx.obj.get('progress')
+
+    if start_date > end_date:
+        raise click.UsageError(f'--from {start_date} must be smaller or equal then --to {end_date}')
+
+    if 'nrt' in products[0].lower():
+        stream = OPERATIONAL
+        now = datetime.now()
+
+        start_days_difference = abs((now - start_date).days)
+        if start_days_difference > 7:
+            raise ValueError(f'Invalid start_date={start_date}.{stream} stream holds max 7 days of data. ')
+        end_days_difference = abs((now - end_date).days)
+        if end_days_difference > 7:
+            raise ValueError(f'Invalid start_date={end_date}.{stream} stream holds max 7 days of data. ')
+
+    else:
+        stream = ARCHIVE
+    downloaded_files = await bdownload(
+        bbox=bbox, start_date=start_date, end_date=end_date,
+        stream=stream, products=products,
+        dst_dir=dst_dir,progress=progress
+    )
+
+
+
+
+
 # @ntl.command(short_help=f'Execute crisis impact detection (48h Alerts / 72h Assessments)')
 # @click.pass_context
 # async def detect(ctx):
diff --git a/rapida/ntl/nasa/const.py b/rapida/ntl/nasa/const.py
index c4999f1..2ea5c6c 100644
--- a/rapida/ntl/nasa/const.py
+++ b/rapida/ntl/nasa/const.py
@@ -128,7 +128,7 @@ def processing_levels(collections=COLLECTIONS):
     levels = {level for stream, prods in COLLECTIONS.items() for level in stream}
 
 
-    # print(PROCESSING_LEVELS)
-    print(cleaned_dict)
+    print(levels)
+
 
     #print(PRODUCTS)
\ No newline at end of file
diff --git a/rapida/ntl/nasa/io.py b/rapida/ntl/nasa/io.py
index 5541eb7..0dd5140 100644
--- a/rapida/ntl/nasa/io.py
+++ b/rapida/ntl/nasa/io.py
@@ -7,7 +7,13 @@
 from rapida.ntl import cache
 from rapida.util.download_remote_file import download_remote_files
 import logging
+from urllib.parse import urlparse
+from rapida.ntl.nasa.search import stac_search
+from datetime import datetime
+import numbers
+
 logger = logging.getLogger(__name__)
+
 async def download(timestamp: str = None, product: str = None, tile:str=None, dst_dir:str=None, progress:Progress=None):
 
 
@@ -31,3 +37,110 @@ async def download(timestamp: str = None, product: str = None, tile:str=None, ds
     return await download_remote_files(
         file_urls=urls,dst_folder=dst_dir, progress=progress, headers=headers
     )
+
+
+async def download_tile(
+        client: httpx.AsyncClient=None,
+        url: str=None,
+        dest_path: Path=None,
+        semaphore: asyncio.Semaphore=None,
+        progress: Progress=None,
+        max_retries: int = 3
+) -> Path:
+    """
+    Asynchronously downloads a single tile with exponential backoff retries.
+    Uses a semaphore to prevent overwhelming the LANCE servers.
+    """
+    progress_task = None
+    async with semaphore:
+        for attempt in range(max_retries):
+            try:
+                # Start the streaming request
+                async with client.stream("GET", url) as response:
+                    response.raise_for_status()
+
+                    # Initialize progress bar for this specific file
+                    total_bytes = int(response.headers.get("Content-Length", 0))
+                    if progress:
+                        progress_task = progress.add_task(f'Downloading {url}', total=total_bytes,)
+                    tmp_path = dest_path.with_suffix(dest_path.suffix + ".tmp")
+                    with open(tmp_path, "wb") as f:
+                        async for chunk in response.aiter_bytes():
+                            f.write(chunk)
+                            if progress and progress_task:
+                                progress.advance(progress_task, advance=len(chunk))
+                    tmp_path.rename(dest_path)
+                if progress and progress_task:
+                    progress.update(progress_task, description=f"[green]✓ {dest_path.name}")
+
+                return dest_path
+
+            except httpx.HTTPError as e:
+                if attempt == max_retries - 1:
+                    if progress and progress_task:
+                        progress.update(progress_task, description=f"[red]✗ {dest_path.name} (Failed)[/red]")
+                    #progress.console.print(f"[red]Error downloading {url}: {e}[/red]")
+                    return dest_path
+
+                # Exponential backoff before retry (1s, 2s, 4s...)
+                await asyncio.sleep(2 ** attempt)
+
+            finally:
+                if progress and progress_task:
+                    progress.remove_task(progress_task)
+
+async def bulk_download(bbox:tuple[numbers.Number]=None, start_date:datetime=None, end_date:datetime=None,
+                           stream:str = None, products:str=None, dst_dir:str=None, progress=None):
+
+
+    results = stac_search(stream=stream, products=products,
+                       dt=[start_date, end_date], bbox=bbox,push_to_cache=False)
+
+
+    # EarthAccess token
+    ea_token = os.environ.get('EARTHDATA_TOKEN')
+
+    # Add this sanity check
+    if not ea_token:
+        raise ValueError("CRITICAL: EARTHDATA_TOKEN environment variable is not set or is empty!")
+
+    headers = {"Authorization": f"Bearer {ea_token}"}
+    tasks = []
+    semaphore = asyncio.Semaphore(5)
+    progress_task = None
+    if results:
+        dest_path = Path(dst_dir)
+        dest_path.mkdir(parents=True, exist_ok=True)
+        async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:
+            for e in results:
+                *r, url = e
+                path = urlparse(url).path
+
+                # 2. Get the basename from that path
+                file_name = os.path.basename(path)
+
+                filepath = dest_path / file_name
+
+                tasks.append(asyncio.Task(
+                    download_tile(client, url, filepath, semaphore), name=file_name
+                ))
+
+            if progress:
+                progress_task = progress.add_task(description=f'Downloading {len(tasks)} images...', total=len(tasks))
+            downloaded_files = []
+            for task in asyncio.as_completed(tasks, timeout=20*len(tasks)):
+                try:
+                    downloaded_file = await task
+
+                    if progress and progress_task is not None:
+                        progress.update(progress_task,description=f'[green]🡇 {downloaded_file.name}', advance=1)
+                    downloaded_files.append(str(downloaded_file))
+                except Exception as e:
+                    logger.error(e)
+
+                except asyncio.CancelledError as ce:
+                    for atask in tasks:
+                        if not atask.done():
+                            atask.cancel()
+                    await asyncio.gather(*tasks, return_exceptions=True)
+                    raise
\ No newline at end of file
diff --git a/rapida/ntl/nasa/search.py b/rapida/ntl/nasa/search.py
index 0ad844f..89dbd83 100644
--- a/rapida/ntl/nasa/search.py
+++ b/rapida/ntl/nasa/search.py
@@ -10,6 +10,7 @@
 from rich.progress import Progress
 from rapida.ntl.nasa.util import get_intersecting_tiles
 import httpx
+from typing import Optional
 logger = logging.getLogger(__name__)
 
 
@@ -104,7 +105,7 @@ def calculate_local_utc(stream:str, processing_level:str, nominal_date: datetime
     return dt
 
 
-def api_search(stream:str, products:str, dt:datetime, bbox:tuple[float])-> list[str]:
+def api_search(stream:str, products:str, dt:datetime, bbox:tuple[float], push_to_cache:bool=True)-> list[str]:
     tiles = get_intersecting_tiles(bbox=bbox)
     urls = []
     for product in products:
@@ -121,25 +122,33 @@ def api_search(stream:str, products:str, dt:datetime, bbox:tuple[float])-> list[
                 for tile in tiles:
                     if tile in file_name:
                         url = item.get('downloadsLink')
-                        result = url2result(url=url, store=True)
+                        result = url2result(url=url, store=push_to_cache)
                         urls.append(result)
 
 
     return urls
 
-def stac_search(stream:str, processing_level:str, dt:datetime, bbox:tuple[float]):
+def stac_search(stream:str=None, processing_level:Optional[str]=None, products:list[str]=None, dt:datetime=None,
+                bbox:tuple[float]=None, push_to_cache:bool=True):
 
     catalog_name = const.STREAM2CATALOG[stream]
     catalog_collections = const.COLLECTIONS[catalog_name]
     catalog_processing_levels = catalog_collections.keys()
-    assert processing_level in catalog_processing_levels, (f'Invalid processing level {processing_level} for {catalog_name}. \''
+
+    if not products:
+
+        available_collections = sorted(catalog_collections[processing_level], reverse=True)
+
+        assert processing_level in catalog_processing_levels, (f'Invalid processing level {processing_level} for {catalog_name}. \''
                                                            f'Valid processing levels {catalog_processing_levels}')
+    else:
+        available_collections = list(products)
+    logger.info(
+        f'Searching for imagery in catalog "{catalog_name}" collections: {available_collections}')
+
+
 
-    available_collections = sorted(catalog_collections[processing_level], reverse=True) #we preffe
 
-    logger.info(f'Searching for {processing_level} imagery in catalog "{catalog_name}" collections: {available_collections}')
-    logger.debug(f'Searching for {processing_level} imagery in catalog "{catalog_name}" collections: {available_collections} ' \
-                 f'for {dt} and {bbox} geaographic area')
     stac_url = f'{const.CMR_STAC_ROOT}{catalog_name}'
     urls = []
     catalog = Client.open(url=stac_url)
@@ -150,6 +159,7 @@ def stac_search(stream:str, processing_level:str, dt:datetime, bbox:tuple[float]
         bbox=bbox
     )
 
+
     if search_result.matched():
         logger.info(f"Found {search_result.matched()} granule(s) at {stac_url}")
         items = search_result.item_collection()
@@ -160,7 +170,7 @@ def stac_search(stream:str, processing_level:str, dt:datetime, bbox:tuple[float]
                 # Look for the .h5 file, but specifically grab the HTTPS link
                 if asset.href.endswith('.h5') and asset.href.startswith('https'):
                     url = asset.href
-                    result = url2result(url=url, store=True)
+                    result = url2result(url=url, store=push_to_cache)
                     urls.append(result)
 
         return urls
@@ -172,6 +182,7 @@ def search(
         bbox: tuple[float, float, float, float],
         stream: str = None,
         route:str = None,
+        push_to_cache:bool=False,
         max_concurrency: int = 5,
         progress: Progress = None
 ):

From bce3fda849e7928b686376f82d38aca07f15faf7 Mon Sep 17 00:00:00 2001
From: Ioan Ferencik <ioan.ferencik@undp.org>
Date: Fri, 29 May 2026 12:02:57 +0200
Subject: [PATCH 9/9] add earthdata toke env var

---
 .env.example | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.env.example b/.env.example
index 79dd779..e7c8cb0 100644
--- a/.env.example
+++ b/.env.example
@@ -1,6 +1,6 @@
 TENANT_ID=
 CLIENT_ID=
-
+EARTHDATA_TOKEN=
 # azure blob storage account name for rapida tool. default is 'undpgeohub'
 AZURE_STORAGE_ACCOUNT=
 # container name of azure blob storage for publishing. default is 'rapida'