feat(ecmwf): Whole india archive (#104)

* feat(ecmwf): Whole india archive

* fix(tests): india

local_archives/nwp/ecmwf/__init__.py

@@ -1,6 +1,6 @@
 import dagster as dg
 
-from . import ecmwf_malta, ecmwf_nw_india, ecmwf_uk
+from . import ecmwf_malta, ecmwf_nw_india, ecmwf_uk, ecmwf_india
 
 uk_assets = dg.load_assets_from_modules(
     modules=[ecmwf_uk],
@@ -18,8 +18,14 @@
     group_name="ecmwf_malta",
 )
 
+india_assets = dg.load_assets_from_modules(
+    modules=[ecmwf_india],
+    group_name="ecmwf_india",
+)
+
 all_assets: list[dg.AssetsDefinition] = [
     *uk_assets,
     *nw_india_assets,
     *malta_assets,
+    *india_assets,
 ]

local_archives/nwp/ecmwf/ecmwf_india.py

@@ -0,0 +1,34 @@
+"""ECMWF India data pipeline."""
+import dagster as dg
+from nwp_consumer.internal import FetcherInterface
+from nwp_consumer.internal.inputs.ecmwf import mars
+
+from local_archives.nwp._generic_definitions_factory import (
+    MakeDefinitionsOptions,
+    MakeDefinitionsOutputs,
+    make_definitions,
+)
+
+fetcher: FetcherInterface = mars.MARSClient(
+    area="india",
+    hours=55,
+)
+
+partitions: dg.TimeWindowPartitionsDefinition = dg.TimeWindowPartitionsDefinition(
+    start="2020-01-01T00:00",
+    cron_schedule="0 0,12 * * *",  # 00:00 and 12:00
+    fmt="%Y-%m-%dT%H:%M",
+    end_offset=-(4 * 2),  # ECMWF only available 4 days back (2 partitions per day)
+)
+
+defs: MakeDefinitionsOutputs = make_definitions(
+    opts=MakeDefinitionsOptions(
+        area="india",
+        source="ecmwf",
+        partitions=partitions,
+        fetcher=fetcher,
+    ),
+)
+
+ecmwf_india_raw_archive = defs.raw_asset
+ecmwf_india_zarr_archive = defs.zarr_asset

tests/compile_test.py

@@ -14,4 +14,4 @@ def test_nwp_asset_key_prefixes() -> None:
         # The second element should be the provider
         assert asset.key.path[1] in ["ecmwf", "metoffice", "eumetsat", "cams", "ceda", "meteomatics"]
         # The third element should be the region
-        assert asset.key.path[2] in ["uk", "eu", "global", "nw_india", "malta"]
+        assert asset.key.path[2] in ["uk", "eu", "global", "nw_india", "malta", "india"]