dwdown

dwdown is a Python package for downloading weather forecast and historical observation data from the Deutscher Wetterdienst (DWD) open data server, processing GRIB2 files, and uploading the results to S3-compatible object storage such as MinIO. The package also provides parallel transfers, post-download verification (file size for HTTP downloads, ETag for object storage), and status notifications.

The source code is available in the GitHub repository.

Features

• ForecastDownloader: downloads NWP forecast files (ICON, ICON-D2, etc.) from the DWD open data server with timestep filtering.
• HistoricalDownloader: downloads historical climate observation archives.
• MOSMIXDownloader: downloads MOSMIX point-forecast files (MOSMIX_L, MOSMIX_S, MOSMIX-SNOW_S).
• OSUploader: uploads files to S3-compatible or MinIO object storage using parallel workers, and S3-compatible ETag integrity verification (size pre-check + composite hash).
• OSDownloader: downloads files from S3-compatible or MinIO object storage using parallel workers, and S3-compatible ETag integrity verification (size pre-check + composite hash).
• GribFileManager: decompresses BZ2 archives and converts GRIB2 files to CSV.
• DataMerger: filters and merges CSV data frames across variables and levels.
• Notifier: sends status messages about downloads, uploads, and errors to a Gotify server.
• Logging: writes a per-run log file listing the downloaded, failed, and corrupted files.
• Integrity checks: all three HTTP downloaders verify the downloaded file size against the size reported by DWD; object-storage transfers are verified via S3-compatible ETag.
• Parallel processing: downloads and uploads run in parallel with a configurable number of workers. GRIB processing is currently sequential.

Installation

git clone https://github.com/trholy/dwdown.git
cd dwdown
pip install .

Documentation

Read the documentation on GitLab Pages.

Usage

HistoricalDownloader

Fetch historical observations from German weather stations.

from dwdown.download import HistoricalDownloader


# Initialize HistoricalDownloader
scraper = HistoricalDownloader(
    base_url=None,                       # Base URL for historical data (defaults to DWD open data)
    files_path=None,                     # Path for downloaded files (defaults to download_files)
    extracted_files_path=None,           # Path for extracted files (defaults to extracted_files)
    log_files_path="log_files",          # Path for log files
    encoding=None,                       # Encoding for station description files (defaults to windows-1252)
    station_description_file_name=None,  # Station description filename (uses DWD default)
    delay=1,                             # 1 second delay between downloads
    retry=0,                             # Don't retry failed downloads
    timeout=30                           # 30 second timeout for requests
)

# Download station descriptions
scraper.download_station_description()

# Read station descriptions
station_descriptions = scraper.read_station_description()
print(station_descriptions)

# Get download links for specific stations
links = scraper.get_links(
    station_ids=['00001', '00003'],  # Zero-padded 5-digit station IDs
    prefix="tageswerte_KL",         # File prefix for daily weather data
    suffix="_hist.zip"              # File suffix for historical zip files
)
print(links)

# Download files — post-download size is verified against DWD-reported size
scraper.download(check_for_existence=True)

# Print status after download
print("Successfully downloaded files:", scraper.downloaded_files)
print("Failed downloads:", scraper.failed_files)
print("Download might be corrupted:", scraper.corrupted_files)

# Unpack ZIP files
scraper.extract(unpack_hist_data_only=True, check_for_existence=True)

# Read and save data as CSV
df = scraper.read_data(save_as_csv=True)
print(df)

MOSMIXDownloader

Fetch MOSMIX forecasts, which are derived from global weather models and statistically downscaled to land-based climate stations using their historical observations.

from dwdown.download import MOSMIXDownloader

# Initialize MOSMIXDownloader
# mosmix_type can be "MOSMIX_L" (single- or all-stations files) or "MOSMIX_S" (all-stations files only)
scraper = MOSMIXDownloader(
    mosmix_type="MOSMIX_L",
    base_url=None,             # Base URL constructed automatically based on mosmix_type
    files_path=None,           # Path for downloaded files (defaults to download_files)
    extracted_files_path=None, # Path for extracted files (defaults to extracted_files)
    log_files_path="log_files",
    delay=1,
    retry=0,
    timeout=30
)

# Get download links for a specific station.
# For MOSMIX_L, station_ids selects individual station files.
# To pin a specific forecast run, pass a YYYYMMDDhh token via include_pattern —
# e.g. include_pattern=["2026061109"] matches MOSMIX_L_2026061109_01001.kmz
# (single-station) and MOSMIX_L_2026061109.kmz (all-stations) alike.
# Omit include_pattern to download the latest available run.
links = scraper.get_links(
    station_ids=['01001'],  # Jan Mayen — only relevant for MOSMIX_L
)
print(f"Found {len(links)} links:", links)

# Download files — post-download size is verified against DWD-reported size
scraper.download(check_for_existence=True)

# Print status after download
print("Successfully downloaded files:", scraper.downloaded_files)
print("Failed downloads:", scraper.failed_files)
print("Download might be corrupted:", scraper.corrupted_files)

# Unpack KMZ files to KML
scraper.extract(check_for_existence=True)

# Read KML data into DataFrames
data = scraper.read_data(save_as_csv=True)

if data:
    for filename, df in data.items():
        print(f"Processed {filename}:")
        print(df.head())
else:
    print("No data processed.")

ForecastDownloader

Fetch numerical weather predictions (NWP) from DWD.

from dwdown.download import ForecastDownloader

variables = [
    'aswdifd_s',
    'relhum',
    'smi',
]

for variable in variables:

    # Initialize ForecastDownloader
    dwd_downloader = ForecastDownloader(
        url=f"https://opendata.dwd.de/weather/nwp/icon-d2/grib/09/{variable}/",
        retry=0,
        delay=0.1,
        n_jobs=4,
        files_path=f"download_files/09/{variable}",
        log_files_path="log_files"
    )

    # Fetch download links
    # timesteps: explicit list of integer forecast hours to include; None = no filter
    dwd_downloader.get_links(
        exclude_pattern=["icosahedral"],
        timesteps=list(range(49)),  # hours 0–48
    )

    # Download files
    dwd_downloader.download(check_for_existence=True)

    # Print status after download
    print("Successfully downloaded files:", dwd_downloader.downloaded_files)
    print("Failed downloads:", dwd_downloader.failed_files)
    print("Download might be corrupted:", dwd_downloader.corrupted_files)

OSUploader

The OSUploader class uploads files to MinIO or any other S3-compatible storage. After each transfer, integrity is verified via S3-compatible ETag (size pre-check + composite hash matching MinIO's multipart ETag algorithm).

from dwdown.upload import OSUploader

# Initialize OSUploader
uploader = OSUploader(
    endpoint="your-minio-server.com",
    access_key="your-access-key",
    secret_key="your-secret-key",
    files_path="download_files",     # Local directory containing files to upload
    bucket_name="weather-forecasts", # Target bucket
    secure=False,                    # True for HTTPS, False for HTTP
    log_files_path="log_files",
    n_jobs=4                         # Parallel upload workers
)

# Upload files — integrity is verified via S3-compatible ETag (size pre-check + composite hash)
uploader.upload()

# Delete local files that were successfully uploaded
uploader.delete()

# Print status after upload
print("Successfully uploaded files:", uploader.uploaded_files)
print("Upload might be corrupted:", uploader.corrupted_files)

OSDownloader

The OSDownloader class downloads files from MinIO or any other S3-compatible storage. Integrity is verified via S3-compatible ETag after each transfer. Note that unlike the HTTP downloaders, there is no separate failed_files list — all failures (network errors and integrity mismatches) are collected in corrupted_files.

from dwdown.download import OSDownloader

# Initialize OSDownloader
minio_downloader = OSDownloader(
    endpoint="your-minio-server.com",
    access_key="your-access-key",
    secret_key="your-secret-key",
    files_path="download_files",     # Local directory to save downloaded files
    bucket_name="weather-forecasts", # Source bucket
    secure=False,                    # True for HTTPS, False for HTTP
    log_files_path="log_files",
    n_jobs=4                         # Parallel download workers
)

# Download files — integrity is verified via S3-compatible ETag (size pre-check + composite hash)
minio_downloader.download(check_for_existence=True)

# Print status after download
# Note: OSDownloader has no separate failed_files list — all failures
# (network errors and integrity mismatches) are collected in corrupted_files.
print("Successfully downloaded files:", minio_downloader.downloaded_files)
print("Download might be corrupted:", minio_downloader.corrupted_files)

# Optionally delete the local copies of the downloaded files
# minio_downloader.delete()

GribFileManager and DataMerger

GribFileManager takes care of decompressing and converting GRIB2 files; DataMerger then merges and filters the resulting CSV data frames across variables and levels.

from dwdown.processing import DataMerger, GribFileManager


# Initialize GribFileManager
processor = GribFileManager(
    files_path="download_files",          # Directory containing .bz2/.grib2 files
    extracted_files_path="extracted_files",
    converted_files_path="csv_files",
)

# Retrieve downloaded filenames
file_names = processor.get_filenames()

# Decompress BZ2, convert GRIB2 to CSV, and optionally apply a geographic bounding box
processor.get_csv(
    file_names=file_names,
    apply_geo_filtering=True,
    start_lat=50.840,
    end_lat=51.000,
    start_lon=11.470,
    end_lon=11.690,
)


# Variables to build merged dataframe from
variables = [
    'aswdifd_s',
    'relhum',
    'smi',
]

# Mapping from DWD variable names to GRIB (ecCodes) short names
mapping_dictionary = {
    'aswdifd_s': 'ASWDIFD_S',
    'relhum': 'r',
    'smi': 'SMI',
}

# Per-variable level filter: only keep files whose level is in the given set
additional_patterns = {
    "relhum": [200, 975, 1000],
    "smi": [0, 9, 27],
}

# Initialize DataMerger
data_editor = DataMerger(
    files_path='csv_files/09/',
    required_columns={'latitude', 'longitude', 'valid_time'},
    join_method='inner',
    mapping_dictionary=mapping_dictionary,
    additional_patterns=additional_patterns,
)

df = data_editor.merge(
    time_step=0,
    variables=variables
)
print("Processed DataFrame:", df)

df.to_csv('processed_dataframe.csv')

Notifier

The Notifier class sends status updates via a Gotify server.

from minio import Minio
from dwdown.notify import Notifier


notifier = Notifier(
    server_url="your-gotify-server.com",
    token="your-access-token",
    priority=5,
    secure=True
)

minio_client = Minio(
    endpoint="your-minio-server.com",
    access_key="your-access-key",
    secret_key="your-secret-key",
    secure=False
)

buckets = minio_client.list_buckets()

status_dict = {}

for bucket in buckets:
    bucket_name = bucket.name
    print(f"Processing bucket: {bucket_name}")
    objects = minio_client.list_objects(bucket_name, recursive=True)
    status_dict[bucket_name] = [len([obj.object_name for obj in objects])]

notifier.send_notification(
    message=status_dict,
    script_name="downloader"
)

Directory structure

./
├── .git
├── .gitignore
├── .gitlab-ci.yml
├── LICENSE
├── README.md
├── THIRD_PARTY_LICENSES.txt
├── docs
│   ├── data
│   │   └── MappingStore.md
│   ├── download
│   │   ├── ForecastDownloader.md
│   │   ├── HistoricalDownloader.md
│   │   ├── MosmixDownloader.md
│   │   └── OSDownloader.md
│   ├── notify
│   │   └── Notifier.md
│   ├── processing
│   │   ├── DataMerger.md
│   │   └── GribFileManager.md
│   ├── upload
│   │   └── OSUploader.md
│   └── utils
│       ├── DataFrameOperator.md
│       ├── DateTimeUtils.md
│       ├── FileHandler.md
│       ├── LogHandler.md
│       ├── NetworkHandlers.md
│       ├── OSHandler.md
│       └── Utilities.md
├── example_usage
│   ├── 00_dwd_forecast_scraper.py
│   ├── 00b_dwd_hist-station-data_scraper.py
│   ├── 00c_dwd_mosmix_scraper.py
│   ├── 01_os_uploader.py
│   ├── 02_os_downloader.py
│   └── 03_data_processing.py
├── img
│   └── example_workflow.png
├── mkdocs.yml
├── pyproject.toml
├── setup.py
├── src
│   └── dwdown
│       ├── __init__.py
│       ├── data
│       │   ├── __init__.py
│       │   └── mapping.py
│       ├── download
│       │   ├── __init__.py
│       │   ├── forecast_download.py
│       │   ├── historical_download.py
│       │   ├── mosmix_download.py
│       │   └── os_download.py
│       ├── notify
│       │   ├── __init__.py
│       │   └── notifier.py
│       ├── processing
│       │   ├── __init__.py
│       │   ├── data_merging.py
│       │   └── grib_data_handling.py
│       ├── upload
│       │   ├── __init__.py
│       │   └── os_upload.py
│       └── utils
│           ├── __init__.py
│           ├── date_time_utilis.py
│           ├── df_utilis.py
│           ├── file_handling.py
│           ├── general_utilis.py
│           ├── log_handling.py
│           ├── network_handling.py
│           └── os_handling.py
└── tests
    ├── integration
    │   ├── __init__.py
    │   ├── test_historical_workflow.py
    │   ├── test_mosmix_workflow.py
    │   ├── test_nwp_forecast_workflow.py
    │   └── test_os_workflow.py
    ├── test_ForecastDownloader.py
    ├── test_HistoricalDownloader.py
    ├── test_MOSMIXDownloader.py
    ├── test_OSDownloader.py
    ├── test_OSUploader.py
    ├── test_date_time_utilis.py
    ├── test_file_handling.py
    ├── test_log_handling.py
    ├── test_mapping.py
    ├── test_network_handling.py
    ├── test_notifier.py
    ├── test_os_handling.py
    ├── test_processing.py
    └── test_utils.py

License

This project is licensed under the MIT License. See the LICENSE file for more details.

Contributing

Contributions are welcome; please open an issue or a pull request.