perf: use atlite streaming

config/config.default.yaml

@@ -187,7 +187,8 @@ electricity:
 # docs in https://pypsa-eur.readthedocs.io/en/latest/configuration.html#atlite
 atlite:
   default_cutout: "europe-2013-sarah3-era5"
-  nprocesses: 16
+  backend: auto
+  nprocesses: 1
   show_progress: false
   plot_availability_matrix: false
   cutouts:

config/schema.default.json

@@ -174,6 +174,16 @@
           "default": "europe-2013-sarah3-era5",
           "description": "Defines a default cutout. Can refer to a single cutout or a list of cutouts."
         },
+        "backend": {
+          "default": "auto",
+          "description": "Backend for atlite conversion. 'auto' selects automatically, 'dask' uses Dask scheduler, 'streaming' uses streaming backend.",
+          "enum": [
+            "auto",
+            "dask",
+            "streaming"
+          ],
+          "type": "string"
+        },
         "nprocesses": {
           "default": 16,
           "description": "Number of parallel processes in cutout preparation.",
@@ -8786,6 +8796,16 @@
           "default": "europe-2013-sarah3-era5",
           "description": "Defines a default cutout. Can refer to a single cutout or a list of cutouts."
         },
+        "backend": {
+          "default": "auto",
+          "description": "Backend for atlite conversion. 'auto' selects automatically, 'dask' uses Dask scheduler, 'streaming' uses streaming backend.",
+          "enum": [
+            "auto",
+            "dask",
+            "streaming"
+          ],
+          "type": "string"
+        },
         "nprocesses": {
           "default": 16,
           "description": "Number of parallel processes in cutout preparation.",

pixi.toml

@@ -42,8 +42,10 @@ sync-locks = """
     pixi workspace export conda-environment -e default envs/environment.yaml -n pypsa-eur
 """
 
+[pypi-dependencies]
+atlite = { git = "https://github.com/PyPSA/atlite.git", branch = "perf/conversion" }
+
 [dependencies]
-atlite = ">=0.3"
 bokeh = ">=3.8.0"
 cartopy = ">=0.25.0"
 copernicusmarine = ">=2.2.4"
@@ -120,7 +122,6 @@ args = ["dir", {"arg" = "output", "default" = "html"}]
 cmd = "sphinx-build -T -b {{ output }} doc {{ dir }}/{{ output }} "
 
 [feature.doc.dependencies]
-atlite = "==0.4.1"
 cartopy = "==0.25.0"
 dask = "==2025.11.0"
 descartes = "==1.1.0"

rules/collect.smk

@@ -30,31 +30,45 @@ rule process_costs:
         ),
 
 
-rule cluster_networks:
+rule create_renewable_profiles:
+    input:
+        expand(
+            resources("profile_{clusters}_{tech}.nc"),
+            tech=[
+                tech
+                for tech in config["electricity"]["renewable_carriers"]
+                if tech != "hydro"
+            ],
+            **config["scenario"],
+            run=config["run"]["name"],
+        ),
     message:
-        "Collecting clustered network files"
+        "Collection renewable profiles."
+
+
+rule cluster_networks:
     input:
         expand(
             resources("networks/base_s_{clusters}.nc"),
             **config["scenario"],
             run=config["run"]["name"],
         ),
+    message:
+        "Collecting clustered network files"
 
 
 rule prepare_elec_networks:
-    message:
-        "Collecting prepared electricity network files"
     input:
         expand(
             resources("networks/base_s_{clusters}_elec_{opts}.nc"),
             **config["scenario"],
             run=config["run"]["name"],
         ),
+    message:
+        "Collecting prepared electricity network files"
 
 
 rule prepare_sector_networks:
-    message:
-        "Collecting prepared sector-coupled network files"
     input:
         expand(
             resources(
@@ -63,41 +77,43 @@ rule prepare_sector_networks:
             **config["scenario"],
             run=config["run"]["name"],
         ),
+    message:
+        "Collecting prepared sector-coupled network files"
 
 
 rule solve_elec_networks:
-    message:
-        "Collecting solved electricity network files"
     input:
         expand(
             RESULTS + "networks/base_s_{clusters}_elec_{opts}.nc",
             **config["scenario"],
             run=config["run"]["name"],
         ),
+    message:
+        "Collecting solved electricity network files"
 
 
 rule solve_sector_networks:
-    message:
-        "Collecting solved sector-coupled network files"
     input:
         expand(
             RESULTS
             + "networks/base_s_{clusters}_{opts}_{sector_opts}_{planning_horizons}.nc",
             **config["scenario"],
             run=config["run"]["name"],
         ),
+    message:
+        "Collecting solved sector-coupled network files"
 
 
 rule solve_sector_networks_perfect:
-    message:
-        "Collecting solved sector-coupled network files with perfect foresight"
     input:
         expand(
             RESULTS
             + "maps/static/base_s_{clusters}_{opts}_{sector_opts}-costs-all_{planning_horizons}.pdf",
             **config["scenario"],
             run=config["run"]["name"],
         ),
+    message:
+        "Collecting solved sector-coupled network files with perfect foresight"
 
 
 def balance_map_paths(kind, w):
@@ -117,11 +133,11 @@ def balance_map_paths(kind, w):
 
 
 rule plot_balance_maps:
-    message:
-        "Plotting energy balance maps"
     input:
         static=lambda w: balance_map_paths("static", w),
         interactive=lambda w: balance_map_paths("interactive", w),
+    message:
+        "Plotting energy balance maps"
 
 
 rule plot_balance_maps_static:
@@ -135,11 +151,11 @@ rule plot_balance_maps_interactive:
 
 
 rule plot_power_networks_clustered:
-    message:
-        "Plotting clustered power network topology"
     input:
         expand(
             resources("maps/power-network-s-{clusters}.pdf"),
             **config["scenario"],
             run=config["run"]["name"],
         ),
+    message:
+        "Plotting clustered power network topology"

scripts/build_renewable_profiles.py

@@ -89,7 +89,6 @@
 """
 
 import logging
-import time
 from itertools import product
 
 import geopandas as gpd
@@ -120,6 +119,9 @@
     configure_logging(snakemake)
     set_scenario_config(snakemake)
 
+    logging.getLogger("atlite").setLevel(logging.WARNING)
+    logging.getLogger("pyogrio").setLevel(logging.WARNING)
+
     nprocesses = int(snakemake.threads)
     noprogress = snakemake.config["run"].get("disable_progressbar", True)
     noprogress = noprogress or not snakemake.config["atlite"]["show_progress"]
@@ -137,9 +139,6 @@
     correction_factor = params.get("correction_factor", 1.0)
     capacity_per_sqkm = params["capacity_per_sqkm"]
 
-    if correction_factor != 1.0:
-        logger.info(f"correction_factor is set as {correction_factor}")
-
     if nprocesses > 1:
         client = Client(n_workers=nprocesses, threads_per_worker=1)
     else:
@@ -173,26 +172,15 @@
     )
 
     func = getattr(cutout, resource.pop("method"))
-    if client is not None:
+    backend = snakemake.config["atlite"]["backend"]
+    if client is not None and backend == "dask":
         resource["dask_kwargs"] = {"scheduler": client}
 
-    logger.info(
-        f"Calculate average capacity factor per grid cell for technology {technology}..."
-    )
-    start = time.time()
-
-    capacity_factor = correction_factor * func(capacity_factor=True, **resource)
-
-    duration = time.time() - start
-    logger.info(
-        f"Completed average capacity factor calculation per grid cell for technology {technology} ({duration:2.2f}s)"
+    capacity_factor = correction_factor * func(
+        aggregate_time="mean", backend=backend, **resource
     )
 
     nbins = params.get("resource_classes", 1)
-    logger.info(
-        f"Create masks for {nbins} resource classes for technology {technology}..."
-    )
-    start = time.time()
 
     fn = snakemake.input.resource_regions
     resource_regions = gpd.read_file(fn).set_index("name").rename_axis("bus").geometry
@@ -239,20 +227,10 @@
         class_regions.index.names = ["bus", "bin"]
     class_regions.to_file(snakemake.output.class_regions)
 
-    duration = time.time() - start
-    logger.info(
-        f"Completed resource class calculation for technology {technology} ({duration:2.2f}s)"
-    )
-
     layout = capacity_factor * area * capacity_per_sqkm
 
     profiles = []
     for year, model in models.items():
-        logger.info(
-            f"Calculate weighted capacity factor time series for model {model} for technology {technology}..."
-        )
-        start = time.time()
-
         resource[tech] = model
 
         matrix = (availability * class_masks).stack(
@@ -265,6 +243,8 @@
             index=matrix.indexes["bus_bin"],
             per_unit=True,
             return_capacity=False,
+            aggregate_time=None,
+            backend=backend,
             **resource,
         )
         profile = profile.unstack("bus_bin")
@@ -274,17 +254,10 @@
 
         profiles.append(profile.rename("profile"))
 
-        duration = time.time() - start
-        logger.info(
-            f"Completed weighted capacity factor time series calculation for model {model} for technology {technology} ({duration:2.2f}s)"
-        )
-
     profiles = xr.merge(profiles)
 
-    logger.info(f"Calculating maximal capacity per bus for technology {technology}")
     p_nom_max = capacity_per_sqkm * availability * class_masks @ area
 
-    logger.info(f"Calculate average distances for technology {technology}.")
     layoutmatrix = (layout * availability * class_masks).stack(
         bus_bin=["bus", "bin"], spatial=["y", "x"]
     )

scripts/determine_availability_matrix.py

@@ -56,7 +56,6 @@
 
 import functools
 import logging
-import time
 
 import atlite
 import geopandas as gpd
@@ -158,17 +157,9 @@
             snakemake.input.country_shapes, buffer=buffer, invert=True
         )
 
-    logger.info(f"Calculate landuse availability for {technology}...")
-    start = time.time()
-
     kwargs = dict(nprocesses=nprocesses, disable_progressbar=noprogress)
     availability = cutout.availabilitymatrix(regions, excluder, **kwargs)
 
-    duration = time.time() - start
-    logger.info(
-        f"Completed landuse availability calculation for {technology} ({duration:2.2f}s)"
-    )
-
     if params.get("plot_availability_matrix", False):
         logger.info(f"Plotting landuse availability matrix for {technology}.")
         band, transform = shape_availability(

scripts/lib/validation/config/atlite.py

@@ -8,6 +8,8 @@
 See docs in https://pypsa-eur.readthedocs.io/en/latest/configuration.html#atlite
 """
 
+from typing import Literal
+
 from pydantic import BaseModel, Field, field_validator
 
 from scripts.lib.validation.config._base import ConfigModel
@@ -113,6 +115,10 @@ class AtliteConfig(BaseModel):
         "europe-2013-sarah3-era5",
         description="Defines a default cutout. Can refer to a single cutout or a list of cutouts.",
     )
+    backend: Literal["auto", "dask", "streaming"] = Field(
+        "auto",
+        description="Backend for atlite conversion. 'auto' selects automatically, 'dask' uses Dask scheduler, 'streaming' uses streaming backend.",
+    )
     nprocesses: int = Field(
         16,
         description="Number of parallel processes in cutout preparation.",