More fixes for IODC downloading

sat/assets/eumetsat/common.py

@@ -1,9 +1,10 @@
 """
 EO:EUM:DAT:MSG:HRSEVIRI-IODC
 """
-from satip.eumetsat import DownloadManager
+from satip.eumetsat import DownloadManager, eumetsat_filename_to_datetime
 from satip.utils import filter_dataset_ids_on_current_files
 import pandas as pd
+import os
 
 from dagster import Config
 
@@ -22,12 +23,19 @@ def download_product_range(api_key: str, api_secret: str, data_dir: str, product
     date_range = pd.date_range(start=start_str,
                                end=end_str,
                                freq="30min")
+    filenames_downloaded = []
+    for filename in os.listdir(data_dir):
+        filenames_downloaded.append(filename.split("/")[-1])
     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)
+        filtered_datasets = []
+        for dataset in datasets:
+            if dataset["id"] not in filenames_downloaded:
+                filtered_datasets.append(dataset)
+        datasets = filtered_datasets
         download_manager.download_datasets(datasets, product_id=product_id)

sat/jobs.py

@@ -2,18 +2,21 @@
 from dagster import AssetSelection, define_asset_job, EnvVar
 
 from sat.assets.eumetsat.common import EumetsatConfig
+from sat.assets import download_eumetsat_iodc_data
 
 base_path = "/mnt/storage_c/IODC/"
 
-
+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=pd.Timestamp().utcnow().strftime('%Y-%m-%d'))
 
 asset_jobs = []
-asset_job = define_asset_job(f"download_iodc_raw_files", AssetSelection.all(),
+asset_job = define_asset_job(f"download_iodc_raw_files", AssetSelection.assets(download_eumetsat_iodc_data),
                              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()},},})
+                                 'ops': {"download_eumetsat_iodc_data": {
+                                     "config": {"api_key": config.api_key, "api_secret": config.api_secret, "data_dir": config.data_dir,
+                                                "start_date": config.start_date, "end_date": config.end_date}}, }, })
 
 asset_jobs.append(asset_job)