fix(pv): Remove asset key prefixes

openclimatefix · Jan 20, 2025 · 0221b96 · 0221b96
1 parent 2304c2f
commit 0221b96
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 36 deletions.
diff --git a/src/dagster_dags/pv/__init__.py b/src/dagster_dags/pv/__init__.py
@@ -1,2 +1 @@
-from .passiv import *
 
diff --git a/src/dagster_dags/pv/passiv/passiv_monthly.py b/src/dagster_dags/pv/passiv/passiv_monthly.py
@@ -7,7 +7,6 @@
 
 import dagster as dg
 import pandas as pd
-import pytz
 from huggingface_hub import HfFileSystem
 from huggingface_hub.hf_api import HfApi
 
@@ -66,7 +65,7 @@ def get_monthly_passiv_data(
 
     # dont include the last end date
     generation_data = generation_data[
-        generation_data.datetime_GMT < end_date.replace(tzinfo=pytz.utc)
+        generation_data.datetime_GMT < end_date.replace(tzinfo=dt.UTC)
     ]
 
     # save to parquet file
@@ -91,40 +90,30 @@ def get_monthly_passiv_data(
 
 
 @dg.asset(
-    key=["pv", "passiv", "monthly_30min"],
+    name="passiv_monthly_30min",
     automation_condition=dg.AutomationCondition.eager(),
-    partitions_def=dg.TimeWindowPartitionsDefinition(
-        fmt="%Y-%m",
-        start="2010-01",
-        cron_schedule="0 12 1 * *",  # 1st day of the month, at 12 oclock
+    partitions_def=dg.MonthlyPartitionsDefinition(
+        start_date="2010-01-01",
+        hour_offset=12,
     ),
 )
 def pv_passiv_monthly_30min(context: dg.AssetExecutionContext) -> None:
-    """PV Passiv archive monthlyasset."""
-    partition_date_str = context.partition_key
-    start_date = dt.datetime.strptime(partition_date_str, "%Y-%m").replace(tzinfo=dt.UTC)
-    start_date = pytz.utc.localize(start_date)
-
+    """PV Passiv archive monthly asset."""
+    start_date: dt.datetime = context.partition_time_window.start
     get_monthly_passiv_data(start_date, period=30)
 
 
-
-
 @dg.asset(
-    key=["pv", "passiv", "monthly_5min"],
+    name="passiv_monthly_5min",
     automation_condition=dg.AutomationCondition.eager(),
-    partitions_def=dg.TimeWindowPartitionsDefinition(
-        fmt="%Y-%m",
-        start="2018-01",
-        cron_schedule="0 12 1 * *",  # 1st day of the month, at 12 oclock
+    partitions_def=dg.MonthlyPartitionsDefinition(
+        start_date="2018-01-01",
+        hour_offset=12,
     ),
 )
 def pv_passiv_monthly_5min(context: dg.AssetExecutionContext) -> None:
-    """PV Passiv archive monthlyasset."""
-    partition_date_str = context.partition_key
-    start_date = dt.datetime.strptime(partition_date_str, "%Y-%m").replace(tzinfo=dt.UTC)
-    start_date = pytz.utc.localize(start_date)
-
+    """PV Passiv archive monthly asset."""
+    start_date: dt.datetime = context.partition_time_window.start
     get_monthly_passiv_data(start_date, period=5)
 
 

diff --git a/src/dagster_dags/pv/passiv/passiv_year.py b/src/dagster_dags/pv/passiv/passiv_year.py
@@ -8,7 +8,6 @@
 
 import dagster as dg
 import pandas as pd
-import pytz
 from huggingface_hub import HfFileSystem
 from huggingface_hub.hf_api import HfApi
 
@@ -75,7 +74,7 @@ def get_yearly_passiv_data(
 
 
 @dg.asset(
-    key=["pv", "passiv", "yearly_5min"],
+    name="passiv_yearly_5min",
     automation_condition=dg.AutomationCondition.eager(),
     partitions_def=dg.TimeWindowPartitionsDefinition(
         fmt="%Y",
@@ -85,15 +84,12 @@ def get_yearly_passiv_data(
 )
 def pv_passiv_yearly_5min(context: dg.AssetExecutionContext) -> None:
     """PV Passiv archive yearly data."""
-    partition_date_str = context.partition_key
-    start_date = dt.datetime.strptime(partition_date_str, "%Y").replace(tzinfo=dt.UTC)
-    start_date = pytz.utc.localize(start_date)
-
+    start_date: dt.datetime = context.partition_time_window.start
     get_yearly_passiv_data(start_date, period=5)
 
 
 @dg.asset(
-    key=["pv", "passiv", "yearly_30min"],
+    name="passiv_yearly_30min",
     automation_condition=dg.AutomationCondition.eager(),
     partitions_def=dg.TimeWindowPartitionsDefinition(
         fmt="%Y",
@@ -103,9 +99,6 @@ def pv_passiv_yearly_5min(context: dg.AssetExecutionContext) -> None:
 )
 def pv_passiv_yearly_30min(context: dg.AssetExecutionContext) -> None:
     """PV Passiv archive yearly data."""
-    partition_date_str = context.partition_key
-    start_date = dt.datetime.strptime(partition_date_str, "%Y").replace(tzinfo=dt.UTC)
-    start_date = pytz.utc.localize(start_date)
-
+    start_date: dt.datetiome = context.partition_time_window.start
     get_yearly_passiv_data(start_date, period=30)