Update calls to consumer for synchronous stdout capture (#15)

.github/workflows/ci.yml

@@ -21,6 +21,8 @@ jobs:
   # * Installs dependencies and caches them
   build-venv:
     runs-on: ubuntu-latest
+    container: quay.io/condaforge/miniforge3:latest
+
     steps:
       - name: Checkout repository
         uses: actions/checkout@v3
@@ -38,16 +40,17 @@ jobs:
       # Should mirror the build-venv stage in the Containerfile
       - name: Build venv
         run: |
-          sudo apt -qq update && sudo apt -qq install -y build-essential libgeos-dev
-          python -m venv ./venv
-          ./venv/bin/pip install --upgrade -q pip wheel setuptools
+          apt -qq update && apt -qq install -y build-essential
+          conda create -p ./venv python=3.10
+          ./venv/bin/python -m pip install --upgrade -q pip wheel setuptools
         if: steps.restore-cache.outputs.cache-hit != 'true'
 
       # Should mirror the build-reqs stage in the Containerfile
       # * Except this installs the dev dependencies and binaries as well
       - name: Install all dependencies
         run: |
-          ./venv/bin/pip install -q --no-binary=dags .[dev]
+          conda install -p ./venv -q -y eccodes zarr
+          ./venv/bin/python -m pip install -q .[dev]
         if: steps.restore-cache.outputs.cache-hit != 'true'
 
       # Cache the virtualenv for future runs
@@ -57,7 +60,7 @@ jobs:
           path: ./venv
           key: ${{ steps.restore-cache.outputs.cache-primary-key }}
         if: steps.restore-cache.outputs.cache-hit != 'true'
-
+  
   # Define a unittest job that runs on all branches and PRs
   test-unit:
     runs-on: ubuntu-latest

nwp/assets/ecmwf/mars.py

@@ -1,7 +1,4 @@
-import os
-import subprocess
-import sys
-
+import nwp_consumer.cmd.main as consumer
 from dagster import Config, OpExecutionContext, op
 from dagster_docker import execute_docker_container
 
@@ -35,28 +32,35 @@ def nwp_consumer_docker_op(context: OpExecutionContext, config: NWPConsumerConfi
 
 @op
 def nwp_consumer_download_op(context: OpExecutionContext, config: NWPConsumerConfig):
-    process = subprocess.run(
-        [f"{os.path.dirname(sys.executable)}/nwp-consumer", "download",
-         f'--source={config.source}', f'--from={config.date_from}', f'--to={config.date_to}',
-         f'--rdir={config.raw_dir}', f'--zdir={config.zarr_dir}'],
-        capture_output=True,
-        text=True
-    )
-    print(process.stdout)
-    print(process.stderr)
-    process.check_returncode()
+    """Download the data from the source."""
+    consumer.run({
+        "download": True,
+        "convert": False,
+        "consume": False,
+        "check": False,
+        "--source": config.source,
+        "--sink": "local",
+        "--from": config.date_from,
+        "--to": config.date_to,
+        "--rdir": config.raw_dir,
+        "--zdir": config.zarr_dir,
+        "--create-latest": False,
+    })
     return config
 
 @op
 def nwp_consumer_convert_op(context: OpExecutionContext, downloadedConfig: NWPConsumerConfig):
-    process = subprocess.run(
-        [f"{os.path.dirname(sys.executable)}/nwp-consumer", "convert",
-         f'--source={downloadedConfig.source}', f'--from={downloadedConfig.date_from}',
-         f'--to={downloadedConfig.date_to}',
-         f'--rdir={downloadedConfig.raw_dir}', f'--zdir={downloadedConfig.zarr_dir}'],
-        capture_output=True,
-        text=True
-    )
-    print(process.stdout)
-    print(process.stderr)
-
+    """Convert the downloaded data to zarr format."""
+    consumer.run({
+        "download": False,
+        "convert": True,
+        "consume": False,
+        "check": False,
+        "--source": downloadedConfig.source,
+        "--sink": "local",
+        "--from": downloadedConfig.date_from,
+        "--to": downloadedConfig.date_to,
+        "--rdir": downloadedConfig.raw_dir,
+        "--zdir": downloadedConfig.zarr_dir,
+        "--create-latest": False,
+    })

pyproject.toml

@@ -25,7 +25,7 @@ dependencies = [
     "dagster-docker == 0.20.11",
     "huggingface-hub == 0.16.4",
     "numpy >= 1.23.0",
-    "nwp-consumer >= 0.1.13",
+    "nwp-consumer == 0.1.17",
     "ocf-blosc2 == 0.0.3",
     "pathlib == 1.0.1",
     "requests >= 2.28.0",