Add simple IODC downloader (#6)

* Add simple IODC downloader

* Update pyproject.toml

pyproject.toml

@@ -28,7 +28,8 @@ dependencies = [
     "pathlib == 1.0.1",
     "requests == 2.31.0",
     "xarray == 2023.2.0",
-    "zarr == 2.14.2"
+    "zarr == 2.14.2",
+    "satip == 2.11.10",
 ]
 
 [project.optional-dependencies]

sat/__init__.py

@@ -1,9 +1,10 @@
 from dagster import Definitions, load_assets_from_modules
 
-from . import assets
+from sat import assets, jobs
 
 all_assets = load_assets_from_modules([assets])
 
 defs = Definitions(
     assets=all_assets,
+    jobs=jobs.asset_jobs,
 )

sat/assets/__init__.py

@@ -0,0 +1 @@
+from sat.assets.eumetsat.iodc import download_eumetsat_iodc_data

sat/assets/eumetsat/__init__.py

@@ -0,0 +1 @@
+from .common import download_product_range, EumetsatConfig

sat/assets/eumetsat/common.py

@@ -0,0 +1,33 @@
+"""
+EO:EUM:DAT:MSG:HRSEVIRI-IODC
+"""
+from satip.eumetsat import DownloadManager
+from satip.utils import filter_dataset_ids_on_current_files
+import pandas as pd
+
+from dagster import Config
+
+
+class EumetsatConfig(Config):
+    api_key: str
+    api_secret: str
+    data_dir: str
+    start_date: str
+    end_date: str
+
+def download_product_range(api_key: str, api_secret: str, data_dir: str, product_id: str, start_date: pd.Timestamp, end_date: pd.Timestamp):
+    download_manager = DownloadManager(user_key=api_key, user_secret=api_secret, data_dir=data_dir)
+    start_str = start_date.strftime("%Y-%m-%d")
+    end_str = end_date.strftime("%Y-%m-%d")
+    date_range = pd.date_range(start=start_str,
+                               end=end_str,
+                               freq="30min")
+    for date in date_range:
+        start_date = pd.Timestamp(date) - pd.Timedelta("1min")
+        end_date = pd.Timestamp(date) + pd.Timedelta("1min")
+        datasets = download_manager.identify_available_datasets(
+            start_date=start_date.tz_localize(None).strftime("%Y-%m-%d-%H-%M-%S"),
+            end_date=end_date.tz_localize(None).strftime("%Y-%m-%d-%H-%M-%S"),
+        )
+        datasets = filter_dataset_ids_on_current_files(datasets, data_dir)
+        download_manager.download_datasets(datasets, product_id=product_id)

sat/assets/eumetsat/iodc.py

@@ -0,0 +1,12 @@
+from dagster import asset  # import the `dagster` library
+from . import download_product_range, EumetsatConfig
+import pandas as pd
+
+@asset
+def download_eumetsat_iodc_data(config: EumetsatConfig) -> None:
+    download_product_range(api_key=config.api_key,
+                           api_secret=config.api_secret,
+                           data_dir=config.data_dir,
+                           product_id="EO:EUM:DAT:MSG:HRSEVIRI-IODC",
+                           start_date=pd.Timestamp(config.start_date),
+                           end_date=pd.Timestamp(config.end_date))

sat/jobs.py

@@ -0,0 +1,19 @@
+import pandas as pd
+from dagster import AssetSelection, define_asset_job, EnvVar
+
+from sat.assets.eumetsat.common import EumetsatConfig
+
+base_path = "/mnt/storage_c/IODC/"
+
+
+
+asset_jobs = []
+asset_job = define_asset_job(f"download_iodc_raw_files", AssetSelection.all(),
+                             config={
+                                 'ops': {"download_eumetsat_iodc_data": {"config": EumetsatConfig(api_key=EnvVar("EUMETSAT_API_KEY"),
+                                                                                                  api_secret=EnvVar("EUMETSAT_API_SECRET"),
+                                                                                                  data_dir=base_path,
+                                                                                                  start_date="2017-02-01",
+                                                                                                  end_date=str(pd.Timestamp().utcnow())).to_fields_dict()},},})
+
+asset_jobs.append(asset_job)