feat: 🎨 clean models

datadex/assets/huggingface.py

@@ -25,8 +25,7 @@ def hf_asset(data: pd.DataFrame, hf: HuggingFaceResource) -> None:
     "threatened_animal_species",
     "country_year_indicators",
     "spain_ipc",
-    "spain_aemet_stations",
-    "spain_aemet_weather_data",
+    "spain_aemet_historical_weather",
 ]
 
 assets = []

datadex/assets/spain.py

@@ -74,13 +74,33 @@ def spain_ipc() -> pd.DataFrame:
 
 
 @asset(group_name="spain_open_data")
-def spain_aemet_stations(aemet_api: AEMETAPI) -> pd.DataFrame:
+def spain_aemet_stations_data(aemet_api: AEMETAPI) -> pd.DataFrame:
     """
     Spain AEMET stations data.
     """
 
     df = pd.DataFrame(aemet_api.get_all_stations())
 
+    # Clean latitud and longitud
+    df["latitude_sign"] = df["latitud"].str[-1]
+    df["longitude_sign"] = df["longitud"].str[-1]
+
+    df["latitud"] = pd.to_numeric(
+        df["latitud"].str[:-1].str.replace(",", "."), errors="coerce"
+    )
+    df["longitud"] = pd.to_numeric(
+        df["longitud"].str[:-1].str.replace(",", "."), errors="coerce"
+    )
+
+    df["latitud"] = df["latitud"] * df["latitude_sign"].apply(
+        lambda x: 1 if x == "N" else -1
+    )
+    df["longitud"] = df["longitud"] * df["longitude_sign"].apply(
+        lambda x: 1 if x == "E" else -1
+    )
+
+    df = df.drop(columns=["latitude_sign", "longitude_sign"])
+
     return df
 
 
@@ -107,6 +127,21 @@ def spain_aemet_weather_data(
 
         df = pd.concat([df, mdf], ignore_index=True)
 
-    # df["fecha"] = pd.to_datetime(df["fecha"], format="%Y-%m-%d")
+    df["fecha"] = pd.to_datetime(df["fecha"], format="%Y-%m-%d")
+
+    float_columns = [
+        "prec",
+        "presMax",
+        "presMin",
+        "racha",
+        "sol",
+        "tmax",
+        "tmed",
+        "tmin",
+        "velmedia",
+    ]
+
+    df[float_columns] = df[float_columns].apply(lambda x: x.str.replace(",", "."))
+    df[float_columns] = df[float_columns].apply(pd.to_numeric, errors="coerce")
 
     return df

dbt/models/sources.yml

@@ -15,3 +15,11 @@ sources:
         meta:
           dagster:
             asset_key: ["world_bank_wdi"]
+      - name: spain_aemet_weather_data
+        meta:
+          dagster:
+            asset_key: ["spain_aemet_weather_data"]
+      - name: spain_aemet_stations_data
+        meta:
+          dagster:
+            asset_key: ["spain_aemet_stations_data"]

dbt/models/spain_aemet_historical_weather.sql

@@ -0,0 +1,31 @@
+select
+    cast(w.fecha as date) as fecha,
+    w.indicativo,
+    w.nombre,
+    w.provincia,
+    s.latitud,
+    s.longitud,
+    w.altitud,
+    w.tmed,
+    w.prec,
+    w.tmin,
+    w.horatmin,
+    w.tmax,
+    w.horatmax,
+    w.dir,
+    w.velmedia,
+    w.racha,
+    w.horaracha,
+    w.presMax,
+    w.horaPresMax,
+    w.presMin,
+    w.horaPresMin,
+    w.hrMedia,
+    w.hrMax,
+    w.horaHrMax,
+    w.hrMin,
+    w.horaHrMin,
+    w.sol
+from {{ source('main', 'spain_aemet_weather_data') }} as w
+left join {{ source('main', 'spain_aemet_stations_data') }} as s
+    on w.indicativo = s.indicativo

dbt/models/spain_aemet_historical_weather_schema.yml

@@ -0,0 +1,9 @@
+version: 2
+
+models:
+  - name: spain_aemet_historical_weather
+    description: "Historical weather data for Spain. Cleaned and augmented with station metadata."
+    config:
+      meta:
+        dagster:
+          group: spain_open_data