[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

pypsa_eur_cli/pypsa_eur/cli_helpers.py

@@ -1,5 +1,4 @@
 import hashlib
-import json
 import time
 from datetime import datetime
 from pathlib import Path
@@ -13,27 +12,27 @@ def generate_job_id(configfile: Path | None = None, name: str | None = None) ->
     if name:
         # If name is provided via CLI, use it directly
         return name
-    
+
     if configfile and configfile.exists():
         # Load the configuration file
         with open(configfile) as f:
             config_data = yaml.safe_load(f)
-        
+
         # Extract the original run name from the config
         original_run_name = config_data.get("run", {}).get("name", "pypsa_eur")
     else:
         original_run_name = "pypsa_eur"
-    
+
     # Generate timestamp-based unique ID
     timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-    
+
     # Create a short hash for uniqueness
     hash_input = f"{original_run_name}{timestamp}{time.time()}"
     short_hash = hashlib.sha256(hash_input.encode()).hexdigest()[:6]
-    
+
     # Combine everything
     final_job_id = f"{timestamp}_{original_run_name}_{short_hash}"
-    
+
     return final_job_id
 
 
@@ -76,4 +75,4 @@ def recursive_merge(d1: dict, d2: dict) -> dict:
             recursive_merge(d1[key], value)  # Recursively merge dictionaries
         else:
             d1[key] = value  # Overwrite or add the value
-    return d1
+    return d1

pypsa_eur_cli/pypsa_eur/main.py

@@ -1,20 +1,18 @@
-import os
 import subprocess
 import tempfile
 from pathlib import Path
-from typing import Optional
 
 import typer
 import yaml
+from tqdm import tqdm
+
 from pypsa_eur.cli_helpers import (
     check_in_project_root,
     comma_separated_list,
     generate_job_id,
     recursive_merge,
 )
 from pypsa_eur.progress_tracker import ProgressTracker
-from snakemake.utils import validate
-from tqdm import tqdm
 
 app = typer.Typer(
     help="PyPSA-Eur CLI: Command Line Interface for European Power System Analysis",
@@ -24,65 +22,65 @@
 
 @app.command(name="run")
 def run(
-        name: str = typer.Option(
-            None,
-            "--name",
-            "-n",
-            help="Name for this run (will be used as job ID).",
-        ),
-        configfile: Path = typer.Option(
-            None,
-            "--configfile",
-            "-c",
-            help="Path to configuration file (default: config/config.default.yaml).",
-        ),
-        verbose: bool = typer.Option(
-            False,
-            "--verbose",
-            "-v",
-            help="Show live Snakemake output during workflow execution.",
-        ),
-        cores: int = typer.Option(
-            None,
-            "--cores",
-            help="Number of cores to use for Snakemake. If not specified, all available cores will be used.",
-        ),
-        snapshots: str = typer.Option(
-            None,
-            "--snapshots",
-            "-s",
-            help="Override snapshots configuration (format: 'start,end' e.g., '2013-01-01,2014-01-01').",
-        ),
-        planning_horizons: str = typer.Option(
-            None,
-            "--planning-horizons",
-            help="Override planning horizons as comma-separated list (e.g., 2030,2040,2050).",
-        ),
-        clusters: str = typer.Option(
-            None,
-            "--clusters",
-            help="Override network clusters as comma-separated list (e.g., 39,128,256).",
-        ),
-        opts: str = typer.Option(
-            None,
-            "--opts",
-            help="Override opts configuration as comma-separated list.",
-        ),
-        sector_opts: str = typer.Option(
-            None,
-            "--sector-opts",
-            help="Override sector_opts configuration as comma-separated list.",
-        ),
-        foresight: str = typer.Option(
-            None,
-            "--foresight",
-            help="Override foresight configuration (overnight, myopic, or perfect).",
-        ),
-        countries: str = typer.Option(
-            None,
-            "--countries",
-            help="Override countries configuration as comma-separated list (e.g., DE,FR,ES).",
-        ),
+    name: str = typer.Option(
+        None,
+        "--name",
+        "-n",
+        help="Name for this run (will be used as job ID).",
+    ),
+    configfile: Path = typer.Option(
+        None,
+        "--configfile",
+        "-c",
+        help="Path to configuration file (default: config/config.default.yaml).",
+    ),
+    verbose: bool = typer.Option(
+        False,
+        "--verbose",
+        "-v",
+        help="Show live Snakemake output during workflow execution.",
+    ),
+    cores: int = typer.Option(
+        None,
+        "--cores",
+        help="Number of cores to use for Snakemake. If not specified, all available cores will be used.",
+    ),
+    snapshots: str = typer.Option(
+        None,
+        "--snapshots",
+        "-s",
+        help="Override snapshots configuration (format: 'start,end' e.g., '2013-01-01,2014-01-01').",
+    ),
+    planning_horizons: str = typer.Option(
+        None,
+        "--planning-horizons",
+        help="Override planning horizons as comma-separated list (e.g., 2030,2040,2050).",
+    ),
+    clusters: str = typer.Option(
+        None,
+        "--clusters",
+        help="Override network clusters as comma-separated list (e.g., 39,128,256).",
+    ),
+    opts: str = typer.Option(
+        None,
+        "--opts",
+        help="Override opts configuration as comma-separated list.",
+    ),
+    sector_opts: str = typer.Option(
+        None,
+        "--sector-opts",
+        help="Override sector_opts configuration as comma-separated list.",
+    ),
+    foresight: str = typer.Option(
+        None,
+        "--foresight",
+        help="Override foresight configuration (overnight, myopic, or perfect).",
+    ),
+    countries: str = typer.Option(
+        None,
+        "--countries",
+        help="Override countries configuration as comma-separated list (e.g., DE,FR,ES).",
+    ),
 ):
     """
     Run the PyPSA-Eur workflow with specified configuration.
@@ -106,11 +104,9 @@ def run(
         "retrieve_ship_raster": "Retrieving shipping raster data",
         "retrieve_databundle": "Retrieving data bundle",
         "retrieve_egrigcs3_pm_production": "Retrieving EGRIGCS3 power production data",
-
         # Shape and geography building
         "build_shapes": "Building geographical shapes",
         "build_ship_raster": "Building shipping density raster",
-
         # Network building and preparation
         "base_network": "Building base network structure",
         "build_osm_network": "Building network from OpenStreetMap data",
@@ -120,15 +116,13 @@ def run(
         "cluster_network": "Clustering network nodes",
         "add_transmission_projects_and_dlr": "Adding transmission projects and dynamic line rating",
         "build_transmission_projects": "Building transmission projects",
-
         # Electricity system components
         "build_electricity_demand": "Building electricity demand time series",
         "build_electricity_demand_base": "Building base electricity demand",
         "build_powerplants": "Building power plant database",
         "build_line_rating": "Building dynamic line ratings",
         "add_electricity": "Adding electrical parameters",
         "prepare_network": "Preparing network for optimization",
-
         # Renewable energy
         "build_renewable_profiles": "Building renewable generation profiles",
         "determine_availability_matrix": "Determining renewable availability matrix",
@@ -137,13 +131,11 @@ def run(
         "build_temperature_profiles": "Building temperature profiles",
         "build_cop_profiles": "Building heat pump COP profiles",
         "build_solar_thermal_profiles": "Building solar thermal profiles",
-
         # Solving and optimization
         "solve_network": "Solving network optimization",
         "solve_operations_network": "Solving operational dispatch",
         "solve_elec_networks": "Solving electricity network",
         "solve_sector_network": "Solving sector-coupled network",
-
         # Sector coupling
         "prepare_sector_network": "Preparing sector-coupled network",
         "build_industrial_demand": "Building industrial demand",
@@ -153,13 +145,11 @@ def run(
         "build_industry_sector": "Building industry sector model",
         "build_heating_sector": "Building heating sector model",
         "build_transport_sector": "Building transport sector model",
-
         # Post-processing and validation
         "make_summary": "Creating results summary",
         "plot_network": "Plotting network",
         "plot_summary": "Plotting summary statistics",
         "validate_network": "Validating network consistency",
-
         # Other rules
         "copy_config": "Copying configuration",
         "build_cutout": "Building weather cutout",
@@ -221,7 +211,9 @@ def run(
             config_data.setdefault("costs", {})["year"] = ph_list[0]
 
     if clusters:
-        config_data.setdefault("scenario", {})["clusters"] = comma_separated_list(clusters)
+        config_data.setdefault("scenario", {})["clusters"] = comma_separated_list(
+            clusters
+        )
 
     if opts:
         config_data.setdefault("scenario", {})["opts"] = comma_separated_list(opts)
@@ -444,4 +436,4 @@ def callback():
 
 
 if __name__ == "__main__":
-    app()
+    app()

pypsa_eur_cli/pypsa_eur/progress_tracker.py

@@ -1,6 +1,5 @@
 import re
 import sys
-from pathlib import Path
 
 from colorama import Fore, Style
 from tqdm import tqdm
@@ -117,7 +116,12 @@ def parse_line(self, line):
         if "rule " in line and ":" in line:
             parts = line.split()
             for i, part in enumerate(parts):
-                if part in ["rule", "localrule", "checkpoint", "localcheckpoint"] and i + 1 < len(parts):
+                if part in [
+                    "rule",
+                    "localrule",
+                    "checkpoint",
+                    "localcheckpoint",
+                ] and i + 1 < len(parts):
                     rule_name = parts[i + 1].strip(":")
                     statement_identifier = f"{rule_name}_detected"
                     if statement_identifier not in self.printed_statements:
@@ -150,7 +154,9 @@ def parse_line(self, line):
 
                 if statement_identifier not in self.printed_statements:
                     # Print wildcard details as additional info
-                    tqdm.write(f"  {Fore.YELLOW}└─ with {wildcards_str}{Style.RESET_ALL}")
+                    tqdm.write(
+                        f"  {Fore.YELLOW}└─ with {wildcards_str}{Style.RESET_ALL}"
+                    )
                     self.printed_statements.add(statement_identifier)
 
             return True
@@ -171,4 +177,4 @@ def clean_up(self):
 
     def __repr__(self):
         """Return a string representation of the progress tracker state."""
-        return f"ProgressTracker()"
+        return "ProgressTracker()"