Use cloudnet-api-client to download raw files

Author: tukiains

pyproject.toml

@@ -20,6 +20,7 @@ classifiers = [
 dependencies = [
   "ceilopyter",
   "cftime",
+  "cloudnet-api-client",
   "cloudnetpy[extras]>=1.69.9",
   "doppy>=0.3.5",
   "influxdb-client",

scripts/cloudnet.py

@@ -16,6 +16,7 @@
 from typing import TypeVar
 from uuid import UUID
 
+from cloudnet_api_client import APIClient
 from processing import utils
 from processing.dvas import Dvas
 from processing.fetch import fetch
@@ -189,7 +190,8 @@ def process_main(args):
     storage_api = StorageApi(config, session)
     pid_utils = PidUtils(config, session)
     dvas = Dvas(config, md_api)
-    processor = Processor(md_api, storage_api, pid_utils, dvas)
+    client = APIClient(f"{config.dataportal_url}/api/", session)
+    processor = Processor(md_api, storage_api, pid_utils, dvas, client)
 
     if args.cmd == "fetch":
         _print_fetch_header(args)

scripts/worker.py

@@ -9,6 +9,7 @@
 from tempfile import TemporaryDirectory
 from threading import Event
 
+from cloudnet_api_client import APIClient
 from processing import utils
 from processing.config import Config
 from processing.dvas import Dvas
@@ -68,8 +69,9 @@ def __init__(self, config: Config):
         self.storage_api = StorageApi(self.config, self.session)
         self.pid_utils = PidUtils(self.config, self.session)
         self.dvas = Dvas(self.config, self.md_api)
+        self.client = APIClient(f"{config.dataportal_url}/api/", self.session)
         self.processor = Processor(
-            self.md_api, self.storage_api, self.pid_utils, self.dvas
+            self.md_api, self.storage_api, self.pid_utils, self.dvas, self.client
         )
         self.logger = MemoryLogger()
         self.n_processed_tasks = 0

src/processing/fetch.py

@@ -63,7 +63,7 @@ def get_raw_instrument_metadata(self) -> list:
         return [m for m in metadata if not m["filename"].lower().endswith(".lv0")]
 
     def get_raw_model_metadata(self) -> list:
-        url = f"{self.api_url}raw-model-files"
+        url = f"{self.api_url}/raw-model-files"
         payload = {
             **self.payload,
             "status": ["uploaded", "processed"],

src/processing/processor.py

@@ -1,11 +1,11 @@
 import datetime
 import logging
-import re
 import uuid
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Literal
 
+import cloudnet_api_client as cac
 from cloudnetpy.exceptions import PlottingError
 from cloudnetpy.model_evaluation.plotting.plotting import generate_L3_day_plots
 from cloudnetpy.plotting import Dimensions, PlotParameters, generate_figure
@@ -87,11 +87,13 @@ def __init__(
         storage_api: StorageApi,
         pid_utils: PidUtils,
         dvas: Dvas,
+        client: cac.APIClient,
     ):
         self.md_api = md_api
         self.storage_api = storage_api
         self.pid_utils = pid_utils
         self.dvas = dvas
+        self.client = client
 
     def get_site(self, site_id: str) -> Site:
         site = self.md_api.get(f"api/sites/{site_id}")
@@ -214,6 +216,7 @@ def download_instrument(
         else:
             start_date, end_date = date
         start_date_ext, end_date_ext = start_date, end_date
+
         if time_offset is not None:
             if largest_only:
                 raise ValueError("Cannot use both time_offset and largest_only")
@@ -223,54 +226,49 @@ def download_instrument(
                 start_date_ext -= datetime.timedelta(days=1)
             elif time_offset > TIMEDELTA_ZERO:
                 end_date_ext += datetime.timedelta(days=1)
-        payload = self._get_payload(
-            site=site_id,
-            date=(start_date_ext, end_date_ext),
-            instrument=instrument_id,
+
+        metadata = self.client.raw_metadata(
+            site_id,
+            date_from=start_date_ext,
+            date_to=end_date_ext,
+            instrument_id=instrument_id,
             instrument_pid=instrument_pid,
-            skip_created=True,
             filename_prefix=filename_prefix,
             filename_suffix=filename_suffix,
+            status=["uploaded", "processed"],
         )
-        upload_metadata = self.md_api.get("api/raw-files", payload)
-        if include_pattern is not None:
-            upload_metadata = _include_records_with_pattern_in_filename(
-                upload_metadata, include_pattern
+        if include_pattern:
+            metadata = self.client.filter(metadata, include_pattern=include_pattern)
+        if exclude_pattern:
+            metadata = self.client.filter(metadata, exclude_pattern=exclude_pattern)
+        if include_tag_subset:
+            metadata = self.client.filter(
+                metadata, include_tag_subset=include_tag_subset
             )
-        if exclude_pattern is not None:
-            upload_metadata = _exclude_records_with_pattern_in_filename(
-                upload_metadata, exclude_pattern
+        if exclude_tag_subset:
+            metadata = self.client.filter(
+                metadata, exclude_tag_subset=exclude_tag_subset
             )
-        if include_tag_subset is not None:
-            upload_metadata = [
-                record
-                for record in upload_metadata
-                if include_tag_subset.issubset(set(record["tags"]))
-            ]
-        if exclude_tag_subset is not None:
-            upload_metadata = [
-                record
-                for record in upload_metadata
-                if not exclude_tag_subset.issubset(set(record["tags"]))
-            ]
-        if not upload_metadata:
+
+        if not metadata:
             if allow_empty:
                 return [], []
             else:
                 raise utils.RawDataMissingError
+
         if largest_only:
-            upload_metadata = [max(upload_metadata, key=lambda item: int(item["size"]))]
-        full_paths, uuids = self.storage_api.download_raw_data(
-            upload_metadata, directory
-        )
+            metadata = [max(metadata, key=lambda item: int(item.size))]
+
+        full_paths = self.client.download(metadata, directory, progress=False)
+        uuids = [f.uuid for f in metadata]
+
         if time_offset is not None:
             uuids = [
-                meta["uuid"]
-                for meta in upload_metadata
-                if start_date
-                <= datetime.date.fromisoformat(meta["measurementDate"])
-                <= end_date
+                meta.uuid
+                for meta in metadata
+                if start_date <= meta.measurement_date <= end_date
             ]
+
         if largest_only:
             return full_paths[0], uuids
         return full_paths, uuids
@@ -563,24 +561,6 @@ def _dimensions2dict(dimensions: Dimensions) -> dict:
     }
 
 
-def _include_records_with_pattern_in_filename(metadata: list, pattern: str) -> list:
-    """Includes only records with certain pattern."""
-    return [
-        row
-        for row in metadata
-        if re.search(pattern, row["filename"], flags=re.IGNORECASE)
-    ]
-
-
-def _exclude_records_with_pattern_in_filename(metadata: list, pattern: str) -> list:
-    """Excludes records with certain pattern."""
-    return [
-        row
-        for row in metadata
-        if not re.search(pattern, row["filename"], flags=re.IGNORECASE)
-    ]
-
-
 def _full_product_to_l3_product(full_product: str):
     """Returns l3 product name."""
     return full_product.split("-")[1]