Repository: ecmwf/earthkit-data
Pull request #184 (2023-09-15 14:11:54)

WEkEO data source
This pull request allow access to datasets hosted on  [WEkEO](https://www.wekeo.eu) wrapping the [hda](https://github.com/ecmwf/hda) api.

It comes with two wrappers

- **_wekeocds_** can be used for accessing CDS datasets served on wekeo using the cdsapi standard "grammar" 
- _**wekeo**_ allow to access any dataset present in wekeo using the wekeo "grammar"

Example of _from_source_ usage for wekeocds:
```
s = from_source(
        "wekeocds",
        "EO:ECMWF:DAT:REANALYSIS_ERA5_SINGLE_LEVELS",
        variable=["2m_temperature", "mean_sea_level_pressure"],
        product_type=["reanalysis"],
        year=["2012"],
        month=["12"],
        day=["12"],
        time=["12:00"],
        format="grib"
    )
```
Example of _from_source_ usage for wekeo:
```
rq = 
s = from_source(
        "wekeo",
        "EO:CLMS:DAT:CGLS_GLOBAL_NDVI300_V1_333M",
        request={
            "datasetId": "EO:CLMS:DAT:CGLS_GLOBAL_NDVI300_V1_333M",
            "dateRangeSelectValues": [
                {
                  "name": "dtrange",
                  "start": "2014-01-01T00:00:00.000Z",
                  "end": "2014-01-01T23:59:59.999Z"
                }
            ]
      }
    )
```
Both examples are used in the test suite


FussyDuck: [![CLA assistant check](https://bol-claassistant.ecmwf.int:443/pull/badge/not_signed)](https://bol-claassistant.ecmwf.int:443/ecmwf/earthkit-data?pullRequest=184) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://bol-claassistant.ecmwf.int:443/ecmwf/earthkit-data?pullRequest=184) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://bol-claassistant.ecmwf.int:443/check/ecmwf/earthkit-data?pullRequest=184) it.</sub>
sandorkertesz: I just made an experiment by removing my `~/.hdarc` and let earthkit-data help me create it when I use the `wekeo`/`cdswekeo` source the next time. However it failed when I typed in the password:

```

An API key is needed to access this dataset. Please visit
https://www.wekeo.eu to register or sign-in
then visit https://www.wekeo.eu to retrieve you API key.

API url:
User name: my_user_name
Password: my_password
Traceback (most recent call last):
  File "/Users/cgr/git/earthkit-data-PR-other/_dev/wikio.py", line 33, in <module>
    s = from_source(
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/__init__.py", line 143, in from_source
    src = get_source(name, *args, **kwargs)
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/__init__.py", line 124, in __call__
    source = klass(*args, **kwargs)
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/core/__init__.py", line 21, in __call__
    obj.__init__(*args, **kwargs)
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/wekeocds.py", line 135, in __init__
    self.client()  # Trigger password prompt before thraeding
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/wekeocds.py", line 121, in client
    return client()
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/wekeocds.py", line 97, in client
    prompt.check()
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/prompt.py", line 131, in check
    self.ask_user_and_save()
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/prompt.py", line 146, in ask_user_and_save
    input = self.ask_user()
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/prompt.py", line 143, in ask_user
    return self.validate(prompt.ask_user())
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/prompt.py", line 76, in ask_user
    value = validate(value)
  File "/Users/cgr/git/earthkit-data-PR-other/earthkit/data/sources/prompt.py", line 53, in __call__
    assert re.fullmatch(self.pattern, value), (self.pattern, value)
AssertionError: ('[\\-0-9a-z\\!\\@\\#\\$\\%\\&\\*]{5,30}', 'my_password')
```
My password is definitely a valid one and worked when stored in the `~/.hdarc`. It contains "!", "$" and "&" ampersand on top of letters and numbers. It seems to be the a problem in the regex in `HDAAPIKeyPrompt`:

```python
validate=r"[\-0-9a-z\!\@\#\$\%\&\*]{5,30}",
```
 it should use `A-z` instead of `a-z`.
gbiavati: Thanks again for the review.
I implemented the suggested changes and tested that they are working.
the redundant code has been removed and the regular expressions have been improved.
sandorkertesz: Thank you for the update. I can confirm everything works and ready to be merged. Only my last comment should be sorted out.

Patch:
diff --git a/earthkit/data/readers/netcdf.py b/earthkit/data/readers/netcdf.py
index 90cd068..5bf7333 100644
--- a/earthkit/data/readers/netcdf.py
+++ b/earthkit/data/readers/netcdf.py
@@ -144,6 +144,122 @@ class DataSet:
         return self._bbox[(lat, lon)]
 
 
+def get_fields_from_ds(
+    ds,
+    field_type=None,
+    check_only=False,
+):  # noqa C901
+    # Select only geographical variables
+    has_lat = False
+    has_lon = False
+
+    fields = []
+
+    skip = set()
+
+    for name in ds.data_vars:
+        v = ds[name]
+        skip.update(getattr(v, "coordinates", "").split(" "))
+        skip.update(getattr(v, "bounds", "").split(" "))
+        skip.update(getattr(v, "grid_mapping", "").split(" "))
+
+    for name in ds.data_vars:
+        if name in skip:
+            continue
+
+        v = ds[name]
+
+        coordinates = []
+
+        # self.log.info('Scanning file: %s var=%s coords=%s', self.path, name, v.coords)
+
+        info = [value for value in v.coords if value not in v.dims]
+        non_dim_coords = {}
+        for coord in v.coords:
+            if coord not in v.dims:
+                non_dim_coords[coord] = ds[coord].values
+                continue
+
+            c = ds[coord]
+
+            # self.log.info("COORD %s %s %s %s", coord, type(coord), hasattr(c, 'calendar'), c)
+
+            standard_name = getattr(c, "standard_name", "")
+            axis = getattr(c, "axis", "")
+            long_name = getattr(c, "long_name", "")
+
+            use = False
+
+            if (
+                standard_name.lower() in GEOGRAPHIC_COORDS["x"]
+                or (long_name == "longitude")
+                or (axis == "X")
+            ):
+                has_lon = True
+                use = True
+
+            if (
+                standard_name.lower() in GEOGRAPHIC_COORDS["y"]
+                or (long_name == "latitude")
+                or (axis == "Y")
+            ):
+                has_lat = True
+                use = True
+
+            # print(f"  standard_name={standard_name}")
+
+            # Of course, not every one sets the standard_name
+            if (
+                standard_name in ["time", "forecast_reference_time"]
+                or long_name in ["time"]
+                or axis == "T"
+            ):
+                # we might not be able to convert time to datetime
+                try:
+                    coordinates.append(TimeCoordinate(c, coord in info))
+                    use = True
+                except ValueError:
+                    break
+
+            # TODO: Support other level types
+            if standard_name in [
+                "air_pressure",
+                "model_level_number",
+                "altitude",
+            ] or long_name in [
+                "pressure_level"
+            ]:  # or axis == 'Z':
+                coordinates.append(LevelCoordinate(c, coord in info))
+                use = True
+
+            if axis in ("X", "Y"):
+                use = True
+
+            if not use:
+                coordinates.append(OtherCoordinate(c, coord in info))
+
+        if not (has_lat and has_lon):
+            # self.log.info("NetCDFReader: skip %s (Not a 2 field)", name)
+            continue
+
+        for values in product(*[c.values for c in coordinates]):
+            slices = []
+            for value, coordinate in zip(values, coordinates):
+                slices.append(coordinate.make_slice(value))
+
+            if check_only:
+                return True
+
+            fields.append(field_type(ds, name, slices, non_dim_coords))
+
+    # if not fields:
+    #     raise Exception("NetCDFReader no 2D fields found in %s" % (self.path,))
+
+    if check_only:
+        return False
+    return fields
+
+
 class XArrayFieldGeography(Geography):
     def __init__(self, metadata, da, ds, variable):
         self.metadata = metadata
@@ -253,7 +369,6 @@ class XArrayMetadata(RawMetadata):
 class XArrayField(Field):
     def __init__(self, ds, variable, slices, non_dim_coords):
         super().__init__()
-
         self._ds = ds
         self._da = ds[variable]
 
@@ -343,118 +458,20 @@ class XArrayFieldListCore(FieldList):
             self._scan()
         return self._fields
 
+    def has_fields(self):
+        if self._fields is None:
+            return get_fields_from_ds(
+                DataSet(self.ds), field_type=self.FIELD_TYPE, check_only=True
+            )
+        else:
+            return len(self._fields)
+
     def _scan(self):
         if self._fields is None:
             self._fields = self._get_fields()
 
     def _get_fields(self):
-        return self._get_fields_from_ds(DataSet(self.ds))
-
-    def _get_fields_from_ds(self, ds):  # noqa C901
-        # Select only geographical variables
-        has_lat = False
-        has_lon = False
-
-        fields = []
-
-        skip = set()
-
-        for name in ds.data_vars:
-            v = ds[name]
-            skip.update(getattr(v, "coordinates", "").split(" "))
-            skip.update(getattr(v, "bounds", "").split(" "))
-            skip.update(getattr(v, "grid_mapping", "").split(" "))
-
-        for name in ds.data_vars:
-            if name in skip:
-                continue
-
-            v = ds[name]
-
-            coordinates = []
-
-            # self.log.info('Scanning file: %s var=%s coords=%s', self.path, name, v.coords)
-
-            info = [value for value in v.coords if value not in v.dims]
-            non_dim_coords = {}
-            for coord in v.coords:
-                if coord not in v.dims:
-                    non_dim_coords[coord] = ds[coord].values
-                    continue
-
-                c = ds[coord]
-
-                # self.log.info("COORD %s %s %s %s", coord, type(coord), hasattr(c, 'calendar'), c)
-
-                standard_name = getattr(c, "standard_name", "")
-                axis = getattr(c, "axis", "")
-                long_name = getattr(c, "long_name", "")
-
-                use = False
-
-                if (
-                    standard_name.lower() in GEOGRAPHIC_COORDS["x"]
-                    or (long_name == "longitude")
-                    or (axis == "X")
-                ):
-                    has_lon = True
-                    use = True
-
-                if (
-                    standard_name.lower() in GEOGRAPHIC_COORDS["y"]
-                    or (long_name == "latitude")
-                    or (axis == "Y")
-                ):
-                    has_lat = True
-                    use = True
-
-                # print(f"  standard_name={standard_name}")
-
-                # Of course, not every one sets the standard_name
-                if (
-                    standard_name in ["time", "forecast_reference_time"]
-                    or long_name in ["time"]
-                    or axis == "T"
-                ):
-                    # we might not be able to convert time to datetime
-                    try:
-                        coordinates.append(TimeCoordinate(c, coord in info))
-                        use = True
-                    except ValueError:
-                        break
-
-                # TODO: Support other level types
-                if standard_name in [
-                    "air_pressure",
-                    "model_level_number",
-                    "altitude",
-                ] or long_name in [
-                    "pressure_level"
-                ]:  # or axis == 'Z':
-                    coordinates.append(LevelCoordinate(c, coord in info))
-                    use = True
-
-                if axis in ("X", "Y"):
-                    use = True
-
-                if not use:
-                    coordinates.append(OtherCoordinate(c, coord in info))
-
-            if not (has_lat and has_lon):
-                # self.log.info("NetCDFReader: skip %s (Not a 2 field)", name)
-                continue
-
-            for values in product(*[c.values for c in coordinates]):
-                slices = []
-                for value, coordinate in zip(values, coordinates):
-                    slices.append(coordinate.make_slice(value))
-
-                fields.append(self.FIELD_TYPE(ds, name, slices, non_dim_coords))
-
-        # if not fields:
-        #     raise Exception("NetCDFReader no 2D fields found in %s" % (self.path,))
-
-        return fields
+        return get_fields_from_ds(DataSet(self.ds), field_type=self.FIELD_TYPE)
 
     def to_pandas(self):
         return self.to_xarray().to_pandas()
@@ -524,8 +541,9 @@ class NetCDFField(XArrayField):
 
 
 class NetCDFFieldList(XArrayFieldListCore):
+    FIELD_TYPE = NetCDFField
+
     def __init__(self, path, *args, **kwargs):
-        self.FIELD_TYPE = NetCDFField
         self.path = path
         # self._fields = None
         super().__init__(None, *args, **kwargs)
@@ -536,7 +554,20 @@ class NetCDFFieldList(XArrayFieldListCore):
         with closing(
             xr.open_mfdataset(self.path, combine="by_coords")
         ) as ds:  # or nested
-            return self._get_fields_from_ds(DataSet(ds))
+            return get_fields_from_ds(DataSet(ds), field_type=self.FIELD_TYPE)
+
+    def has_fields(self):
+        if self._fields is None:
+            import xarray as xr
+
+            with closing(
+                xr.open_mfdataset(self.path, combine="by_coords")
+            ) as ds:  # or nested
+                return get_fields_from_ds(
+                    DataSet(ds), field_type=self.FIELD_TYPE, check_only=True
+                )
+        else:
+            return len(self._fields)
 
     @classmethod
     def merge(cls, sources):
@@ -670,7 +701,7 @@ def _match_magic(magic, deeper_check):
 def reader(source, path, magic=None, deeper_check=False):
     if _match_magic(magic, deeper_check):
         fs = NetCDFFieldListReader(source, path)
-        if len(fs) > 0:
+        if fs.has_fields():
             return fs
         else:
             return NetCDFReader(source, path)
diff --git a/earthkit/data/sources/wekeo.py b/earthkit/data/sources/wekeo.py
new file mode 100644
index 0000000..b4bddcf
--- /dev/null
+++ b/earthkit/data/sources/wekeo.py
@@ -0,0 +1,160 @@
+# (C) Copyright 2020 ECMWF.
+#
+# This software is licensed under the terms of the Apache Licence Version 2.0
+# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
+# In applying this licence, ECMWF does not waive the privileges and immunities
+# granted to it by virtue of its status as an intergovernmental organisation
+# nor does it submit to any jurisdiction.
+#
+
+import os
+
+import hda
+import yaml
+from hda.api import DataOrderRequest
+
+from earthkit.data.core.thread import SoftThreadPool
+from earthkit.data.utils import tqdm
+
+from .file import FileSource
+from .prompt import APIKeyPrompt
+
+
+class HDAAPIKeyPrompt(APIKeyPrompt):
+    register_or_sign_in_url = "https://www.wekeo.eu"
+    retrieve_api_key_url = "https://www.wekeo.eu"
+
+    prompts = [
+        dict(
+            name="url",
+            default="https://wekeo-broker.apps.mercator.dpi.wekeo.eu/databroker",
+            title="API url",
+            validate=r"http.?://.*",
+        ),
+        dict(
+            name="user",
+            example="name",
+            title="User name",
+            hidden=False,
+            validate=r"[0-9a-z]+",
+        ),
+        dict(
+            name="password",
+            example="secretpassword",
+            title="Password",
+            hidden=True,
+            validate=r"[0-9A-z\!\@\#\$\%\&\*]{5,30}",
+        ),
+    ]
+
+    rcfile = "~/.hdarc"
+
+    def save(self, input, file):
+        yaml.dump(input, file, default_flow_style=False)
+
+
+class ApiClient(hda.Client):
+    name = "wekeo"
+
+    def __int__(self, *args, **kwargs):
+        super().__init__(self, *args, **kwargs)
+
+    def retrieve(self, name, request, target=None):
+        matches = self.search(request["request"])
+        out = []
+        for result in matches.results:
+            query = {"jobId": matches.job_id, "uri": result["url"]}
+            url = DataOrderRequest(self).run(query)
+            out.append(
+                os.path.abspath(
+                    self.stream(
+                        result.get("filename"), result.get("size"), target, *url
+                    )
+                )
+            )
+        return out
+
+    def download(self, download_dir: str = "."):
+        for result in self.results:
+            query = {"jobId": self.jobId, "uri": result["url"]}
+            self.debug(result)
+            url = DataOrderRequest(self.client).run(query)
+            self.stream(result.get("filename"), result.get("size"), download_dir, *url)
+
+
+EXTENSIONS = {
+    "grib": ".grib",
+    "netcdf": ".nc",
+}
+
+
+class WekeoRetriever(FileSource):
+    sphinxdoc = """
+    WekeoRetriever
+    """
+
+    @staticmethod
+    def client():
+        prompt = HDAAPIKeyPrompt()
+        prompt.check()
+
+        try:
+            return ApiClient()
+        except Exception as e:
+            if ".hdarc" in str(e):
+                prompt.ask_user_and_save()
+                return ApiClient()
+            raise
+
+    def __init__(self, dataset, *args, **kwargs):
+        super().__init__()
+
+        assert isinstance(dataset, str)
+        if len(args):
+            assert len(args) == 1
+            assert isinstance(args[0], dict)
+            assert not kwargs
+            kwargs = args[0]
+
+        requests = self.requests(**kwargs)
+
+        self.client()  # Trigger password prompt before thraeding
+
+        nthreads = min(self.settings("number-of-download-threads"), len(requests))
+
+        if nthreads < 2:
+            self.path = [self._retrieve(dataset, r) for r in requests]
+        else:
+            with SoftThreadPool(nthreads=nthreads) as pool:
+                futures = [pool.submit(self._retrieve, dataset, r) for r in requests]
+
+                iterator = (f.result() for f in futures)
+                self.path = list(tqdm(iterator, leave=True, total=len(requests)))
+
+    def _retrieve(self, dataset, request):
+        def retrieve(target, args):
+            self.client().retrieve(args[0], args[1], target)
+
+        return self.cache_file(
+            retrieve,
+            (dataset, request),
+            extension=EXTENSIONS.get(request.get("format"), ".cache"),
+        )
+
+    @staticmethod
+    def requests(**kwargs):
+        split_on = kwargs.pop("split_on", None)
+        if split_on is None or not isinstance(kwargs.get(split_on), (list, tuple)):
+            return [kwargs]
+
+        result = []
+
+        for v in kwargs[split_on]:
+            r = dict(**kwargs)
+            r[split_on] = v
+            result.append(r)
+
+        return result
+
+
+source = WekeoRetriever
diff --git a/earthkit/data/sources/wekeocds.py b/earthkit/data/sources/wekeocds.py
new file mode 100644
index 0000000..d86b2ca
--- /dev/null
+++ b/earthkit/data/sources/wekeocds.py
@@ -0,0 +1,131 @@
+# (C) Copyright 2020 ECMWF.
+#
+# This software is licensed under the terms of the Apache Licence Version 2.0
+# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
+# In applying this licence, ECMWF does not waive the privileges and immunities
+# granted to it by virtue of its status as an intergovernmental organisation
+# nor does it submit to any jurisdiction.
+#
+
+import os
+
+from hda.api import DataOrderRequest
+
+from earthkit.data.core.thread import SoftThreadPool
+from earthkit.data.decorators import normalize
+from earthkit.data.utils import tqdm
+
+from .file import FileSource
+from .wekeo import EXTENSIONS
+from .wekeo import ApiClient as WekeoClient
+from .wekeo import HDAAPIKeyPrompt
+
+
+class ApiClient(WekeoClient):
+    name = "wekeocds"
+
+    def __int__(self, *args, **kwargs):
+        super().__init__(self, *args, **kwargs)
+
+    def retrieve(self, name, request, target=None):
+        rq = {
+            "datasetId": name,
+            "multiStringSelectValues": [
+                {
+                    "name": _name,
+                    "value": _value if isinstance(_value, list) else [_value],
+                }
+                for _name, _value in request.items()
+            ],
+        }
+        if "area" in request:
+            rq.update({"boundingBoxValues": {"name": "area", "bbox": request["area"]}})
+        matches = self.search(rq)
+        out = []
+        for result in matches.results:
+            query = {"jobId": matches.job_id, "uri": result["url"]}
+            # matches.debug(result)
+            url = DataOrderRequest(self).run(query)
+            out.append(
+                self.stream(result.get("filename"), result.get("size"), target, *url)
+            )
+        return [os.path.abspath(_) for _ in out]
+
+
+class WekeoCdsRetriever(FileSource):
+    sphinxdoc = """
+    WekeoCdsRetriever
+    """
+
+    @staticmethod
+    def client():
+        prompt = HDAAPIKeyPrompt()
+        prompt.check()
+        try:
+            return ApiClient()
+        except Exception as e:
+            if ".hdarc" in str(e):
+                prompt.ask_user_and_save()
+                return ApiClient()
+
+            raise
+
+    def __init__(self, dataset, *args, **kwargs):
+        super().__init__()
+
+        assert isinstance(dataset, str)
+        if len(args):
+            assert len(args) == 1
+            assert isinstance(args[0], dict)
+            assert not kwargs
+            kwargs = args[0]
+
+        requests = self.requests(**kwargs)
+
+        self.client()  # Trigger password prompt before thraeding
+
+        nthreads = min(self.settings("number-of-download-threads"), len(requests))
+
+        if nthreads < 2:
+            self.path = [self._retrieve(dataset, r) for r in requests]
+        else:
+            with SoftThreadPool(nthreads=nthreads) as pool:
+                futures = [pool.submit(self._retrieve, dataset, r) for r in requests]
+
+                iterator = (f.result() for f in futures)
+                self.path = list(tqdm(iterator, leave=True, total=len(requests)))
+
+    def _retrieve(self, dataset, request):
+        def retrieve(target, args):
+            self.client().retrieve(args[0], args[1], target)
+
+        return self.cache_file(
+            retrieve,
+            (dataset, request),
+            extension=EXTENSIONS.get(request.get("format"), ".cache"),
+        )
+
+    @normalize("date", "date-list(%Y-%m-%d)")
+    @normalize("area", "bounding-box(list)")
+    def requests(self, **kwargs):
+        if "year" in kwargs:
+            if "month" not in kwargs:
+                kwargs["month"] = [f"{i+1:02}" for i in range(0, 12)]
+            if "day" not in kwargs:
+                kwargs["day"] = [f"{i+1:02}" for i in range(0, 31)]
+
+        split_on = kwargs.pop("split_on", None)
+        if split_on is None or not isinstance(kwargs.get(split_on), (list, tuple)):
+            return [kwargs]
+
+        result = []
+
+        for v in kwargs[split_on]:
+            r = dict(**kwargs)
+            r[split_on] = v
+            result.append(r)
+
+        return result
+
+
+source = WekeoCdsRetriever
diff --git a/earthkit/data/wrappers/xarray.py b/earthkit/data/wrappers/xarray.py
index 4d088ea..189e7c6 100644
--- a/earthkit/data/wrappers/xarray.py
+++ b/earthkit/data/wrappers/xarray.py
@@ -145,7 +145,7 @@ def wrapper(data, *args, **kwargs):
 
     if ds is not None:
         fs = netcdf.XArrayFieldList(ds, **kwargs)
-        if len(fs) > 0:
+        if fs.has_fields():
             return fs
         else:
             return XArrayDatasetWrapper(ds, *args, **kwargs)
diff --git a/environment.yml b/environment.yml
index 48bb9e4..2637183 100644
--- a/environment.yml
+++ b/environment.yml
@@ -20,6 +20,7 @@ dependencies:
 - jupyterlab
 - ecmwf-api-client>=1.6.1
 - cdsapi
+- hda
 - pip:
   - git+https://github.com/ecmwf/multiurl
   - git+https://github.com/ecmwf/pyfdb
diff --git a/setup.cfg b/setup.cfg
index ba175aa..1a8ff01 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -29,6 +29,7 @@ install_requires =
     dask
     entrypoints
     filelock
+    hda
     jinja2
     markdown
     multiurl


Test patch:
diff --git a/earthkit/data/testing.py b/earthkit/data/testing.py
index 61bb044..7dd8904 100644
--- a/earthkit/data/testing.py
+++ b/earthkit/data/testing.py
@@ -80,6 +80,7 @@ def modules_installed(*modules):
 
 NO_MARS = not os.path.exists(os.path.expanduser("~/.ecmwfapirc"))
 NO_CDS = not os.path.exists(os.path.expanduser("~/.cdsapirc"))
+NO_HDA = not os.path.exists(os.path.expanduser("~/.hdarc"))
 IN_GITHUB = os.environ.get("GITHUB_WORKFLOW") is not None
 try:
     import ecmwf.opendata  # noqa
diff --git a/tests/readers/test_netcdf_reader.py b/tests/readers/test_netcdf_reader.py
index 9011383..1ebe5d1 100644
--- a/tests/readers/test_netcdf_reader.py
+++ b/tests/readers/test_netcdf_reader.py
@@ -248,6 +248,14 @@ def test_netcdf_non_fieldlist():
     ek_ch4_l2.to_xarray()
 
 
+@pytest.mark.no_eccodes
+def test_netcdf_lazy_fieldlist_scan():
+    ds = from_source("file", earthkit_examples_file("test.nc"))
+    assert ds._fields is None
+    assert len(ds) == 2
+    assert len(ds._fields) == 2
+
+
 if __name__ == "__main__":
     from earthkit.data.testing import main
 
diff --git a/tests/sources/test_wekeo.py b/tests/sources/test_wekeo.py
new file mode 100644
index 0000000..370f837
--- /dev/null
+++ b/tests/sources/test_wekeo.py
@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+
+# (C) Copyright 2020 ECMWF.
+#
+# This software is licensed under the terms of the Apache Licence Version 2.0
+# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
+# In applying this licence, ECMWF does not waive the privileges and immunities
+# granted to it by virtue of its status as an intergovernmental organisation
+# nor does it submit to any jurisdiction.
+#
+
+import pytest
+
+from earthkit.data import from_source
+from earthkit.data.testing import NO_HDA
+
+
+@pytest.mark.long_test
+@pytest.mark.download
+@pytest.mark.skipif(NO_HDA, reason="No access to WEKEO")
+def test_wekeo_download():
+    s = from_source(
+        "wekeo",
+        "EO:CLMS:DAT:CGLS_GLOBAL_NDVI300_V1_333M",
+        request={
+            "datasetId": "EO:CLMS:DAT:CGLS_GLOBAL_NDVI300_V1_333M",
+            "dateRangeSelectValues": [
+                {
+                    "name": "dtrange",
+                    "start": "2014-01-01T00:00:00.000Z",
+                    "end": "2014-01-01T23:59:59.999Z",
+                }
+            ],
+        },
+    )
+    assert len(s) == 1
+
+
+if __name__ == "__main__":
+    from earthkit.data.testing import main
+
+    main(__file__)
diff --git a/tests/sources/test_wekeocds.py b/tests/sources/test_wekeocds.py
new file mode 100644
index 0000000..0e5e9fa
--- /dev/null
+++ b/tests/sources/test_wekeocds.py
@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+
+# (C) Copyright 2020 ECMWF.
+#
+# This software is licensed under the terms of the Apache Licence Version 2.0
+# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
+# In applying this licence, ECMWF does not waive the privileges and immunities
+# granted to it by virtue of its status as an intergovernmental organisation
+# nor does it submit to any jurisdiction.
+#
+
+import pytest
+
+from earthkit.data import from_source
+from earthkit.data.testing import NO_HDA
+
+
+@pytest.mark.long_test
+@pytest.mark.download
+@pytest.mark.skipif(NO_HDA, reason="No access to WEKEO")
+def test_wekeo_grib_1():
+    s = from_source(
+        "wekeocds",
+        "EO:ECMWF:DAT:REANALYSIS_ERA5_SINGLE_LEVELS",
+        variable=["2m_temperature", "mean_sea_level_pressure"],
+        product_type=["reanalysis"],
+        year=["2012"],
+        month=["12"],
+        day=["12"],
+        time=["13:00"],
+        format="grib",
+    )
+    assert len(s) == 2
+
+
+@pytest.mark.long_test
+@pytest.mark.download
+@pytest.mark.skipif(NO_HDA, reason="No access to CDS")
+def test_wekeo_grib_2():
+    s = from_source(
+        "wekeocds",
+        "EO:ECMWF:DAT:REANALYSIS_ERA5_SINGLE_LEVELS",
+        variable=["2m_temperature", "mean_sea_level_pressure"],
+        product_type=["reanalysis"],
+        year=["2012"],
+        month=["12"],
+        day=["12"],
+        time=["13:00"],
+        format="grib",
+        split_on="variable",
+    )
+    assert len(s) == 2
+
+
+@pytest.mark.long_test
+@pytest.mark.download
+@pytest.mark.skipif(NO_HDA, reason="No access to CDS")
+def test_wekeo_grib_3():
+    s = from_source(
+        "wekeocds",
+        "EO:ECMWF:DAT:REANALYSIS_ERA5_SINGLE_LEVELS",
+        variable=["2m_temperature", "mean_sea_level_pressure"],
+        product_type=["reanalysis"],
+        year=["2012"],
+        month=["12"],
+        day=["12", "13", "14", "15"],
+        time=["13:00"],
+        format="grib",
+    )
+    assert len(s) == 8
+
+
+@pytest.mark.long_test
+@pytest.mark.download
+@pytest.mark.skipif(NO_HDA, reason="No access to CDS")
+def test_wekeo_netcdf():
+    s = from_source(
+        "wekeocds",
+        "EO:ECMWF:DAT:REANALYSIS_ERA5_SINGLE_LEVELS",
+        variable=["2m_temperature", "mean_sea_level_pressure"],
+        product_type=["reanalysis"],
+        year=["2012"],
+        month=["12"],
+        day=["12"],
+        time=["13:00"],
+        format="netcdf",
+    )
+    assert len(s) == 2
+
+
+if __name__ == "__main__":
+    from earthkit.data.testing import main
+
+    main(__file__)
diff --git a/tests/wrappers/test_xarray.py b/tests/wrappers/test_xarray.py
index fc8e739..3088132 100644
--- a/tests/wrappers/test_xarray.py
+++ b/tests/wrappers/test_xarray.py
@@ -9,15 +9,14 @@
 # nor does it submit to any jurisdiction.
 #
 
-
-import logging
+import pytest
 
 from earthkit.data import from_object, wrappers
+from earthkit.data.testing import earthkit_examples_file
 from earthkit.data.wrappers import xarray as xr_wrapper
 
-LOG = logging.getLogger(__name__)
-
 
+@pytest.mark.no_eccodes
 def test_dataset_wrapper():
     import xarray as xr
 
@@ -29,6 +28,7 @@ def test_dataset_wrapper():
     assert isinstance(_wrapper, xr_wrapper.XArrayDatasetWrapper)
 
 
+@pytest.mark.no_eccodes
 def test_dataarray_wrapper():
     import xarray as xr
 
@@ -38,3 +38,13 @@ def test_dataarray_wrapper():
     assert isinstance(_wrapper, xr_wrapper.XArrayDataArrayWrapper)
     _wrapper = from_object(xr.DataArray())
     assert isinstance(_wrapper, xr_wrapper.XArrayDataArrayWrapper)
+
+
+@pytest.mark.no_eccodes
+def test_xarray_lazy_fieldlist_scan():
+    import xarray as xr
+
+    ds = from_object(xr.open_dataset(earthkit_examples_file("test.nc")))
+    assert ds._fields is None
+    assert len(ds) == 2
+    assert len(ds._fields) == 2