Warning

This documentation is under active development. Some sections may be incomplete.

Examples#

This section provides practical examples of using the Fleet Electrification BC framework for different scenarios and use cases.

Basic Usage Examples#

Single Scenario Run#

import models.fleet_simulator.main as fleet
import models.fleet_EV_load_simulator.scripts.main as fleet_EV
from models.PyPSA_BC import run_pypsa_bc as pypsa_bc

# Define simulation timespan
timespan = {
    'start_date': "2021-01-01",
    'end_date': "2021-01-07"  # One week simulation
}

# Step 1: Generate fleet patterns
print("Running fleet pattern simulation...")
commercial_pattern_args = {
    **timespan, 
    'samples': 10000  # 10,000 vehicles
}
fleet.run_fleet_simulation(**commercial_pattern_args, show_log=False)

# Step 2: Convert to EV loads
print("Converting to EV loads...")
fleet_EV_args = {
    **timespan,
    'ev_charging': "coordinated",
    'ev_population': 0.8,  # 80% EV penetration
}
fleet_EV.get_pypsa_ev_components(**fleet_EV_args)

# Step 3: Run power system analysis
print("Running PyPSA analysis...")
pypsa_model_build_args = {
    'copperplate': True,
    'update_data': False,
    'resource_options': 'full_potential',
    'total_load_scaling_factor': 1.4,
    'run_tag': 'basic_example'
}
results = pypsa_bc.run(**fleet_EV_args, **pypsa_model_build_args)
print("Analysis complete!")

Quick Development Test#

# Reduced complexity for faster testing
timespan_test = {
    'start_date': "2021-01-01",
    'end_date': "2021-01-02"  # Single day
}

# Small sample for quick iteration
quick_args = {
    **timespan_test,
    'samples': 1000,  # Reduced samples
    'ev_population': 0.5,
    'ev_charging': "uncoordinated"
}

# Run quick test
fleet.run_fleet_simulation(**{k: v for k, v in quick_args.items() 
                             if k in ['start_date', 'end_date', 'samples']})
fleet_EV.get_pypsa_ev_components(**quick_args)

Scenario Comparison Examples#

Charging Strategy Comparison#

import itertools
import pandas as pd

# Define scenarios
charging_strategies = ["uncoordinated", "coordinated", "v2g"]
ev_penetrations = [0.5, 0.8, 1.0]

# Results storage
scenario_results = {}

# Run all combinations
for strategy, penetration in itertools.product(charging_strategies, ev_penetrations):
    scenario_name = f"{strategy}_{int(penetration*100)}pct"
    print(f"Running scenario: {scenario_name}")
    
    # Configure scenario
    scenario_args = {
        **timespan,
        'ev_charging': strategy,
        'ev_population': penetration,
        'run_tag': scenario_name
    }
    
    # Run simulation
    fleet_EV.get_pypsa_ev_components(**scenario_args)
    
    pypsa_args = {
        **pypsa_model_build_args,
        'run_tag': scenario_name
    }
    
    results = pypsa_bc.run(**scenario_args, **pypsa_args)
    scenario_results[scenario_name] = results

# Compare results
comparison_df = pd.DataFrame({
    name: {
        'total_capacity': result['total_capacity'],
        'total_cost': result['total_cost'],
        'renewable_share': result['renewable_share']
    } 
    for name, result in scenario_results.items()
}).T

print(comparison_df)

Hybrid Coordination Analysis#

# Analyze different coordination levels
coordination_levels = [10, 30, 50, 70, 90]
results_hybrid = {}

for coord_pct in coordination_levels:
    scenario_name = f"hybrid_coord_{coord_pct}"
    print(f"Running {coord_pct}% coordination scenario...")
    
    # Configure hybrid scenario
    hybrid_args = {
        **timespan,
        'ev_charging': "hybrid",
        'coordination_percent': coord_pct,
        'ev_population': 0.8,
        'run_tag': scenario_name
    }
    
    # Execute simulation
    fleet_EV.get_pypsa_ev_components(**hybrid_args)
    
    pypsa_args = {
        **pypsa_model_build_args,
        'run_tag': scenario_name
    }
    
    results_hybrid[scenario_name] = pypsa_bc.run(**hybrid_args, **pypsa_args)

# Create elbow curve data
import matplotlib.pyplot as plt

coord_levels = [10, 30, 50, 70, 90]
vre_capacity = [results_hybrid[f"hybrid_coord_{c}"]['vre_capacity'] 
                for c in coord_levels]

plt.figure(figsize=(10, 6))
plt.plot(coord_levels, vre_capacity, 'o-', linewidth=2, markersize=8)
plt.xlabel('Coordination Level (%)')
plt.ylabel('Required VRE Capacity (GW)')
plt.title('Elbow Curve: VRE Capacity vs Coordination Level')
plt.grid(True, alpha=0.3)
plt.savefig('vis/elbow_curve_example.png', dpi=300, bbox_inches='tight')
plt.show()

Advanced Configuration Examples#

Custom Load Scaling#

# Test different load scaling factors
load_factors = [1.0, 1.2, 1.4, 1.6, 1.8, 2.0]
scaling_results = {}

base_args = {
    **timespan,
    'ev_charging': "coordinated", 
    'ev_population': 0.8
}

for factor in load_factors:
    scenario_name = f"load_scale_{int(factor*100)}"
    print(f"Running load scaling factor {factor}...")
    
    pypsa_args = {
        **pypsa_model_build_args,
        'total_load_scaling_factor': factor,
        'run_tag': scenario_name
    }
    
    fleet_EV.get_pypsa_ev_components(**base_args)
    scaling_results[scenario_name] = pypsa_bc.run(**base_args, **pypsa_args)

# Analyze scaling impact
scaling_df = pd.DataFrame({
    name: {
        'load_factor': result['load_scaling_factor'],
        'peak_demand': result['peak_demand'],
        'total_generation': result['total_generation']
    }
    for name, result in scaling_results.items()
}).T

print("Load Scaling Analysis:")
print(scaling_df)

Multi-Year Analysis#

# Analyze multiple years
years = [2021, 2022, 2023]
annual_results = {}

for year in years:
    # Define year timespan
    year_timespan = {
        'start_date': f"{year}-01-01",
        'end_date': f"{year}-12-31"
    }
    
    print(f"Processing year {year}...")
    
    # Annual scenario configuration
    annual_args = {
        **year_timespan,
        'ev_charging': "coordinated",
        'ev_population': 0.6 + (year-2021)*0.1,  # Increasing penetration
        'samples': 50000,
        'run_tag': f"year_{year}"
    }
    
    # Run annual simulation
    fleet.run_fleet_simulation(**{k: v for k, v in annual_args.items() 
                                 if k in ['start_date', 'end_date', 'samples']})
    fleet_EV.get_pypsa_ev_components(**annual_args)
    
    pypsa_args = {
        **pypsa_model_build_args,
        'run_tag': f"year_{year}"
    }
    
    annual_results[year] = pypsa_bc.run(**annual_args, **pypsa_args)

# Multi-year comparison
years_df = pd.DataFrame({
    year: {
        'ev_penetration': result['ev_penetration'],
        'peak_ev_load': result['peak_ev_load'],
        'total_capacity': result['total_capacity']
    }
    for year, result in annual_results.items()
}).T

print("Multi-Year Analysis:")
print(years_df)

Visualization Examples#

Load Profile Visualization#

import matplotlib.pyplot as plt
import models.pypsa_ev_vis as vis

# Create load profile plots
def plot_load_comparison(results_dict):
    """Compare load profiles across scenarios"""
    
    fig, axes = plt.subplots(2, 2, figsize=(15, 10))
    axes = axes.flatten()
    
    for i, (name, result) in enumerate(results_dict.items()):
        if i < 4:  # Only plot first 4 scenarios
            ax = axes[i]
            
            # Extract load data
            load_profile = result['load_profile']
            time_index = result['time_index']
            
            ax.plot(time_index, load_profile, linewidth=2)
            ax.set_title(f'Load Profile: {name}')
            ax.set_xlabel('Time')
            ax.set_ylabel('Load (MW)')
            ax.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.savefig('vis/load_profile_comparison.png', dpi=300)
    plt.show()

# Use the function
plot_load_comparison(scenario_results)

Capacity Analysis#

# Capacity breakdown visualization
def plot_capacity_breakdown(results):
    """Plot generation capacity by technology"""
    
    technologies = ['Solar', 'Wind', 'Hydro', 'Gas', 'Storage']
    capacities = [results[tech.lower()+'_capacity'] for tech in technologies]
    
    plt.figure(figsize=(10, 6))
    bars = plt.bar(technologies, capacities, 
                   color=['gold', 'skyblue', 'blue', 'gray', 'green'])
    
    # Add value labels on bars
    for bar, capacity in zip(bars, capacities):
        plt.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 50,
                f'{capacity:.0f} MW', ha='center', va='bottom')
    
    plt.title('Generation Capacity by Technology')
    plt.ylabel('Capacity (MW)')
    plt.grid(True, alpha=0.3, axis='y')
    plt.savefig('vis/capacity_breakdown.png', dpi=300, bbox_inches='tight')
    plt.show()

# Example usage
plot_capacity_breakdown(scenario_results['coordinated_80pct'])

Integration with Notebooks#

Jupyter Notebook Setup#

# Cell 1: Setup and imports
%matplotlib inline
import sys
sys.path.append('../')  # Adjust path as needed

import models.fleet_simulator.main as fleet
import models.fleet_EV_load_simulator.scripts.main as fleet_EV
from models.PyPSA_BC import run_pypsa_bc as pypsa_bc

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

# Set plotting style
plt.style.use('seaborn-v0_8')
sns.set_palette("husl")
# Cell 2: Configuration
# Define scenario parameters
SCENARIOS = {
    'low_penetration': {
        'ev_population': 0.3,
        'ev_charging': 'uncoordinated'
    },
    'medium_penetration': {
        'ev_population': 0.6, 
        'ev_charging': 'coordinated'
    },
    'high_penetration': {
        'ev_population': 0.9,
        'ev_charging': 'v2g'
    }
}

# Simulation timespan
TIMESPAN = {
    'start_date': "2021-06-01",
    'end_date': "2021-06-07"
}
# Cell 3: Run simulations
results_notebook = {}

for scenario_name, config in SCENARIOS.items():
    print(f"\n{'='*50}")
    print(f"Running scenario: {scenario_name}")
    print(f"{'='*50}")
    
    # Combine configuration
    scenario_args = {**TIMESPAN, **config, 'run_tag': scenario_name}
    
    # Execute workflow
    fleet_EV.get_pypsa_ev_components(**scenario_args)
    
    pypsa_args = {
        'copperplate': True,
        'update_data': False,
        'resource_options': 'full_potential',
        'total_load_scaling_factor': 1.2,
        'run_tag': scenario_name
    }
    
    results_notebook[scenario_name] = pypsa_bc.run(**scenario_args, **pypsa_args)
    print(f"Completed: {scenario_name}")
# Cell 4: Analysis and visualization
# Create summary table
summary_data = []
for name, result in results_notebook.items():
    summary_data.append({
        'Scenario': name,
        'EV Penetration': f"{result['ev_penetration']:.0%}",
        'Peak Load (MW)': result['peak_load'],
        'Total Capacity (MW)': result['total_capacity'],
        'Renewable Share': f"{result['renewable_share']:.1%}",
        'Total Cost (M$)': result['total_cost']/1e6
    })

summary_df = pd.DataFrame(summary_data)
print("Scenario Comparison Summary:")
print(summary_df.to_string(index=False))

Error Handling Examples#

Robust Simulation Runner#

import logging
from typing import Dict, Any

# Configure logging
logging.basicConfig(level=logging.INFO,
                   format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def run_robust_simulation(scenario_config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Run simulation with comprehensive error handling
    
    Args:
        scenario_config: Dictionary containing scenario parameters
        
    Returns:
        Dictionary containing results or error information
    """
    try:
        # Validate inputs
        required_keys = ['start_date', 'end_date', 'ev_charging', 'ev_population']
        missing_keys = [key for key in required_keys if key not in scenario_config]
        
        if missing_keys:
            raise ValueError(f"Missing required parameters: {missing_keys}")
        
        # Validate date format
        from datetime import datetime
        for date_key in ['start_date', 'end_date']:
            try:
                datetime.strptime(scenario_config[date_key], '%Y-%m-%d')
            except ValueError:
                raise ValueError(f"Invalid date format for {date_key}. Use YYYY-MM-DD")
        
        # Validate EV penetration
        ev_pop = scenario_config['ev_population']
        if not 0 <= ev_pop <= 1:
            raise ValueError(f"EV population must be between 0 and 1, got {ev_pop}")
        
        logger.info(f"Starting simulation: {scenario_config.get('run_tag', 'unnamed')}")
        
        # Run fleet simulation
        fleet_args = {k: v for k, v in scenario_config.items() 
                     if k in ['start_date', 'end_date', 'samples']}
        fleet_args.setdefault('samples', 10000)
        
        fleet.run_fleet_simulation(**fleet_args, show_log=False)
        logger.info("Fleet simulation completed")
        
        # Run EV load simulation
        ev_args = {k: v for k, v in scenario_config.items() 
                  if k in ['start_date', 'end_date', 'ev_charging', 'ev_population']}
        
        fleet_EV.get_pypsa_ev_components(**ev_args)
        logger.info("EV load simulation completed")
        
        # Run PyPSA model
        pypsa_args = {
            'copperplate': scenario_config.get('copperplate', True),
            'update_data': scenario_config.get('update_data', False),
            'resource_options': scenario_config.get('resource_options', 'full_potential'),
            'total_load_scaling_factor': scenario_config.get('total_load_scaling_factor', 1.0),
            'run_tag': scenario_config.get('run_tag', 'simulation')
        }
        
        results = pypsa_bc.run(**ev_args, **pypsa_args)
        logger.info("PyPSA simulation completed successfully")
        
        return {
            'status': 'success',
            'results': results,
            'config': scenario_config
        }
        
    except Exception as e:
        error_msg = f"Simulation failed: {str(e)}"
        logger.error(error_msg)
        
        return {
            'status': 'error',
            'error': str(e),
            'error_type': type(e).__name__,
            'config': scenario_config
        }

# Example usage with error handling
test_scenarios = [
    {
        'start_date': '2021-01-01',
        'end_date': '2021-01-03',
        'ev_charging': 'coordinated',
        'ev_population': 0.8,
        'run_tag': 'test_1'
    },
    {
        'start_date': '2021-01-01',
        'end_date': '2021-01-03', 
        'ev_charging': 'invalid_strategy',  # This will cause an error
        'ev_population': 0.8,
        'run_tag': 'test_2'
    }
]

# Run scenarios with error handling
for scenario in test_scenarios:
    result = run_robust_simulation(scenario)
    
    if result['status'] == 'success':
        print(f"✅ {scenario['run_tag']}: Success")
    else:
        print(f"❌ {scenario['run_tag']}: {result['error']}")

This comprehensive examples section provides practical guidance for users to implement various scenarios and use cases with the Fleet Electrification BC framework.