FIX: various fixes and uniformization for s3 reading/writing

Author: wolfidan

src/pyrad_proc/pyrad/flow/flow_aux.py

@@ -1000,16 +1000,24 @@ def _generate_prod(dataset, cfg, prdname, prdfunc, dsname, voltime, runinfo=None
         filenames = prdfunc(dataset, prdcfg)
         if isinstance(filenames, str):  # convert to list if needed
             filenames = [filenames]
-        if "s3copypath" in prdcfg and filenames is not None:  # copy to S3
-            s3AccessPolicy = (
-                prdcfg["s3AccessPolicy"] if "s3AccessPolicy" in prdcfg else None
-            )
+        if (
+            "s3BucketWrite" in prdcfg
+            and "s3EndpointWrite" in prdcfg
+            and filenames is not None
+        ):  # copy to S3
+            s3AccessPolicy = prdcfg.get("s3AccessPolicy", None)
+            s3path = prdcfg.get("s3PathWrite", None)
             for fname in filenames:
                 if (
                     prdcfg["basepath"] in fname
                 ):  # only products saved to standard basepath
                     write_to_s3(
-                        fname, prdcfg["basepath"], prdcfg["s3copypath"], s3AccessPolicy
+                        fname,
+                        prdcfg["basepath"],
+                        prdcfg["s3EndpointWrite"],
+                        prdcfg["s3BucketWrite"],
+                        s3path,
+                        s3AccessPolicy,
                     )
         return False
     except Exception as inst:
@@ -1221,6 +1229,9 @@ def _create_cfg_dict(cfgfile):
     if "MFScale" not in cfg:
         cfg.update({"MFScale": 0})
 
+    if not cfg["datapath"]:  # empty datapath in case of s3 reading
+        cfg["datapath"] = ["" for rad in range(cfg["NumRadars"])]
+
     # parameters necessary to read correctly MF grid binary files
     if "BinFileParams" not in cfg:
         bin_file_params = {
@@ -1333,35 +1344,35 @@ def _create_datacfg_dict(cfg):
     datacfg.update({"DataTypeIDInFiles": cfg["DataTypeIDInFiles"]})
 
     # s3 buckets
-    if "bucket" in cfg:
+    if "s3BucketRead" in cfg:
         try:
-            datacfg["s3_key"] = os.environ["S3_IN_KEY"]
-            datacfg["s3_secret_key"] = os.environ["S3_IN_SECRET"]
+            datacfg["s3KeyRead"] = os.environ["S3_KEY_READ"]
+            datacfg["s3SecretRead"] = os.environ["S3_SECRET_READ"]
         except KeyError:
             warn(
-                "Define environment variables S3_IN_KEY and S3_IN_SECRET"
+                "Define environment variables S3_KEY_READ and S3_SECRET_READ"
                 " to get input data from S3 buckets."
             )
 
-        if "s3path" in cfg:
-            datacfg.update({"s3path": cfg["s3path"]})
+        if "s3PathRead" in cfg:
+            datacfg.update({"s3PathRead": cfg["s3PathRead"]})
         else:
-            warn("Unable to read data from s3 bucket. Define s3path")
-        if "s3_url" in cfg:
-            datacfg.update({"s3_url": cfg["s3_url"]})
+            warn("Unable to read data from s3 bucket. Define s3PathRead")
+        if "s3EndpointRead" in cfg:
+            datacfg.update({"s3EndpointRead": cfg["s3EndpointRead"]})
         else:
-            warn("Unable to read data from s3 bucket. Define s3_url")
+            warn("Unable to read data from s3 bucket. Define s3EndpointRead")
 
         if "rm_s3_file" in cfg:
             datacfg.update({"rm_s3_file": cfg["rm_s3_file"]})
 
         if (
-            "s3path" in datacfg
-            and "s3_url" in datacfg
-            and "s3_key" in datacfg
-            and "s3_secret_key" in datacfg
+            "s3PathRead" in datacfg
+            and "s3EndpointRead" in datacfg
+            and "s3KeyRead" in datacfg
+            and "s3SecretRead" in datacfg
         ):
-            datacfg.update({"bucket": cfg["bucket"]})
+            datacfg.update({"s3BucketRead": cfg["s3BucketRead"]})
 
     # Modify size of radar or radar spectra object
     datacfg.update({"elmin": cfg.get("elmin", None)})
@@ -1582,11 +1593,14 @@ def _create_prdcfg_dict(cfg, dataset, product, voltime, runinfo=None):
     prdcfg.update({"imgformat": cfg["imgformat"]})
     prdcfg.update({"RadarName": cfg["RadarName"]})
 
-    if "s3copypath" in cfg:
-        prdcfg.update({"s3copypath": cfg["s3copypath"]})
+    if "s3EndpointWrite" in cfg:
+        prdcfg.update({"s3EndpointWrite": cfg["s3EndpointWrite"]})
+    if "s3BucketWrite" in cfg:
+        prdcfg.update({"s3BucketWrite": cfg["s3BucketWrite"]})
+    if "s3PathWrite" in cfg:
+        prdcfg.update({"s3PathWrite": cfg["s3PathWrite"]})
     if "s3AccessPolicy" in cfg:
         prdcfg.update({"s3AccessPolicy": cfg["s3AccessPolicy"]})
-
     if "RadarBeamwidth" in cfg:
         prdcfg.update({"RadarBeamwidth": cfg["RadarBeamwidth"]})
     if "ppiImageConfig" in cfg:
@@ -1641,9 +1655,13 @@ def _get_datatype_list(cfg, radarnr="RADAR001"):
         if "datatype" not in cfg[dataset]:
             continue
         if isinstance(cfg[dataset]["datatype"], str):
-            (radarnr_descr, datagroup, datatype_aux, dataset_save, product_save) = (
-                get_datatype_fields(cfg[dataset]["datatype"])
-            )
+            (
+                radarnr_descr,
+                datagroup,
+                datatype_aux,
+                dataset_save,
+                product_save,
+            ) = get_datatype_fields(cfg[dataset]["datatype"])
             if datagroup != "PROC" and radarnr_descr == radarnr:
                 if (dataset_save is None) and (product_save is None):
                     datatypesdescr.add(
@@ -1667,9 +1685,13 @@ def _get_datatype_list(cfg, radarnr="RADAR001"):
                     )
         else:
             for datatype in cfg[dataset]["datatype"]:
-                (radarnr_descr, datagroup, datatype_aux, dataset_save, product_save) = (
-                    get_datatype_fields(datatype)
-                )
+                (
+                    radarnr_descr,
+                    datagroup,
+                    datatype_aux,
+                    dataset_save,
+                    product_save,
+                ) = get_datatype_fields(datatype)
                 if datagroup != "PROC" and radarnr_descr == radarnr:
                     if dataset_save is None and product_save is None:
                         datatypesdescr.add(
@@ -1799,7 +1821,7 @@ def _get_masterfile_list(datatypesdescr, starttimes, endtimes, datacfg, scan_lis
         )
         return [], None, None
 
-    if "bucket" in datacfg:
+    if "s3BucketRead" in datacfg:
         masterfilelist = get_file_list_s3(
             masterdatatypedescr, starttimes, endtimes, datacfg, scan=masterscan
         )

src/pyrad_proc/pyrad/io/io_aux.py

@@ -2122,12 +2122,12 @@ def get_scan_files_to_merge_s3(
 
     s3_client = boto3.client(
         "s3",
-        endpoint_url=cfg["s3_url"],
-        aws_access_key_id=cfg["s3_key"],
-        aws_secret_access_key=cfg["s3_secret_key"],
+        endpoint_url=cfg["s3EndpointRead"],
+        aws_access_key_id=cfg["s3KeyRead"],
+        aws_secret_access_key=cfg["s3SecretRead"],
         verify=False,
     )
-    response = s3_client.list_objects_v2(Bucket=cfg["bucket"])
+    response = s3_client.list_objects_v2(Bucket=cfg["s3BucketRead"])
 
     dayinfo = voltime.strftime("%y%j")
     timeinfo = voltime.strftime("%H%M")
@@ -2751,12 +2751,12 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
 
     s3_client = boto3.client(
         "s3",
-        endpoint_url=cfg["s3_url"],
-        aws_access_key_id=cfg["s3_key"],
-        aws_secret_access_key=cfg["s3_secret_key"],
+        endpoint_url=cfg["s3EndpointRead"],
+        aws_access_key_id=cfg["s3KeyRead"],
+        aws_secret_access_key=cfg["s3SecretRead"],
         verify=False,
     )
-    response = s3_client.list_objects_v2(Bucket=cfg["bucket"])
+    response = s3_client.list_objects_v2(Bucket=cfg["s3BucketRead"])
 
     for starttime, endtime in zip(starttimes, endtimes):
         startdate = starttime.replace(hour=0, minute=0, second=0, microsecond=0)
@@ -2775,9 +2775,9 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
                     )
                 daydir = (starttime + datetime.timedelta(days=i)).strftime(fpath_strf)
                 if daydir == "":
-                    pattern = f'{cfg["s3path"]}*{scan}*'
+                    pattern = f'{cfg["s3PathRead"]}*{scan}*'
                 else:
-                    pattern = f'{cfg["s3path"]}{daydir}/*{scan}*'
+                    pattern = f'{cfg["s3PathRead"]}{daydir}/*{scan}*'
                 dayfilelist = []
                 for content in response["Contents"]:
                     if fnmatch.fnmatch(content["Key"], pattern):
@@ -2805,7 +2805,7 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
                         f'M{cfg["RadarRes"][ind_rad]}'
                         f'{cfg["RadarName"][ind_rad]}{dayinfo}'
                     )
-                    datapath = f'{cfg["s3path"]}{dayinfo}/{basename}/'
+                    datapath = f'{cfg["s3PathRead"]}{dayinfo}/{basename}/'
 
                     # check that M files exist. if not search P files
                     pattern = f"{datapath}{basename}*{scan}*"
@@ -2818,7 +2818,7 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
                             f'P{cfg["RadarRes"][ind_rad]}'
                             f'{cfg["RadarName"][ind_rad]}{dayinfo}'
                         )
-                        datapath = f'{cfg["s3path"]}{dayinfo}/{basename}/'
+                        datapath = f'{cfg["s3PathRead"]}{dayinfo}/{basename}/'
                         pattern = f"{datapath}{basename}*{scan}*"
                         for content in response["Contents"]:
                             if fnmatch.fnmatch(content["Key"], pattern):
@@ -2837,9 +2837,9 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
                         fpath_strf
                     )
                     if daydir == "":
-                        pattern = f'{cfg["s3path"]}*{scan}*'
+                        pattern = f'{cfg["s3PathRead"]}*{scan}*'
                     else:
-                        pattern = f'{cfg["s3path"]}{daydir}/*{scan}*'
+                        pattern = f'{cfg["s3PathRead"]}{daydir}/*{scan}*'
 
                     dayfilelist = []
                     for content in response["Contents"]:
@@ -2859,9 +2859,9 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
                         fpath_strf
                     )
                     if daydir == "":
-                        pattern = f'{cfg["s3path"]}{scan}/*'
+                        pattern = f'{cfg["s3PathRead"]}{scan}/*'
                     else:
-                        pattern = f'{cfg["s3path"]}{scan}/{daydir}/*'
+                        pattern = f'{cfg["s3PathRead"]}{scan}/{daydir}/*'
 
                     dayfilelist = []
                     for content in response["Contents"]:
@@ -2874,7 +2874,7 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
                         f'{cfg["RadarName"][ind_rad]}{dayinfo}'
                     )
                     datapath = (
-                        f'{cfg["s3path"]}M{cfg["RadarRes"][ind_rad]}'
+                        f'{cfg["s3PathRead"]}M{cfg["RadarRes"][ind_rad]}'
                         f'{cfg["RadarName"][ind_rad]}'
                     )
 
@@ -2890,7 +2890,7 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
                             f'{cfg["RadarName"][ind_rad]}{dayinfo}'
                         )
                         datapath = (
-                            f'{cfg["s3path"]}P{cfg["RadarRes"][ind_rad]}'
+                            f'{cfg["s3PathRead"]}P{cfg["RadarRes"][ind_rad]}'
                             f'{cfg["RadarName"][ind_rad]}'
                         )
                         pattern = f"{datapath}{basename}*{scan}*"
@@ -2910,7 +2910,7 @@ def get_file_list_s3(datadescriptor, starttimes, endtimes, cfg, scan=None):
                 if not os.path.isdir(datapath):
                     os.makedirs(datapath)
                 fname_aux = f"{datapath}{os.path.basename(filename)}"
-                s3_client.download_file(cfg["bucket"], filename, fname_aux)
+                s3_client.download_file(cfg["s3PathRead"], filename, fname_aux)
                 _, tend_sweeps, _, _ = get_sweep_time_coverage(fname_aux)
                 for tend in tend_sweeps:
                     if starttime <= tend <= endtime: