Merge pull request #105 from MET-OM/fix-cmems-current

Fix cmems current

Author: bjorkqvi

dnora/read/cmems_functions.py

@@ -1,24 +1,20 @@
-from dnora.read.ds_read_functions import  setup_temp_dir
-from dnora.type_manager.dnora_types import DnoraDataType
 import os
 import pandas as pd
 import xarray as xr
 from dnora import msg
+
 def ds_cmems_read(
     start_time: pd.Timestamp,
     end_time: pd.Timestamp,
     url: str,
     ## Partial variables from ProductReader
     lon: tuple[float],
     lat: tuple[float],
-    name: str,
-    data_type: DnoraDataType,
     ## Partial variables in ProductConfiguration
     **kwargs
 
 ):   
-    temp_dir = setup_temp_dir(data_type, name)
-
+    
     cred_file = os.path.expanduser("~/.copernicusmarine/.copernicusmarine-credentials")
     try:
         import copernicusmarine
@@ -32,8 +28,7 @@ def ds_cmems_read(
         )
 
         copernicusmarine.login()
-
-    copernicusmarine.subset(
+    ds = copernicusmarine.open_dataset(
         dataset_id=kwargs.get('dataset_id'),
         variables=kwargs.get('variables'),
         minimum_longitude=lon[0],
@@ -44,11 +39,7 @@ def ds_cmems_read(
         end_datetime=end_time.strftime("%Y-%m-%dT%H:%M:00"),
         minimum_depth=kwargs.get('minimum_depth'),
         maximum_depth=kwargs.get('maximum_depth'),
-        output_directory=temp_dir,
-        credentials_file=cred_file,
-        force_download=True,
-        output_filename=f"{name}_CMEMS_temp.nc",
+
     )
 
-    ds = xr.open_dataset(f"{temp_dir}/{name}_CMEMS_temp.nc")
-    return ds
+    return ds

dnora/read/current/__init__.py

@@ -1 +1 @@
-from . import metno
+from . import metno, cmems

dnora/read/current/cmems.py

@@ -0,0 +1,40 @@
+from dnora.type_manager.data_sources import DataSource
+from dnora.read.file_structure import FileStructure
+
+from functools import partial
+from dnora.read.cmems_functions import ds_cmems_read
+from dnora.read.product_readers import ProductReader
+from dnora.read.product_configuration import ProductConfiguration
+from dnora.process.gridded import FillNaNs
+from dnora.read.depreciation_decorator import deprecated_class_call
+import geo_parameters as gp
+
+@deprecated_class_call("CMEMS", "cmems", "current")
+class Global(ProductReader):
+    """This product is a L4 REP and NRT global total velocity field at 0m and 15m together wiht its individual components 
+    (geostrophy and Ekman) and related uncertainties. It consists of the zonal and meridional velocity at a 1h frequency 
+    and at 1/4 degree regular grid. The total velocity fields are obtained by combining CMEMS satellite Geostrophic surface 
+    currents and modelled Ekman currents at the surface and 15m depth (using ERA5 wind stress in REP and ERA5* in NRT). 
+    1 hourly product, daily and monthly means are available. This product has been initiated in the frame of CNES/CLS projects. 
+    Then it has been consolidated during the Globcurrent project (funded by the ESA User Element Program)..
+
+    DOI (product): https://doi.org/10.48670/mds-00327
+    https://https://data.marine.copernicus.eu/product/MULTIOBS_GLO_PHY_MYNRT_015_003/description
+    """
+
+    product_configuration = ProductConfiguration(
+        ds_creator_function=partial(
+            ds_cmems_read,
+            dataset_id="cmems_obs-mob_glo_phy-cur_my_0.25deg_PT1H-i",
+            variables=["uo","vo"],
+            minimum_depth=0,
+            maximum_depth=0,
+        ),
+        ds_aliases={"uo": gp.ocean.XCurrent, "vo": gp.ocean.YCurrent},
+        default_data_source=DataSource.REMOTE,
+    )
+
+    file_structure = FileStructure()
+
+    def post_processing(self):
+        return FillNaNs(0)

dnora/read/ds_read_functions.py

@@ -278,7 +278,8 @@ def read_ds_list(
         )
 
         if ds is not None:
-            msg.from_file(url)
+            if url:
+                msg.from_file(url)
             if not ds_list:
                 keys = list(ds.sizes.keys())
                 expected_shape = tuple(

dnora/read/product_configuration.py

@@ -32,7 +32,7 @@ def get_constant_url(folder, filename, file_times, **kwargs) -> list[str]:
 
 @dataclass
 class ProductConfiguration:
-    filename: str = "model_output_%Y%m.nc"
+    filename: str = None
     default_filenames: dict[DataSource, str] = field(default_factory=dict)
     default_folders: dict[DataSource, str] = field(default_factory=dict)
     tile: Optional[str] = None

dnora/read/waterlevel/ec.py

@@ -39,7 +39,7 @@ def download_GTSM_from_cds(start_time, end_time, folder="dnora_wlv_temp") -> str
 
     c = cdsapi.Client()
 
-    filename = f"{folder}/EC_GTSM_ERA5.tar.gz"
+    filename = f"{folder}/EC_GTSM_ERA5.zip"
     # cds_command_test = {
     #     'product_type': 'reanalysis',
     #     'format': 'netcdf',
@@ -75,12 +75,12 @@ def download_GTSM_from_cds(start_time, end_time, folder="dnora_wlv_temp") -> str
         months = months[0]
 
     cds_command = {
-        "data_format": "tgz",
         "variable": ["total_water_level"],
         "experiment": "reanalysis",
         "temporal_aggregation": "hourly",
         "year": years,  # 1979-2018
         "month": months,
+        "version": ["v3"]
     }
 
     c.retrieve("sis-water-level-change-timeseries-cmip6", cds_command, filename)
@@ -120,26 +120,23 @@ def __call__(
         out_file = download_GTSM_from_cds(start_time, end_time, folder=temp_folder)
 
         temppath = os.path.dirname(out_file)
-        # first unpack the tar.gz file.
-        nc_file = (
-            subprocess.run(["tar", "-ztf", out_file], stdout=subprocess.PIPE)
-            .stdout.decode("utf-8")
-            .split("\n")[0:-1]
-        )
-        nc_file = sorted([ff.strip("\r") for ff in nc_file])
-        # print(nc_file)
-        subprocess.run(
-            ["tar", "-xzvf", out_file, "--directory", temppath], stdout=subprocess.PIPE
-        )  # Extract tar file
-
+   
+        import zipfile
+        with zipfile.ZipFile(out_file, "r") as zip_ref:
+            nc_files = zip_ref.namelist()
+            
+            # Extract all files to the specified directory
+            zip_ref.extractall(temppath)
+
+        
         lon_local = np.arange(lon[0], lon[1], 0.1)
         lat_local = np.arange(lat[0], lat[1], 0.1)
         grid_x, grid_y = np.meshgrid(lon_local, lat_local, indexing="xy")
 
-        print(nc_file)
+        print(nc_files)
         grid_tot = []
         time_tot = []
-        for ncfile in nc_file:
+        for ncfile in nc_files:
             # print(os.path.join(temppath,nc_file))
             waterlevel = xr.open_dataset(
                 os.path.join(temppath, ncfile), engine="netcdf4"
@@ -190,7 +187,8 @@ def __call__(
         # )
 
         # print(waterlevel_gridded)
-        coord_dict = {"lon": lon_local, "lat": lat_local, "time": time}
+
+        coord_dict = {"lon": lon_local, "lat": lat_local, "time": time_tot}
         data_dict = {"eta": grid_tot}
         meta_dict = {"description": "Waterlevel from GTSM/ERA5"}
 

dnora/utils/io.py

@@ -55,22 +55,25 @@ def get_url(
         folder = time_stamp.strftime(folder)
     url = []
     for fn in filename:
-        url_temp = Path(folder).joinpath(fn)
-        # If we are on a Windows machine, then / will be replaced by \
-        # If we have an url, then we still want /
-        if "http" in str(url_temp) or "ftp" in str(url_temp):
-            url_temp = url_temp.as_posix()
-        else:
-            url_temp = str(url_temp)
-
-        url_temp = re.sub("https:/", "https://", url_temp, 1)
-        url_temp = re.sub("http:/", "http://", url_temp, 1)
-        url_temp = re.sub("ftp:/", "ftp://", url_temp, 1)
-        if time_stamp is not None:
-            for floor_hour in range(1, 24):
-                hfloor = int(np.floor(time_stamp.hour / floor_hour) * floor_hour)
-                url_temp = re.sub(f"\\[{floor_hour}\\]", f"{hfloor:02.0f}", url_temp)
+        if fn: # Otherwise '' will become '.'
+            url_temp = Path(folder).joinpath(fn)
+            # If we are on a Windows machine, then / will be replaced by \
+            # If we have an url, then we still want /
+            if "http" in str(url_temp) or "ftp" in str(url_temp):
+                url_temp = url_temp.as_posix()
+            else:
+                url_temp = str(url_temp)
 
+            url_temp = re.sub("https:/", "https://", url_temp, 1)
+            url_temp = re.sub("http:/", "http://", url_temp, 1)
+            url_temp = re.sub("ftp:/", "ftp://", url_temp, 1)
+            if time_stamp is not None:
+                for floor_hour in range(1, 24):
+                    hfloor = int(np.floor(time_stamp.hour / floor_hour) * floor_hour)
+                    url_temp = re.sub(f"\\[{floor_hour}\\]", f"{hfloor:02.0f}", url_temp)
+        else:
+            url_temp = fn
+        
         url.append(url_temp)
     if len(url) == 1 and not get_list:
         return os.path.expanduser(url[0])

tests/test_readers/test_remote_waterlevel_readers.py

@@ -30,10 +30,10 @@ def cleanup():
 @pytest.mark.remote
 def test_era5(grid, timevec):
     cleanup()
-    model = dn.modelrun.ModelRun(grid, year=2023, month=4, day=1)
+    model = dn.modelrun.ModelRun(grid, start_time='2023-04-30 00:00', end_time='2023-05-01 23:00')
     model.import_waterlevel(dn.read.waterlevel.ec.GTSM_ERA5())
 
-    assert np.all(model.waterlevel().time() == timevec)
+    assert np.all(model.waterlevel().time() == pd.date_range('2023-04-30 00:00', '2023-05-01 23:00', freq="1h"))
 
     grid_is_covered(grid, model.waterlevel())
     cleanup()