bluemath_tk.downloaders package

Subpackages

Module contents

Project: BlueMath_tk Sub-Module: downloaders Author: GeoOcean Research Group, Universidad de Cantabria Repository: https://github.com/GeoOcean/BlueMath_tk.git Status: Under development (Working)

class bluemath_tk.downloaders.BaseDownloader(product: str, base_path_to_download: str, debug: bool = True)[source]

Bases: BlueMathModel

Abstract base class for BlueMath downloaders.

All downloaders should: 1. Have a download_data method that routes to product-specific methods 2. Have product-specific methods like download_data_<product> 3. Use DownloadResult to track download status

product

The product name (e.g., “SWOT”, “ERA5”).

Type:: str

product_config

Product configuration dictionary.

Type:: dict

base_path_to_download

Base path where downloaded files are stored.

Type:: str

debug

If True, logger is set to DEBUG level.

Type:: bool

property base_path_to_download: str

create_download_result(start_time: datetime | None = None) → DownloadResult[source]

Create a new DownloadResult instance.

Parameters:: start_time (Optional[datetime], optional) – The start time of the download operation. If None, the current time is used.
Returns:: A new DownloadResult instance.
Return type:: DownloadResult

property debug: bool

abstractmethod download_data(*args, **kwargs) → DownloadResult[source]

Download data for the product.

Routes to product-specific methods like download_data_<product>().

Parameters:

*args – Arguments passed to product-specific download method.
**kwargs – Keyword arguments (e.g., force, dry_run).

Returns:

Result with information about downloaded, skipped, and error files.

Return type:

DownloadResult

finalize_download_result(result: DownloadResult, message: str | None = None) → DownloadResult[source]

Finalize a DownloadResult with end time and summary message.

Parameters:

result (DownloadResult) – The DownloadResult to finalize.
message (Optional[str], optional) – The message to add to the DownloadResult.

Returns:

The finalized DownloadResult.

Return type:

DownloadResult

list_datasets() → List[str][source]

List all available datasets for the product.

Returns:: List of available dataset names.
Return type:: List[str]

property product: str

abstract property product_config: dict

class bluemath_tk.downloaders.DownloadResult(success: bool = False, downloaded_files: ~typing.List[str] = <factory>, skipped_files: ~typing.List[str] = <factory>, error_files: ~typing.List[str] = <factory>, errors: ~typing.List[~typing.Dict[str, ~typing.Any]] = <factory>, metadata: ~typing.Dict[str, ~typing.Any] = <factory>, message: str = '', start_time: ~datetime.datetime | None = None, end_time: ~datetime.datetime | None = None, duration_seconds: float | None = None)[source]

Bases: object

Standardized result structure for download operations.

This class provides a consistent interface for download results across all downloaders, making it easier to handle success/failure cases and track downloaded files.

success

Whether the download operation completed successfully.

Type:: bool

downloaded_files

List of file paths that were successfully downloaded.

Type:: List[str]

skipped_files

List of file paths that were skipped (e.g., already exist, incomplete).

Type:: List[str]

error_files

List of file paths that failed to download.

Type:: List[str]

errors

List of error dictionaries containing error details. Each dict has keys: ‘file’, ‘error’, ‘timestamp’.

Type:: List[Dict[str, Any]]

metadata

Additional metadata about the download operation.

Type:: Dict[str, Any]

message

Human-readable summary message.

Type:: str

start_time

When the download operation started.

Type:: Optional[datetime]

end_time

When the download operation ended.

Type:: Optional[datetime]

duration_seconds

Total duration of the download operation in seconds.

Type:: Optional[float]

Examples

>>> result = DownloadResult(
...     success=True,
...     downloaded_files=["/path/to/file1.nc", "/path/to/file2.nc"],
...     message="Downloaded 2 files successfully"
... )
>>> print(result.message)
Downloaded 2 files successfully
>>> print(f"Success rate: {result.success_rate:.1%}")
Success rate: 100.0%

add_downloaded(file_path: str)[source]: Add a successfully downloaded file.

add_error(file_path: str, error: Exception, context: Dict[str, Any] = None)[source]

Add an error to the result.

Parameters:

file_path (str) – Path to the file that caused the error.
error (Exception) – The exception that occurred.
context (Dict[str, Any], optional) – Additional context about the error.

add_skipped(file_path: str, reason: str = '')[source]

Add a skipped file.

Parameters:

file_path (str) – Path to the skipped file.
reason (str, optional) – Reason why the file was skipped.

downloaded_files: List[str]

duration_seconds: float | None = None

end_time: datetime | None = None

error_files: List[str]

errors: List[Dict[str, Any]]

property has_errors: bool: Whether any errors occurred.

message: str = ''

metadata: Dict[str, Any]

skipped_files: List[str]

start_time: datetime | None = None

success: bool = False

property success_rate: float: Success rate as a fraction (0.0 to 1.0).

to_dict() → Dict[str, Any][source]: Convert the result to a dictionary.

property total_files: int: Total number of files processed.