Downloader

The InSARHub Downloader module provides a streamlined interface for searching and downloading satellite data.

• Import downloader

  Import the Downloader class to access all downloader functionality

  from insarhub import Downloader
  

• View available downloaders

  List all registered downloaders

  Downloader.available()
  

Available Downloaders

ASF_Base_DownloaderS1_SLCS1_BurstNISAR_GSLCNISAR_RSLCNISAR_GUNW

InSARHub wrapped asf_search as one of its download backends. The ASF_Base_Downloader is implemented on top of a reusable base configuration class, which provides the full searching, filtering, and downloading logic of asf_search.

Source code in src/insarhub/downloader/asf_base.py

class ASF_Base_Downloader(BaseDownloader):
    """
    Simplify searching and downloading satellite data using ASF Search API.
    """
    description = "Generic ASF Search API downloader. Supports Sentinel-1, ALOS, NISAR, and more."
    default_config = ASF_Base_Config

    # Config fields that are NOT asf.search() keywords. search() builds its query
    # from asdict(config), so any subclass field that configures post-search
    # behaviour instead of the query itself must be listed here -- otherwise it
    # is forwarded to asf.search() as an unknown kwarg, the call raises, and the
    # retry loop below burns ~17 minutes of exponential backoff before surfacing
    # it. Subclasses extend this set rather than overriding search().
    _NON_SEARCH_FIELDS: frozenset = frozenset(
        {"workdir", "name", "bbox", "granule_names", "ssl_verify", "max_workers"})

    #: What the second half of a stack key IS, for user-facing output. Frame-based
    #: datasets key on frameNumber; SLC-BURST has no frameNumber at all and keys on
    #: fullBurstID, so printing "frame 124_264305_IW2" misnames the value and
    #: implies a number the user could pass to --frame. Subclasses override.
    stack_key_label: str = "frame"

    #: Plural noun for what this downloader searches, used in progress output.
    #: The base class is dataset-agnostic, so it stays generic; every subclass
    #: names its own product ("bursts", "GUNWs") rather than reporting the
    #: S1_SLC wording for everything.
    product_label: str = "products"

    _DATASET_GROUP_KEYS = {
        'SENTINEL-1': ('pathNumber', 'frameNumber'),
        'ALOS':       ('pathNumber', 'frameNumber'),
        'NISAR':      ('pathNumber', 'frameNumber'),  # NISAR carries frameNumber, not frameID
        'BURST':      ('pathNumber', 'burstID'),
    }
    _DATASET_PROPERTY_KEYS = {
        'SENTINEL-1': {
            'relativeOrbit': 'pathNumber',
            'absoluteOrbit': 'absoluteOrbit',
            'polarization':  'polarization',
            'flightDirection': 'flightDirection',
        },
        'ALOS': {
            'relativeOrbit': 'pathNumber',
            'absoluteOrbit': 'absoluteOrbit',
            'polarization':  'polarization',
            'flightDirection': 'flightDirection',
        },
        'NISAR': {
            'relativeOrbit': 'pathNumber',   # NISAR exposes track as pathNumber, like S1
            'absoluteOrbit': 'absoluteOrbit',
            'polarization':  'polarization',
            'flightDirection': 'flightDirection',
        },
    }

    def __init__(self, config: ASF_Base_Config | None = None): 

        """
        Initialize the Downloader with search parameters. Options was adapted from asf_search searching api. 
        You may check https://docs.asf.alaska.edu/asf_search/searching/ for more info, below only list customized parameters.
        """
        print(f"""
This downloader relies on the ASF API. Please ensure you to create an account at https://search.asf.alaska.edu/. 
If a .netrc file is not provide under your home directory, you will be prompt to enter your ASF username and password. 
Check documentation for how to setup .netrc file.\n""")
        super().__init__(config)

        if self.config.dataset is None and self.config.platform is None and not getattr(self.config, 'granule_names', None):
            raise ValueError(f"{Fore.RED}Dataset or platform must be specified for ASF search (or provide granule_names).")

        self.config.intersectsWith = _to_wkt(self.config.intersectsWith)


    def _asf_authorize(self):
        self._has_asf_netrc = self._check_netrc(keyword='machine urs.earthdata.nasa.gov')
        if not self._has_asf_netrc:
            while True:
                _username = input("Enter your ASF username: ")
                _password = getpass.getpass("Enter your ASF password: ")
                try:
                    self._session = asf.ASFSession().auth_with_creds(_username, _password)
                    self._session.verify = self.config.ssl_verify
                except ASFAuthenticationError:
                    print(f"{Fore.RED}Authentication failed. Please check your credentials and try again.\n")
                    continue
                print(f"{Fore.GREEN}Authentication successful.\n")
                netrc_path = Path.home().joinpath(".netrc")
                asf_entry = f"\nmachine urs.earthdata.nasa.gov\n    login {_username}\n    password {_password}\n"
                with open(netrc_path, 'a') as f:
                    f.write(asf_entry)
                print(f"{Fore.GREEN}Credentials saved to {netrc_path}. You can now use the downloader without entering credentials again.\n")
                break
        else:
            self._session = asf.ASFSession()
            self._session.verify = self.config.ssl_verify

    def _check_netrc(self, keyword: str) -> bool:
        """Check if .netrc file exists in the home directory with the specified keyword.

        Args:
            keyword (str): The machine name to search for in .netrc file.

        Returns:
            bool: True if .netrc file exists and contains the keyword, False otherwise.
        """
        netrc_path = Path.home().joinpath('.netrc')
        if not netrc_path.is_file():            
            print(f"{Fore.RED}No .netrc file found in your home directory. Will prompt login.\n")
            return False
        else: 
            with netrc_path.open() as f:
                content = f.read()
                if keyword in content:
                    return True
                else:
                    print(f"{Fore.RED}no machine name {keyword} found .netrc file. Will prompt login.\n")
                    return False


    def _stack_key_label_title(self) -> str:
        """:attr:`stack_key_label` with only its first character upper-cased.

        ``str.capitalize()`` would lower-case the rest and turn ``"Burst_ID"``
        into ``"Burst_id"``.
        """
        label = self.stack_key_label
        return label[:1].upper() + label[1:]

    def _stack_key_matches(self, key: tuple, target: tuple) -> bool:
        """Does stack ``key`` satisfy a user-supplied ``(path, selector)`` target?

        The hook behind :meth:`filter`'s ``path_frame`` argument (and so behind
        the CLI's ``--stacks``). The base rule is plain equality, which is right
        wherever both halves of the key are numbers. Subclasses whose second half
        is not a frame number -- :class:`~insarhub.downloader.s1_burst.S1_Burst`
        keys on ``fullBurstID`` -- override this to define what a selector means
        for them, rather than every caller having to know the key shape.
        """
        return tuple(key) == tuple(target)

    def _get_group_key(self, result) -> tuple:
        """Derive grouping key based on available properties, with fallback.

        Args:
            result: Search result object containing properties.

        Returns:
            tuple: A tuple of (path_number, frame_identifier) used for grouping results.
        """
        props = result.properties
        # Burst product — any burst ID field set in config takes highest priority
        if any([
            self.config.absoluteBurstID,
            self.config.fullBurstID,
            self.config.operaBurstID,
            self.config.relativeBurstID,
        ]):
            return (props.get('pathNumber'), props.get('burstID'))

        if self.config.asfFrame is not None:
            # 'asfFrame' is a search filter parameter, not a scene property name.
            # Use 'frameNumber' (the actual returned property) for consistent grouping.
            return (props.get('pathNumber'), props.get('frameNumber'))

        if self.config.frame is not None:
            return (props.get('pathNumber'), props.get('frameNumber'))

        # Dataset-level mapping
        if self.config.dataset:
            datasets = [self.config.dataset] if isinstance(self.config.dataset, str) else self.config.dataset
            for ds in datasets:
                ds_upper = ds.upper()
                if ds_upper in self._DATASET_GROUP_KEYS:
                    pk, fk = self._DATASET_GROUP_KEYS[ds_upper]
                    return (props.get(pk), props.get(fk))
        # Platform-level fallback mapping      
        if self.config.platform:
            platforms = [self.config.platform] if isinstance(self.config.platform, str) else self.config.platform
            for pl in platforms:
                pl_upper = pl.upper()
                if 'SENTINEL' in pl_upper:
                    return (props.get('pathNumber'), props.get('frameNumber'))
                if 'ALOS' in pl_upper:
                    return (props.get('pathNumber'), props.get('frameNumber'))
                if 'NISAR' in pl_upper:
                    return (props.get('pathNumber'), props.get('frameNumber'))
        # last resort — group everything under the platform name
        return (props.get('pathNumber'), props.get('frameNumber'))

    # Platforms whose user-facing frame is the ASF frame (CMR FRAME_NUMBER == the
    # 'frameNumber' property we group by), rather than the ESA frame. See
    # _uses_asf_frame() / the asf_search 13.0.0 compatibility note in search().
    _ASF_FRAME_TOKENS = ('SENTINEL-1', 'SENTINEL1', 'ALOS', 'NISAR', 'SEASAT')

    def _uses_asf_frame(self) -> bool:
        """True when this dataset/platform numbers frames by the ASF frame.

        For Sentinel-1 / ALOS / NISAR / SEASAT the frame a user specifies is the ASF
        frame (the ``frameNumber`` property InSARHub groups by), which asf_search
        queries via ``asfFrame`` (CMR ``FRAME_NUMBER``) -- not ``frame`` (CMR
        ``CENTER_ESA_FRAME``). See :meth:`_apply_asf_frame_compat`.
        """
        vals: list = []
        for attr in ('dataset', 'platform'):
            v = getattr(self.config, attr, None)
            if v is None:
                continue
            vals.extend(v if isinstance(v, (list, tuple)) else [v])
        blob = ' '.join(str(x).upper() for x in vals)
        return any(tok in blob for tok in self._ASF_FRAME_TOKENS)

    def _apply_asf_frame_compat(self, search_opts: dict) -> dict:
        """Route ``frame`` -> ``asfFrame`` for ASF-frame platforms (asf_search 13.0.0).

        For Sentinel-1 / ALOS / NISAR the frame a user gives is the ASF frame (== the
        ``frameNumber`` InSARHub groups by). asf_search maps ``frame`` to CMR
        ``CENTER_ESA_FRAME`` and only rewrote it to ``FRAME_NUMBER`` for these platforms
        via ``should_use_asf_frame()``. asf_search **13.0.0** broke that check for a
        generic ``platform=SENTINEL-1`` query: it now tests for a ``shortName[]`` CMR key,
        but the query emits ``shortName`` (no brackets), and its ``platform[]`` fallback
        only lists the per-satellite names (``SENTINEL-1A/-1B/-1C/-1D``), not generic
        ``SENTINEL-1`` -- so neither branch matches, the ``CENTER_ESA_FRAME`` ->
        ``FRAME_NUMBER`` rewrite never fires, and ``frame=`` silently matches nothing.
        ``asfFrame`` maps straight to ``FRAME_NUMBER`` (bypassing that broken check) and
        works on both 12.x and 13.x, so route the frame filter there. We keep this
        workaround inside InSARHub rather than depend on an upstream asf_search fix.
        Mutates and returns ``search_opts``.
        """
        if search_opts.get('frame') is not None and self._uses_asf_frame():
            if search_opts.get('asfFrame') is None:
                search_opts['asfFrame'] = search_opts.pop('frame')
                print(f"{Fore.YELLOW}Note: querying ASF frame via 'asfFrame' for "
                      f"asf_search {getattr(asf, '__version__', '?')} compatibility "
                      f"('frame' targets the ESA frame and no longer matches for this "
                      f"platform).{Fore.RESET}")
            else:
                # Both set: asfFrame is authoritative for these platforms; drop the
                # ambiguous ESA-frame filter so it can't zero out the query.
                search_opts.pop('frame')
        return search_opts

    def _get_property_keys(self) -> dict:
        """Return the correct result.properties key mapping based on config.

        Returns:
            dict: Mapping of property names to their corresponding keys in search results.
        """
        if self.config.dataset:
            datasets = [self.config.dataset] if isinstance(self.config.dataset, str) else self.config.dataset
            for ds in datasets:
                ds_upper = ds.upper()
                if ds_upper in self._DATASET_PROPERTY_KEYS:
                    return self._DATASET_PROPERTY_KEYS[ds_upper]

        if self.config.platform:
            platforms = [self.config.platform] if isinstance(self.config.platform, str) else self.config.platform
            for pl in platforms:
                if 'SENTINEL' in pl.upper():
                    return self._DATASET_PROPERTY_KEYS['SENTINEL-1']
                if 'ALOS' in pl.upper():
                    return self._DATASET_PROPERTY_KEYS['ALOS']
                if 'NISAR' in pl.upper():
                    return self._DATASET_PROPERTY_KEYS['NISAR']

        # Default to Sentinel-1 keys as they are most common
        return self._DATASET_PROPERTY_KEYS['SENTINEL-1']

    @property
    def session(self):
        """Get or create an authenticated ASF session.

        Returns:
            ASFSession: Authenticated session for ASF downloads.
        """
        if not hasattr(self, '_session') or self._session is None:
            self._asf_authorize()
        return self._session

    @property
    def active_results(self):
        """Get the currently active results (filtered or full search results).

        Returns the subset of results if a filter/pick is active, 
        otherwise returns the full search results.

        Returns:
            dict: Dictionary of active search results grouped by (path, frame).

        Raises:
            ValueError: If no search results are available.
        """
        if not hasattr(self, 'results'):
             raise ValueError(f"{Fore.RED}No search results found. Please run search() first.")
        return self._subset if self._subset is not None else self.results

    def search(self) -> dict:
        """Search for data using the ASF Search API with the provided parameters.

        When ``config.granule_names`` is set the search is performed by granule
        name instead of the normal parameter search.  ``granule_names`` may be:

        * A ``list[str]`` of scene/granule names (with or without extensions).
        * A ``str`` containing a single name, a comma-separated list of names,
          or a path to a CSV / XLSX / TXT file on disk.

        Returns:
            dict: Dictionary of search results grouped by (path, frame) tuples.

        Raises:
            ValueError: If search returns no results.
            Exception: If search fails after 10 retry attempts.
        """
        self._subset = None

        granule_names = getattr(self.config, 'granule_names', None)
        if granule_names:
            print(f"{Fore.GREEN}Granule_names provided, Performing search by granule name(s) from {self.config.granule_names}...{Fore.RESET}")
            from insarhub.utils.tool import parse_scene_names_from_file
            raw_inputs = granule_names if isinstance(granule_names, list) else [n.strip() for n in granule_names.split(',') if n.strip()]
            names: list[str] = []
            for item in raw_inputs:
                p = Path(item)
                if p.exists():
                    names.extend(parse_scene_names_from_file(str(p)))
                else:
                    names.append(item)
            return self._search_by_name(names)

        print(f"Searching for {self.product_label}....")
        search_opts = {k: v for k, v in asdict(self.config).items()
                       if v is not None and k not in self._NON_SEARCH_FIELDS}
        if 'end' in search_opts and isinstance(search_opts['end'], str):
            search_opts['end'] = _end_of_day(search_opts['end'])

        search_opts = self._apply_asf_frame_compat(search_opts)

        if os.environ.get("INSARHUB_DEBUG_SEARCH"):
            print(f"[debug] asf.search opts: {search_opts}")

        for attempt in range(1, 11):
            try:
                self.results = asf.search(**search_opts)
                break
            except Exception as e:
                print(f"{Fore.RED}Search failed: {e}")
                if os.environ.get("INSARHUB_DEBUG_SEARCH"):
                    import traceback; traceback.print_exc()
                if attempt == 10:
                    raise
                time.sleep(2 ** attempt)

        if not self.results:
            raise ValueError(f'{Fore.RED}Search does not return any result, please check input parameters or Internet connection')
        else:
            print(f"{Fore.GREEN} -- A total of {len(self.results)} results found. \n")

        grouped = defaultdict(list)
        for result in self.results:
            key = self._get_group_key(result)
            grouped[key].append(result)
        self.results = grouped
        if len(grouped) > 1:
            print(f"{Fore.YELLOW}The AOI crosses {len(grouped)} stacks")
        return grouped

    def _search_by_name(self, scene_names: list[str]) -> dict:
        """Populate results from a list of scene/granule names or filenames.

        Accepts names with or without file extensions (e.g. ``.zip``).
        Uses ``asf_search.granule_search()`` so no config parameters are needed
        and works for any ASF-supported dataset (S1 SLC, S1 Burst, ALOS, etc.).

        Args:
            scene_names: Scene or filename strings, e.g.
                ``["S1A_IW_SLC__1SDV_20201227T133500_..._5DB4",
                   "S1A_IW_SLC__1SDV_20201227T133500_..._5DB4.zip"]``

        Returns:
            Grouped results dict keyed by ``(relativeOrbit, frame)``.
        """
        # Strip common file extensions so granule_search can find them
        clean = [Path(n).stem if '.' in n else n for n in scene_names]
        raw = asf.granule_search(clean)
        if not raw:
            raise ValueError(f"No ASF results found for the {len(clean)} provided scene name(s).")

        # granule_search returns all product types per granule (SLC + METADATA_SLC, etc.).
        # Exclude metadata-only products, then deduplicate by sceneName.
        _EXCLUDE_LEVELS = {'METADATA_SLC', 'METADATA'}
        seen: set[str] = set()
        deduped = []
        for result in raw:
            if result.properties.get('processingLevel', '') in _EXCLUDE_LEVELS:
                continue
            sname = result.properties.get('sceneName', '')
            if sname not in seen:
                seen.add(sname)
                deduped.append(result)

        grouped: dict = defaultdict(list)
        for result in deduped:
            key = self._get_group_key(result)
            grouped[key].append(result)
        self.results = grouped
        print(f"{Fore.GREEN} -- Found {len(deduped)} scenes across {len(grouped)} stack(s).\n")
        if len(deduped) < len(clean):
            missing = len(clean) - len(deduped)
            print(f"{Fore.YELLOW} -- {missing} scene(s) not found on ASF.\n")
        return grouped

    def reset(self):
        """Reset the view to include all search results.

        Clears any active filters and restores the full result set.
        """
        self._subset = None
        print(f"{Fore.GREEN}Selection reset. Now viewing all {len(self.results)} stacks.")

    def summary(self, ls=False):
        """Summarize the active results, separated by flight direction.

        Args:
            ls (bool, optional): If True, list individual scene names and dates. 
                Defaults to False.
        """
        if not hasattr(self, 'results'):
            self.search()

        active_results = self.active_results

        if not active_results:
            print(f"{Fore.YELLOW}No results to summarize.")
            return

        ascending_stacks = {}
        descending_stacks = {}

        for key, items in active_results.items():
            if not items: continue
            direction = items[0].properties.get('flightDirection', 'UNKNOWN').upper()

            if direction == 'ASCENDING':
                ascending_stacks[key] = items
            elif direction == 'DESCENDING':
                descending_stacks[key] = items

        def _print_group(label, data_dict, color_code):
            if not data_dict:
                return
            print(f"\n{color_code}=== {label} ORBITS ({len(data_dict)} Stacks) ==={Fore.RESET}")
            sorted_keys = sorted(data_dict.keys())

            for key in sorted_keys:
                    items = data_dict[key]
                    count = len(items)

                    # Calculate time range
                    dates = [isoparse(i.properties['startTime']) for i in items]
                    start_date = min(dates).date()
                    end_date = max(dates).date()

                    print(f"relativeOrbit {key[0]} {self.stack_key_label} {key[1]} | Count: {count} | {start_date} --> {end_date}")

                    if ls:
                        # Sort scenes by date
                        items_sorted = sorted(items, key=lambda x: isoparse(x.properties['startTime']))
                        for scene in items_sorted:
                            scene_date = isoparse(scene.properties['startTime']).date()
                            print(f"    {Fore.LIGHTBLACK_EX}{scene.properties['sceneName']} ({scene_date}){Fore.RESET}")
        if ascending_stacks:
            _print_group("ASCENDING", ascending_stacks, Fore.MAGENTA)

        if descending_stacks:
            _print_group("DESCENDING", descending_stacks, Fore.CYAN)

        print("") # Final newline


    def footprint(self, save_path: str | None = None):
        """Display or save the search result footprints and AOI using matplotlib.

        Args:
            save_path (str, optional): Path to save the figure. If None, displays interactively.
                Defaults to None.
        """
        results_to_plot = self.active_results
        if not results_to_plot:
            print(f"{Fore.RED}No results to plot.")
            return

        transformer = Transformer.from_crs("EPSG:4326", "EPSG:3857", always_xy=True)
        N = len(results_to_plot)
        cmap = matplotlib.colormaps['hsv'].resampled(N+1)

        fig, ax = plt.subplots(1, 1, figsize=(10,10), dpi=150)

        geom_aoi = transform(transformer.transform, wkt.loads(self.config.intersectsWith))
        global_minx, global_miny, global_maxx, global_maxy = geom_aoi.bounds
        plotting.plot_polygon(geom_aoi, ax=ax, edgecolor='red', facecolor='none', linewidth=2, linestyle='--')

        label_x_aoi = global_maxx - 0.01 * (global_maxx - global_minx)
        label_y_aoi = global_maxy - 0.01 * (global_maxy - global_miny)
        plt.text(label_x_aoi, label_y_aoi,
             f"AOI",
             horizontalalignment='right', verticalalignment='top',
             fontsize=12, color='red', fontweight='bold',
             bbox=dict(facecolor='white', alpha=0.7, edgecolor='none', boxstyle='round,pad=0.3'))

        for i, (key, results) in enumerate(results_to_plot.items()):
            geom = transform(transformer.transform, shape(results[0].geometry))
            minx, miny, maxx, maxy = geom.bounds

            global_minx = min(global_minx, minx)
            global_miny = min(global_miny, miny)
            global_maxx = max(global_maxx, maxx)
            global_maxy = max(global_maxy, maxy)

            label_x = maxx - 0.01 * (maxx - minx)
            label_y = maxy - 0.01 * (maxy - miny)

            plt.text(label_x, label_y,
             f"Path: {key[0]}\n{self._stack_key_label_title()}: {key[1]}\nStack: {len(results)}",
             horizontalalignment='right', verticalalignment='top',
             fontsize=12, color=cmap(i), fontweight='bold',
             bbox=dict(facecolor='white', alpha=0.7, edgecolor='none', boxstyle='round,pad=0.3'))

            for result in results:
                geom = transform(transformer.transform, shape(result.geometry))
                x, y = geom.exterior.xy
                ax.plot(x, y, color=cmap(i))

        ctx.add_basemap(ax, source=ctx.providers.OpenStreetMap.Mapnik, headers=_OSM_TILE_HEADERS)

        ax.set_xlim(global_minx, global_maxx)
        ax.set_ylim(global_miny, global_maxy)

        ax.set_axis_off()
        if save_path is not None:
            save_path = Path(save_path).expanduser().resolve()
            plt.savefig(save_path.as_posix(), dpi=300, bbox_inches='tight')
            print(f"Footprint figure saved to {save_path}")
        else:
            plt.subplots_adjust(top = 1, bottom = 0, right = 1, left = 0, hspace = 0, wspace = 0)
            plt.show()

    def filter(self, 
                path_frame : tuple | list[tuple] | None = None,
                start: str | None = None,
                end: str | None = None,
                frame: int | list[int] | None = None, 
                asfFrame: int | list[int] | None = None, 
                flightDirection: str | None = None,
                relativeOrbit: int | list[int] | None = None,
                absoluteOrbit: int | list[int] | None = None,
                lookDirection: str | None = None,
                polarization: str | list[str] | None = None,
                processingLevel: str | None = None,
                beamMode: str | None = None,
                season: list[int] | None = None,
                min_coverage: float | None = None,
                min_count: int | None = None,
                max_count: int | None = None,
                latest_n: int | None = None,
                earliest_n: int | None = None
               ) -> dict:
        """Filter active results by various properties after search.

        Args:
            path_frame (tuple | list[tuple], optional): A single (path, frame) tuple or list of tuples.
                Defaults to None.
            start (str, optional): Start date string, e.g. '2021-01-01'. Defaults to None.
            end (str, optional): End date string, e.g. '2023-12-31'. Defaults to None.
            frame (int | list[int], optional): Sensor native frame number(s), e.g. 50. Defaults to None.
            asfFrame (int | list[int], optional): ASF internal frame number(s), e.g. 50. Defaults to None.
            flightDirection (str, optional): 'ASCENDING' or 'DESCENDING'. Defaults to None.
            relativeOrbit (int | list[int], optional): Relative orbit number(s) to keep. Defaults to None.
            absoluteOrbit (int | list[int], optional): Absolute orbit number(s) to keep. Defaults to None.
            lookDirection (str, optional): 'LEFT' or 'RIGHT'. Defaults to None.
            polarization (str | list[str], optional): Polarization(s) to keep, e.g. 'VV' or ['VV', 'VH']. 
                Defaults to None.
            processingLevel (str, optional): Processing level to keep, e.g. 'SLC'. Defaults to None.
            beamMode (str, optional): Beam mode to keep, e.g. 'IW'. Defaults to None.
            season (list[int], optional): List of months (1-12) to keep, e.g. [6, 7, 8] for summer. 
                Defaults to None.
            min_coverage (float, optional): Minimum fractional overlap (0-1) between scene and AOI. 
                Defaults to None.
            min_count (int, optional): Drop stacks with fewer than this many scenes after filtering. 
                Defaults to None.
            max_count (int, optional): Keep at most this many scenes per stack (from earliest). 
                Defaults to None.
            latest_n (int, optional): Keep the N most recent scenes per stack. Defaults to None.
            earliest_n (int, optional): Keep the N earliest scenes per stack. Defaults to None.

        Returns:
            dict: Filtered results grouped by (path, frame).

        Raises:
            ValueError: If no search results are available.
        """

        if not hasattr(self, 'results'):
            raise ValueError(f"{Fore.RED}No search results found. Please run search() first.")

        source = self.active_results
        filtered = defaultdict(list)
        prop_keys = self._get_property_keys()

        # --- Pre-process filter values ---
        if path_frame is not None:
            targets = {path_frame} if isinstance(path_frame, tuple) else set(path_frame)
        else:
            targets = None

        start_dt = isoparse(start).replace(tzinfo=None)             if start else None
        end_dt   = isoparse(_end_of_day(end)).replace(tzinfo=None)  if end   else None
        frames     = {frame}    if isinstance(frame, int)    else set(frame)    if frame    else None
        asf_frames = {asfFrame} if isinstance(asfFrame, int) else set(asfFrame) if asfFrame else None
        relative_orbits  = {relativeOrbit}  if isinstance(relativeOrbit, int)  else set(relativeOrbit)  if relativeOrbit  else None
        absolute_orbits  = {absoluteOrbit}  if isinstance(absoluteOrbit, int)  else set(absoluteOrbit)  if absoluteOrbit  else None
        polarizations    = {polarization}   if isinstance(polarization, str)   else set(polarization)   if polarization   else None
        season_months    = set(season) if season else None

        if min_coverage is not None:
            aoi_geom = wkt.loads(self.config.intersectsWith)

        for key, items in source.items():
            if targets is not None and not any(self._stack_key_matches(key, t) for t in targets):
                continue

            if flightDirection:
                stack_dir = items[0].properties.get('flightDirection', '').upper()
                if stack_dir != flightDirection.upper():
                    continue

            if lookDirection:
                stack_look = items[0].properties.get('lookDirection', '').upper()
                if stack_look != lookDirection.upper():
                    continue

            if beamMode:
                stack_beam = items[0].properties.get('beamMode', '').upper()
                if stack_beam != beamMode.upper():
                    continue

            if processingLevel:
                stack_proc = items[0].properties.get('processingLevel', '').upper()
                if stack_proc != processingLevel.upper():
                    continue
        # --- Scene-level filters ---
            filtered_items = []
            for item in items:
                props = item.properties

                scene_dt = isoparse(props['startTime']).replace(tzinfo=None)
                # Date range
                if start_dt and scene_dt < start_dt:
                    continue
                if end_dt and scene_dt > end_dt:
                    continue

                # Native frame filter
                if frames is not None:
                    if props.get('frameNumber') not in frames:
                        continue

                # ASF frame filter
                if asf_frames is not None:
                    if props.get('asfFrame') not in asf_frames:
                        continue
                # Season (month filter)
                if season_months and scene_dt.month not in season_months:
                    continue

                # Relative orbit
                if relative_orbits and props.get(prop_keys['relativeOrbit']) not in relative_orbits:
                    continue

                # Absolute orbit
                if absolute_orbits and props.get(prop_keys['absoluteOrbit']) not in absolute_orbits:
                    continue

                # Polarization — props value may be a string like 'VV+VH'
                if polarizations:
                    scene_pols = set(props.get(prop_keys['polarization'], '').replace('+', ' ').split())
                    if not polarizations.intersection(scene_pols):
                        continue

                if min_coverage is not None:
                    scene_geom = shape(item.geometry)
                    intersection = aoi_geom.intersection(scene_geom)
                    coverage = intersection.area / aoi_geom.area
                    if coverage < min_coverage:
                        continue

                filtered_items.append(item)
            if not filtered_items:
                continue


            filtered_items = sorted(filtered_items, key=lambda x: isoparse(x.properties['startTime']))

            if earliest_n is not None:
                filtered_items = filtered_items[:earliest_n]
            elif latest_n is not None:
                filtered_items = filtered_items[-latest_n:]
            elif max_count is not None:
                filtered_items = filtered_items[:max_count]

            if min_count is not None and len(filtered_items) < min_count:
                print(f"{Fore.YELLOW}Stack Path {key[0]} {self._stack_key_label_title()} {key[1]} dropped: only {len(filtered_items)} scenes (min_count={min_count}).")
                continue

            filtered[key] = filtered_items

        # Commit the subset even when it is EMPTY. Leaving _subset as None on a
        # total miss made active_results silently fall back to the unfiltered
        # search (see the property above), so a filter that matched nothing read
        # downstream as a filter that was never asked for -- callers then
        # summarised, paired and downloaded every stack the user had just
        # excluded, behind one yellow warning line.
        self._subset = dict(filtered)

        if not filtered:
            print(f"{Fore.YELLOW}Warning: No results matched the given filters.")
        else:
            total_scenes = sum(len(v) for v in filtered.values())
            print(f"{Fore.GREEN}Filter applied. {len(filtered)} stacks, {total_scenes} total scenes remaining.")

        return filtered

    def dem(self, save_path: str | None = None):
        """Download DEM for co-registration uses.

        Args:
            save_path (str, optional): Directory to save DEM files. If None, uses config.workdir.
                Defaults to None.

        Returns:
            tuple: (X, p) where X is the DEM array and p is the rasterio profile.
        """
        output_dir = Path(save_path).expanduser().resolve() if save_path else self.config.workdir
        _dem_is_stack = (output_dir / "insarhub_config.json").exists()

        for key, results in self.active_results.items():
            _dem_sub = Path() if _dem_is_stack else Path(f'p{key[0]}_f{key[1]}')
            download_path = output_dir.joinpath('dem', _dem_sub)
            download_path.mkdir(exist_ok=True, parents=True)
            geom = shape(results[0].geometry)
            west_lon, south_lat, east_lon, north_lat =  geom.bounds
            bbox = [ west_lon, south_lat, east_lon, north_lat]
            X, p = dem_stitcher.stitch_dem(
                bbox, 
                dem_name='glo_30',
                dst_area_or_point='Point',
                dst_ellipsoidal_height=True
            )

            with rio.open(download_path.joinpath(f'dem_p{key[0]}_f{key[1]}.tif'), 'w', **p) as ds:
                    ds.write(X,1)
                    ds.update_tags(AREA_OR_POINT='Point')
        return X, p

    #: False for products ASF publishes no baseline stack for. ``ASFProduct``
    #: ``.stack()`` needs a baseline-stack reference that NISAR granules do not
    #: carry, so no perpendicular baseline can be derived for them at all.
    has_perpendicular_baseline: bool = True

    def _warn_if_pairs_will_not_be_used(self) -> None:
        """Say so up front when a pair network cannot be built or will be ignored.

        Both cases end with the user's ``--select-pairs`` having no effect on
        processing, but for different reasons, so they are reported separately:

        * No perpendicular baseline (NISAR): ASF publishes no baseline stack for
          these granules, so there is nothing to prune pairs by. The pair graph
          cannot be built the way it is for Sentinel-1.

        * The consuming processor builds its own network (ISCE3_Burst,
          ISCE3_NISAR): dolphin forms interferograms from the phase-linked SLCs
          using an index/temporal network (``max_bandwidth``, default 3, or
          ``max_temporal_baseline``) and has no perpendicular-baseline option.
          Those processors accept a ``pairs`` argument and never read it, so a
          selected network is silently discarded.

        Warn rather than raise: writing ``stack_*.json`` and ``network_*.png``
        is still useful for inspecting coverage, it just does not drive
        processing.
        """
        from insarhub.core.registry import Processor

        name = type(self).name

        if not getattr(type(self), "has_perpendicular_baseline", True):
            print(
                f"{Fore.YELLOW}Note: ASF publishes no baseline stack for "
                f"{name} products, so no perpendicular baseline can be "
                f"derived and NO pair graph is produced -- select_pairs "
                f"returns an empty network, not a temporal-only one. "
                f"(Sentinel-1 bursts are unaffected: they get bperp from their "
                f"own orbits, so their pairs are built normally.){Fore.RESET}"
            )

        consumers = sorted(
            pname for pname, pcls in Processor._registry.items()
            if getattr(pcls, "compatible_downloader", None) == name
            and getattr(pcls, "builds_own_network", False)
        )
        if consumers:
            glob = getattr(Processor._registry[consumers[0]], "input_glob", "*")
            print(
                f"{Fore.YELLOW}Note: {', '.join(consumers)} build their own "
                f"interferogram network from slc/{glob} -- dolphin phase "
                f"linking, then an index/temporal network, with no "
                f"perpendicular-baseline criterion -- so a selected pair list "
                f"is NOT used for processing. Shape the network with the "
                f"processor's pl_ifg_network / n_connections / "
                f"max_temporal_baseline instead. The stack file and network "
                f"plot are still written, for inspection.{Fore.RESET}"
            )

    def select_pairs(
        self,
        dt_targets: tuple        = None,
        dt_tol: int              = None,
        dt_max: int              = None,
        pb_max: float            = None,
        min_degree: int          = None,
        max_degree: int          = None,
        force_connect: bool      = None,
        max_workers: int         = None,
        aoi_wkt: str | None = None,
        merge: bool = False,
        burst: bool = False,
        safe_dir: str | None = None,
        eof_dir: str | None = None,
        poeorb_cache: str | None = None,
        quality_check: bool = True,
        plot_network: bool = True,
    ) -> tuple:
        """Compute interferogram pairs for all active stacks.

        Args:
            dt_targets (tuple, optional): Target temporal spacings in days. Defaults to (6, 12, 24, 36, 48, 72, 96).
            dt_tol (int, optional): Tolerance in days around each target spacing. Defaults to 3.
            dt_max (int, optional): Maximum temporal baseline in days. Defaults to 120.
            pb_max (float, optional): Maximum perpendicular baseline in meters. Defaults to 150.0.
            min_degree (int, optional): Minimum number of connections per scene. Defaults to 3.
            max_degree (int, optional): Maximum number of connections per scene. Defaults to 5.
            force_connect (bool, optional): Force connectivity for isolated scenes. Defaults to True.
            max_workers (int, optional): Threads for API baseline fallback. Defaults to 4.
            aoi_wkt (str, optional): AOI geometry in WKT for quality scoring. Defaults to search AOI.
            merge (bool, optional): When True, stacks sharing the same relative
                orbit (path) are combined into one pairing network before
                temporal/baseline selection — matching how ISCE2's stackSentinel
                treats multiple frames of one track/pass as a single continuous
                acquisition. Stacks on different paths are never combined
                (cross-track pairs have no physical baseline). Use together with
                ``download(merge=True)``, which puts all scenes in one
                ``merged/slc/`` directory. Defaults to False.
            burst (bool, optional): Select pairs for an SLC-BURST stack. Nodes
                become acquisition dates and baselines are computed from the
                bursts' own startTime/orbits — no parent-SLC lookup. Defaults
                to False.
            safe_dir (str, optional): Burst mode: directory of assembled
                ``.SAFE`` dirs whose annotation orbits supply bperp (offline).
            eof_dir (str, optional): Burst mode: directory of precise-orbit
                ``.EOF`` files used for bperp (offline).
            poeorb_cache (str, optional): Burst mode: directory for POEORB
                downloads keyed by date + mission (online fallback).
            quality_check (bool, optional): After pairing, write the stack
                file(s) and build the PairQualityDB verdict for every possible
                pair (the slow, network-heavy step). ``False`` skips it.
                Defaults to True.
            plot_network (bool, optional): Save ``network_*.png`` with the
                pair network, coloured healthy/concern when ``quality_check`` is
                True (falls back to temporal-baseline colouring otherwise).
                Defaults to True.

        Returns:
            tuple: ``(pairs, baselines, scene_bperp, pair_status,
            quality_factors)``
                - pairs: dict keyed by (path, frame) for multi-stack — or
                  (path, "merged") per distinct path when merge=True — or a
                  flat list for a single stack.
                - baselines: temporal baselines
                - scene_bperp: perpendicular baselines per scene
                - pair_status: ``{pair_key: "healthy"|"concern"}`` (list case)
                  or ``{(path, frame): {pair_key: status}}`` (dict case);
                  ``None`` when ``quality_check`` is False or it failed.
                - quality_factors: the events behind each verdict, same keying.
        """
        self._warn_if_pairs_will_not_be_used()

        # None → pull from the single source of truth
        from insarhub.utils.defaults import SELECT_PAIRS_DEFAULTS as _SP
        if dt_targets             is None: dt_targets             = _SP["dt_targets"]
        if dt_tol                 is None: dt_tol                 = _SP["dt_tol"]
        if dt_max                 is None: dt_max                 = _SP["dt_max"]
        if pb_max                 is None: pb_max                 = _SP["pb_max"]
        if min_degree             is None: min_degree             = _SP["min_degree"]
        if max_degree             is None: max_degree             = _SP["max_degree"]
        if force_connect          is None: force_connect          = _SP["force_connect"]
        if max_workers            is None: max_workers            = _SP["max_workers"]
        from insarhub.utils.tool import select_pairs as _select_pairs

        if not hasattr(self, 'results'):
            raise ValueError("No search results found. Please run search() first.")

        search_input = self.active_results
        if merge and isinstance(search_input, dict) and len(search_input) > 1:
            # Group stacks by relative orbit (path) — frames of the same
            # path/pass overlap and can be validly combined; different paths
            # never share a baseline and must never be merged together.
            paths = {path for (path, _frame) in search_input.keys()}
            if len(paths) > 1:
                raise ValueError(
                    f"merge=True requires all stacks to share one relative orbit "
                    f"(path), got {sorted(paths)}. Different tracks have unrelated "
                    f"viewing geometry and cannot be interferometrically paired — "
                    f"narrow the search to a single path before merging."
                )
            path = next(iter(paths))
            frames = [frame for (_path, frame) in search_input.keys()]
            merged_prods = [p for prods in search_input.values() for p in prods]
            tag = StackPaths.merge_tag(frames)
            search_input = {(path, tag): merged_prods}
            print(f"{Fore.CYAN}merge=True: combined frame(s) {sorted(frames)} "
                  f"of path {path} ({len(merged_prods)} scenes) into one stack "
                  f"({tag}).\n")

        _aoi_wkt = aoi_wkt or getattr(self.config, "intersectsWith", None)
        _sp_result = _select_pairs(
            search_input,
            dt_targets=dt_targets,
            dt_tol=dt_tol,
            dt_max=dt_max,
            pb_max=pb_max,
            min_degree=min_degree,
            max_degree=max_degree,
            force_connect=force_connect,
            max_workers=max_workers,
            aoi_wkt=_aoi_wkt,
            burst=burst,
            safe_dir=safe_dir,
            eof_dir=eof_dir,
            poeorb_cache=poeorb_cache,
        )
        pairs      = _sp_result[0]
        baselines  = _sp_result[1]
        scene_bperp: dict = _sp_result[2] if len(_sp_result) > 2 else {}

        if not (quality_check or plot_network):
            return pairs, baselines, scene_bperp, None, None

        # ── Finalize: write stack files, score pairs, plot network ────────
        from dataclasses import asdict
        from insarhub.utils.config_io import write_insarhub_config
        from insarhub.utils.stack_io import finalize_stack
        from insarhub.utils.tool import group_scenes_by_stack, write_workflow_marker

        scenes_by_stack = group_scenes_by_stack(self.active_results, merge=merge)
        workdir = Path(self.config.workdir).expanduser()
        workdir.mkdir(parents=True, exist_ok=True)
        _sp = StackPaths(workdir)
        _dl_is_stack = (workdir / "insarhub_config.json").exists()

        quality_scores: dict | None
        quality_factors: dict | None
        if isinstance(pairs, dict):
            quality_scores = {}
            quality_factors = {}
            for (path, frame), group_pairs in pairs.items():
                is_merged = StackPaths.is_merge_key(frame)
                label = f"P{path} ({frame})" if is_merged else f"P{path}/F{frame}"
                tag = _sp.dir_for(path, frame).name
                subdir = workdir if _dl_is_stack else workdir / tag
                subdir.mkdir(parents=True, exist_ok=True)
                write_workflow_marker(subdir, downloader=type(self).name)
                cfg = {k: v for k, v in asdict(self.config).items() if k != "workdir"}
                cfg["relativeOrbit"] = path
                if not is_merged:
                    cfg["frame"] = frame
                # Record the AOI when the user did not draw one. Without it
                # _load_aoi() has nothing to read and pair quality has no
                # location to fetch weather or coherence for.
                if not cfg.get("intersectsWith"):
                    from insarhub.utils.pair_quality._geom import footprint_wkt_from_products
                    if is_merged:
                        _prods = [p for (k_path, _k_frame), prods in self.active_results.items()
                                  if k_path == path for p in prods]
                    else:
                        _prods = self.active_results.get((path, frame), [])
                    _wkt = footprint_wkt_from_products(_prods)
                    if _wkt:
                        cfg["scene_footprint_wkt"] = _wkt
                write_insarhub_config(subdir, {"downloader": {"type": type(self).name, "config": cfg}})
                sp = scene_bperp.get((path, frame)) or {}
                stack_scenes = scenes_by_stack.get((path, frame), [])
                stack_path = subdir / _sp.stack_file_for(path, frame).name
                qs, qf = finalize_stack(
                    subdir, stack_path, group_pairs, sp, stack_scenes,
                    key=(path, frame),
                    baselines=baselines[(path, frame)],
                    title=f"Interferogram Network — {label}",
                    save_path=subdir / f"network_{tag}.png",
                    quality_check=quality_check,
                    plot_network=plot_network,
                )
                if qs is not None:
                    quality_scores[(path, frame)] = qs
                    quality_factors[(path, frame)] = qf
        else:
            sp = scene_bperp if isinstance(scene_bperp, dict) else {}
            stack_scenes = scenes_by_stack.get((0, 0), [])
            stack_path = workdir / _sp.stack_file(0, 0).name
            quality_scores, quality_factors = finalize_stack(
                workdir, stack_path, pairs, sp, stack_scenes,
                key=(0, 0),
                baselines=baselines,
                title="Interferogram Network",
                save_path=workdir / "network.png",
                quality_check=quality_check,
                plot_network=plot_network,
            )

        return pairs, baselines, scene_bperp, quality_scores, quality_factors

    def _mark_stack_dir(self, stack_dir, extra_cfg: dict | None = None) -> None:
        """Write the two files that make a directory a recognised stack.

        ``insarhub_config.json`` + the workflow marker are what the GUI, the CLI
        and every processor use to identify a stack folder and recover which
        downloader produced it. Both ``download()`` implementations need this,
        and S1_Burst cannot reuse the base's copy because it overrides
        ``download()`` wholesale (burst2safe owns the transfer), so it lives
        here rather than being written out twice.
        """
        from dataclasses import asdict as _asdict
        from pathlib import Path as _Path

        from insarhub.utils.config_io import write_insarhub_config as _wic
        from insarhub.utils.tool import write_workflow_marker

        stack_dir = _Path(stack_dir)
        try:
            stack_dir.mkdir(parents=True, exist_ok=True)
            write_workflow_marker(stack_dir, downloader=type(self).name)
            cfg = {k: v for k, v in _asdict(self.config).items() if k != "workdir"}
            cfg.update(extra_cfg or {})
            _wic(stack_dir, {"downloader": {"type": type(self).name, "config": cfg}})
        except Exception as exc:                                    # noqa: BLE001
            logger.error("%s: could not write stack config to %s: %s",
                         type(self).name, stack_dir, exc)

    def download(self, save_path: str | None = None, max_workers: int = None,
                 stop_event=None, on_progress=None,
                 scenes=None, merge: bool = False):
        """Download search results to the specified output directory.

        Args:
            save_path (str, optional): Download path. Defaults to config.workdir.
            max_workers (int, optional): Concurrent downloads. When None, falls
                back to ``config.max_workers`` (the GUI's per-downloader
                "Download" setting), then to 3. Resolving it here rather than in
                each caller means the config field takes effect from every entry
                point; previously only the S1_Burst routes passed it explicitly,
                so a GUI-set value was silently ignored for S1_SLC.
            scenes: Restrict download to a subset of scenes. Accepts any of:
                - ``list | set`` of scene name strings
                - The direct output of ``select_pairs()`` — either a
                  ``list[[ref, sec], ...]`` (single-stack) or a
                  ``dict{(path, frame): [[ref, sec], ...]}`` (multi-stack).
                  Unique scene names are extracted automatically.
                  When ``None`` (default) all search results are downloaded.
            merge (bool): When True, all stacks are downloaded into a single
                ``merged/slc/`` subdirectory instead of per-stack ``p{path}_f{frame}/``
                subdirs. Useful when combining multiple overlapping stacks for ISCE/MintPy.

        Raises:
            ValueError: If no search results are available.
        """
        from insarhub.utils.defaults import DOWNLOAD_DEFAULTS as _DL
        if max_workers is None:
            max_workers = getattr(self.config, "max_workers", None) or _DL["max_workers"]
        max_workers = max(1, int(max_workers))
        from concurrent.futures import ThreadPoolExecutor, as_completed
        output_dir = Path(save_path).expanduser().resolve() if save_path else self.config.workdir
        output_dir.mkdir(exist_ok=True, parents=True)

        self.download_dir = output_dir

        if not hasattr(self, 'results'):
            raise ValueError(f"{Fore.RED}No search results found. Please run search() first.")

        if stop_event is None:
            stop_event = threading.Event()

        scene_filter = _parse_scene_filter(scenes)

        # merge=True: all stacks of one path land in output_dir/p{path}_{tag}/slc/,
        # where tag encodes every constituent frame number — this must match
        # select_pairs(merge=True)'s own directory naming so the stack file
        # and the downloaded SLCs end up co-located. active_results is always
        # a dict[(path, frame), list[ASFProduct]] (see its docstring/contract).
        _sp = StackPaths(output_dir)
        # A single stack is never a merge: merge naming (p{path}_merged_f...) is
        # only meaningful for 2+ frames sharing one path. This mirrors
        # select_pairs(merge=...)'s own `len > 1` gate, so a single-frame
        # --merge download lands in p{path}_f{frame}/slc alongside the plain
        # stack file rather than in a mismatched p{path}_merged_f{frame}/.
        do_merge = (merge and isinstance(self.active_results, dict)
                    and len(self.active_results) > 1)
        if do_merge:
            paths = {path for (path, _frame) in self.active_results.keys()}
            if len(paths) > 1:
                raise ValueError(
                    f"merge=True requires all stacks to share one relative orbit "
                    f"(path), got {sorted(paths)}. Different tracks have unrelated "
                    f"viewing geometry and cannot be combined into one stack — "
                    f"narrow the search to a single path before merging."
                )
            path = next(iter(paths))
            frames = [frame for (_path, frame) in self.active_results.keys()]
            merged_dir = _sp.merge_dir(path, frames)
            merged_dir.mkdir(parents=True, exist_ok=True)
            self._mark_stack_dir(merged_dir)

        jobs = []
        stack_paths: dict = {}
        _dir_is_stack = (self.download_dir / "insarhub_config.json").exists()
        for key, results in self.active_results.items():
            if do_merge:
                stack_path    = merged_dir
                download_path = stack_path / "slc"
            elif _dir_is_stack:
                stack_path    = self.download_dir
                download_path = stack_path / "slc"
            else:
                stack_path    = StackPaths(self.download_dir).stack_dir(key[0], key[1])
                download_path = stack_path / "slc"
            download_path.mkdir(parents=True, exist_ok=True)
            stack_paths[key] = download_path
            if not merge:
                # key[1] is only a FRAME NUMBER for frame-based datasets. Burst
                # stacks key on fullBurstID, so writing it as 'frame' puts a
                # string into an int-range search field and the folder's next
                # search dies in asf_search's validator. Persist it only when it
                # really is an integer frame.
                extra = {'relativeOrbit': key[0]}
                try:
                    extra['frame'] = int(key[1])
                except (TypeError, ValueError):
                    pass
                self._mark_stack_dir(stack_path, extra)
            for result in results:
                if scene_filter is None or result.properties['sceneName'] in scene_filter:
                    jobs.append((key, result, download_path))

        total_jobs   = len(jobs)
        success_count = 0
        failure_count = 0
        failed_files  = []

        active_files: dict[int, Path] = {}
        active_files_lock = threading.Lock()

        total_scenes = sum(len(v) for v in self.active_results.values())
        filter_note  = (f" (filtered to {total_jobs} of {total_scenes})"
                        if scene_filter is not None and total_jobs != total_scenes else "")
        print(f"Downloading {total_jobs} scene(s) across "
              f"{len(self.active_results)} stack(s)"
              f"{filter_note} ({max_workers} concurrent)...\n")

        def _stream_download_interruptible(url, file_path, expected_bytes, 
                                        pbar_position, scene_name):
            """Stream download that checks stop_event on every chunk."""
            from tqdm import tqdm
            from asf_search.download.download import _try_get_response

            thread_session = asf.ASFSession()
            thread_session.cookies.update(self.session.cookies)
            thread_session.headers.update(self.session.headers)
            thread_session.verify = self.config.ssl_verify

            for attempt in range(1, 4):
                if stop_event.is_set():
                    raise InterruptedError("Download cancelled by user.")
                try:
                    response = _try_get_response(session=thread_session, url=url)
                    total_bytes = int(response.headers.get('content-length', expected_bytes))

                    with tqdm(
                        total=total_bytes,
                        unit='B',
                        unit_scale=True,
                        unit_divisor=1024,
                        desc=f"[Worker {pbar_position+1}] {scene_name}",
                        bar_format='{desc:<60}{percentage:3.0f}%|{bar:25}{r_bar}',
                        colour='green',
                        position=pbar_position,
                        leave=True,
                    ) as pbar:
                        with open(file_path, 'wb') as f:
                            for chunk in response.iter_content(chunk_size=65536):
                                # Check stop event on EVERY chunk — this is the key
                                if stop_event.is_set():
                                    response.close()  # abort the connection immediately
                                    raise InterruptedError("Download cancelled by user.")
                                if chunk:
                                    f.write(chunk)
                                    pbar.update(len(chunk))
                    return  # success

                except InterruptedError:
                    raise  # propagate immediately, don't retry
                except Exception as e:
                    if file_path.exists():
                        file_path.unlink()
                    if attempt == 3:
                        raise
                    time.sleep(2 ** attempt)

        def _download_job(args):
            key, result, download_path, position = args
            file_id   = result.properties['fileID']
            size_b    = _product_byte_size(result.properties)
            size_mb   = (size_b / (1024 * 1024)) if size_b else 0
            filename  = result.properties.get('fileName', f"{file_id}.zip")
            file_path = download_path / filename

            scene_name = result.properties.get('sceneName', file_id)

            if stop_event.is_set():
                return file_id, 'cancelled', 0, None

            # Skip if already complete
            if size_b and file_path.exists() and file_path.stat().st_size == size_b:
                return file_id, 'skipped', size_mb, None

            # Remove incomplete file
            if file_path.exists():
                file_path.unlink()

            with active_files_lock:
                active_files[position] = file_path

            try:
                start_time = time.time()
                _stream_download_interruptible(
                    url=result.properties['url'],
                    file_path=file_path,
                    expected_bytes=size_b or 0,
                    pbar_position=position,
                    scene_name=scene_name,
                )

                actual_size = file_path.stat().st_size
                if size_b and actual_size != size_b:
                    raise IOError(f"Size mismatch: expected {size_b}, got {actual_size} bytes.")

                elapsed = time.time() - start_time
                speed   = size_mb / elapsed if elapsed > 0 else 0
                return file_id, 'success', speed, None

            except InterruptedError:
                return file_id, 'cancelled', 0, None

            except Exception as e:
                if file_path.exists():
                    file_path.unlink()
                return file_id, 'failed', 0, str(e)
            finally:
                with active_files_lock:
                    active_files.pop(position, None)
        job_args = [
            (key, result, download_path, i % max_workers) 
            for i, (key, result, download_path) in enumerate(jobs)
        ]

        executor = ThreadPoolExecutor(max_workers=max_workers)
        futures  = {executor.submit(_download_job, args): args for args in job_args}

        completed_count = 0
        try:
            for future in as_completed(futures):
                file_id, status, value, error = future.result()
                completed_count += 1
                pct = int(completed_count / total_jobs * 100) if total_jobs else 100

                if status == 'success':
                    print(f"  {Fore.GREEN}✔ {file_id} ({value:.1f} MB/s)")
                    success_count += 1
                    if on_progress:
                        on_progress(f"[{completed_count}/{total_jobs}] ✔ {file_id}", pct)
                elif status == 'skipped':
                    print(f"  {Fore.YELLOW}⏭ {file_id} ({value:.1f} MB, already exists)")
                    success_count += 1
                    if on_progress:
                        on_progress(f"[{completed_count}/{total_jobs}] ⏭ {file_id} (exists)", pct)
                elif status == 'cancelled':
                    pass  # silently skip cancelled jobs
                else:
                    print(f"  {Fore.RED}✘ {file_id} — {error}")
                    failure_count += 1
                    failed_files.append(file_id)
                    if on_progress:
                        on_progress(f"[{completed_count}/{total_jobs}] ✘ {file_id}", pct)
        except KeyboardInterrupt:
            print(f"\n{Fore.YELLOW}⚠ Download interrupted by user. Cancelling pending jobs...")
            stop_event.set()
            # Cancel all pending futures that haven't started yet
            for future in futures:
                future.cancel()

            # Shut down without waiting for running threads to finish
            executor.shutdown(wait=False, cancel_futures=True)

            # Clean up any partial files being actively written
            with active_files_lock:
                for position, file_path in active_files.items():
                    if file_path.exists():
                        print(f"  {Fore.RED}Removing partial file: {file_path.name}")
                        file_path.unlink()

            print(f"{Fore.YELLOW}Download cancelled. "
                    f"{success_count} scenes completed before interrupt.")
            return

        else:
            executor.shutdown(wait=True)

        # Final summary
        print("\n" + "─" * 60)
        print(f"Download complete: {Fore.GREEN}{success_count}/{total_jobs} succeeded{Fore.RESET}", end="")
        if failure_count:
            print(f", {Fore.RED}{failure_count}/{total_jobs} failed{Fore.RESET}")
            print(f"\nFailed files:")
            for f in failed_files:
                print(f"  {Fore.RED}- {f}")
        if len(stack_paths) == 1:
            print(f"\nFiles saved to: {next(iter(stack_paths.values()))}")
        else:
            print(f"\nFiles saved to:")
            for key, p in stack_paths.items():
                print(f"  path={key[0]} frame={key[1]}: {p}")

Usage

• Create downloader with parameters

  Initialize a downloader instance with search criteria

  s1 = Downloader.create('ASF_Base_Downloader',
                          intersectsWith=[-113.05, 37.74, -112.68, 38.00],
                          dataset='SENTINEL-1',
                          instrument='C-SAR',
                          beamMode='IW',
                          polarization=['VV', 'VV+VH'],
                          processingLevel='SLC',
                          start='2020-01-01',
                          end='2020-12-31',
                          relativeOrbit=100,
                          frame=466,
                          workdir='path/to/dir')
  

  OR

  params = {
      "intersectsWith": [-113.05, 37.74, -112.68, 38.00],
      "dataset": "SENTINEL-1",
      "instrument": "C-SAR",
      "beamMode": "IW",
      "polarization": ["VV", "VV+VH"],
      "processingLevel": "SLC",
      "start": "2020-01-01",
      "end": "2020-12-31",
      "relativeOrbit": 100,
      "frame": 466,
      "workdir": "path/to/dir"
  }
  dl = Downloader.create('ASF_Base_Downloader', **params)
  

  OR

  from insarhub.config import ASF_Base_Config
  cfg = ASF_Base_Config(intersectsWith=[-113.05, 37.74, -112.68, 38.00],
                        dataset='SENTINEL-1',
                        instrument='C-SAR',
                        beamMode='IW',
                        polarization=['VV', 'VV+VH'],
                        processingLevel='SLC',
                        start='2020-01-01',
                        end='2020-12-31',
                        relativeOrbit=100,
                        frame=466,
                        workdir='path/to/dir')
  dl = Downloader.create('ASF_Base_Downloader', config=cfg)
  

  The base config ASF_Base_Config contains all parameters from asf_search keywords. For detailed descriptions refer to the official ASF Search documentation.

  Source code in src/insarhub/config/defaultconfig.py

  @dataclass
  class ASF_Base_Config:
      '''
      Dataclass containing all configuration options for asf_search.
  
      This class provides a unified interface for configuring ASF (Alaska Satellite Facility) 
      search parameters.
      '''
      name: str = "ASF_Base_Config"
      dataset: str | list[str] | None = None
      platform: str | list[str] | None = None
      instrument: str | None = None
      absoluteBurstID: int | list[int] | None = None
      absoluteOrbit: int | list[int] | None = None
      asfFrame: int | list[int] | None = None
      beamMode: str | None = None
      beamSwath: str | list[str] | None = None
      campaign: str | None = None
      maxDoppler: float | None = None
      minDoppler: float | None = None
      maxFaradayRotation: float | None = None
      minFaradayRotation: float | None = None
      flightDirection: str | None = None
      flightLine: str | None = None
      frame: int | list[int] | None = None
      frameCoverage: str | None = None
      fullBurstID: str | list[str] | None = None
      groupID: str | None = None
      jointObservation: bool | None = None
      lookDirection: str | None = None
      offNadirAngle: float | list[float] | None = None
      operaBurstID: str | list[str] | None = None
      polarization: str | list[str] | None = None
      mainBandPolarization: str | list[str] | None = None
      sideBandPolarization: str | list[str] | None = None
      processingLevel: str | None = None
      productionConfiguration: str | list[str] | None = None
      rangeBandwidth: str | list[str] | None = None
      relativeBurstID: str | list[str] | None = None
      relativeOrbit: int | list[int] | None = None
      intersectsWith: str | None = None  
      processingDate: str | None = None
      start: str | None = None
      end: str | None = None
      season: list[int] | None = None
      stack_from_id: str | None = None
      maxResults: int | None = None
      granule_names: str | list[str] | None = None
      workdir: Path | str = field(default_factory=lambda: Path.cwd())
      ssl_verify: bool = True
      # Per-downloader download concurrency. Each downloader owns this; there is
      # deliberately no global setting shadowing it, because the two downloaders
      # parallelise different work: S1_SLC opens concurrent HTTP transfers, while
      # S1_Burst assembles whole dates (each one a burst2safe run that unpacks and
      # rewrites a multi-GB product). A value that suits one starves or overloads
      # the other.
      max_workers: int = 3
  
      # ── UI metadata consumed by the API / settings panel ─────────────────────
      # The downloader CONFIG panel is deliberately minimal: only the operational
      # knobs (download concurrency + result cap). Every actual SEARCH parameter
      # (dataset/level are fixed per downloader; flightDirection / path / frame /
      # polarization / dates / AOI) lives in the Search Filters panel instead
      # (per-downloader `search_filter_schema` + the TopBar), so the two don't
      # duplicate ~30 asf_search fields. The remaining `_ui_fields` entries below
      # stay defined (harmless) in case a downloader re-groups them.
      _ui_groups: ClassVar[list] = [
          {"label": "Download",
           "fields": ["max_workers", "maxResults", "ssl_verify"]},
      ]
      _ui_fields: ClassVar[dict] = {
          "max_workers":     {"type": "number", "min": 1, "max": 16, "step": 1,
                              "hint": "Concurrent download threads."},
          "ssl_verify":      {"type": "bool",
                              "hint": "Verify ASF's SSL certificate. Turn OFF only if downloads "
                                      "fail with an expired-certificate error."},
          # Dataset
          "dataset":         {"type": "text",
                              "hint": "Dataset to search (e.g. SENTINEL-1, ALOS, NISAR)"},
          "platform":        {"type": "text",
                              "hint": "Platform name (e.g. S1A, ALOS)"},
          "instrument":      {"type": "text",
                              "hint": "Instrument name (e.g. C-SAR)"},
          # SAR Parameters
          "beamMode":        {"type": "select", "options": ["", "IW", "EW", "SM", "WV"],
                              "hint": "SAR acquisition mode"},
          "beamSwath":       {"type": "text",
                              "hint": "Beam swath identifier"},
          "processingLevel": {"type": "select",
                              "options": ["", "SLC", "GRD", "GRD_HD", "GRD_MS",
                                          "BURST", "RTC_HI_RES", "RTC_LOW_RES"],
                              "hint": "Processing level"},
          "polarization":    {"type": "text",
                              "hint": "Polarization(s), e.g. VV or VV+VH"},
          "mainBandPolarization": {"type": "text",
                              "hint": "Main band polarization (NISAR dual-band)"},
          "sideBandPolarization": {"type": "text",
                              "hint": "Side band polarization (NISAR dual-band)"},
          "lookDirection":   {"type": "select", "options": ["", "LEFT", "RIGHT"],
                              "hint": "Radar look direction"},
          "flightDirection": {"type": "select", "options": ["", "ASCENDING", "DESCENDING"],
                              "hint": "Orbit direction (empty = both)"},
          "flightLine":      {"type": "text",
                              "hint": "Flight line identifier"},
          # Orbit & Frame
          "relativeOrbit":   {"type": "text",
                              "hint": "Relative orbit (path) number(s), e.g. 64 or 64,65"},
          "absoluteOrbit":   {"type": "text",
                              "hint": "Absolute orbit number(s)"},
          "frame":           {"type": "text",
                              "hint": "Sensor native frame number(s)"},
          "asfFrame":        {"type": "text",
                              "hint": "ASF internal frame number(s)"},
          "frameCoverage":   {"type": "text",
                              "hint": "Frame coverage filter"},
          # Burst IDs
          "absoluteBurstID": {"type": "text",
                              "hint": "Absolute burst ID(s)"},
          "relativeBurstID": {"type": "text",
                              "hint": "Relative burst ID(s)"},
          "fullBurstID":     {"type": "text",
                              "hint": "Full burst ID, e.g. T064_135524_IW1"},
          "operaBurstID":    {"type": "text",
                              "hint": "OPERA burst ID(s)"},
          # Temporal & Location
          "start":           {"type": "text",
                              "hint": "Default start date (ISO 8601, e.g. 2020-01-01)"},
          "end":             {"type": "text",
                              "hint": "Default end date (ISO 8601, e.g. 2022-12-31)"},
          "processingDate":  {"type": "text",
                              "hint": "Processing date filter (ISO 8601)"},
          "season":          {"type": "text",
                              "hint": "Day-of-year range for seasonal filtering, e.g. 1,90"},
          "intersectsWith":  {"type": "text",
                              "hint": "WKT geometry for spatial intersection"},
          "stack_from_id":   {"type": "text",
                              "hint": "Build stack from a reference scene ID"},
          "maxResults":      {"type": "auto_number", "min": 1, "max": 50000, "step": 100,
                              "hint": "Maximum number of search results returned"},
          "granule_names":   {"type": "text",
                              "hint": "Granule/scene names (comma-separated), or a path to a CSV/XLSX/TXT file. "
                                      "When set, overrides normal parameter-based search."},
          # Advanced
          "campaign":        {"type": "text",
                              "hint": "Campaign name filter (UAVSAR / airborne datasets)"},
          "groupID":         {"type": "text",
                              "hint": "Group ID filter"},
          "maxDoppler":      {"type": "auto_number",
                              "hint": "Maximum Doppler centroid frequency (Hz)"},
          "minDoppler":      {"type": "auto_number",
                              "hint": "Minimum Doppler centroid frequency (Hz)"},
          "maxFaradayRotation": {"type": "auto_number",
                              "hint": "Maximum Faraday rotation angle (degrees)"},
          "minFaradayRotation": {"type": "auto_number",
                              "hint": "Minimum Faraday rotation angle (degrees)"},
          "offNadirAngle":   {"type": "text",
                              "hint": "Off-nadir angle(s), e.g. 34.3 or 21.5,26.2"},
          "jointObservation":{"type": "bool",
                              "hint": "Filter for joint ALOS PALSAR/AVNIR-2 observations"},
          "productionConfiguration": {"type": "text",
                              "hint": "Production configuration identifier"},
          "rangeBandwidth":  {"type": "text",
                              "hint": "Range bandwidth filter"},
      }
      # ─────────────────────────────────────────────────────────────────────────
  
      def __post_init__(self):
          if isinstance(self.workdir, str):
              self.workdir = Path(self.workdir).expanduser().resolve()
  

• Search

  Query the satellite archive and retrieve available scenes matching your criteria

  results = dl.search()
  

  Raises:

  Type	Description
  ValueError	If search returns no results.
  Exception	If search fails after 10 retry attempts.

• Filter

  Refine existing search results by applying additional constraints

  filter_result = dl.filter(start='2020-02-01')
  

  Parameters:

  Name	Type	Description	Default
  path_frame	tuple | list[tuple]	A single (path, frame) tuple or list of tuples. Defaults to None.	None
  start	str	Start date string, e.g. '2021-01-01'. Defaults to None.	None
  end	str	End date string, e.g. '2023-12-31'. Defaults to None.	None
  frame	int | list[int]	Sensor native frame number(s), e.g. 50. Defaults to None.	None
  asfFrame	int | list[int]	ASF internal frame number(s), e.g. 50. Defaults to None.	None
  flightDirection	str	'ASCENDING' or 'DESCENDING'. Defaults to None.	None
  relativeOrbit	int | list[int]	Relative orbit number(s) to keep. Defaults to None.	None
  absoluteOrbit	int | list[int]	Absolute orbit number(s) to keep. Defaults to None.	None
  lookDirection	str	'LEFT' or 'RIGHT'. Defaults to None.	None
  polarization	str | list[str]	Polarization(s) to keep, e.g. 'VV' or ['VV', 'VH']. Defaults to None.	None
  processingLevel	str	Processing level to keep, e.g. 'SLC'. Defaults to None.	None
  beamMode	str	Beam mode to keep, e.g. 'IW'. Defaults to None.	None
  season	list[int]	List of months (1-12) to keep, e.g. [6, 7, 8] for summer. Defaults to None.	None
  min_coverage	float	Minimum fractional overlap (0-1) between scene and AOI. Defaults to None.	None
  min_count	int	Drop stacks with fewer than this many scenes after filtering. Defaults to None.	None
  max_count	int	Keep at most this many scenes per stack (from earliest). Defaults to None.	None
  latest_n	int	Keep the N most recent scenes per stack. Defaults to None.	None
  earliest_n	int	Keep the N earliest scenes per stack. Defaults to None.	None

  Raises:

  Type	Description
  ValueError	If no search results are available.

• Reset filter

  Restore search results to the original unfiltered state

  dl.reset()
  

• Summary

  Display statistics and overview of current search results

  dl.summary()
  

  Parameters:

  Name	Type	Description	Default
  ls	bool	If True, list individual scene names and dates. Defaults to False.	False

• View Footprint

  Visualize geographic coverage of search results on an interactive map

  dl.footprint()
  

  Parameters:

  Name	Type	Description	Default
  save_path	str	Path to save the figure. If None, displays interactively. Defaults to None.	None

• Download

  Download all scenes from current search results to local storage

  dl.download()
  

  Parameters:

  Name	Type	Description	Default
  save_path	str	Download path. Defaults to config.workdir.	None
  max_workers	int	Concurrent downloads. When None, falls back to config.max_workers (the GUI's per-downloader "Download" setting), then to 3. Resolving it here rather than in each caller means the config field takes effect from every entry point; previously only the S1_Burst routes passed it explicitly, so a GUI-set value was silently ignored for S1_SLC.	None
  scenes		Restrict download to a subset of scenes. Accepts any of: - list | set of scene name strings - The direct output of select_pairs() — either a list[[ref, sec], ...] (single-stack) or a dict{(path, frame): [[ref, sec], ...]} (multi-stack). Unique scene names are extracted automatically. When None (default) all search results are downloaded.	None
  merge	bool	When True, all stacks are downloaded into a single merged/slc/ subdirectory instead of per-stack p{path}_f{frame}/ subdirs. Useful when combining multiple overlapping stacks for ISCE/MintPy.	False

  Raises:

  Type	Description
  ValueError	If no search results are available.

• DEM Download

  Download DEM covering all scenes from current search results

  dl.dem()
  

  Parameters:

  Name	Type	Description	Default
  save_path	str	Directory to save DEM files. If None, uses config.workdir. Defaults to None.	None

• Select Pairs

  Compute interferogram pairs for all active stacks based on temporal and perpendicular baseline constraints. Scenes with poor acquisition conditions (heavy rain, snow) are excluded automatically when avoid_low_quality_days=True (default).

  from insarhub.utils import plot_pair_network
  pairs, baselines, scene_bperp, _ = dl.select_pairs(
      dt_targets=(6, 12, 24, 36, 48, 72, 96),
      dt_tol=3,
      dt_max=120,
      pb_max=150.0,
      min_degree=3,
      max_degree=5,
      force_connect=True,
      avoid_low_quality_days=True,
      precip_mm_threshold=25.0,
      snow_threshold=0.5,
  )
  fig = plot_pair_network(pairs, baselines, scene_bperp)
  fig.show()
  

  Parameters:

  Name	Type	Description	Default
  dt_targets	tuple	Target temporal spacings in days. Defaults to (6, 12, 24, 36, 48, 72, 96).	None
  dt_tol	int	Tolerance in days around each target spacing. Defaults to 3.	None
  dt_max	int	Maximum temporal baseline in days. Defaults to 120.	None
  pb_max	float	Maximum perpendicular baseline in meters. Defaults to 150.0.	None
  min_degree	int	Minimum number of connections per scene. Defaults to 3.	None
  max_degree	int	Maximum number of connections per scene. Defaults to 5.	None
  force_connect	bool	Force connectivity for isolated scenes. Defaults to True.	None
  max_workers	int	Threads for API baseline fallback. Defaults to 4.	None
  aoi_wkt	str	AOI geometry in WKT for quality scoring. Defaults to search AOI.	None
  merge	bool	When True, stacks sharing the same relative orbit (path) are combined into one pairing network before temporal/baseline selection — matching how ISCE2's stackSentinel treats multiple frames of one track/pass as a single continuous acquisition. Stacks on different paths are never combined (cross-track pairs have no physical baseline). Use together with download(merge=True), which puts all scenes in one merged/slc/ directory. Defaults to False.	False
  burst	bool	Select pairs for an SLC-BURST stack. Nodes become acquisition dates and baselines are computed from the bursts' own startTime/orbits — no parent-SLC lookup. Defaults to False.	False
  safe_dir	str	Burst mode: directory of assembled .SAFE dirs whose annotation orbits supply bperp (offline).	None
  eof_dir	str	Burst mode: directory of precise-orbit .EOF files used for bperp (offline).	None
  poeorb_cache	str	Burst mode: directory for POEORB downloads keyed by date + mission (online fallback).	None
  quality_check	bool	After pairing, write the stack file(s) and build the PairQualityDB verdict for every possible pair (the slow, network-heavy step). False skips it. Defaults to True.	True
  plot_network	bool	Save network_*.png with the pair network, coloured healthy/concern when quality_check is True (falls back to temporal-baseline colouring otherwise). Defaults to True.	True

S1_SLC is a specialized downloader that extends ASF_Base_Downloader, preconfigured specifically for downloading Sentinel-1 SLC data.

Source code in src/insarhub/downloader/s1_slc.py

class S1_SLC(ASF_Base_Downloader):
    name = "S1_SLC"
    description = "Sentinel-1 SLC scene search and download via ASF."
    default_config = S1_SLC_Config
    product_label = "SLCs"

    search_filter_schema = [
        {"name": "flightDirection", "label": "Flight Direction", "kind": "select",
         "group": "Additional Filters", "choices": ["ASCENDING", "DESCENDING"]},
        {"name": "platform", "label": "Platform", "kind": "select",
         "group": "Additional Filters",
         "choices": ["Sentinel-1A", "Sentinel-1B", "Sentinel-1C", "Sentinel-1D"]},
        {"name": "polarization", "label": "Polarization", "kind": "select",
         "group": "Additional Filters",
         "choices": ["VV", "VV+VH", "HH", "HH+HV"]},
        {"name": "relativeOrbit", "label": "Path", "kind": "range", "group": "Path and Frame Filters"},
        {"name": "asfFrame", "label": "Frame", "kind": "range", "group": "Path and Frame Filters"},
    ]

    """
    A class to search and download Sentinel-1 data using ASF Search API."""

    def download(self, save_path: str | None = None, max_workers: int = None,
                 force_cdse: bool = False, download_orbit: bool = False,
                 stop_event=None, on_progress=None, merge: bool = False):
        """Download SLC data and optionally associated orbit files.

        Args:
            save_path (str | None): Optional path to save the downloaded files. Defaults to None.
            max_workers (int): Parallel download workers. None lets the base
                resolve config.max_workers, then the built-in default.
            force_cdse (bool): If True, forces downloading orbit files from CDSE instead of ASF. Defaults to False.
            download_orbit (bool): If True, also downloads orbit files after scenes. Defaults to False.
            stop_event: Optional threading.Event to cancel the download.
            on_progress: Optional callback(message, pct) called after each file completes.
            merge (bool): If True, all stacks download into a single merged/slc/ directory.
        """
        super().download(save_path=save_path, max_workers=max_workers,
                         stop_event=stop_event, on_progress=on_progress, merge=merge)
        if download_orbit:
            self.download_orbit(force_cdse=force_cdse, merge=merge)

    def download_orbit(self, force_cdse: bool = False, save_dir: str | None = None,
                       stop_event=None, scenes=None, merge: bool = False):
        """Download orbit files for the current search results.

        Downloads from ASF by default (no credentials required).  Pass
        ``force_cdse=True`` to use the Copernicus Data Space Ecosystem (CDSE)
        server instead — CDSE typically publishes precise orbits a few hours
        earlier but requires an account at https://dataspace.copernicus.eu/
        configured in your ``.netrc`` file.

        Args:
            force_cdse (bool): Use CDSE instead of ASF. Defaults to False.
            save_dir (str | None): Directory to save orbit files. Defaults to workdir if not specified.
            scenes: Restrict to a subset of scenes. Accepts scene name strings, or the
                direct output of ``select_pairs()`` (list or dict). Same format as
                ``download(scenes=...)``. When ``None`` all scenes get orbit files.
        """
        use_asf = not force_cdse
        print(f"Downloading orbit files from {'ASF' if use_asf else 'CDSE'}…")

        if force_cdse:
            self._has_cdse_netrc = self._check_netrc(keyword='machine dataspace.copernicus.eu')
            if self._has_cdse_netrc:
                print(f"{Fore.GREEN}CDSE credentials found in .netrc.\n")
            else:
                while True:
                    self._cdse_username = input("Enter your CDSE username: ")
                    self._cdse_password = getpass.getpass("Enter your CDSE password: ")
                    if not self._check_cdse_credentials(self._cdse_username, self._cdse_password):
                        print(f"{Fore.RED}Authentication failed. Please check your credentials and try again.\n")
                        continue
                    netrc_path = Path.home().joinpath(".netrc")
                    cdse_entry = f"\nmachine dataspace.copernicus.eu\n    login {self._cdse_username}\n    password {self._cdse_password}\n"
                    with open(netrc_path, 'a') as f:
                        f.write(cdse_entry)
                    print(f"{Fore.GREEN}Credentials saved to {netrc_path}.\n")
                    break

        from insarhub.downloader.asf_base import _parse_scene_filter
        scene_filter = _parse_scene_filter(scenes)

        base_dir = Path(save_dir) if save_dir else (getattr(self, 'download_dir', None) or Path(getattr(self.config, 'workdir', None) or Path.cwd()))
        all_items = [
            (key, result)
            for key, results in self.results.items()  # type: ignore[union-attr]
            for result in results
            if scene_filter is None or result.properties['sceneName'] in scene_filter
        ]
        with tqdm(all_items, desc="Orbit files", unit="scene", bar_format="{l_bar}{bar:20}{r_bar}") as pbar:
            for key, result in pbar:
                if stop_event is not None and stop_event.is_set():
                    tqdm.write("Orbit download stopped.")
                    break
                _base = Path(base_dir)
                _is_stack = (_base / "insarhub_config.json").exists()
                # Single stack is never a merge -- gate on 2+ frames, mirroring
                # download()/select_pairs() so a single-frame --merge -O keeps
                # orbits with their per-frame SLCs instead of a stray merged/.
                _do_merge = merge and isinstance(self.results, dict) and len(self.results) > 1
                if _do_merge:
                    download_path = _base / 'merged' / 'slc'
                elif save_dir:
                    download_path = Path(save_dir) / 'slc'
                elif _is_stack:
                    download_path = _base / 'slc'
                else:
                    download_path = _base / f'p{key[0]}_f{key[1]}' / 'slc'
                download_path.mkdir(parents=True, exist_ok=True)
                scene_name = result.properties['sceneName']
                short_name = scene_name[:40] + "..."
                acq_time = scene_name.replace("__", "_").split("_")[4]
                already_have = False
                for eof in download_path.glob("*.EOF"):
                    parts = eof.stem.split("_V")
                    if len(parts) == 2:
                        validity = parts[1].split("_")
                        if len(validity) == 2 and validity[0] <= acq_time <= validity[1]:
                            pbar.set_postfix_str(f"skip {short_name}")
                            already_have = True
                            break
                if already_have:
                    continue
                pbar.set_postfix_str(f"fetch {short_name}")
                _save = download_path.as_posix()
                try:
                    info = download_eofs(sentinel_file=scene_name, save_dir=_save, force_asf=use_asf)
                except Exception as e:
                    if use_asf:
                        pbar.set_postfix_str(f"ASF fail, try CDSE {short_name}")
                        try:
                            info = download_eofs(sentinel_file=scene_name, save_dir=_save, force_asf=False)
                        except Exception as e2:
                            tqdm.write(f"{Fore.RED}[ERROR] {scene_name}: {e2}")
                            info = []
                    else:
                        tqdm.write(f"{Fore.RED}[ERROR] {scene_name}: {e}")
                        info = []
                if info:
                    pbar.set_postfix_str(f"ok {short_name}")
                else:
                    tqdm.write(f"{Fore.YELLOW}[WARN] No orbit file found for: {scene_name}")

    def _check_cdse_credentials(self, username: str, password: str) -> bool:
        url = "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token"
        data = {
            "grant_type": "password",
            "client_id": "cdse-public",
            "username": username,
            "password": password
        }
        resp = requests.post(url, data=data)
        return resp.status_code == 200 and "access_token" in resp.json()

Usage

• Create downloader with parameters

  Initialize a downloader instance with search criteria

  s1 = Downloader.create('S1_SLC',
                          intersectsWith=[-113.05, 37.74, -112.68, 38.00],
                          start='2020-01-01',
                          end='2020-12-31',
                          relativeOrbit=100,
                          frame=466,
                          workdir='path/to/dir')
  

  OR

  params = {
      "intersectsWith": [-113.05, 37.74, -112.68, 38.00],
      "start": "2020-01-01",
      "end": "2020-12-31",
      "relativeOrbit": 100,
      "frame": 466,
      "workdir": "path/to/dir"
  }
  dl = Downloader.create('S1_SLC', **params)
  

  OR

  from insarhub.config import S1_SLC_Config
  cfg = S1_SLC_Config(intersectsWith=[-113.05, 37.74, -112.68, 38.00],
                      start="2020-01-01",
                      end="2020-12-31",
                      relativeOrbit=100,
                      frame=466,
                      workdir="path/to/dir")
  dl = Downloader.create('S1_SLC', config=cfg)
  

  The config S1_SLC_Config contains pre-defined parameters specifically for Sentinel-1 data. For detailed descriptions refer to the official ASF Search documentation.

  Source code in src/insarhub/config/defaultconfig.py

  @dataclass
  class S1_SLC_Config(ASF_Base_Config):
      name:str = "S1_SLC_Config"
      dataset: str | list[str] | None =  constants.DATASET.SENTINEL1
      instrument: str | None = constants.INSTRUMENT.C_SAR
      beamMode:str | None = constants.BEAMMODE.IW
      polarization: str|list[str] | None = field(default_factory=lambda: [constants.POLARIZATION.VV, constants.POLARIZATION.VV_VH])
      processingLevel: str | None = constants.PRODUCT_TYPE.SLC
  

• Search

  results = dl.search()
  

  Raises:

  Type	Description
  ValueError	If search returns no results.
  Exception	If search fails after 10 retry attempts.

• Filter

  filter_result = dl.filter(start='2020-02-01')
  

  Parameters:

  Name	Type	Description	Default
  path_frame	tuple | list[tuple]	A single (path, frame) tuple or list of tuples. Defaults to None.	None
  start	str	Start date string, e.g. '2021-01-01'. Defaults to None.	None
  end	str	End date string, e.g. '2023-12-31'. Defaults to None.	None
  frame	int | list[int]	Sensor native frame number(s), e.g. 50. Defaults to None.	None
  asfFrame	int | list[int]	ASF internal frame number(s), e.g. 50. Defaults to None.	None
  flightDirection	str	'ASCENDING' or 'DESCENDING'. Defaults to None.	None
  relativeOrbit	int | list[int]	Relative orbit number(s) to keep. Defaults to None.	None
  absoluteOrbit	int | list[int]	Absolute orbit number(s) to keep. Defaults to None.	None
  lookDirection	str	'LEFT' or 'RIGHT'. Defaults to None.	None
  polarization	str | list[str]	Polarization(s) to keep, e.g. 'VV' or ['VV', 'VH']. Defaults to None.	None
  processingLevel	str	Processing level to keep, e.g. 'SLC'. Defaults to None.	None
  beamMode	str	Beam mode to keep, e.g. 'IW'. Defaults to None.	None
  season	list[int]	List of months (1-12) to keep, e.g. [6, 7, 8] for summer. Defaults to None.	None
  min_coverage	float	Minimum fractional overlap (0-1) between scene and AOI. Defaults to None.	None
  min_count	int	Drop stacks with fewer than this many scenes after filtering. Defaults to None.	None
  max_count	int	Keep at most this many scenes per stack (from earliest). Defaults to None.	None
  latest_n	int	Keep the N most recent scenes per stack. Defaults to None.	None
  earliest_n	int	Keep the N earliest scenes per stack. Defaults to None.	None

  Raises:

  Type	Description
  ValueError	If no search results are available.

• Reset filter

  dl.reset()
  

• Summary

  dl.summary()
  

  Parameters:

  Name	Type	Description	Default
  ls	bool	If True, list individual scene names and dates. Defaults to False.	False

• View Footprint

  dl.footprint()
  

  Parameters:

  Name	Type	Description	Default
  save_path	str	Path to save the figure. If None, displays interactively. Defaults to None.	None

• Download

  dl.download()
  

  Parameters:

  Name	Type	Description	Default
  save_path	str | None	Optional path to save the downloaded files. Defaults to None.	None
  max_workers	int	Parallel download workers. None lets the base resolve config.max_workers, then the built-in default.	None
  force_cdse	bool	If True, forces downloading orbit files from CDSE instead of ASF. Defaults to False.	False
  download_orbit	bool	If True, also downloads orbit files after scenes. Defaults to False.	False
  stop_event		Optional threading.Event to cancel the download.	None
  on_progress		Optional callback(message, pct) called after each file completes.	None
  merge	bool	If True, all stacks download into a single merged/slc/ directory.	False

• DEM Download

  dl.dem()
  

  Parameters:

  Name	Type	Description	Default
  save_path	str	Directory to save DEM files. If None, uses config.workdir. Defaults to None.	None

• Select Pairs

  from insarhub.utils import plot_pair_network
  pairs, baselines, scene_bperp, _ = s1.select_pairs(
      dt_targets=(6, 12, 24, 36, 48, 72, 96),
      dt_tol=3,
      dt_max=120,
      pb_max=150.0,
      min_degree=3,
      max_degree=5,
      force_connect=True,
      avoid_low_quality_days=True,
      precip_mm_threshold=25.0,
      snow_threshold=0.5,
  )
  fig = plot_pair_network(pairs, baselines, scene_bperp)
  fig.show()
  

  Parameters:

  Name	Type	Description	Default
  dt_targets	tuple	Target temporal spacings in days. Defaults to (6, 12, 24, 36, 48, 72, 96).	None
  dt_tol	int	Tolerance in days around each target spacing. Defaults to 3.	None
  dt_max	int	Maximum temporal baseline in days. Defaults to 120.	None
  pb_max	float	Maximum perpendicular baseline in meters. Defaults to 150.0.	None
  min_degree	int	Minimum number of connections per scene. Defaults to 3.	None
  max_degree	int	Maximum number of connections per scene. Defaults to 5.	None
  force_connect	bool	Force connectivity for isolated scenes. Defaults to True.	None
  max_workers	int	Threads for API baseline fallback. Defaults to 4.	None
  aoi_wkt	str	AOI geometry in WKT for quality scoring. Defaults to search AOI.	None
  merge	bool	When True, stacks sharing the same relative orbit (path) are combined into one pairing network before temporal/baseline selection — matching how ISCE2's stackSentinel treats multiple frames of one track/pass as a single continuous acquisition. Stacks on different paths are never combined (cross-track pairs have no physical baseline). Use together with download(merge=True), which puts all scenes in one merged/slc/ directory. Defaults to False.	False
  burst	bool	Select pairs for an SLC-BURST stack. Nodes become acquisition dates and baselines are computed from the bursts' own startTime/orbits — no parent-SLC lookup. Defaults to False.	False
  safe_dir	str	Burst mode: directory of assembled .SAFE dirs whose annotation orbits supply bperp (offline).	None
  eof_dir	str	Burst mode: directory of precise-orbit .EOF files used for bperp (offline).	None
  poeorb_cache	str	Burst mode: directory for POEORB downloads keyed by date + mission (online fallback).	None
  quality_check	bool	After pairing, write the stack file(s) and build the PairQualityDB verdict for every possible pair (the slow, network-heavy step). False skips it. Defaults to True.	True
  plot_network	bool	Save network_*.png with the pair network, coloured healthy/concern when quality_check is True (falls back to temporal-baseline colouring otherwise). Defaults to True.	True

Extends ASF_Base_Downloader for ASF's SLC-BURST dataset. A burst is roughly 1/9th of a full IW slice, so an AOI-limited burst stack pulls far less data than the equivalent S1_SLC search. Pair it with the ISCE3_Burst processor and the ISCE3_Dolphin_S1_PL analyzer.

Only download() differs from S1_SLC: it hands the selected granules to burst2safe, which assembles them into valid .SAFE directories.

Burst stacks are keyed by fullBurstID, not frame

ASF returns no frameNumber on SLC-BURST products, so a frame filter matches nothing and is excluded from the query. A burst stack is identified by fullBurstID (e.g. 056_118970_IW2); the downloader folder is named p<path>_iw<s>_b<id> accordingly.

Source code in src/insarhub/downloader/s1_burst.py

class S1_Burst(ASF_Base_Downloader):
    name = "S1_Burst"
    description = "Sentinel-1 SLC-BURST search and download, assembled into .SAFE via burst2safe."
    default_config = S1_Burst_Config

    # These configure burst2safe assembly, not the ASF query -- forwarding them
    # to asf.search() makes the call fail (and the base retry loop then burns
    # ~17 minutes of backoff before reporting it).
    # frame/asfFrame are excluded deliberately, not just unused: ASF returns NO
    # frameNumber on SLC-BURST products, so a frame filter can only ever match
    # nothing. Worse, a burst stack is keyed by fullBurstID, and the base
    # download() persists its group key as {'frame': key[1]} -- which for a
    # burst is a STRING ("056_118970_IW2", or the "?_?" placeholder when ASF
    # omits burstIndex/subswath). That lands in the folder's insarhub_config
    # and the next search sends it to asf.search(frame=...), which parses frame
    # as an int range and dies with:
    #     Invalid int or range: invalid literal for int() with base 10: '?'
    # Dropping them here means the value can never reach the query no matter
    # what a stale or hand-edited config contains.
    _NON_SEARCH_FIELDS = ASF_Base_Downloader._NON_SEARCH_FIELDS | {
        "swaths", "mode", "min_bursts", "all_anns", "keep_files",
        "frame", "asfFrame"}

    # A burst stack key's second half is a fullBurstID, never a frame number --
    # calling it "frame" in the summary both misnames it and invites the user to
    # feed it back to --frame, which is exactly the int-parse crash described above.
    stack_key_label = "Burst_ID"
    product_label = "bursts"

    search_filter_schema = [
        {"name": "flightDirection", "label": "Flight Direction", "kind": "select",
         "group": "Additional Filters", "choices": ["ASCENDING", "DESCENDING"]},
        {"name": "platform", "label": "Platform", "kind": "select",
         "group": "Additional Filters",
         "choices": ["Sentinel-1A", "Sentinel-1B", "Sentinel-1C", "Sentinel-1D"]},
        {"name": "polarization", "label": "Polarization", "kind": "select",
         "group": "Additional Filters",
         "choices": ["VV", "VV+VH", "HH", "HH+HV"]},
        {"name": "relativeOrbit", "label": "Path", "kind": "range",
         "group": "Path and Frame Filters"},
        {"name": "fullBurstID", "label": "Burst ID", "kind": "text",
         "group": "Path and Frame Filters"},
    ]
    # NOTE: no "beamSwath" filter here, deliberately. ASF leaves beamSwath
    # EMPTY on SLC-BURST products, so passing any value matches nothing --
    # measured over a 3-burst AOI on 2024-01-01..2024-03-01: AOI alone -> 13
    # granules, AOI + beamSwath=IW2 -> 0. Offering it as a search filter meant
    # picking a subswath in the GUI silently returned an empty search.
    # Subswath selection is a client-side concern and lives in `swaths`
    # (Settings -> Burst Assembly), applied in _group_by_date below.

    # ------------------------------------------------------------------
    # helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _granule_of(result) -> str | None:
        """ASF granule name for a search result, tolerating shape differences."""
        props = getattr(result, "properties", None) or {}
        return props.get("fileID") or props.get("sceneName") or props.get("granuleName")

    @staticmethod
    def _swath_of(granule: str) -> str | None:
        """Subswath (IW1/IW2/IW3, EW1..EW5) parsed from the granule name.

        ASF does NOT populate ``beamSwath`` on SLC-BURST products -- it comes
        back None -- so the granule name is the only reliable source::

            S1_264306_IW3_20240915T043118_VV_F23F-BURST
                      ^^^
        """
        for tok in str(granule).split("_"):
            if len(tok) == 3 and tok[:2] in ("IW", "EW") and tok[2].isdigit():
                return tok
        return None

    @staticmethod
    def _date_of(result) -> str | None:
        """YYYYMMDD acquisition date, used to group bursts into one SAFE per date."""
        props = getattr(result, "properties", None) or {}
        start = props.get("startTime") or props.get("stopTime")
        if not start:
            return None
        # startTime is ISO-8601: 2024-09-15T16:12:34.000000Z
        return str(start)[:10].replace("-", "")

    @staticmethod
    def _flatten(results) -> list:
        """Flatten ``active_results``.

        ``ASF_Base_Downloader.active_results`` is a property returning a dict
        grouped by (path, frame) -- one entry per stack when an AOI spans more
        than one. Burst assembly is per-acquisition, so the grouping is redone
        by date here; flatten first.
        """
        if isinstance(results, dict):
            return [r for group in results.values() for r in group]
        return list(results)

    @staticmethod
    def _path_of(result) -> int | None:
        """Relative orbit (track). ASF leaves ``relativeOrbit`` empty on BURST
        products; the value lives in ``pathNumber``, and also as the first field
        of ``burst.fullBurstID`` ("124_264312_IW2")."""
        props = getattr(result, "properties", None) or {}
        p = props.get("pathNumber") or props.get("relativeOrbit")
        if p is not None:
            return int(p)
        fid = (props.get("burst") or {}).get("fullBurstID") or ""
        head = str(fid).split("_")[0]
        return int(head) if head.isdigit() else None

    def _get_group_key(self, result) -> tuple:
        """Group bursts by ``(path, fullBurstID)``: one fixed burst position.

        The base implementation groups by ``(pathNumber, frameNumber)``, but
        ASF burst granules carry NO ``frameNumber`` -- so every burst of a
        track collapsed onto ``(124, None)`` and the GUI rendered them all as
        "Path 0 · Frame 0". A burst stack is a burst position that repeats
        every revisit, uniquely identified by its OPERA ``fullBurstID``
        ("124_266256_IW3" = path_burstIndex_subswath).
        """
        props = result.properties
        path = props.get("pathNumber")
        if path is None:
            path = self._path_of(result)
        b = props.get("burst") or {}
        full = (props.get("fullBurstID") or b.get("fullBurstID")
                or props.get("relativeBurstID") or b.get("relativeBurstID"))
        if not full:
            # Last resort: the granule name always carries the subswath, and is
            # a real identifier. The previous fallback built "?_?" from missing
            # burstIndex/subswath -- a placeholder that reads as data, groups
            # every unidentifiable burst together, and (via the base download's
            # group-key persistence) ended up in configs as frame="?_?".
            g = self._granule_of(result) or ""
            full = g or f"unknown_{self._date_of(result) or 'nodate'}"
            logger.warning("S1_Burst: result carries no burst ID; grouping by "
                           "granule name %r instead", full)
        return (path, full)

    @staticmethod
    def _burst_id_parts(text) -> tuple[int, int, str] | None:
        """``"087_185682_IW2"`` -> ``(87, 185682, "IW2")``; None if not that shape.

        Path and index come back as ints so the zero-padded form ASF prints
        (``087_...``) and the bare form a user types (``87_...``) compare equal.
        """
        parts = str(text).strip().upper().split("_")
        if len(parts) != 3:
            return None
        path, index, swath = parts
        if not (path.isdigit() and index.isdigit()):
            return None
        return int(path), int(index), swath

    def _stack_key_matches(self, key: tuple, target: tuple) -> bool:
        """Match a burst stack against a user ``PATH:SELECTOR`` token.

        Burst stacks key on ``fullBurstID`` (see :meth:`_get_group_key`), so the
        second half of a ``--stacks`` token is not a number and the base class's
        plain-equality rule can never match it. Three spellings of one stack are
        accepted, widest last::

            124:124_264305_IW2   full burst ID, exactly as summary() prints it
            124:264305_IW2       burst index + subswath
            124:264305           bare burst index -- EVERY subswath at that index

        The bare index is deliberately one-to-many: ASF reuses an index across
        subswaths (``124_266256_IW2`` and ``124_266256_IW3`` both exist), so it
        selects both. Name the subswath to pin exactly one.

        A stack that fell back to grouping by granule name (no burst ID on the
        product at all) matches only on a verbatim string, since it has no index
        or subswath to compare against.
        """
        try:
            if int(key[0]) != int(target[0]):
                return False
        except (TypeError, ValueError):
            return False   # a stack with no path can only be reached by AOI, not by token

        selector = str(target[1]).strip().upper()
        full     = str(key[1]).strip().upper()
        if not selector:
            return False
        if selector == full:
            return True

        key_parts = self._burst_id_parts(full)
        if key_parts is None:
            return False            # granule-name fallback key: verbatim match only
        _, index, swath = key_parts

        sel_parts = self._burst_id_parts(selector)
        if sel_parts is not None:
            return sel_parts == key_parts

        bits = selector.split("_")
        if len(bits) == 2 and bits[0].isdigit():
            return int(bits[0]) == index and bits[1] == swath
        return selector.isdigit() and int(selector) == index

    @staticmethod
    def folder_name(path: int, subswath: str | None = None,
                    burst_id: int | None = None) -> str:
        """Job-folder name for a burst selection.

        A single burst is one fixed burst position -> a per-burst folder
        ``p<path>_iw<s>_b<id>``, where ``id`` is the OPERA relative burst ID
        (unique per subswath across the orbit, e.g. 266256). A whole track
        (several bursts merged) is one ``p<path>`` folder, because burst2safe
        assembles one SAFE per (date, path) anyway -- the track is the unit the
        ISCE3_Burst processor consumes.
        """
        if subswath and burst_id is not None:
            sw = str(subswath).lower()          # "IW3" -> "iw3"
            num = sw[2:] if sw[:2] in ("iw", "ew") else sw   # -> "3"
            return f"p{path}_iw{num}_b{burst_id}"
        return f"p{path}"

    def _group_by_date(self, results) -> dict[str, list[str]]:
        """{"YYYYMMDD" or "YYYYMMDD_pNNN": [granule, ...]} -- one SAFE per group.

        Grouped by acquisition date AND path. A SAFE is a single pass, and
        burst2safe enforces "all bursts must have the same absolute orbit"; a
        date-only key would merge two paths imaged on the same day into one
        call, which raises and -- because assembly failures are caught per group
        -- would silently drop BOTH paths for that date.

        Two paths on one date is uncommon for a small AOI (over Hawaii, paths
        14/87/124 all fall on different days) but is normal where passes
        converge at high latitude or where an AOI sees both ascending and
        descending on the same day. The key stays the bare date in the common
        single-path case so output names are unchanged.
        """
        # A bare string is tolerated: swaths round-trips through JSON configs
        # and a hand-edited "IW2" would otherwise iterate as {"I","W","2"} and
        # drop every granule.
        _sw = getattr(self.config, "swaths", None) or []
        want = {s.upper() for s in ([_sw] if isinstance(_sw, str) else _sw)}
        # s1reader derives the OPERA burst ID from the IW2 mid-burst sensing
        # time, so it opens the IW2 ANNOTATION unconditionally -- whichever
        # subswath you asked for:
        #     ValueError: burst iw2-slc-vv not in SAFE: <dir>
        #
        # But annotation is all it needs, not IW2 measurement data. Verified by
        # assembling an IW3-ONLY SAFE with all_anns=True (measurement: iw3;
        # annotation: iw1+iw2+iw3) -- s1reader loaded all 4 IW3 bursts fine.
        # So:
        #   all_anns=True  -> slice-level annotation for every subswath is
        #                     included anyway; adding IW2 DATA would roughly
        #                     double the download for nothing.
        #   all_anns=False -> only the requested subswaths' annotation is kept,
        #                     so IW2 must be pulled in or the SAFE is unreadable.
        if (want and str(getattr(self.config, "mode", "IW")).upper() == "IW"
                and "IW2" not in want
                and not bool(getattr(self.config, "all_anns", False))):
            print(f"[S1_Burst] adding IW2 to swaths {sorted(want)}: s1reader needs the "
                  f"IW2 annotation for the burst-ID reference time, and all_anns is "
                  f"off. Set all_anns=True to keep {sorted(want)} only and avoid "
                  f"downloading IW2 data.")
            want = want | {"IW2"}
        groups: dict[str, list[str]] = defaultdict(list)
        skipped = dropped = 0
        for r in self._flatten(results):
            g, d = self._granule_of(r), self._date_of(r)
            if not g or not d:
                skipped += 1
                continue
            # config.swaths must be applied HERE. It cannot be applied at search
            # time (ASF leaves beamSwath empty on BURST products), and it cannot
            # be delegated to burst2safe either: when burst2safe is given an
            # explicit granule list it assembles exactly those granules and
            # ignores its own `swaths` argument. Verified against a live
            # download -- swaths=["IW3"] still produced an IW2+IW3 SAFE.
            if want and self._swath_of(g) not in want:
                dropped += 1
                continue
            groups[(d, self._path_of(r))].append(g)
        if skipped:
            logger.warning("S1_Burst: %d result(s) lacked a granule name or start "
                           "time and were skipped", skipped)
        if dropped:
            print(f"[S1_Burst] swath filter {sorted(want)}: dropped {dropped} burst(s)")

        # Flatten (date, path) -> label. Keep the bare date when a date has only
        # one path (the normal case), so names match what callers already expect.
        per_date: dict[str, int] = defaultdict(int)
        for (d, _p) in groups:
            per_date[d] += 1
        out: dict[str, list[str]] = {}
        for (d, path), gr in sorted(groups.items(), key=lambda kv: (kv[0][0], kv[0][1] or 0)):
            key = d if per_date[d] == 1 else f"{d}_p{path if path is not None else 'NA'}"
            out[key] = gr
        multi = [d for d, n in per_date.items() if n > 1]
        if multi:
            print(f"[S1_Burst] {len(multi)} date(s) span multiple paths; assembling "
                  f"one SAFE per path: {sorted(multi)}")
        return out

    # ------------------------------------------------------------------
    # pair selection
    # ------------------------------------------------------------------

    #: parent SLC granule embedded in every burst's download URL
    _PARENT_RE = re.compile(
        r"(S1[ABCD]_IW_SLC__\w{4}_\d{8}T\d{6}_\d{8}T\d{6}_\d{6}_\w{6}_\w{4})")

    def parent_slcs(self, results=None) -> dict[str, set[str]]:
        """{parent SLC granule: {YYYYMMDD, ...}} for the current burst results.

        ASF serves each burst from its parent slice, and names it in the URL::

            https://sentinel1-burst.asf.alaska.edu/S1A_IW_SLC__1SDV_2024...96A1/...

        That is the only link back to a product ASF actually publishes baselines
        for -- burst granules themselves carry ``perpendicularBaseline: None``.
        """
        out: dict[str, set[str]] = defaultdict(set)
        for r in self._flatten(self.active_results if results is None else results):
            m = self._PARENT_RE.search((getattr(r, "properties", {}) or {}).get("url") or "")
            d = self._date_of(r)
            if m and d:
                out[m.group(1)].add(d)
        return dict(out)

    def sequential_pairs(self, n_connections: int = 3, dates=None) -> list[tuple[str, str]]:
        """Tutorial-exact sequential pairing: each date to its next N neighbours.

        Byte-for-byte the rule in the COMPASS stack notebook's
        ``utils.generate_ifgram_pairs``::

            max_step = min(n_connections + 1, len(dates))
            for i in range(len(dates) - 1):
                for j in range(i + 1, min(i + max_step, len(dates))):
                    pairs.append((dates[i], dates[j]))

        Use this when the goal is to reproduce the tutorial exactly.
        :meth:`select_pairs` cannot: it is target-driven (``dt_targets``) rather
        than rule-driven, so the closest it gets on the 9-date Hawaii stack is a
        23-pair SUPERSET of these 21 (at ``pb_max=400, max_degree=6``), never
        the set itself.

        Note what this deliberately ignores: perpendicular baseline. On that
        same stack it emits ``20240927_20241009`` (12 d but 300 m dBperp) and
        ``20240903_20240915`` (12 d, 208 m), which :meth:`select_pairs` rejects
        as geometrically decorrelated. That is the tutorial's behaviour, not a
        defect here -- but it is the reason the two disagree.

        Args:
            n_connections: Neighbours ahead to connect each date to.
            dates: Explicit YYYYMMDD list; defaults to the dates in the current
                search results.

        Returns:
            Sorted ``[(YYYYMMDD, YYYYMMDD), ...]``.
        """
        if dates is None:
            dates = sorted({d for d in (self._date_of(r)
                                        for r in self._flatten(self.active_results)) if d})
        else:
            dates = sorted(set(dates))
        if len(dates) < 2:
            print(f"[S1_Burst] only {len(dates)} date(s); no pairs to form")
            return []
        max_step = min(n_connections + 1, len(dates))
        pairs = [(dates[i], dates[j])
                 for i in range(len(dates) - 1)
                 for j in range(i + 1, min(i + max_step, len(dates)))]
        print(f"[S1_Burst] sequential pairing (tutorial rule): {len(dates)} dates, "
              f"n_connections={n_connections} -> {len(pairs)} pairs")
        return sorted(pairs)

    def select_pairs(self, *args, **kwargs):
        """Baseline-aware pair selection for bursts, computed burst-natively.

        Why this override exists: ASF publishes NO baseline metadata for
        SLC-BURST products -- ``perpendicularBaseline``, ``temporalBaseline``
        and ``insarStackId`` are all None -- so the inherited implementation
        (which reads state vectors / baselines off each product) selects
        against data that does not exist. Measured on a 9-date Hawaii stack:
        the naive path returned 16 pairs instead of 21, silently dropping two
        12-day pairs (the highest-coherence ones) and leaving the final date
        with "0 / 3 connections available".

        The burst-native path in :func:`insarhub.utils.select_pairs` treats
        each acquisition **date** as the pairing node (one date = one stitched
        SAFE after burst2safe): temporal baseline comes from the burst's
        ``startTime`` and perpendicular baseline from per-date orbit state
        vectors (assembled ``.SAFE`` annotation, local ``.EOF``, or POEORB by
        date+mission). No parent-SLC lookup is performed.

        Returns:
            The base implementation's structure, with scene names replaced by
            ``YYYYMMDD`` acquisition dates -- the identifier that is meaningful
            for a burst stack (a date has many bursts, so no single burst
            granule can stand for it).
        """
        # Any configured orbit sources from the current workdir (post-download).
        workdir = getattr(self.config, "workdir", None)
        safe_dir = eof_dir = None
        if workdir:
            from pathlib import Path as _Path
            _w = _Path(workdir)
            slc = _w / "slc"
            if slc.is_dir():
                safe_dir = str(slc)
                eof_dir = str(slc)      # S1_Burst writes .EOF beside the SAFEs
        kwargs.setdefault("burst", True)
        if safe_dir and "safe_dir" not in kwargs:
            kwargs["safe_dir"] = safe_dir
        if eof_dir and "eof_dir" not in kwargs:
            kwargs["eof_dir"] = eof_dir
        kwargs.setdefault("poeorb_cache", _Path.home() / ".insarhub" / "poeorb")
        return super().select_pairs(*args, **kwargs)

    @staticmethod
    def write_ifgram_list(pairs, path) -> Path:
        """Write pairs as ``ifgram_list.txt`` (one ``YYYYMMDD_YYYYMMDD`` per line).

        This is the format the COMPASS stack notebook's
        ``generate_ifgram_pairs`` produces and section 3.3 consumes, so a
        baseline-aware selection can be dropped in place of the purely
        sequential one.
        """
        flat = ([p for v in pairs.values() for p in v]
                if isinstance(pairs, dict) else list(pairs))
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("w") as fh:
            fh.write("# date12\n")
            for a, b in sorted(set(flat)):
                fh.write(f"{a}_{b}\n")
        print(f"[S1_Burst] wrote {len(set(flat))} pair(s) -> {path}")
        return path

    # ------------------------------------------------------------------
    # download
    # ------------------------------------------------------------------

    def download(self, save_path: str | None = None, max_workers: int | None = None,
                 download_orbit: bool = False, force_cdse: bool = False,
                 stop_event=None, on_progress=None, merge: bool = False):
        """Download the selected bursts and assemble them into .SAFE directories.

        Unlike :class:`S1_SLC`, this does not stream files straight from ASF --
        ``burst2safe`` owns the download so it can also fetch each burst's
        annotation/calibration/noise XML and merge them into a coherent product.
        ``max_workers`` is therefore accepted for interface parity but not used;
        burst2safe manages its own concurrency.

        Args:
            save_path: Destination root. Defaults to the configured workdir.
            max_workers: Accepted for parity with S1_SLC; unused (see above).
            download_orbit: Also fetch the matching precise orbits (.EOF), which
                every downstream processor needs.
            force_cdse: Fetch orbits from CDSE instead of ASF.
            stop_event: threading.Event to cancel between date groups.
            on_progress: callback(message, pct) after each date group.
            merge: Assemble every stack into one directory instead of per-stack
                subfolders.

        Returns:
            list[Path]: the assembled .SAFE directories.
        """
        try:
            from burst2safe.burst2safe import burst2safe
        except ImportError as exc:                                  # noqa: BLE001
            raise ImportError(
                "S1_Burst requires the 'burst2safe' package, which provides the "
                "burst -> SAFE assembly step. Install it with:\n"
                "    conda install -c conda-forge burst2safe\n"
                f"(original error: {exc})") from exc

        results = self._flatten(self.active_results)   # property, dict-grouped
        if not results:
            print("[S1_Burst] No search results to download. Run search() first.")
            return []

        # Default to <workdir>/slc, the same layout S1_SLC produces: scenes and
        # their .EOF orbits together in one lowercase "slc" directory.
        out_dir = (Path(save_path) if save_path
                   else Path(getattr(self.config, "workdir", ".") or ".") / "slc")
        out_dir.mkdir(parents=True, exist_ok=True)

        groups = self._group_by_date(results)
        if not groups:
            print("[S1_Burst] No usable burst granules in the current results.")
            return []

        cfg = self.config
        pols = cfg.polarization
        if isinstance(pols, str):
            pols = [pols]
        pols = list(pols) if pols else None

        # --worker N (CLI) / max_workers param overrides the config field;
        # otherwise fall back to cfg.max_workers (GUI-set), then 3.
        nw = max(1, int(max_workers or getattr(self.config, "max_workers", None) or 3))

        n_bursts = sum(len(v) for v in groups.values())
        # Match S1_SLC's summary line. This override never calls
        # super().download(), so the base's banner (asf_base.download) never
        # printed for bursts and the two downloaders reported differently.
        print(f"Downloading {n_bursts} burst(s) across {len(groups)} date group(s)"
              f" ({nw} concurrent)...\n")
        print(f"[S1_Burst] assembling .SAFE in {out_dir}")

        # A date normally maps to exactly one group. It splits when the same
        # date carries more than one relative orbit -- real at high latitude or
        # where an AOI sees both passes, but also what happens when _path_of()
        # cannot parse a granule and returns None, which silently invents a
        # 'pNA' group. Either way the group count then exceeds the scene count,
        # so say which it is rather than leaving an unexplained off-by-N.
        _split = sorted(k for k in groups if "_p" in k)
        if _split:
            _na = [k for k in _split if k.endswith("_pNA")]
            print(f"[S1_Burst] {len(_split)} date(s) split across paths: "
                  f"{', '.join(_split)}")
            if _na:
                logger.warning(
                    "S1_Burst: %d group(s) have an unparseable relative orbit "
                    "(%s). These come from granules whose path could not be read "
                    "and will assemble separately, inflating the group count "
                    "above the number of dates.", len(_na), ", ".join(_na))

        # Same bookkeeping the base download() does for S1_SLC: a stack folder
        # is identified by its insarhub_config.json + workflow marker, and
        # without them the folder is not recognised as a stack by the GUI, the
        # CLI, or the processors. S1_Burst writes its SAFEs into <stack>/slc,
        # so the stack folder is out_dir's parent.
        self._mark_stack_dir(out_dir.parent if out_dir.name == "slc" else out_dir)
        # Authorize the ASF session once before the worker threads race on it.
        _ = self.session
        safes: list[Path] = []
        failed: list[str] = []

        # --worker N runs up to N dates in parallel. Each worker downloads its
        # date's bursts (one per-burst bar at its own position) then assembles
        # the SAFE; the shared assembly bar below counts completed dates.
        from concurrent.futures import ThreadPoolExecutor, as_completed
        date_tasks = [
            (idx, date_str, granules,
             [r for r in results if self._granule_of(r) in set(granules)])
            for idx, (date_str, granules) in enumerate(groups.items())
        ]
        with tqdm(total=len(groups), desc="[S1_Burst] assembling SAFE",
                  unit="SAFE", leave=True, colour="green", position=0) as pbar:
            with ThreadPoolExecutor(max_workers=nw) as ex:
                futures = {
                    ex.submit(self._date_worker, idx, date_str, granules,
                              prods, out_dir, cfg, pols, pbar,
                              stop_event, on_progress, len(groups), nw):
                        (date_str, granules)
                    for idx, date_str, granules, prods in date_tasks
                }
                for fut in as_completed(futures):
                    date_str, granules = futures[fut]
                    try:
                        safe = fut.result()
                    except Exception as exc:                    # noqa: BLE001
                        logger.error("S1_Burst: worker failed for %s: %s",
                                     date_str, exc)
                        safe = None
                    if safe is not None:
                        safes.append(safe)
                    else:
                        failed.append(date_str)
                    if stop_event is not None and stop_event.is_set():
                        for f in futures:
                            f.cancel()
                        break
        if failed:
            print(f"[S1_Burst] {len(failed)} date(s) failed to assemble: "
                  f"{', '.join(sorted(failed))}")
        print(f"[S1_Burst] assembled {len(safes)} / {len(groups)} .SAFE directory(ies).")
        self._report_burst_consistency(groups)

        if download_orbit and safes:
            # Orbits land ALONGSIDE the .SAFE directories, matching S1_SLC,
            # which writes .EOF into the same slc/ folder as the scenes rather
            # than a sibling orbits/ dir.
            self.download_orbit_for_safes(safes, force_cdse=force_cdse)
        return safes

    @staticmethod
    def _burst_of(granule: str) -> str | None:
        """Burst position ``<relativeBurstID>_<subswath>`` from a granule name.

        ``S1_118970_IW2_20240102T131002_VV_A124-BURST`` -> ``118970_IW2``.
        """
        toks = str(granule).split("_")
        for i, t in enumerate(toks):
            if len(t) == 3 and t[:2] in ("IW", "EW") and t[2].isdigit():
                return f"{toks[i - 1]}_{t}" if i else t
        return None

    def _report_burst_consistency(self, groups: dict) -> None:
        """Warn about dates that lack a burst other dates have.

        Runs after assembly, once every date's real content is known. A date
        missing one of the stack's burst positions still assembles -- with
        ``min_bursts=1`` it becomes a short SAFE covering less ground -- and
        nothing downstream flags it. It surfaces much later, and obscurely: the
        burst stacks end up with different date lists, so the interferogram
        network built from one burst prescribes pairs another cannot form, and
        those pairs silently stitch from a single burst into half-width
        products.

        Reported here because this is the first point where the answer is
        knowable and still cheap to act on -- before geocoding hours of data.
        """
        by_date: dict[str, set[str]] = {}
        for key, granules in groups.items():
            date = key[0] if isinstance(key, tuple) else key
            for g in granules:
                b = self._burst_of(g)
                if b:
                    by_date.setdefault(str(date), set()).add(b)
        if not by_date:
            return

        all_bursts = set().union(*by_date.values())
        short = {d: sorted(all_bursts - b) for d, b in by_date.items()
                 if all_bursts - b}
        if not short:
            print(f"[S1_Burst] burst coverage consistent: all "
                  f"{len(by_date)} date(s) have the same {len(all_bursts)} "
                  f"burst position(s)")
            return

        print(f"{Fore.YELLOW}[S1_Burst] WARNING: {len(short)} of "
              f"{len(by_date)} date(s) are missing a burst that other dates "
              f"have. ASF has no data for those burst/date combinations.")
        for d in sorted(short):
            print(f"    {d}  missing {', '.join(short[d])}")
        print(f"  These assemble as SHORT SAFEs (min_bursts="
              f"{getattr(self.config, 'min_bursts', 1)}) covering less ground. "
              f"Downstream, ISCE3_Burst excludes them from the interferogram "
              f"network so it stays formable on every burst -- so they cost "
              f"you those dates. Set min_bursts to {len(all_bursts)} to skip "
              f"them at download time instead.{Fore.RESET}")
        logger.warning("S1_Burst: %d date(s) with incomplete burst coverage: %s",
                       len(short), ", ".join(sorted(short)))

    def _stream_burst_download(self, file_id: str, url: str, dst: Path,
                               expected_bytes: int | None, position: int,
                               stop_event=None) -> bool:
        """Stream one burst's ``.tiff`` with its own tqdm bar.

        Saves to ``<out_dir>/<fileID>.tiff`` — the exact location burst2safe
        expects for its data files, so it skips them during assembly.
        Returns True on success; partial files are removed on failure.
        """
        from asf_search.download.download import _try_get_response
        import asf_search as asf

        # asf_search reports properties["bytes"] as an int for most granules and
        # as a str for some, so the resume check below raised TypeError ('>=' not
        # supported between 'int' and 'str') for whichever dates happened to
        # carry the str form. That escaped to the caller's outer handler, which
        # abandoned the date AFTER burst2safe had copied the measurement tiff
        # but before it wrote the annotation XMLs -- leaving an annotation-less
        # <granule>_0000.SAFE on disk that s1reader later rejects with
        # "burst <id> not in SAFE". Coerce, and treat an unusable value as
        # unknown rather than fatal.
        try:
            expected_bytes = int(expected_bytes) if expected_bytes else None
        except (TypeError, ValueError):
            expected_bytes = None

        if dst.exists() and expected_bytes and dst.stat().st_size >= expected_bytes:
            return True
        thread_session = asf.ASFSession()
        thread_session.cookies.update(self.session.cookies)
        thread_session.headers.update(self.session.headers)
        thread_session.verify = getattr(self.config, "ssl_verify", True)
        try:
            response = _try_get_response(session=thread_session, url=url)
            total = int(response.headers.get("content-length", expected_bytes or 0))
            # desc/format match ASF_Base_Downloader.download's per-file bar, but
            # leave=False is deliberate and differs from S1_SLC: S1_SLC has no
            # aggregate bar, so its per-file bars can persist harmlessly. Here
            # the "assembling SAFE" bar is pinned at position 0, and every
            # left-behind inner bar pushes it down the screen until it scrolls
            # away. Transient inner bars keep the aggregate at the top.
            with tqdm(total=total, unit="B", unit_scale=True, unit_divisor=1024,
                      desc=f"[Worker {position}] {file_id}", leave=False,
                      position=position, colour="green",
                      bar_format="{desc:<60}{percentage:3.0f}%|{bar:25}{r_bar}") as bar:
                with open(dst, "wb") as f:
                    for chunk in response.iter_content(chunk_size=65536):
                        if stop_event is not None and stop_event.is_set():
                            response.close()
                            raise InterruptedError("Download cancelled by user.")
                        if chunk:
                            f.write(chunk)
                            bar.update(len(chunk))
            return True
        except InterruptedError:
            dst.unlink(missing_ok=True)
            raise
        except Exception:                                        # noqa: BLE001
            dst.unlink(missing_ok=True)
            return False

    def _date_worker(self, idx: int, date_str: str, granules: list[str],
                     prods, out_dir: Path, cfg, pols, asm_bar,
                     stop_event, on_progress, n_dates: int, n_workers: int = 1):
        """Download one date's bursts (sequentially) and assemble its SAFE.

        Runs in its own thread (one per active date, ``--worker N`` = N dates
        in parallel). Each burst gets a per-burst bar on this worker's own
        terminal line; the shared ``asm_bar`` at position 0 counts completed
        dates.

        The bar position is the worker SLOT (``idx % n_workers``), not the date
        index. tqdm reserves one terminal line per position, so using the date
        index asked for as many lines as there are dates -- 112 on a real stack
        -- while only ``n_workers`` are ever live, leaving the bars scattered
        down the screen with large gaps. S1_SLC wraps the same way
        (``i % max_workers`` in ASF_Base_Downloader.download).
        """
        from burst2safe.burst2safe import burst2safe

        try:
            for r in prods:
                props = r.properties
                file_id = self._granule_of(r)
                url = props.get("url")
                if not file_id or not url:
                    continue
                if stop_event is not None and stop_event.is_set():
                    return None
                self._stream_burst_download(
                    file_id, url, out_dir / f"{file_id}.tiff",
                    props.get("bytes"),
                    position=(idx % max(1, n_workers)) + 1,
                    stop_event=stop_event)

            buf = io.StringIO()
            safe, exc = None, None
            try:
                # NOTE: no swaths= here. burst2safe ignores it when given an
                # explicit granule list; the filter is already applied in
                # _group_by_date(), so `granules` is exactly what we want.
                with redirect_stdout(buf):
                    safe = burst2safe(
                        granules=granules,
                        polarizations=pols,
                        mode=getattr(cfg, "mode", "IW"),
                        min_bursts=int(getattr(cfg, "min_bursts", 1)),
                        all_anns=bool(getattr(cfg, "all_anns", False)),
                        keep_files=bool(getattr(cfg, "keep_files", False)),
                        work_dir=out_dir,
                    )
            except Exception as e:                              # noqa: BLE001
                # One bad date should not lose the rest of the stack. The
                # granule-list path re-searches ASF by name and reads each
                # product's UMM "InputGranules"; when that field is missing
                # (a burst2safe/asf_search fragility) it raises KeyError. Try
                # the orbit+extent group path as a fallback first.
                exc = e
                with redirect_stdout(buf):
                    safe = self._try_group_assembly(
                        date_str, granules, prods, out_dir, cfg)
            asm_bar.update(1)
            if safe is not None:
                asm_bar.set_postfix_str(f"{date_str} ✓ {Path(safe).name}")
                if exc is not None:
                    asm_bar.write(f"[S1_Burst] {date_str}: recovered via "
                                  f"group fallback -> {Path(safe).name}")
                if on_progress:
                    on_progress(f"assembled {date_str}",
                                int(100.0 * asm_bar.n / max(1, n_dates)))
                return Path(safe)
            tail = "\n".join((buf.getvalue() or "").strip().splitlines()[-8:])
            logger.error("S1_Burst: assembly failed for %s (%s): %s%s",
                         date_str, ", ".join(granules), exc,
                         f"\n  burst2safe output:\n{tail}" if tail else "")
            self._warn_failed_date(date_str, granules, prods, exc)
            self._cleanup_failed_date(date_str, granules, prods, out_dir, cfg)
            if on_progress:
                on_progress(f"assembly failed {date_str}", 0)
            return None
        except Exception as exc:                                # noqa: BLE001
            # Clean up here too, not just on the handled burst2safe failure
            # above: an unexpected raise can land mid-assembly, and a partial
            # .SAFE (measurement written, annotation/ empty) is worse than no
            # .SAFE -- the date silently poisons cslc instead of being absent.
            logger.error("S1_Burst: date %s raised: %s", date_str, exc)
            try:
                self._cleanup_failed_date(date_str, granules, prods, out_dir, cfg)
                self._remove_partial_safe(date_str, out_dir)
            except Exception:                                   # noqa: BLE001
                logger.exception("S1_Burst: cleanup after %s failed", date_str)
            return None

    def _warn_failed_date(self, date_str: str, granules: list[str],
                          prods, exc: Exception) -> None:
        """Explain *why* a date's assembly failed, when the cause is known.

        burst2safe's error text is the source of truth for the common failure
        modes; this makes them legible in the log instead of a bare ``ValueError``:
        - non-consecutive burst IDs (a burst missing in the requested
          polarization) -> name the gap and offer the fix
        - ``InputGranules`` KeyError -> burst2safe/asf_search UMM fragility,
          already covered by the group fallback
        """
        msg = str(exc)
        if "consecutive burst IDs" in msg:
            # e.g. "All bursts must have consecutive burst IDs. Found: [118969, 118971]."
            found = re.findall(r"\d+", msg)
            ids = sorted(set(int(x) for x in found)) if found else []
            gap = ""
            if len(ids) >= 2:
                missing = [i for i in range(ids[0], ids[-1] + 1) if i not in set(ids)]
                if missing:
                    pol = getattr(self.config, "polarization", None)
                    if isinstance(pol, str):
                        pol = [pol]
                    pol_s = ",".join(sorted(pol)) if pol else "the selected polarization"
                    gap = (f" — burst(s) {missing} have no {pol_s} product on ASF, "
                           f"so the remaining bursts cannot form one SAFE. "
                           f"Drop that date, widen the AOI, or add the missing "
                           f"polarization.")
            logger.warning(
                "S1_Burst: %s skipped: bursts %s are not consecutive%s",
                date_str, ids, gap)
        elif "InputGranules" in msg:
            logger.warning(
                "S1_Burst: %s skipped: burst2safe could not resolve the parent "
                "SLC granules (UMM 'InputGranules' missing). The group fallback "
                "was already attempted and also failed.", date_str)
        else:
            logger.warning("S1_Burst: %s skipped: %s", date_str, msg.splitlines()[0])

    @staticmethod
    def _remove_partial_safe(date_str: str, out_dir: Path) -> None:
        """Delete a .SAFE for ``date_str`` that has no annotation XMLs.

        burst2safe writes measurement/ before annotation/, so a run that dies
        in between leaves a directory that looks like a product and is not one.
        Nothing downstream detects it: the campaign counts *.SAFE and calls the
        download satisfied, then cslc fails the whole site on the one bad date.
        Only annotation-less directories are removed -- a complete .SAFE for the
        same date is left alone.
        """
        for safe in out_dir.glob(f"*_{date_str}T*.SAFE"):
            if any((safe / "annotation").glob("*.xml")):
                continue
            shutil.rmtree(safe, ignore_errors=True)
            logger.warning("S1_Burst: removed partial .SAFE %s", safe.name)

    def _cleanup_failed_date(self, date_str: str, granules: list[str],
                             prods, out_dir: Path, cfg) -> None:
        """Remove a failed date's pre-downloaded per-burst ``.tiff`` files.

        On success burst2safe deletes its own data files (``keep_files=False``),
        but on failure the pre-downloaded ``<fileID>.tiff`` files stay behind and
        silently consume ~500 MB each. Remove them here so a failed date leaves
        no orphans; honour ``keep_files`` for users who explicitly asked to keep
        them.
        """
        if bool(getattr(cfg, "keep_files", False)):
            return
        removed: list[str] = []
        for r in prods:
            file_id = self._granule_of(r)
            if not file_id:
                continue
            tiff = out_dir / f"{file_id}.tiff"
            try:
                if tiff.exists():
                    tiff.unlink()
                    removed.append(tiff.name)
            except OSError as e:
                logger.warning("S1_Burst: could not remove %s: %s", tiff, e)
        if removed:
            print(f"[S1_Burst] {date_str}: removed {len(removed)} failed "
                  f"burst .tiff file(s): {', '.join(sorted(removed))}")

    def _try_group_assembly(self, date_str: str, granules: list[str],
                            prods, out_dir: Path, cfg) -> Path | None:
        """Fallback assembly via burst2safe's orbit + extent group path.

        The granule-list path (``burst2safe(granules=...)``) re-searches ASF by
        name and reads each product's UMM ``InputGranules`` to recover the
        parent SLC; when that field is absent on a product the whole call dies
        with ``KeyError('InputGranules')`` (a burst2safe/asf_search fragility)
        and the date's SAFE is silently skipped. The group path searches by
        ``absoluteOrbit`` + footprint instead -- a parameter search that returns
        complete products -- so it can assemble the same bursts even when the
        name-search UMM is incomplete.

        ``prods`` is the date's already-matched ASF products (from the original
        search), not the re-search.

        Returns the assembled SAFE path, or None on any failure.
        """
        try:
            from burst2safe.burst2safe import burst2safe
            from shapely.geometry import shape
            from shapely.ops import unary_union

            prods = [r for r in prods
                     if self._granule_of(r) in set(granules)]
            if not prods:
                logger.error("S1_Burst: group fallback for %s: no products "
                             "matched %s", date_str, granules)
                return None
            orbit = prods[0].properties.get("orbit")
            if orbit is None:
                logger.error("S1_Burst: group fallback for %s: no absolute "
                             "orbit on %s", date_str, prods[0].properties.get("fileID"))
                return None
            geom = unary_union([shape(r.geometry) for r in prods])
            pols = cfg.polarization
            if isinstance(pols, str):
                pols = [pols]
            swaths = getattr(cfg, "swaths", None) or None
            safe = burst2safe(
                orbit=int(orbit),
                extent=geom,
                polarizations=list(pols) if pols else None,
                swaths=swaths,
                mode=getattr(cfg, "mode", "IW"),
                min_bursts=int(getattr(cfg, "min_bursts", 1)),
                all_anns=bool(getattr(cfg, "all_anns", False)),
                keep_files=bool(getattr(cfg, "keep_files", False)),
                work_dir=out_dir,
            )
            print(f"      (fallback group assembly, orbit {orbit}) -> {Path(safe).name}")
            return Path(safe)
        except Exception as exc:                                     # noqa: BLE001
            logger.error("S1_Burst: group fallback failed for %s: %s", date_str, exc)
            return None

    # ------------------------------------------------------------------
    # orbits
    # ------------------------------------------------------------------

    def download_orbit(self, force_cdse: bool = False, save_dir: str | None = None,
                       stop_event=None, scenes=None, merge: bool = False):
        """Fetch orbits for the .SAFE directories already assembled in save_dir.

        Named to match S1_SLC so the generic CLI/GUI paths find it -- both do
        ``hasattr(downloader, "download_orbit")`` and call it with ``save_dir``
        (main.py's --orbit-files handling, ScenePanel's orbit button). Without
        this method those paths silently no-op for bursts.

        Unlike S1_SLC's version it does not use the search results: burst
        granules are not validly-named Sentinel scenes, so orbits are resolved
        from the assembled SAFEs on disk instead (see
        download_orbit_for_safes).
        """
        root = Path(save_dir) if save_dir else Path(getattr(self.config, "workdir", ".") or ".")
        safes = sorted(root.rglob("*.SAFE"))
        if not safes:
            print(f"[S1_Burst] no .SAFE directories under {root}; "
                  f"run download() first, orbits are resolved from assembled SAFEs")
            return []
        return self.download_orbit_for_safes(safes, force_cdse=force_cdse)

    def download_orbit_for_safes(self, safes, save_dir=None, force_cdse: bool = False):
        """Fetch precise orbits for ASSEMBLED SAFEs, not for burst granules.

        S1_SLC.download_orbit() derives orbit names from the search results,
        which works because its results are whole scenes. Burst granules are
        named differently::

            S1_264306_IW3_20240915T043118_VV_F23F-BURST

        and sentineleof rejects them outright ("Invalid Sentinel filename"), so
        delegating to S1_SLC yields one error per burst and no orbits. The SAFE
        directories burst2safe produces *are* validly named, so orbits are
        resolved from those instead -- one per acquisition rather than one per
        burst, which is also what the notebook's `eof --search-path <slc_dir>`
        does.
        """
        from eof.download import download_eofs

        # Alongside the SAFEs by default -- S1_SLC puts .EOF in the same slc/
        # directory as the scenes, and downstream tools scan one folder.
        if save_dir is None:
            parents = {Path(s).parent for s in safes}
            save_dir = parents.pop() if len(parents) == 1 else Path(".")
        save_dir = Path(save_dir)
        save_dir.mkdir(parents=True, exist_ok=True)

        got: list[Path] = []
        with tqdm(total=len(safes), desc="[S1_Burst] downloading orbits",
                  unit="SAFE", leave=True, colour="cyan") as pbar:
            for safe in safes:
                safe = Path(safe)
                pbar.set_postfix_str(safe.name)
                try:
                    got += download_eofs(sentinel_file=str(safe),
                                         save_dir=str(save_dir),
                                         orbit_type="precise",
                                         force_asf=not force_cdse)
                except Exception as exc:                            # noqa: BLE001
                    logger.error("S1_Burst: orbit download failed for %s: %s",
                                 safe.name, exc)
                pbar.update(1)
        uniq = sorted({Path(p).name for p in got})
        print(f"[S1_Burst] orbits: {len(uniq)} file(s) -> {save_dir}")
        for n in uniq:
            print(f"    {n}")
        return [save_dir / n for n in uniq]

_NON_SEARCH_FIELDS = ASF_Base_Downloader._NON_SEARCH_FIELDS | {'swaths', 'mode', 'min_bursts', 'all_anns', 'keep_files', 'frame', 'asfFrame'} class-attribute instance-attribute

_PARENT_RE = re.compile('(S1[ABCD]_IW_SLC__\\w{4}_\\d{8}T\\d{6}_\\d{8}T\\d{6}_\\d{6}_\\w{6}_\\w{4})') class-attribute instance-attribute

default_config = S1_Burst_Config class-attribute instance-attribute

description = 'Sentinel-1 SLC-BURST search and download, assembled into .SAFE via burst2safe.' class-attribute instance-attribute

name = 'S1_Burst' class-attribute instance-attribute

product_label = 'bursts' class-attribute instance-attribute

search_filter_schema = [{'name': 'flightDirection', 'label': 'Flight Direction', 'kind': 'select', 'group': 'Additional Filters', 'choices': ['ASCENDING', 'DESCENDING']}, {'name': 'platform', 'label': 'Platform', 'kind': 'select', 'group': 'Additional Filters', 'choices': ['Sentinel-1A', 'Sentinel-1B', 'Sentinel-1C', 'Sentinel-1D']}, {'name': 'polarization', 'label': 'Polarization', 'kind': 'select', 'group': 'Additional Filters', 'choices': ['VV', 'VV+VH', 'HH', 'HH+HV']}, {'name': 'relativeOrbit', 'label': 'Path', 'kind': 'range', 'group': 'Path and Frame Filters'}, {'name': 'fullBurstID', 'label': 'Burst ID', 'kind': 'text', 'group': 'Path and Frame Filters'}] class-attribute instance-attribute

stack_key_label = 'Burst_ID' class-attribute instance-attribute

_burst_id_parts(text) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def _burst_id_parts(text) -> tuple[int, int, str] | None:
    """``"087_185682_IW2"`` -> ``(87, 185682, "IW2")``; None if not that shape.

    Path and index come back as ints so the zero-padded form ASF prints
    (``087_...``) and the bare form a user types (``87_...``) compare equal.
    """
    parts = str(text).strip().upper().split("_")
    if len(parts) != 3:
        return None
    path, index, swath = parts
    if not (path.isdigit() and index.isdigit()):
        return None
    return int(path), int(index), swath

_burst_of(granule) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def _burst_of(granule: str) -> str | None:
    """Burst position ``<relativeBurstID>_<subswath>`` from a granule name.

    ``S1_118970_IW2_20240102T131002_VV_A124-BURST`` -> ``118970_IW2``.
    """
    toks = str(granule).split("_")
    for i, t in enumerate(toks):
        if len(t) == 3 and t[:2] in ("IW", "EW") and t[2].isdigit():
            return f"{toks[i - 1]}_{t}" if i else t
    return None

_cleanup_failed_date(date_str, granules, prods, out_dir, cfg)

Source code in src/insarhub/downloader/s1_burst.py

def _cleanup_failed_date(self, date_str: str, granules: list[str],
                         prods, out_dir: Path, cfg) -> None:
    """Remove a failed date's pre-downloaded per-burst ``.tiff`` files.

    On success burst2safe deletes its own data files (``keep_files=False``),
    but on failure the pre-downloaded ``<fileID>.tiff`` files stay behind and
    silently consume ~500 MB each. Remove them here so a failed date leaves
    no orphans; honour ``keep_files`` for users who explicitly asked to keep
    them.
    """
    if bool(getattr(cfg, "keep_files", False)):
        return
    removed: list[str] = []
    for r in prods:
        file_id = self._granule_of(r)
        if not file_id:
            continue
        tiff = out_dir / f"{file_id}.tiff"
        try:
            if tiff.exists():
                tiff.unlink()
                removed.append(tiff.name)
        except OSError as e:
            logger.warning("S1_Burst: could not remove %s: %s", tiff, e)
    if removed:
        print(f"[S1_Burst] {date_str}: removed {len(removed)} failed "
              f"burst .tiff file(s): {', '.join(sorted(removed))}")

_date_of(result) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def _date_of(result) -> str | None:
    """YYYYMMDD acquisition date, used to group bursts into one SAFE per date."""
    props = getattr(result, "properties", None) or {}
    start = props.get("startTime") or props.get("stopTime")
    if not start:
        return None
    # startTime is ISO-8601: 2024-09-15T16:12:34.000000Z
    return str(start)[:10].replace("-", "")

_date_worker(idx, date_str, granules, prods, out_dir, cfg, pols, asm_bar, stop_event, on_progress, n_dates, n_workers=1)

Source code in src/insarhub/downloader/s1_burst.py

def _date_worker(self, idx: int, date_str: str, granules: list[str],
                 prods, out_dir: Path, cfg, pols, asm_bar,
                 stop_event, on_progress, n_dates: int, n_workers: int = 1):
    """Download one date's bursts (sequentially) and assemble its SAFE.

    Runs in its own thread (one per active date, ``--worker N`` = N dates
    in parallel). Each burst gets a per-burst bar on this worker's own
    terminal line; the shared ``asm_bar`` at position 0 counts completed
    dates.

    The bar position is the worker SLOT (``idx % n_workers``), not the date
    index. tqdm reserves one terminal line per position, so using the date
    index asked for as many lines as there are dates -- 112 on a real stack
    -- while only ``n_workers`` are ever live, leaving the bars scattered
    down the screen with large gaps. S1_SLC wraps the same way
    (``i % max_workers`` in ASF_Base_Downloader.download).
    """
    from burst2safe.burst2safe import burst2safe

    try:
        for r in prods:
            props = r.properties
            file_id = self._granule_of(r)
            url = props.get("url")
            if not file_id or not url:
                continue
            if stop_event is not None and stop_event.is_set():
                return None
            self._stream_burst_download(
                file_id, url, out_dir / f"{file_id}.tiff",
                props.get("bytes"),
                position=(idx % max(1, n_workers)) + 1,
                stop_event=stop_event)

        buf = io.StringIO()
        safe, exc = None, None
        try:
            # NOTE: no swaths= here. burst2safe ignores it when given an
            # explicit granule list; the filter is already applied in
            # _group_by_date(), so `granules` is exactly what we want.
            with redirect_stdout(buf):
                safe = burst2safe(
                    granules=granules,
                    polarizations=pols,
                    mode=getattr(cfg, "mode", "IW"),
                    min_bursts=int(getattr(cfg, "min_bursts", 1)),
                    all_anns=bool(getattr(cfg, "all_anns", False)),
                    keep_files=bool(getattr(cfg, "keep_files", False)),
                    work_dir=out_dir,
                )
        except Exception as e:                              # noqa: BLE001
            # One bad date should not lose the rest of the stack. The
            # granule-list path re-searches ASF by name and reads each
            # product's UMM "InputGranules"; when that field is missing
            # (a burst2safe/asf_search fragility) it raises KeyError. Try
            # the orbit+extent group path as a fallback first.
            exc = e
            with redirect_stdout(buf):
                safe = self._try_group_assembly(
                    date_str, granules, prods, out_dir, cfg)
        asm_bar.update(1)
        if safe is not None:
            asm_bar.set_postfix_str(f"{date_str} ✓ {Path(safe).name}")
            if exc is not None:
                asm_bar.write(f"[S1_Burst] {date_str}: recovered via "
                              f"group fallback -> {Path(safe).name}")
            if on_progress:
                on_progress(f"assembled {date_str}",
                            int(100.0 * asm_bar.n / max(1, n_dates)))
            return Path(safe)
        tail = "\n".join((buf.getvalue() or "").strip().splitlines()[-8:])
        logger.error("S1_Burst: assembly failed for %s (%s): %s%s",
                     date_str, ", ".join(granules), exc,
                     f"\n  burst2safe output:\n{tail}" if tail else "")
        self._warn_failed_date(date_str, granules, prods, exc)
        self._cleanup_failed_date(date_str, granules, prods, out_dir, cfg)
        if on_progress:
            on_progress(f"assembly failed {date_str}", 0)
        return None
    except Exception as exc:                                # noqa: BLE001
        # Clean up here too, not just on the handled burst2safe failure
        # above: an unexpected raise can land mid-assembly, and a partial
        # .SAFE (measurement written, annotation/ empty) is worse than no
        # .SAFE -- the date silently poisons cslc instead of being absent.
        logger.error("S1_Burst: date %s raised: %s", date_str, exc)
        try:
            self._cleanup_failed_date(date_str, granules, prods, out_dir, cfg)
            self._remove_partial_safe(date_str, out_dir)
        except Exception:                                   # noqa: BLE001
            logger.exception("S1_Burst: cleanup after %s failed", date_str)
        return None

_flatten(results) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def _flatten(results) -> list:
    """Flatten ``active_results``.

    ``ASF_Base_Downloader.active_results`` is a property returning a dict
    grouped by (path, frame) -- one entry per stack when an AOI spans more
    than one. Burst assembly is per-acquisition, so the grouping is redone
    by date here; flatten first.
    """
    if isinstance(results, dict):
        return [r for group in results.values() for r in group]
    return list(results)

_get_group_key(result)

Source code in src/insarhub/downloader/s1_burst.py

def _get_group_key(self, result) -> tuple:
    """Group bursts by ``(path, fullBurstID)``: one fixed burst position.

    The base implementation groups by ``(pathNumber, frameNumber)``, but
    ASF burst granules carry NO ``frameNumber`` -- so every burst of a
    track collapsed onto ``(124, None)`` and the GUI rendered them all as
    "Path 0 · Frame 0". A burst stack is a burst position that repeats
    every revisit, uniquely identified by its OPERA ``fullBurstID``
    ("124_266256_IW3" = path_burstIndex_subswath).
    """
    props = result.properties
    path = props.get("pathNumber")
    if path is None:
        path = self._path_of(result)
    b = props.get("burst") or {}
    full = (props.get("fullBurstID") or b.get("fullBurstID")
            or props.get("relativeBurstID") or b.get("relativeBurstID"))
    if not full:
        # Last resort: the granule name always carries the subswath, and is
        # a real identifier. The previous fallback built "?_?" from missing
        # burstIndex/subswath -- a placeholder that reads as data, groups
        # every unidentifiable burst together, and (via the base download's
        # group-key persistence) ended up in configs as frame="?_?".
        g = self._granule_of(result) or ""
        full = g or f"unknown_{self._date_of(result) or 'nodate'}"
        logger.warning("S1_Burst: result carries no burst ID; grouping by "
                       "granule name %r instead", full)
    return (path, full)

_granule_of(result) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def _granule_of(result) -> str | None:
    """ASF granule name for a search result, tolerating shape differences."""
    props = getattr(result, "properties", None) or {}
    return props.get("fileID") or props.get("sceneName") or props.get("granuleName")

_group_by_date(results)

Source code in src/insarhub/downloader/s1_burst.py

def _group_by_date(self, results) -> dict[str, list[str]]:
    """{"YYYYMMDD" or "YYYYMMDD_pNNN": [granule, ...]} -- one SAFE per group.

    Grouped by acquisition date AND path. A SAFE is a single pass, and
    burst2safe enforces "all bursts must have the same absolute orbit"; a
    date-only key would merge two paths imaged on the same day into one
    call, which raises and -- because assembly failures are caught per group
    -- would silently drop BOTH paths for that date.

    Two paths on one date is uncommon for a small AOI (over Hawaii, paths
    14/87/124 all fall on different days) but is normal where passes
    converge at high latitude or where an AOI sees both ascending and
    descending on the same day. The key stays the bare date in the common
    single-path case so output names are unchanged.
    """
    # A bare string is tolerated: swaths round-trips through JSON configs
    # and a hand-edited "IW2" would otherwise iterate as {"I","W","2"} and
    # drop every granule.
    _sw = getattr(self.config, "swaths", None) or []
    want = {s.upper() for s in ([_sw] if isinstance(_sw, str) else _sw)}
    # s1reader derives the OPERA burst ID from the IW2 mid-burst sensing
    # time, so it opens the IW2 ANNOTATION unconditionally -- whichever
    # subswath you asked for:
    #     ValueError: burst iw2-slc-vv not in SAFE: <dir>
    #
    # But annotation is all it needs, not IW2 measurement data. Verified by
    # assembling an IW3-ONLY SAFE with all_anns=True (measurement: iw3;
    # annotation: iw1+iw2+iw3) -- s1reader loaded all 4 IW3 bursts fine.
    # So:
    #   all_anns=True  -> slice-level annotation for every subswath is
    #                     included anyway; adding IW2 DATA would roughly
    #                     double the download for nothing.
    #   all_anns=False -> only the requested subswaths' annotation is kept,
    #                     so IW2 must be pulled in or the SAFE is unreadable.
    if (want and str(getattr(self.config, "mode", "IW")).upper() == "IW"
            and "IW2" not in want
            and not bool(getattr(self.config, "all_anns", False))):
        print(f"[S1_Burst] adding IW2 to swaths {sorted(want)}: s1reader needs the "
              f"IW2 annotation for the burst-ID reference time, and all_anns is "
              f"off. Set all_anns=True to keep {sorted(want)} only and avoid "
              f"downloading IW2 data.")
        want = want | {"IW2"}
    groups: dict[str, list[str]] = defaultdict(list)
    skipped = dropped = 0
    for r in self._flatten(results):
        g, d = self._granule_of(r), self._date_of(r)
        if not g or not d:
            skipped += 1
            continue
        # config.swaths must be applied HERE. It cannot be applied at search
        # time (ASF leaves beamSwath empty on BURST products), and it cannot
        # be delegated to burst2safe either: when burst2safe is given an
        # explicit granule list it assembles exactly those granules and
        # ignores its own `swaths` argument. Verified against a live
        # download -- swaths=["IW3"] still produced an IW2+IW3 SAFE.
        if want and self._swath_of(g) not in want:
            dropped += 1
            continue
        groups[(d, self._path_of(r))].append(g)
    if skipped:
        logger.warning("S1_Burst: %d result(s) lacked a granule name or start "
                       "time and were skipped", skipped)
    if dropped:
        print(f"[S1_Burst] swath filter {sorted(want)}: dropped {dropped} burst(s)")

    # Flatten (date, path) -> label. Keep the bare date when a date has only
    # one path (the normal case), so names match what callers already expect.
    per_date: dict[str, int] = defaultdict(int)
    for (d, _p) in groups:
        per_date[d] += 1
    out: dict[str, list[str]] = {}
    for (d, path), gr in sorted(groups.items(), key=lambda kv: (kv[0][0], kv[0][1] or 0)):
        key = d if per_date[d] == 1 else f"{d}_p{path if path is not None else 'NA'}"
        out[key] = gr
    multi = [d for d, n in per_date.items() if n > 1]
    if multi:
        print(f"[S1_Burst] {len(multi)} date(s) span multiple paths; assembling "
              f"one SAFE per path: {sorted(multi)}")
    return out

_path_of(result) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def _path_of(result) -> int | None:
    """Relative orbit (track). ASF leaves ``relativeOrbit`` empty on BURST
    products; the value lives in ``pathNumber``, and also as the first field
    of ``burst.fullBurstID`` ("124_264312_IW2")."""
    props = getattr(result, "properties", None) or {}
    p = props.get("pathNumber") or props.get("relativeOrbit")
    if p is not None:
        return int(p)
    fid = (props.get("burst") or {}).get("fullBurstID") or ""
    head = str(fid).split("_")[0]
    return int(head) if head.isdigit() else None

_remove_partial_safe(date_str, out_dir) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def _remove_partial_safe(date_str: str, out_dir: Path) -> None:
    """Delete a .SAFE for ``date_str`` that has no annotation XMLs.

    burst2safe writes measurement/ before annotation/, so a run that dies
    in between leaves a directory that looks like a product and is not one.
    Nothing downstream detects it: the campaign counts *.SAFE and calls the
    download satisfied, then cslc fails the whole site on the one bad date.
    Only annotation-less directories are removed -- a complete .SAFE for the
    same date is left alone.
    """
    for safe in out_dir.glob(f"*_{date_str}T*.SAFE"):
        if any((safe / "annotation").glob("*.xml")):
            continue
        shutil.rmtree(safe, ignore_errors=True)
        logger.warning("S1_Burst: removed partial .SAFE %s", safe.name)

_report_burst_consistency(groups)

Source code in src/insarhub/downloader/s1_burst.py

def _report_burst_consistency(self, groups: dict) -> None:
    """Warn about dates that lack a burst other dates have.

    Runs after assembly, once every date's real content is known. A date
    missing one of the stack's burst positions still assembles -- with
    ``min_bursts=1`` it becomes a short SAFE covering less ground -- and
    nothing downstream flags it. It surfaces much later, and obscurely: the
    burst stacks end up with different date lists, so the interferogram
    network built from one burst prescribes pairs another cannot form, and
    those pairs silently stitch from a single burst into half-width
    products.

    Reported here because this is the first point where the answer is
    knowable and still cheap to act on -- before geocoding hours of data.
    """
    by_date: dict[str, set[str]] = {}
    for key, granules in groups.items():
        date = key[0] if isinstance(key, tuple) else key
        for g in granules:
            b = self._burst_of(g)
            if b:
                by_date.setdefault(str(date), set()).add(b)
    if not by_date:
        return

    all_bursts = set().union(*by_date.values())
    short = {d: sorted(all_bursts - b) for d, b in by_date.items()
             if all_bursts - b}
    if not short:
        print(f"[S1_Burst] burst coverage consistent: all "
              f"{len(by_date)} date(s) have the same {len(all_bursts)} "
              f"burst position(s)")
        return

    print(f"{Fore.YELLOW}[S1_Burst] WARNING: {len(short)} of "
          f"{len(by_date)} date(s) are missing a burst that other dates "
          f"have. ASF has no data for those burst/date combinations.")
    for d in sorted(short):
        print(f"    {d}  missing {', '.join(short[d])}")
    print(f"  These assemble as SHORT SAFEs (min_bursts="
          f"{getattr(self.config, 'min_bursts', 1)}) covering less ground. "
          f"Downstream, ISCE3_Burst excludes them from the interferogram "
          f"network so it stays formable on every burst -- so they cost "
          f"you those dates. Set min_bursts to {len(all_bursts)} to skip "
          f"them at download time instead.{Fore.RESET}")
    logger.warning("S1_Burst: %d date(s) with incomplete burst coverage: %s",
                   len(short), ", ".join(sorted(short)))

_stack_key_matches(key, target)

Source code in src/insarhub/downloader/s1_burst.py

def _stack_key_matches(self, key: tuple, target: tuple) -> bool:
    """Match a burst stack against a user ``PATH:SELECTOR`` token.

    Burst stacks key on ``fullBurstID`` (see :meth:`_get_group_key`), so the
    second half of a ``--stacks`` token is not a number and the base class's
    plain-equality rule can never match it. Three spellings of one stack are
    accepted, widest last::

        124:124_264305_IW2   full burst ID, exactly as summary() prints it
        124:264305_IW2       burst index + subswath
        124:264305           bare burst index -- EVERY subswath at that index

    The bare index is deliberately one-to-many: ASF reuses an index across
    subswaths (``124_266256_IW2`` and ``124_266256_IW3`` both exist), so it
    selects both. Name the subswath to pin exactly one.

    A stack that fell back to grouping by granule name (no burst ID on the
    product at all) matches only on a verbatim string, since it has no index
    or subswath to compare against.
    """
    try:
        if int(key[0]) != int(target[0]):
            return False
    except (TypeError, ValueError):
        return False   # a stack with no path can only be reached by AOI, not by token

    selector = str(target[1]).strip().upper()
    full     = str(key[1]).strip().upper()
    if not selector:
        return False
    if selector == full:
        return True

    key_parts = self._burst_id_parts(full)
    if key_parts is None:
        return False            # granule-name fallback key: verbatim match only
    _, index, swath = key_parts

    sel_parts = self._burst_id_parts(selector)
    if sel_parts is not None:
        return sel_parts == key_parts

    bits = selector.split("_")
    if len(bits) == 2 and bits[0].isdigit():
        return int(bits[0]) == index and bits[1] == swath
    return selector.isdigit() and int(selector) == index

_stream_burst_download(file_id, url, dst, expected_bytes, position, stop_event=None)

Source code in src/insarhub/downloader/s1_burst.py

def _stream_burst_download(self, file_id: str, url: str, dst: Path,
                           expected_bytes: int | None, position: int,
                           stop_event=None) -> bool:
    """Stream one burst's ``.tiff`` with its own tqdm bar.

    Saves to ``<out_dir>/<fileID>.tiff`` — the exact location burst2safe
    expects for its data files, so it skips them during assembly.
    Returns True on success; partial files are removed on failure.
    """
    from asf_search.download.download import _try_get_response
    import asf_search as asf

    # asf_search reports properties["bytes"] as an int for most granules and
    # as a str for some, so the resume check below raised TypeError ('>=' not
    # supported between 'int' and 'str') for whichever dates happened to
    # carry the str form. That escaped to the caller's outer handler, which
    # abandoned the date AFTER burst2safe had copied the measurement tiff
    # but before it wrote the annotation XMLs -- leaving an annotation-less
    # <granule>_0000.SAFE on disk that s1reader later rejects with
    # "burst <id> not in SAFE". Coerce, and treat an unusable value as
    # unknown rather than fatal.
    try:
        expected_bytes = int(expected_bytes) if expected_bytes else None
    except (TypeError, ValueError):
        expected_bytes = None

    if dst.exists() and expected_bytes and dst.stat().st_size >= expected_bytes:
        return True
    thread_session = asf.ASFSession()
    thread_session.cookies.update(self.session.cookies)
    thread_session.headers.update(self.session.headers)
    thread_session.verify = getattr(self.config, "ssl_verify", True)
    try:
        response = _try_get_response(session=thread_session, url=url)
        total = int(response.headers.get("content-length", expected_bytes or 0))
        # desc/format match ASF_Base_Downloader.download's per-file bar, but
        # leave=False is deliberate and differs from S1_SLC: S1_SLC has no
        # aggregate bar, so its per-file bars can persist harmlessly. Here
        # the "assembling SAFE" bar is pinned at position 0, and every
        # left-behind inner bar pushes it down the screen until it scrolls
        # away. Transient inner bars keep the aggregate at the top.
        with tqdm(total=total, unit="B", unit_scale=True, unit_divisor=1024,
                  desc=f"[Worker {position}] {file_id}", leave=False,
                  position=position, colour="green",
                  bar_format="{desc:<60}{percentage:3.0f}%|{bar:25}{r_bar}") as bar:
            with open(dst, "wb") as f:
                for chunk in response.iter_content(chunk_size=65536):
                    if stop_event is not None and stop_event.is_set():
                        response.close()
                        raise InterruptedError("Download cancelled by user.")
                    if chunk:
                        f.write(chunk)
                        bar.update(len(chunk))
        return True
    except InterruptedError:
        dst.unlink(missing_ok=True)
        raise
    except Exception:                                        # noqa: BLE001
        dst.unlink(missing_ok=True)
        return False

_swath_of(granule) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def _swath_of(granule: str) -> str | None:
    """Subswath (IW1/IW2/IW3, EW1..EW5) parsed from the granule name.

    ASF does NOT populate ``beamSwath`` on SLC-BURST products -- it comes
    back None -- so the granule name is the only reliable source::

        S1_264306_IW3_20240915T043118_VV_F23F-BURST
                  ^^^
    """
    for tok in str(granule).split("_"):
        if len(tok) == 3 and tok[:2] in ("IW", "EW") and tok[2].isdigit():
            return tok
    return None

_try_group_assembly(date_str, granules, prods, out_dir, cfg)

Source code in src/insarhub/downloader/s1_burst.py

def _try_group_assembly(self, date_str: str, granules: list[str],
                        prods, out_dir: Path, cfg) -> Path | None:
    """Fallback assembly via burst2safe's orbit + extent group path.

    The granule-list path (``burst2safe(granules=...)``) re-searches ASF by
    name and reads each product's UMM ``InputGranules`` to recover the
    parent SLC; when that field is absent on a product the whole call dies
    with ``KeyError('InputGranules')`` (a burst2safe/asf_search fragility)
    and the date's SAFE is silently skipped. The group path searches by
    ``absoluteOrbit`` + footprint instead -- a parameter search that returns
    complete products -- so it can assemble the same bursts even when the
    name-search UMM is incomplete.

    ``prods`` is the date's already-matched ASF products (from the original
    search), not the re-search.

    Returns the assembled SAFE path, or None on any failure.
    """
    try:
        from burst2safe.burst2safe import burst2safe
        from shapely.geometry import shape
        from shapely.ops import unary_union

        prods = [r for r in prods
                 if self._granule_of(r) in set(granules)]
        if not prods:
            logger.error("S1_Burst: group fallback for %s: no products "
                         "matched %s", date_str, granules)
            return None
        orbit = prods[0].properties.get("orbit")
        if orbit is None:
            logger.error("S1_Burst: group fallback for %s: no absolute "
                         "orbit on %s", date_str, prods[0].properties.get("fileID"))
            return None
        geom = unary_union([shape(r.geometry) for r in prods])
        pols = cfg.polarization
        if isinstance(pols, str):
            pols = [pols]
        swaths = getattr(cfg, "swaths", None) or None
        safe = burst2safe(
            orbit=int(orbit),
            extent=geom,
            polarizations=list(pols) if pols else None,
            swaths=swaths,
            mode=getattr(cfg, "mode", "IW"),
            min_bursts=int(getattr(cfg, "min_bursts", 1)),
            all_anns=bool(getattr(cfg, "all_anns", False)),
            keep_files=bool(getattr(cfg, "keep_files", False)),
            work_dir=out_dir,
        )
        print(f"      (fallback group assembly, orbit {orbit}) -> {Path(safe).name}")
        return Path(safe)
    except Exception as exc:                                     # noqa: BLE001
        logger.error("S1_Burst: group fallback failed for %s: %s", date_str, exc)
        return None

_warn_failed_date(date_str, granules, prods, exc)

Source code in src/insarhub/downloader/s1_burst.py

def _warn_failed_date(self, date_str: str, granules: list[str],
                      prods, exc: Exception) -> None:
    """Explain *why* a date's assembly failed, when the cause is known.

    burst2safe's error text is the source of truth for the common failure
    modes; this makes them legible in the log instead of a bare ``ValueError``:
    - non-consecutive burst IDs (a burst missing in the requested
      polarization) -> name the gap and offer the fix
    - ``InputGranules`` KeyError -> burst2safe/asf_search UMM fragility,
      already covered by the group fallback
    """
    msg = str(exc)
    if "consecutive burst IDs" in msg:
        # e.g. "All bursts must have consecutive burst IDs. Found: [118969, 118971]."
        found = re.findall(r"\d+", msg)
        ids = sorted(set(int(x) for x in found)) if found else []
        gap = ""
        if len(ids) >= 2:
            missing = [i for i in range(ids[0], ids[-1] + 1) if i not in set(ids)]
            if missing:
                pol = getattr(self.config, "polarization", None)
                if isinstance(pol, str):
                    pol = [pol]
                pol_s = ",".join(sorted(pol)) if pol else "the selected polarization"
                gap = (f" — burst(s) {missing} have no {pol_s} product on ASF, "
                       f"so the remaining bursts cannot form one SAFE. "
                       f"Drop that date, widen the AOI, or add the missing "
                       f"polarization.")
        logger.warning(
            "S1_Burst: %s skipped: bursts %s are not consecutive%s",
            date_str, ids, gap)
    elif "InputGranules" in msg:
        logger.warning(
            "S1_Burst: %s skipped: burst2safe could not resolve the parent "
            "SLC granules (UMM 'InputGranules' missing). The group fallback "
            "was already attempted and also failed.", date_str)
    else:
        logger.warning("S1_Burst: %s skipped: %s", date_str, msg.splitlines()[0])

download(save_path=None, max_workers=None, download_orbit=False, force_cdse=False, stop_event=None, on_progress=None, merge=False)

Parameters:

Name	Type	Description	Default
save_path	str | None	Destination root. Defaults to the configured workdir.	None
max_workers	int | None	Accepted for parity with S1_SLC; unused (see above).	None
download_orbit	bool	Also fetch the matching precise orbits (.EOF), which every downstream processor needs.	False
force_cdse	bool	Fetch orbits from CDSE instead of ASF.	False
stop_event		threading.Event to cancel between date groups.	None
on_progress		callback(message, pct) after each date group.	None
merge	bool	Assemble every stack into one directory instead of per-stack subfolders.	False

Source code in src/insarhub/downloader/s1_burst.py

def download(self, save_path: str | None = None, max_workers: int | None = None,
             download_orbit: bool = False, force_cdse: bool = False,
             stop_event=None, on_progress=None, merge: bool = False):
    """Download the selected bursts and assemble them into .SAFE directories.

    Unlike :class:`S1_SLC`, this does not stream files straight from ASF --
    ``burst2safe`` owns the download so it can also fetch each burst's
    annotation/calibration/noise XML and merge them into a coherent product.
    ``max_workers`` is therefore accepted for interface parity but not used;
    burst2safe manages its own concurrency.

    Args:
        save_path: Destination root. Defaults to the configured workdir.
        max_workers: Accepted for parity with S1_SLC; unused (see above).
        download_orbit: Also fetch the matching precise orbits (.EOF), which
            every downstream processor needs.
        force_cdse: Fetch orbits from CDSE instead of ASF.
        stop_event: threading.Event to cancel between date groups.
        on_progress: callback(message, pct) after each date group.
        merge: Assemble every stack into one directory instead of per-stack
            subfolders.

    Returns:
        list[Path]: the assembled .SAFE directories.
    """
    try:
        from burst2safe.burst2safe import burst2safe
    except ImportError as exc:                                  # noqa: BLE001
        raise ImportError(
            "S1_Burst requires the 'burst2safe' package, which provides the "
            "burst -> SAFE assembly step. Install it with:\n"
            "    conda install -c conda-forge burst2safe\n"
            f"(original error: {exc})") from exc

    results = self._flatten(self.active_results)   # property, dict-grouped
    if not results:
        print("[S1_Burst] No search results to download. Run search() first.")
        return []

    # Default to <workdir>/slc, the same layout S1_SLC produces: scenes and
    # their .EOF orbits together in one lowercase "slc" directory.
    out_dir = (Path(save_path) if save_path
               else Path(getattr(self.config, "workdir", ".") or ".") / "slc")
    out_dir.mkdir(parents=True, exist_ok=True)

    groups = self._group_by_date(results)
    if not groups:
        print("[S1_Burst] No usable burst granules in the current results.")
        return []

    cfg = self.config
    pols = cfg.polarization
    if isinstance(pols, str):
        pols = [pols]
    pols = list(pols) if pols else None

    # --worker N (CLI) / max_workers param overrides the config field;
    # otherwise fall back to cfg.max_workers (GUI-set), then 3.
    nw = max(1, int(max_workers or getattr(self.config, "max_workers", None) or 3))

    n_bursts = sum(len(v) for v in groups.values())
    # Match S1_SLC's summary line. This override never calls
    # super().download(), so the base's banner (asf_base.download) never
    # printed for bursts and the two downloaders reported differently.
    print(f"Downloading {n_bursts} burst(s) across {len(groups)} date group(s)"
          f" ({nw} concurrent)...\n")
    print(f"[S1_Burst] assembling .SAFE in {out_dir}")

    # A date normally maps to exactly one group. It splits when the same
    # date carries more than one relative orbit -- real at high latitude or
    # where an AOI sees both passes, but also what happens when _path_of()
    # cannot parse a granule and returns None, which silently invents a
    # 'pNA' group. Either way the group count then exceeds the scene count,
    # so say which it is rather than leaving an unexplained off-by-N.
    _split = sorted(k for k in groups if "_p" in k)
    if _split:
        _na = [k for k in _split if k.endswith("_pNA")]
        print(f"[S1_Burst] {len(_split)} date(s) split across paths: "
              f"{', '.join(_split)}")
        if _na:
            logger.warning(
                "S1_Burst: %d group(s) have an unparseable relative orbit "
                "(%s). These come from granules whose path could not be read "
                "and will assemble separately, inflating the group count "
                "above the number of dates.", len(_na), ", ".join(_na))

    # Same bookkeeping the base download() does for S1_SLC: a stack folder
    # is identified by its insarhub_config.json + workflow marker, and
    # without them the folder is not recognised as a stack by the GUI, the
    # CLI, or the processors. S1_Burst writes its SAFEs into <stack>/slc,
    # so the stack folder is out_dir's parent.
    self._mark_stack_dir(out_dir.parent if out_dir.name == "slc" else out_dir)
    # Authorize the ASF session once before the worker threads race on it.
    _ = self.session
    safes: list[Path] = []
    failed: list[str] = []

    # --worker N runs up to N dates in parallel. Each worker downloads its
    # date's bursts (one per-burst bar at its own position) then assembles
    # the SAFE; the shared assembly bar below counts completed dates.
    from concurrent.futures import ThreadPoolExecutor, as_completed
    date_tasks = [
        (idx, date_str, granules,
         [r for r in results if self._granule_of(r) in set(granules)])
        for idx, (date_str, granules) in enumerate(groups.items())
    ]
    with tqdm(total=len(groups), desc="[S1_Burst] assembling SAFE",
              unit="SAFE", leave=True, colour="green", position=0) as pbar:
        with ThreadPoolExecutor(max_workers=nw) as ex:
            futures = {
                ex.submit(self._date_worker, idx, date_str, granules,
                          prods, out_dir, cfg, pols, pbar,
                          stop_event, on_progress, len(groups), nw):
                    (date_str, granules)
                for idx, date_str, granules, prods in date_tasks
            }
            for fut in as_completed(futures):
                date_str, granules = futures[fut]
                try:
                    safe = fut.result()
                except Exception as exc:                    # noqa: BLE001
                    logger.error("S1_Burst: worker failed for %s: %s",
                                 date_str, exc)
                    safe = None
                if safe is not None:
                    safes.append(safe)
                else:
                    failed.append(date_str)
                if stop_event is not None and stop_event.is_set():
                    for f in futures:
                        f.cancel()
                    break
    if failed:
        print(f"[S1_Burst] {len(failed)} date(s) failed to assemble: "
              f"{', '.join(sorted(failed))}")
    print(f"[S1_Burst] assembled {len(safes)} / {len(groups)} .SAFE directory(ies).")
    self._report_burst_consistency(groups)

    if download_orbit and safes:
        # Orbits land ALONGSIDE the .SAFE directories, matching S1_SLC,
        # which writes .EOF into the same slc/ folder as the scenes rather
        # than a sibling orbits/ dir.
        self.download_orbit_for_safes(safes, force_cdse=force_cdse)
    return safes

download_orbit(force_cdse=False, save_dir=None, stop_event=None, scenes=None, merge=False)

Source code in src/insarhub/downloader/s1_burst.py

def download_orbit(self, force_cdse: bool = False, save_dir: str | None = None,
                   stop_event=None, scenes=None, merge: bool = False):
    """Fetch orbits for the .SAFE directories already assembled in save_dir.

    Named to match S1_SLC so the generic CLI/GUI paths find it -- both do
    ``hasattr(downloader, "download_orbit")`` and call it with ``save_dir``
    (main.py's --orbit-files handling, ScenePanel's orbit button). Without
    this method those paths silently no-op for bursts.

    Unlike S1_SLC's version it does not use the search results: burst
    granules are not validly-named Sentinel scenes, so orbits are resolved
    from the assembled SAFEs on disk instead (see
    download_orbit_for_safes).
    """
    root = Path(save_dir) if save_dir else Path(getattr(self.config, "workdir", ".") or ".")
    safes = sorted(root.rglob("*.SAFE"))
    if not safes:
        print(f"[S1_Burst] no .SAFE directories under {root}; "
              f"run download() first, orbits are resolved from assembled SAFEs")
        return []
    return self.download_orbit_for_safes(safes, force_cdse=force_cdse)

download_orbit_for_safes(safes, save_dir=None, force_cdse=False)

Source code in src/insarhub/downloader/s1_burst.py

def download_orbit_for_safes(self, safes, save_dir=None, force_cdse: bool = False):
    """Fetch precise orbits for ASSEMBLED SAFEs, not for burst granules.

    S1_SLC.download_orbit() derives orbit names from the search results,
    which works because its results are whole scenes. Burst granules are
    named differently::

        S1_264306_IW3_20240915T043118_VV_F23F-BURST

    and sentineleof rejects them outright ("Invalid Sentinel filename"), so
    delegating to S1_SLC yields one error per burst and no orbits. The SAFE
    directories burst2safe produces *are* validly named, so orbits are
    resolved from those instead -- one per acquisition rather than one per
    burst, which is also what the notebook's `eof --search-path <slc_dir>`
    does.
    """
    from eof.download import download_eofs

    # Alongside the SAFEs by default -- S1_SLC puts .EOF in the same slc/
    # directory as the scenes, and downstream tools scan one folder.
    if save_dir is None:
        parents = {Path(s).parent for s in safes}
        save_dir = parents.pop() if len(parents) == 1 else Path(".")
    save_dir = Path(save_dir)
    save_dir.mkdir(parents=True, exist_ok=True)

    got: list[Path] = []
    with tqdm(total=len(safes), desc="[S1_Burst] downloading orbits",
              unit="SAFE", leave=True, colour="cyan") as pbar:
        for safe in safes:
            safe = Path(safe)
            pbar.set_postfix_str(safe.name)
            try:
                got += download_eofs(sentinel_file=str(safe),
                                     save_dir=str(save_dir),
                                     orbit_type="precise",
                                     force_asf=not force_cdse)
            except Exception as exc:                            # noqa: BLE001
                logger.error("S1_Burst: orbit download failed for %s: %s",
                             safe.name, exc)
            pbar.update(1)
    uniq = sorted({Path(p).name for p in got})
    print(f"[S1_Burst] orbits: {len(uniq)} file(s) -> {save_dir}")
    for n in uniq:
        print(f"    {n}")
    return [save_dir / n for n in uniq]

folder_name(path, subswath=None, burst_id=None) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def folder_name(path: int, subswath: str | None = None,
                burst_id: int | None = None) -> str:
    """Job-folder name for a burst selection.

    A single burst is one fixed burst position -> a per-burst folder
    ``p<path>_iw<s>_b<id>``, where ``id`` is the OPERA relative burst ID
    (unique per subswath across the orbit, e.g. 266256). A whole track
    (several bursts merged) is one ``p<path>`` folder, because burst2safe
    assembles one SAFE per (date, path) anyway -- the track is the unit the
    ISCE3_Burst processor consumes.
    """
    if subswath and burst_id is not None:
        sw = str(subswath).lower()          # "IW3" -> "iw3"
        num = sw[2:] if sw[:2] in ("iw", "ew") else sw   # -> "3"
        return f"p{path}_iw{num}_b{burst_id}"
    return f"p{path}"

parent_slcs(results=None)

Source code in src/insarhub/downloader/s1_burst.py

def parent_slcs(self, results=None) -> dict[str, set[str]]:
    """{parent SLC granule: {YYYYMMDD, ...}} for the current burst results.

    ASF serves each burst from its parent slice, and names it in the URL::

        https://sentinel1-burst.asf.alaska.edu/S1A_IW_SLC__1SDV_2024...96A1/...

    That is the only link back to a product ASF actually publishes baselines
    for -- burst granules themselves carry ``perpendicularBaseline: None``.
    """
    out: dict[str, set[str]] = defaultdict(set)
    for r in self._flatten(self.active_results if results is None else results):
        m = self._PARENT_RE.search((getattr(r, "properties", {}) or {}).get("url") or "")
        d = self._date_of(r)
        if m and d:
            out[m.group(1)].add(d)
    return dict(out)

select_pairs(*args, **kwargs)

Source code in src/insarhub/downloader/s1_burst.py

def select_pairs(self, *args, **kwargs):
    """Baseline-aware pair selection for bursts, computed burst-natively.

    Why this override exists: ASF publishes NO baseline metadata for
    SLC-BURST products -- ``perpendicularBaseline``, ``temporalBaseline``
    and ``insarStackId`` are all None -- so the inherited implementation
    (which reads state vectors / baselines off each product) selects
    against data that does not exist. Measured on a 9-date Hawaii stack:
    the naive path returned 16 pairs instead of 21, silently dropping two
    12-day pairs (the highest-coherence ones) and leaving the final date
    with "0 / 3 connections available".

    The burst-native path in :func:`insarhub.utils.select_pairs` treats
    each acquisition **date** as the pairing node (one date = one stitched
    SAFE after burst2safe): temporal baseline comes from the burst's
    ``startTime`` and perpendicular baseline from per-date orbit state
    vectors (assembled ``.SAFE`` annotation, local ``.EOF``, or POEORB by
    date+mission). No parent-SLC lookup is performed.

    Returns:
        The base implementation's structure, with scene names replaced by
        ``YYYYMMDD`` acquisition dates -- the identifier that is meaningful
        for a burst stack (a date has many bursts, so no single burst
        granule can stand for it).
    """
    # Any configured orbit sources from the current workdir (post-download).
    workdir = getattr(self.config, "workdir", None)
    safe_dir = eof_dir = None
    if workdir:
        from pathlib import Path as _Path
        _w = _Path(workdir)
        slc = _w / "slc"
        if slc.is_dir():
            safe_dir = str(slc)
            eof_dir = str(slc)      # S1_Burst writes .EOF beside the SAFEs
    kwargs.setdefault("burst", True)
    if safe_dir and "safe_dir" not in kwargs:
        kwargs["safe_dir"] = safe_dir
    if eof_dir and "eof_dir" not in kwargs:
        kwargs["eof_dir"] = eof_dir
    kwargs.setdefault("poeorb_cache", _Path.home() / ".insarhub" / "poeorb")
    return super().select_pairs(*args, **kwargs)

sequential_pairs(n_connections=3, dates=None)

Parameters:

Name	Type	Description	Default
n_connections	int	Neighbours ahead to connect each date to.	3
dates		Explicit YYYYMMDD list; defaults to the dates in the current search results.	None

Source code in src/insarhub/downloader/s1_burst.py

def sequential_pairs(self, n_connections: int = 3, dates=None) -> list[tuple[str, str]]:
    """Tutorial-exact sequential pairing: each date to its next N neighbours.

    Byte-for-byte the rule in the COMPASS stack notebook's
    ``utils.generate_ifgram_pairs``::

        max_step = min(n_connections + 1, len(dates))
        for i in range(len(dates) - 1):
            for j in range(i + 1, min(i + max_step, len(dates))):
                pairs.append((dates[i], dates[j]))

    Use this when the goal is to reproduce the tutorial exactly.
    :meth:`select_pairs` cannot: it is target-driven (``dt_targets``) rather
    than rule-driven, so the closest it gets on the 9-date Hawaii stack is a
    23-pair SUPERSET of these 21 (at ``pb_max=400, max_degree=6``), never
    the set itself.

    Note what this deliberately ignores: perpendicular baseline. On that
    same stack it emits ``20240927_20241009`` (12 d but 300 m dBperp) and
    ``20240903_20240915`` (12 d, 208 m), which :meth:`select_pairs` rejects
    as geometrically decorrelated. That is the tutorial's behaviour, not a
    defect here -- but it is the reason the two disagree.

    Args:
        n_connections: Neighbours ahead to connect each date to.
        dates: Explicit YYYYMMDD list; defaults to the dates in the current
            search results.

    Returns:
        Sorted ``[(YYYYMMDD, YYYYMMDD), ...]``.
    """
    if dates is None:
        dates = sorted({d for d in (self._date_of(r)
                                    for r in self._flatten(self.active_results)) if d})
    else:
        dates = sorted(set(dates))
    if len(dates) < 2:
        print(f"[S1_Burst] only {len(dates)} date(s); no pairs to form")
        return []
    max_step = min(n_connections + 1, len(dates))
    pairs = [(dates[i], dates[j])
             for i in range(len(dates) - 1)
             for j in range(i + 1, min(i + max_step, len(dates)))]
    print(f"[S1_Burst] sequential pairing (tutorial rule): {len(dates)} dates, "
          f"n_connections={n_connections} -> {len(pairs)} pairs")
    return sorted(pairs)

write_ifgram_list(pairs, path) staticmethod

Source code in src/insarhub/downloader/s1_burst.py

@staticmethod
def write_ifgram_list(pairs, path) -> Path:
    """Write pairs as ``ifgram_list.txt`` (one ``YYYYMMDD_YYYYMMDD`` per line).

    This is the format the COMPASS stack notebook's
    ``generate_ifgram_pairs`` produces and section 3.3 consumes, so a
    baseline-aware selection can be dropped in place of the purely
    sequential one.
    """
    flat = ([p for v in pairs.values() for p in v]
            if isinstance(pairs, dict) else list(pairs))
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w") as fh:
        fh.write("# date12\n")
        for a, b in sorted(set(flat)):
            fh.write(f"{a}_{b}\n")
    print(f"[S1_Burst] wrote {len(set(flat))} pair(s) -> {path}")
    return path

• Create downloader with parameters

  s1b = Downloader.create('S1_Burst',
                          intersectsWith=[-106.06, 40.34, -105.70, 40.58],
                          fullBurstID=['056_118970_IW2', '056_118971_IW2'],
                          polarization=['VV'],
                          start='2022-08-04',
                          end='2026-07-21',
                          workdir='path/to/dir')
  

  OR with explicit config:

  from insarhub.config import S1_Burst_Config
  
  cfg = S1_Burst_Config(
      intersectsWith=[-106.06, 40.34, -105.70, 40.58],
      fullBurstID=['056_118970_IW2', '056_118971_IW2'],
      polarization=['VV'],
      start='2022-08-04',
      end='2026-07-21',
      workdir='path/to/dir',
  )
  dl = Downloader.create('S1_Burst', config=cfg)
  

  Attributes:

  Name	Type	Description
  swaths	list[str] | None	Subswaths to keep, e.g. ["IW2", "IW3"]. None = all three.
  mode	str	Acquisition mode passed to burst2safe (IW or EW).
  min_bursts	int	Minimum bursts per assembled SAFE; burst2safe pads with neighbouring bursts to reach it. Guards against a 1-burst SAFE that most downstream tools reject.
  all_anns	bool	Include annotation for every subswath, not just the ones downloaded. Default True; turning it off makes a single-subswath SAFE unreadable by s1reader/COMPASS (see the field comment).
  keep_files	bool	Keep the intermediate per-burst downloads after assembly.

  Source code in src/insarhub/config/defaultconfig.py

  @dataclass
  class S1_Burst_Config(ASF_Base_Config):
      """Sentinel-1 SLC-BURST search + SAFE assembly.
  
      ASF distributes individual TOPS bursts as their own granules (dataset
      SLC-BURST). They are ~1/9th the size of a full slice, so an AOI-limited
      stack downloads far less data than the equivalent SLC search -- the point
      of burst-based processing.
  
      Bursts are not directly consumable by SAFE-expecting tools, so download()
      hands the selected granules to burst2safe, which assembles them into valid
      .SAFE directories (merging annotation/calibration/noise XML and writing a
      manifest). This mirrors the burst2stack CLI, except the granule list comes
      from *this* downloader's own search+filter rather than a second, independent
      search -- so the filters the user applied are the ones that govern what is
      actually assembled.
  
      Attributes:
          swaths: Subswaths to keep, e.g. ["IW2", "IW3"]. None = all three.
          mode: Acquisition mode passed to burst2safe (IW or EW).
          min_bursts: Minimum bursts per assembled SAFE; burst2safe pads with
              neighbouring bursts to reach it. Guards against a 1-burst SAFE that
              most downstream tools reject.
          all_anns: Include annotation for every subswath, not just the ones
              downloaded. Default True; turning it off makes a single-subswath
              SAFE unreadable by s1reader/COMPASS (see the field comment).
          keep_files: Keep the intermediate per-burst downloads after assembly.
      """
      name: str = "S1_Burst_Config"
      dataset: str | list[str] | None = constants.DATASET.SLC_BURST
      instrument: str | None = constants.INSTRUMENT.C_SAR
      beamMode: str | None = constants.BEAMMODE.IW
      polarization: str | list[str] | None = field(
          default_factory=lambda: [constants.POLARIZATION.VV])
      processingLevel: str | None = constants.PRODUCT_TYPE.BURST
  
      # ── burst -> SAFE assembly (burst2safe) ────────────────────────────────
      swaths: list[str] | None = None
      mode: str = "IW"
      min_bursts: int = 1
      # Default True, unlike burst2safe's own False. Annotation XML costs a few
      # hundred KB against ~500 MB of measurement data, and without it a
      # single-subswath SAFE is unreadable: s1reader derives the OPERA burst ID
      # from the IW2 mid-burst sensing time and opens the IW2 annotation whichever
      # swath you asked for. Verified -- an IW3-only SAFE with all_anns loads all
      # 4 IW3 bursts; without it, "ValueError: burst iw2-slc-vv not in SAFE". The
      # alternative is force-adding IW2 DATA, roughly doubling an IW1/IW3-only
      # download. Both COMPASS tutorials pass --all-anns unconditionally.
      all_anns: bool = True
      keep_files: bool = False
      # max_workers is inherited from ASF_Base_Config. For bursts it counts
      # concurrent DATE ASSEMBLIES rather than HTTP transfers: each unit is a
      # burst2safe run that also issues its own ASF search, so a high value
      # invites rate limits. Hence the lower default than S1_SLC.
      max_workers: int = 2
  
      # EXTEND the base groups, never replace them. _ui_groups/_ui_fields are
      # ClassVars, so assigning a fresh list here shadows ASF_Base_Config's
      # entirely -- which dropped AOI, dates, orbit/frame and every other search
      # field from the GUI form, leaving only the five assembly options.
      _ui_groups: ClassVar[list] = ASF_Base_Config._ui_groups + [
          {"label": "Burst assembly",
           "fields": ["swaths", "mode", "min_bursts", "all_anns", "keep_files"]},
      ]
      _ui_fields: ClassVar[dict] = {
          **ASF_Base_Config._ui_fields,
          "swaths":     {"type": "multiselect", "options": ["IW1", "IW2", "IW3"],
                         "hint": "Subswaths to keep. Empty = all three."},
          "mode":       {"type": "select", "options": ["IW", "EW"],
                         "hint": "Acquisition mode passed to burst2safe."},
          "min_bursts": {"type": "number", "step": 1,
                         "hint": "Minimum bursts per assembled SAFE; burst2safe pads with neighbours to reach it."},
          "all_anns":   {"type": "bool",
                         "hint": "Keep annotations for every burst in the slice, not just the selected ones."},
          "keep_files": {"type": "bool",
                         "hint": "Keep intermediate per-burst downloads after assembly."},
          "max_workers": {"type": "number", "min": 1, "max": 8, "step": 1,
                          "hint": "Concurrent DATE ASSEMBLIES, not HTTP transfers: each unit is a burst2safe run that unpacks and rewrites a whole SAFE, and issues its own ASF search. Keep low."},
      }
  

• Search / Filter / Summary / Footprint

  Identical to S1_SLC — these reuse ASF_Base_Downloader and operate on the same ASF burst granule search.

  results = dl.search()
  dl.summary()
  dl.footprint()
  

• Select Pairs

  pairs, baselines, scene_bperp, _ = dl.select_pairs(
      dt_targets=(6, 12, 24, 36, 48, 72, 96),
      dt_tol=3,
      dt_max=120,
      pb_max=150.0,
      force_connect=True,
  )
  

  Parameters:

  Name	Type	Description	Default
  dt_targets	tuple	Target temporal spacings in days. Defaults to (6, 12, 24, 36, 48, 72, 96).	None
  dt_tol	int	Tolerance in days around each target spacing. Defaults to 3.	None
  dt_max	int	Maximum temporal baseline in days. Defaults to 120.	None
  pb_max	float	Maximum perpendicular baseline in meters. Defaults to 150.0.	None
  min_degree	int	Minimum number of connections per scene. Defaults to 3.	None
  max_degree	int	Maximum number of connections per scene. Defaults to 5.	None
  force_connect	bool	Force connectivity for isolated scenes. Defaults to True.	None
  max_workers	int	Threads for API baseline fallback. Defaults to 4.	None
  aoi_wkt	str	AOI geometry in WKT for quality scoring. Defaults to search AOI.	None
  merge	bool	When True, stacks sharing the same relative orbit (path) are combined into one pairing network before temporal/baseline selection — matching how ISCE2's stackSentinel treats multiple frames of one track/pass as a single continuous acquisition. Stacks on different paths are never combined (cross-track pairs have no physical baseline). Use together with download(merge=True), which puts all scenes in one merged/slc/ directory. Defaults to False.	False
  burst	bool	Select pairs for an SLC-BURST stack. Nodes become acquisition dates and baselines are computed from the bursts' own startTime/orbits — no parent-SLC lookup. Defaults to False.	False
  safe_dir	str	Burst mode: directory of assembled .SAFE dirs whose annotation orbits supply bperp (offline).	None
  eof_dir	str	Burst mode: directory of precise-orbit .EOF files used for bperp (offline).	None
  poeorb_cache	str	Burst mode: directory for POEORB downloads keyed by date + mission (online fallback).	None
  quality_check	bool	After pairing, write the stack file(s) and build the PairQualityDB verdict for every possible pair (the slow, network-heavy step). False skips it. Defaults to True.	True
  plot_network	bool	Save network_*.png with the pair network, coloured healthy/concern when quality_check is True (falls back to temporal-baseline colouring otherwise). Defaults to True.	True

• Download

  Download the selected burst granules and assemble them into .SAFE directories via burst2safe.

  dl.download()
  

  Parameters:

  Name	Type	Description	Default
  save_path	str | None	Destination root. Defaults to the configured workdir.	None
  max_workers	int | None	Accepted for parity with S1_SLC; unused (see above).	None
  download_orbit	bool	Also fetch the matching precise orbits (.EOF), which every downstream processor needs.	False
  force_cdse	bool	Fetch orbits from CDSE instead of ASF.	False
  stop_event		threading.Event to cancel between date groups.	None
  on_progress		callback(message, pct) after each date group.	None
  merge	bool	Assemble every stack into one directory instead of per-stack subfolders.	False

Searches and downloads NISAR L2 GSLC (geocoded SLC) products via ASF — one already-geocoded complex SLC frame per date, feeding the ISCE3_NISAR processor and ISCE3_Dolphin_NISAR_PL analyzer directly. Search, filter, footprint and pair selection reuse ASF_Base_Downloader. There is no orbit download: NISAR products carry their own state vectors.

NISAR facets differ from Sentinel-1

NISAR carries polarization per frequency band, not in a single polarization field (which ASF leaves null): filter on mainBandPolarization (frequency A, the wide high-resolution band used for InSAR) and, where needed, sideBandPolarization (frequency B, the 5 MHz band for ionosphere). rangeBandwidth (e.g. 40+5) is the acquisition mode — keep it constant across a stack so every date has the same resolution. frameCoverage (FULL/PARTIAL), relativeOrbit (path) and frame complete the facets. NISAR products also report no beamMode/centerLat/centerLon/granuleType/md5sum, and their bytes is a per-file map rather than a single number.

• Create downloader with parameters

  gslc = Downloader.create('NISAR_GSLC',
                           intersectsWith=[-113.08, 37.68, -112.58, 38.07],
                           mainBandPolarization='HH+HV',
                           rangeBandwidth='40+5',
                           start='2025-11-01',
                           end='2026-09-01',
                           workdir='path/to/dir')
  

  Source code in src/insarhub/config/defaultconfig.py

  @dataclass
  class NISAR_GSLC_Config(ASF_Base_Config):
      """Search/download config for NISAR L2 GSLC (geocoded SLC) via ASF.
  
      GSLC is already geocoded (unlike Sentinel-1 SLC), so it feeds dolphin's
      phase-linking directly through the ISCE3_NISAR processor -- no coregistration
      or COMPASS geocoding stage. L-band (frequencyA) HH is the default single-pol
      for InSAR; override ``polarization`` for HV / dual / quad products.
      """
      name: str = "NISAR_GSLC_Config"
      dataset: str | list[str] | None = constants.DATASET.NISAR
      processingLevel: str | None = constants.PRODUCT_TYPE.GSLC
      # No polarization default: NISAR encodes it as DHDH/SHSH/etc. (not S1's
      # HH/VV), so a fixed default would silently filter out valid products.
      # Filter via the GUI search schema / --polarization when needed.
      #
      # Default to FULL-coverage frames only: NISAR images in segments, so a frame
      # at a segment edge comes back "Partial" (a smaller, clipped footprint). A
      # stack that mixes Full and Partial frames of the same (path, frame) has
      # inconsistent footprint sizes and coverage gaps across dates -- filtering to
      # Full keeps every date on the same frame extent. Set to "" for both.
      frameCoverage: str | None = "FULL"
  

• Search / Download

  Identical to S1_SLC — these reuse ASF_Base_Downloader. The downloaded *GSLC*.h5 frames land in workdir/slc/, where ISCE3_NISAR reads them.

  gslc.search()
  gslc.download()
  

NISAR_RSLC downloads NISAR L1 RSLC (radar-coordinate SLC) products — the rawest InSAR input (frequency A and B), not yet geocoded. It targets GMTSAR's NISAR path (pre_proc_nsr / p2p_processing_nsr, SAT=NSR_A). Same ASF facets as NISAR_GSLC (main/side-band polarization, range bandwidth, frame coverage, path/frame), no orbit download.

Source code in src/insarhub/config/defaultconfig.py

@dataclass
class NISAR_RSLC_Config(ASF_Base_Config):
    """Search/download config for NISAR L1 RSLC (radar-coordinate SLC) via ASF.

    RSLC is the L1 radar SLC (both frequency A and B) -- the rawest InSAR input,
    processed into interferograms by GMTSAR's NISAR path (pre_proc_nsr /
    p2p_processing_nsr, SAT=NSR_A). Same fixed frame grid + FULL-coverage default
    as the other NISAR downloaders.
    """
    name: str = "NISAR_RSLC_Config"
    dataset: str | list[str] | None = constants.DATASET.NISAR
    processingLevel: str | None = constants.PRODUCT_TYPE.RSLC
    frameCoverage: str | None = "FULL"

NISAR_GUNW downloads NISAR L2 GUNW (geocoded unwrapped interferograms) — a ready-made geocoded, unwrapped interferogram pair product, the NISAR analog of a HyP3 Sentinel-1 GUNW. It is single-band (formed on the main band only), so there is no side-band-polarization facet; mainBandPolarization is a single value (HH/HV/VH/VV). Intended to feed MintPy through its prep_nisar loader (no processor needed).

Source code in src/insarhub/config/defaultconfig.py

@dataclass
class NISAR_GUNW_Config(ASF_Base_Config):
    """Search/download config for NISAR L2 GUNW (geocoded unwrapped
    interferograms) via ASF.

    GUNW is a ready-made geocoded, unwrapped interferogram PAIR product (the
    NISAR analog of a HyP3 Sentinel-1 GUNW) -- fed straight into MintPy via its
    ``prep_nisar`` loader, no processor needed. GUNW is single-band (the main
    band only), so unlike GSLC/RSLC there is no side band. FULL coverage by
    default for a consistent stack.
    """
    name: str = "NISAR_GUNW_Config"
    dataset: str | list[str] | None = constants.DATASET.NISAR
    processingLevel: str | None = constants.PRODUCT_TYPE.GUNW
    frameCoverage: str | None = "FULL"