fix(climate): minor fixes & exclude erroring geocodes (#39)

* fix(climate): minor fixes & exclude erroring geocodes

* fix(ci): fix workflows

Author: luabida

.github/workflows/build.yaml

@@ -64,16 +64,14 @@ jobs:
       with:
         convention-name: conventionalcommits
 
-    - name: Install Conda Dependencies
-      uses: conda-incubator/setup-miniconda@v3
+    - uses: conda-incubator/setup-miniconda@v3
       with:
-        miniconda-version: "latest"
-        mamba-version: "*"
+        miniforge-version: latest
         environment-file: conda/env.yaml
         channels: conda-forge,nodefaults
         activate-environment: alertflow
-        use-mamba: true
-        miniforge-variant: Mambaforge
+        auto-update-conda: true
+        conda-solver: libmamba
 
     - name: Create environment variables file
       run: |

alertflow/dags/satellite-weather/brasil.py

@@ -16,9 +16,9 @@
 """
 
 import os
-from pathlib import Path
 from datetime import date, timedelta
 from itertools import chain
+from pathlib import Path
 
 import pendulum
 from airflow import DAG
@@ -45,7 +45,7 @@
     dag_id="COPERNICUS_BRASIL",
     description="ETL of weather data for Brazil",
     tags=["Brasil", "Copernicus"],
-    schedule="@monthly",
+    schedule="@daily",
     default_args=DEFAULT_ARGS,
     start_date=pendulum.datetime(2000, 1, 1),
     catchup=True,
@@ -68,12 +68,21 @@ def fetch_ds(locale, dt, uri, api_key):
                     f" WHERE date = '{str(dt)}'"
                 )
             )
-            table_geocodes = set(chain(*cur.fetchall()))
+            table_geocodes = set(map(int, chain(*cur.fetchall())))
 
-        all_geocodes = set([adm.code for adm in ADM2.filter(adm0=locale)])
+        to_ignore = ["2916104", "2919926", "2605459"]
+        all_geocodes = set(
+            [
+                int(adm.code)
+                for adm in ADM2.filter(adm0=locale)
+                if str(adm.code) not in to_ignore
+            ]
+        )
         geocodes = all_geocodes.difference(table_geocodes)
-        print("TABLE_GEO ", f"[{len(table_geocodes)}]: ", table_geocodes)
-        print("DIFF_GEO: ", f"[{len(geocodes)}]: ", geocodes)
+        print(f"geocodes [{len(geocodes)}]: {geocodes} ")
+
+        if not geocodes:
+            return
 
         basename = str(dt).replace("-", "_") + locale
         with request.reanalysis_era5_land(
@@ -83,7 +92,7 @@ def fetch_ds(locale, dt, uri, api_key):
             locale=locale,
         ) as ds:
             for geocode in geocodes:
-                adm = ADM2.get(code=geocode):
+                adm = ADM2.get(code=geocode)
                 with engine.connect() as conn:
                     ds.cope.to_sql(adm, conn, tablename, "weather")
             file = Path(f"{basename}.zip")

docker/Dockerfile

@@ -49,7 +49,7 @@ ENV PATH "$PATH:/usr/bin/dirname"
 
 COPY --chown=airflow alertflow/airflow.cfg ${AIRFLOW_HOME}/airflow.cfg
 COPY --chown=airflow docker/scripts/entrypoint.sh /entrypoint.sh
-COPY --chown=airflow pyproject.toml ${AIRFLOW_HOME}
+COPY --chown=airflow pyproject.toml README.md ${AIRFLOW_HOME}
 RUN chmod +x /entrypoint.sh
 
 USER airflow
@@ -59,6 +59,6 @@ RUN curl -sSL https://install.python-poetry.org | python3
 WORKDIR ${AIRFLOW_HOME}
 
 RUN poetry config virtualenvs.create false \
-  && poetry install --only main
+  && poetry install --no-root --only main
 
 ENTRYPOINT [ "/entrypoint.sh" ]