API reference#

earthcarekit.download

A download tool for EarthCARE data based on ESA MAAP.

Notes#

This module depends on other internal modules:

ecdownload #

ecdownload(
    file_type: str | Iterable[str],
    baseline: str | None = None,
    orbit_number: int | Iterable[int] | None = None,
    start_orbit_number: int | None = None,
    end_orbit_number: int | None = None,
    frame_id: str | Iterable[str] | None = None,
    orbit_and_frame: str | Iterable[str] | None = None,
    start_orbit_and_frame: str | None = None,
    end_orbit_and_frame: str | None = None,
    timestamp: TimestampLike | Iterable[TimestampLike] | None = None,
    start_time: TimestampLike | None = None,
    end_time: TimestampLike | None = None,
    radius_search: tuple[RadiusMetersFloat, LatFloat, LonFloat] | Iterable | None = None,
    bounding_box: tuple[LatSFloat, LonWFloat, LatNFloat, LonEFloat] | Iterable | None = None,
    geometry: IntersectsLike | None = None,
    path_to_config: str | None = None,
    path_to_data: str | None = None,
    is_log: bool = False,
    is_debug: bool = False,
    is_download: bool = True,
    is_overwrite: bool = False,
    is_unzip: bool = True,
    is_delete: bool = True,
    is_create_subdirs: bool = True,
    is_export_results: bool = False,
    idx_selected_input: int | None = None,
    is_organize_data: bool = False,
    is_include_header: bool | None = None,
    is_only_header: bool | None = None,
    is_reversed_order: bool = False,
    return_results: bool = False,
    verbose: bool = True,
    clear_cache: bool = False,
    config: ECKConfig | None = None,
    **kwargs
) -> ProductDataFrame | None

Search for and download EarthCARE products from the ESA MAAP platform.

Executes two steps: 1. Sends search request to the MAAP catalog using the pystac-client. 2. Downloads matching products.

Parameters:

Name	Type	Description	Default
`file_type`	`str \| Iterable[str]`	Product name(s) (e.g., "ATL_NOM_1B", "ANOM:BA"); case-insensitive.	required
`baseline`	`str \| None`	Default processor baseline (e.g., "BA"). Ignored if already in `file_type`.	`None`
`orbit_number`	`int \| Iterable[int] \| None`	Specific orbit number(s).	`None`
`start_orbit_number`	`int \| None`	Lower orbit limit.	`None`
`end_orbit_number`	`int \| None`	Upper orbit limit.	`None`
`frame_id`	`str \| Iterable[str] \| None`	Frame ID letter(s) ("A"-"H").	`None`
`orbit_and_frame`	`str \| Iterable[str] \| None`	Orbit+frame strings (e.g., "01234F").	`None`
`start_orbit_and_frame`	`str \| None`	Lower orbit+frame limit.	`None`
`end_orbit_and_frame`	`str \| None`	Upper orbit+frame limit.	`None`
`timestamp`	`TimestampLike \| Iterable[TimestampLike] \| None`	Timestamp(s) (e.g., "2024-07-31 13:45").	`None`
`start_time`	`TimestampLike \| None`	Lower time limit.	`None`
`end_time`	`TimestampLike \| None`	Upper time limit.	`None`
`radius_search`	`tuple[RadiusMetersFloat, LatFloat, LonFloat] \| Iterable \| None`	`(radius_m, lat, lon)` for geo radius search.	`None`
`bounding_box`	`tuple[LatSFloat, LonWFloat, LatNFloat, LonEFloat] \| Iterable \| None`	`(latS, lonW, latN, lonE)` for geo box search.	`None`
`geometry`	`IntersectsLike \| None`	Spatial filter (GeoJSON/dict/`__geo_interface__`); overrides `radius_search`/`bounding_box`.	`None`
`path_to_config`	`str \| None`	Custom config file path.	`None`
`path_to_data`	`str \| None`	Custom download directory.	`None`
`is_log`	`bool`	Write logs to `/log` in CWD if True.	`False`
`is_debug`	`bool`	Show debug logs in console if True.	`False`
`is_download`	`bool`	Skip download if False (search only).	`True`
`is_overwrite`	`bool`	Overwrite existing files if True.	`False`
`is_unzip`	`bool`	Skip extraction if False.	`True`
`is_delete`	`bool`	Delete archives after extraction if True.	`True`
`is_create_subdirs`	`bool`	Use config-defined subdirectory structure if True.	`True`
`is_export_results`	`bool`	Save search results to `results.txt` in CWD if True.	`False`
`idx_selected_input`	`int \| None`	Download only the indexed result.	`None`
`is_organize_data`	`bool`	Skip search/download if True.	`False`
`is_include_header`	`bool \| None`	Include `.HDR` files if True.	`None`
`is_only_header`	`bool \| None`	Download only `.HDR` files (overrides `is_include_header`).	`None`
`is_reversed_order`	`bool`	Download latest-first if True.	`False`
`return_results`	`bool`	Return `ProductDataFrame` if True.	`False`
`verbose`	`bool`	Suppress logs to console if False.	`True`
`clear_cache`	`bool`	Bypass cache and refresh product metadata if True.	`False`
`config`	`ECKConfig \| None`	Custom config object.	`None`

Returns:

Type	Description
`ProductDataFrame \| None`	If `return_results` is True, returns results table; otherwise None (default).

Source code in earthcarekit/download/main.py

def ecdownload(
    file_type: str | Iterable[str],
    baseline: str | None = None,
    orbit_number: int | Iterable[int] | None = None,
    start_orbit_number: int | None = None,
    end_orbit_number: int | None = None,
    frame_id: str | Iterable[str] | None = None,
    orbit_and_frame: str | Iterable[str] | None = None,
    start_orbit_and_frame: str | None = None,
    end_orbit_and_frame: str | None = None,
    timestamp: TimestampLike | Iterable[TimestampLike] | None = None,
    start_time: TimestampLike | None = None,
    end_time: TimestampLike | None = None,
    radius_search: tuple[RadiusMetersFloat, LatFloat, LonFloat] | Iterable | None = None,
    bounding_box: (tuple[LatSFloat, LonWFloat, LatNFloat, LonEFloat] | Iterable | None) = None,
    geometry: IntersectsLike | None = None,
    path_to_config: str | None = None,
    path_to_data: str | None = None,
    is_log: bool = False,
    is_debug: bool = False,
    is_download: bool = True,
    is_overwrite: bool = False,
    is_unzip: bool = True,
    is_delete: bool = True,
    is_create_subdirs: bool = True,
    is_export_results: bool = False,
    idx_selected_input: int | None = None,
    is_organize_data: bool = False,
    is_include_header: bool | None = None,
    is_only_header: bool | None = None,
    is_reversed_order: bool = False,
    return_results: bool = False,
    verbose: bool = True,
    clear_cache: bool = False,
    config: ECKConfig | None = None,
    **kwargs,
) -> ProductDataFrame | None:
    """Search for and download EarthCARE products from the ESA MAAP platform.

    Executes two steps:
    1. Sends search request to the [MAAP catalog](https://catalog.maap.eo.esa.int/catalogue/) using the [`pystac-client`](https://pystac-client.readthedocs.io/en/stable/#).
    2. Downloads matching products.

    Args:
        file_type: Product name(s) (e.g., "ATL_NOM_1B", "ANOM:BA"); case-insensitive.
        baseline: Default processor baseline (e.g., "BA"). Ignored if already in ``file_type``.
        orbit_number: Specific orbit number(s).
        start_orbit_number: Lower orbit limit.
        end_orbit_number: Upper orbit limit.
        frame_id: Frame ID letter(s) ("A"-"H").
        orbit_and_frame: Orbit+frame strings (e.g., "01234F").
        start_orbit_and_frame: Lower orbit+frame limit.
        end_orbit_and_frame: Upper orbit+frame limit.
        timestamp: Timestamp(s) (e.g., "2024-07-31 13:45").
        start_time: Lower time limit.
        end_time: Upper time limit.
        radius_search: ``(radius_m, lat, lon)`` for geo radius search.
        bounding_box: ``(latS, lonW, latN, lonE)`` for geo box search.
        geometry: Spatial filter (GeoJSON/dict/``__geo_interface__``); overrides ``radius_search``/``bounding_box``.
        path_to_config: Custom config file path.
        path_to_data: Custom download directory.
        is_log: Write logs to ``/log`` in CWD if True.
        is_debug: Show debug logs in console if True.
        is_download: Skip download if False (search only).
        is_overwrite: Overwrite existing files if True.
        is_unzip: Skip extraction if False.
        is_delete: Delete archives after extraction if True.
        is_create_subdirs: Use config-defined subdirectory structure if True.
        is_export_results: Save search results to ``results.txt`` in CWD if True.
        idx_selected_input: Download only the indexed result.
        is_organize_data: Skip search/download if True.
        is_include_header: Include ``.HDR`` files if True.
        is_only_header: Download only ``.HDR`` files (overrides ``is_include_header``).
        is_reversed_order: Download latest-first if True.
        return_results: Return ``ProductDataFrame`` if True.
        verbose: Suppress logs to console if False.
        clear_cache: Bypass cache and refresh product metadata if True.
        config: Custom config object.

    Returns:
        If ``return_results`` is True, returns results table; otherwise None (default).
    """
    # Handle deprecated arguments
    if "timestamps" in kwargs:
        msg = "'timestamps' is deprecated and will be removed in future versions; use 'timestamp' instead."
        warnings.warn(msg, FutureWarning)
        timestamp = kwargs["timestamps"]
        del kwargs["timestamps"]
    if len(set(kwargs)) != 0:
        raise TypeError(
            f"{ecdownload.__name__}() got an unexpected keyword argument '{list(kwargs)[0]}'"
        )

    time_start_script: pd.Timestamp = pd.Timestamp(
        datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    )
    time_end_script: pd.Timestamp
    execution_time: pd.Timedelta

    def _to_list(input: Any, _type: type[Any] | tuple[type[Any], ...]) -> list | None:
        if isinstance(input, _type):
            return [input]
        elif isinstance(input, Iterable) and not isinstance(input, str):
            return list(input)
        else:
            return None

    _file_type: list[str] | None = _to_list(file_type, str)
    assert isinstance(_file_type, list)
    file_type = _file_type

    orbit_number = _to_list(orbit_number, int)
    frame_id = _to_list(frame_id, str)
    orbit_and_frame = _to_list(orbit_and_frame, str)
    timestamp = _to_list(timestamp, TIMESTAMP_TYPES)
    if isinstance(timestamp, list):
        timestamp = time_utils.times_to_iso(timestamp)

    if isinstance(start_time, TimestampLike):
        start_time = time_utils.time_to_iso(start_time)
    if isinstance(end_time, TimestampLike):
        end_time = time_utils.time_to_iso(end_time)

    if isinstance(radius_search, Iterable):
        radius_search = list(radius_search)

    if isinstance(bounding_box, Iterable):
        bounding_box = list(bounding_box)

    idx_selected: int | None = _cli.parse.selected_index(idx_selected_input)

    logger: Logger | None = None
    if verbose:
        logger = create_logger(
            logger_name=PROGRAM_NAME,
            log_to_file=is_log,
            debug=is_debug,
        )
    if is_log:
        remove_old_logs(100, pd.Timedelta(days=30))

    log_textbox(
        f"EarthCARE Download Tool\n{__title__} {__version__}",
        logger=logger,
        is_mayor=True,
    )

    if logger and not is_organize_data:
        logger.info("Settings")
        logger.info(f"- {is_download=}")
        logger.info(f"- {is_overwrite=}")
        logger.info(f"- {is_unzip=}")
        logger.info(f"- {is_delete=}")
        logger.info(f"- {is_create_subdirs=}")
        logger.info(f"- {is_log=}")
        logger.info(f"- {is_debug=}")
        logger.info(f"- {is_export_results=}")
        logger.info(f"- {idx_selected_input=}")

    if not isinstance(config, ECKConfig):
        config = _cli.parse.path_to_config(path_to_config, logger=logger)

    path_to_data = _cli.parse.path_to_data(path_to_data, logger=logger)
    if isinstance(path_to_data, str):
        config.path_to_data = path_to_data

    if logger and not is_organize_data:
        logger.info(f"- path_to_config=<{str(Path(config.filepath).resolve())}>")
        logger.info(f"- path_to_data=<{str(Path(config.path_to_data).resolve())}>")

    if is_organize_data:
        if logger:
            logger.info("# Organizing local data ...")
        performed_moves = organize_data(
            config=config,
            logger=logger,
        )
        time_end_script = pd.Timestamp(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
        execution_time = time_end_script - time_start_script
        execution_time_str = str(execution_time).split()[-1]
        if logger:
            console_print()
        _moved = len([pm for pm in performed_moves if pm.get("status") == "success"])
        _failed = len([pm for pm in performed_moves if pm.get("status") == "error"])
        _msg = [
            "EXECUTION SUMMARY",
            "---",
            f"Time taken          {execution_time_str}",
            f"Moved files         {_moved}",
            f"Failed moves        {_failed}",
        ]
        log_textbox("\n".join(_msg), logger=logger, show_time=True)
        return None

    if not isinstance(is_include_header, bool):
        is_include_header = config.maap_include_header_file

    search_inputs: _SearchInputs = _cli.parse.search_inputs(
        file_type=file_type,
        baseline=baseline,
        orbit_number=orbit_number,
        start_orbit_number=start_orbit_number,
        end_orbit_number=end_orbit_number,
        frame_id=frame_id,
        orbit_and_frame=orbit_and_frame,
        start_orbit_and_frame=start_orbit_and_frame,
        end_orbit_and_frame=end_orbit_and_frame,
        timestamp=timestamp,
        start_time=start_time,
        end_time=end_time,
        radius_search=radius_search,
        bounding_box=bounding_box,
        geometry=geometry,
        logger=logger,
    )

    planned_requests = maap.get_requests(search_inputs)

    # Ensure MAAP collections
    user_collections = [c.value for c in config.collections]
    user_collections = [c.rstrip("_MAAP") + "_MAAP" for c in user_collections]

    found_products: list[EOProduct] = run_search_requets(
        log_heading_msg="STEP 1/2 - Search products",
        search_requests=planned_requests,
        user_collections=user_collections,
        is_debug=is_debug,
        is_found_files_list_to_txt=is_export_results,
        selected_index=idx_selected,
        selected_index_input=idx_selected_input,
        logger=logger,
        download_only_h5=not is_include_header,
        download_only_hdr=is_only_header or False,
        clear_cache=clear_cache,
    )

    donwload_results: list[_DownloadResult] = run_downloads(
        log_heading_msg="STEP 2/2 - Download products",
        products=found_products,
        config=config,
        is_download=is_download,
        is_overwrite=is_overwrite,
        is_unzip=is_unzip,
        is_delete=is_delete,
        is_create_subdirs=is_create_subdirs,
        logger=logger,
        is_reversed_order=is_reversed_order,
    )

    if logger:
        num_downloads: int = 0
        num_unzips: int = 0
        num_errors: int = 0
        size_msg: str = "N/A"
        avg_speed_mbs: float = 0.0
        download_speed_msg: str = ""
        if len(donwload_results) > 0:
            num_errors = sum([not r.success for r in donwload_results])
            num_downloads = sum([r.downloaded for r in donwload_results])
            num_unzips = sum([r.unzipped for r in donwload_results])
            total_size_mb = sum([r.size_mb for r in donwload_results])
            size_msg = f"{total_size_mb:.2f} MB"
            if total_size_mb >= 1024:
                size_msg = f"{total_size_mb / 1024:.2f} GB"
            avg_speed_mbs = float(np.mean([r.speed_mbs for r in donwload_results]))
            download_speed_msg = f" ({size_msg} at ~{avg_speed_mbs:.2f} MB/s)"

        time_end_script = pd.Timestamp(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
        execution_time = time_end_script - time_start_script
        execution_time_str = str(execution_time).split()[-1]

        console_print()
        _msg = [
            "EXECUTION SUMMARY",
            "---",
            f"Time taken          {execution_time_str}",
            f"API search requests {len(planned_requests)}",
            f"Remote files found  {len(found_products)}",
            f"Files downloaded    {num_downloads}{download_speed_msg}",
            f"Files unzipped      {num_unzips}",
            f"Errors occured      {num_errors}",
        ]
        log_textbox("\n".join(_msg), logger=logger, show_time=True)

    if return_results:
        pdf = get_product_infos([p.name for p in found_products], must_exist=False)
        pdf["start_latitude"] = np.array([p.start_latitude for p in found_products])
        pdf["start_longitude"] = np.array([p.start_longitude for p in found_products])
        pdf["end_latitude"] = np.array([p.end_latitude for p in found_products])
        pdf["end_longitude"] = np.array([p.end_longitude for p in found_products])
        pdf["url_download_h5"] = np.array([p.url_download_h5 for p in found_products], dtype=str)
        pdf["url_download_hdr"] = np.array([p.url_download_hdr for p in found_products], dtype=str)
        pdf["url_quicklook"] = np.array([p.url_quicklook for p in found_products], dtype=str)
        return pdf
    return None