Merge pull request #4411 from airqo-platform/staging

move to production

k8s/auth-service/values-prod.yaml

@@ -6,7 +6,7 @@ app:
 replicaCount: 3
 image:
   repository: eu.gcr.io/airqo-250220/airqo-auth-api
-  tag: prod-fbd4c147-1739233652
+  tag: prod-ae87cecf-1739238823
 nameOverride: ''
 fullnameOverride: ''
 podAnnotations: {}

k8s/device-registry/values-prod.yaml

@@ -6,7 +6,7 @@ app:
 replicaCount: 3
 image:
   repository: eu.gcr.io/airqo-250220/airqo-device-registry-api
-  tag: prod-e7170088-1739227245
+  tag: prod-ae87cecf-1739238823
 nameOverride: ''
 fullnameOverride: ''
 podAnnotations: {}

k8s/device-registry/values-stage.yaml

@@ -6,7 +6,7 @@ app:
 replicaCount: 2
 image:
   repository: eu.gcr.io/airqo-250220/airqo-stage-device-registry-api
-  tag: stage-83e69174-1738960327
+  tag: stage-627aaca8-1739238787
 nameOverride: ''
 fullnameOverride: ''
 podAnnotations: {}

k8s/spatial/values-prod.yaml

@@ -6,7 +6,7 @@ app:
 replicaCount: 3
 image:
   repository: eu.gcr.io/airqo-250220/airqo-spatial-api
-  tag: prod-fbd4c147-1739233652
+  tag: prod-ae87cecf-1739238823
 nameOverride: ''
 fullnameOverride: ''
 podAnnotations: {}

k8s/workflows/values-prod.yaml

@@ -10,7 +10,7 @@ images:
     initContainer: eu.gcr.io/airqo-250220/airqo-workflows-xcom
     redisContainer: eu.gcr.io/airqo-250220/airqo-redis
     containers: eu.gcr.io/airqo-250220/airqo-workflows
-  tag: prod-fbd4c147-1739233652
+  tag: prod-ae87cecf-1739238823
 nameOverride: ''
 fullnameOverride: ''
 podAnnotations: {}

src/workflows/airqo_etl_utils/airqo_utils.py

@@ -14,6 +14,7 @@
 from .datautils import DataUtils
 from .weather_data_utils import WeatherDataUtils
 from typing import List, Dict, Any, Union
+import ast
 
 import logging
 
@@ -392,8 +393,7 @@ def merge_aggregated_weather_data(
                 "distance": station.get("distance", None),
             }
             for _, site in sites.iterrows()
-            for station in site.get("weather_stations", [])
-            if isinstance(station, dict)
+            for station in ast.literal_eval(site.get("weather_stations", []))
         ]
         sites_df = pd.DataFrame(sites_info)
 

src/workflows/airqo_etl_utils/datautils.py

@@ -78,7 +78,7 @@ def get_devices(
         return devices, keys
 
     @staticmethod
-    def get_sites(network: DeviceNetwork = None) -> pd.DataFrame:
+    def get_sites(network: Optional[DeviceNetwork] = None) -> pd.DataFrame:
         """
         Retrieve sites data.
 

src/workflows/airqo_etl_utils/meta_data_utils.py

@@ -8,6 +8,7 @@
 from .weather_data_utils import WeatherDataUtils
 from datetime import datetime, timezone
 from typing import Optional
+import ast
 
 
 class MetaDataUtils:
@@ -271,7 +272,7 @@ def update_nearest_weather_stations(
             {
                 "site_id": site.get("site_id"),
                 "network": site.get("network"),
-                "weather_stations": site.get("weather_stations"),
+                "weather_stations": ast.literal_eval(site.get("weather_stations")),
             }
             for site in updated_sites
         ]

src/workflows/airqo_etl_utils/weather_data_utils.py

@@ -11,6 +11,7 @@
 from .openweather_api import OpenWeatherApi
 from .tahmo_api import TahmoApi
 from .utils import Utils
+import ast
 
 from typing import List, Dict, Optional
 
@@ -141,7 +142,7 @@ def query_raw_data_from_tahmo(
             sites = DataUtils.get_sites()
             station_codes = []
             for _, site in sites.iterrows():
-                weather_stations = site.get("weather_stations", [])
+                weather_stations = ast.literal_eval(site.get("weather_stations", []))
                 station_codes.extend(
                     weather_station.get("code", "")
                     for weather_station in weather_stations

src/workflows/dags/airqo_measurements.py

@@ -658,24 +658,29 @@ def extract_hourly_data(**kwargs) -> pd.DataFrame:
         from airqo_etl_utils.date import date_to_str_hours
 
         # Only used the first time
-        start = kwargs.get("params", {}).get("start_date", "2021-01-01")
-        end_d = kwargs.get("params", {}).get("end_date", "2021-12-31")
-        end_d = datetime.strptime(end_d, "%Y-%m-%d")
-        end_dt = end_d.replace(hour=23, minute=59, second=59)
-        end = datetime.strftime(end_dt, "%Y-%m-%dT%H:%M:%SZ")
+        start = kwargs.get("params", {}).get("start_date", "2021-01-01T00:00:00Z")
+        start = datetime.strptime(start, "%%Y-%m-%dT%H:%M:%SZ")
+        end = kwargs.get("params", {}).get("end_date", "2021-12-31T23:59:59Z")
+        end = datetime.strptime(end, "%Y-%m-%dT%H:%M:%SZ")
 
         previous_date = kwargs["ti"].xcom_pull(key="new_date")
         if not previous_date:
             previous_date = start
 
-        hour_of_day = previous_date + timedelta(hours=1)
-
-        start_date_time = date_to_str_hours(previous_date)
+        hour_of_day = (
+            datetime.strptime(previous_date, "%%Y-%m-%dT%H:%M:%SZ")
+            if not isinstance(previous_date, datetime)
+            else previous_date
+        )
+        start_date_time = date_to_str_hours(hour_of_day)
         end_date_time = datetime.strftime(hour_of_day, "%Y-%m-%dT%H:59:59Z")
 
         if start_date_time > end or end_date_time > end:
             raise AirflowFailException(f"Run expired on {end}")
 
+        if previous_date == start:
+            kwargs["ti"].xcom_push(key="new_date", value=hour_of_day)
+
         return DataUtils.extract_data_from_bigquery(
             DataType.AVERAGED,
             start_date_time=start_date_time,