# coding: utf-8
import numpy as np
import pandas as pd
import warnings
from scipy.interpolate import interp1d
import scipy.integrate as integrate
from scipy.stats import spearmanr
from scipy.integrate import IntegrationWarning
import os
import contextlib
from evpv.chargingsimulator import ChargingSimulator
from evpv.pvsimulator import PVSimulator
# Suppress repeated IntegrationWarning
warnings.filterwarnings("ignore", category=IntegrationWarning)
[docs]
class EVPVSynergies:
"""
A class to analyze energy synergies between electric vehicle (EV) charging demand and photovoltaic (PV) production.
The main metrics calculated include energy coverage, self-sufficiency, self-consumption, and excess PV ratios,
as well as the Spearman correlation between EV and PV profiles over specific days.
This class requires:
- A PVSimulator object, which holds the capacity factor time series data for PV production.
- A ChargingSimulator object, which stores EV charging demand profiles.
- The installed PV capacity (in megawatts, MW) as a float value.
"""
[docs]
def __init__(self, pv: PVSimulator, ev: ChargingSimulator, pv_capacity_MW: float):
"""
Initialize the EVPVSynergies object.
Args:
pv: Object containing PV production calculations.
ev: Object containing EV charging demand calculations.
pv_capacity_MW: PV capacity in megawatts (MW).
"""
print("=========================================")
print(f"INFO \t Creation of a EVPVSynergies object.")
print("=========================================")
self.pv_capacity_MW = pv_capacity_MW
self.pv_capacity_factor = pv
self.ev_calculator = ev # Store the ChargingSimulator object as an instance variable (usefull for recomputing if needed)
self.ev_charging_demand_MW = ev # Store only the interpolate charging demand
print(f"INFO \t Successful initialization of input parameters.")
@property
def ev_charging_demand_MW(self) -> interp1d:
"""interp1d: Interpolation function for EV charging demand."""
return self._ev_charging_demand_MW
@ev_charging_demand_MW.setter
def ev_charging_demand_MW(self, ev_charging_demand_MW: ChargingSimulator):
if isinstance(ev_charging_demand_MW, ChargingSimulator):
demand = ev_charging_demand_MW.temporal_demand_profile_aggregated[['time', 'total']]
else:
raise ValueError("Invalid ev object. Must be ChargingSimulator.")
# Extract the 'Time' and 'Total profile (MW)' columns
time = demand['time']
profile = demand['total'] / 1000.0
self._ev_charging_demand_MW = interp1d(time, profile, kind='linear', fill_value='extrapolate')
@property
def pv_capacity_factor(self) -> dict:
""" dict: Dictionary of interpolation functions for PV capacity factors by day."""
return self._pv_capacity_factor
@pv_capacity_factor.setter
def pv_capacity_factor(self, pv: PVSimulator):
"""pv_capacity_factor (pd.DataFrame): DataFrame containing PV capacity factors."""
df = pv.results['Capacity Factor'].reset_index()
# Rename the columns for convenience (optional, but helpful)
df.columns = ['Timestamp', 'Capacity Factor']
# Convert the first column to datetime format
df['Timestamp'] = pd.to_datetime(df['Timestamp'])
# Extract the 'Month-Day' and 'Hour' from the timestamp
df['Month-Day'] = df['Timestamp'].dt.strftime('%m-%d')
df['Hour'] = df['Timestamp'].dt.hour
# Create a dictionary to hold the interpolation functions for each day
interpolation_functions = {}
# Group data by 'Day'
grouped = df.groupby('Month-Day')
# Create an interpolation function for each day
for day, group in grouped:
hours = group['Hour']
profile = group['Capacity Factor']
# Create the interpolation function for this day
interpolation_function = interp1d(hours, profile, kind='linear', fill_value='extrapolate')
# Store the function in the dictionary with the day as the key
interpolation_functions[day] = interpolation_function
self._pv_capacity_factor = interpolation_functions
@property
def pv_capacity_MW(self) -> float:
""" float: PV capacity in megawatts (MW)."""
return self._pv_capacity_MW
@pv_capacity_MW.setter
def pv_capacity_MW(self, pv_capacity_MW: float):
self._pv_capacity_MW = pv_capacity_MW
# PV Production
[docs]
def pv_power_MW(self, day: str = '01-01') -> callable:
"""Return the PV power in megawatts for a given day as a function of time.
Args:
day (str): The day in 'MM-DD' format to calculate PV power for. Defaults to '01-01'.
Returns:
callable: A lambda function that calculates PV power (MW) at any given time.
"""
return lambda x: self.pv_capacity_factor[day](x) * self.pv_capacity_MW
[docs]
def pv_production(self, day: str = '01-01') -> float:
"""Calculate the total PV production for a given day by integrating over 24 hours.
Args:
day (str): The day in 'MM-DD' format to calculate PV production for. Defaults to '01-01'.
Returns:
float: Total PV production (MWh) for the specified day.
"""
result, error = integrate.quad(self.pv_power_MW(day), 0, 24)
return result
# EV Charging demand
[docs]
def ev_demand(self) -> float:
"""Calculate the total EV charging demand by integrating over 24 hours.
Returns:
float: Total EV charging demand (MWh) for the day.
"""
result, error = integrate.quad(self.ev_charging_demand_MW, 0, 24)
return result
# EV-PV Synergies
[docs]
def energy_coverage_ratio(self, day: str = '01-01') -> float:
"""Calculate the ratio of PV production to EV charging demand for a given day.
Args:
day (str): The day in 'MM-DD' format to calculate the energy coverage ratio for. Defaults to '01-01'.
Returns:
float: Energy coverage ratio for the specified day.
"""
return self.pv_production(day) / self.ev_demand()
[docs]
def self_sufficiency_ratio(self, day: str = '01-01', coincident_power: float = None) -> float:
"""Calculate the self-sufficiency ratio for a given day.
The self-sufficiency ratio is the ratio of coincident power (minimum of PV and EV demand)
to EV demand.
Args:
day (str): The day in 'MM-DD' format to calculate the self-sufficiency ratio for. Defaults to '01-01'.
coincident_power (float): The coincident power if already calculated. Default is None.
Returns:
float: Self-sufficiency ratio for the specified day.
"""
if coincident_power is None:
coincident_power = lambda x: min(self.pv_power_MW(day)(x), self.ev_charging_demand_MW(x))
result, error = integrate.quad(coincident_power, 0, 24)
coincident_power = result
return coincident_power / self.ev_demand()
[docs]
def self_consumption_ratio(self, day: str = '01-01', coincident_power: float = None) -> float:
"""Calculate the self-consumption ratio for a given day.
The self-consumption ratio is the ratio of coincident power to total PV production.
Args:
day (str): The day in 'MM-DD' format to calculate the self-consumption ratio for. Defaults to '01-01'.
coincident_power (float): The coincident power if already calculated. Default is None.
Returns:
float: Self-consumption ratio for the specified day.
"""
if coincident_power is None:
coincident_power = lambda x: min(self.pv_power_MW(day)(x), self.ev_charging_demand_MW(x))
result, error = integrate.quad(coincident_power, 0, 24)
coincident_power = result
return coincident_power / self.pv_production(day)
[docs]
def excess_pv_ratio(self, day: str = '01-01', coincident_power: float = None) -> float:
"""Calculate the excess PV ratio for a given day.
The excess PV ratio is the fraction of PV production that exceeds the EV demand.
Args:
day (str): The day in 'MM-DD' format to calculate the excess PV ratio for. Defaults to '01-01'.
coincident_power (float): The coincident power if already calculated. Default is None.
Returns:
float: Excess PV ratio for the specified day.
"""
if coincident_power is None:
coincident_power = lambda x: min(self.pv_power_MW(day)(x), self.ev_charging_demand_MW(x))
result, error = integrate.quad(coincident_power, 0, 24)
coincident_power = result
pv_prod = self.pv_production(day)
return (pv_prod - coincident_power) / pv_prod
[docs]
def spearman_correlation(self, day: str = '01-01', n_points: int = 100) -> tuple:
"""Calculate the Spearman correlation between PV production and EV charging demand.
Args:
day (str): The day in 'MM-DD' format to calculate the Spearman correlation for. Defaults to '01-01'.
n_points (int): The number of points to sample across the 24-hour period. Defaults to 100.
Returns:
tuple: Spearman correlation coefficient and p-value.
"""
# Define the range and resolution
t_values = np.linspace(0, 24, n_points)
pv_values = self.pv_power_MW(day)(t_values)
ev_values = self.ev_charging_demand_MW(t_values)
# Compute the Spearman rank correlation coefficient
spearman_coef, p_value = spearmanr(pv_values, ev_values)
return spearman_coef, p_value
[docs]
def daily_metrics(self, start_date: str, end_date: str, n_points: int = 100, recompute_probability: float = 0.0) -> pd.DataFrame:
"""Compute all energy and synergy metrics over a given period.
Args:
start_date (str): Start date in 'MM-DD' format.
end_date (str): End date in 'MM-DD' format.
n_points (int): The number of points to sample for each day. Defaults to 100.
recompute_probability (float): Probability (between 0 and 1) of recomputing EV demand for each day.
Returns:
pd.DataFrame: DataFrame containing all metrics for each day within the specified range.
"""
print(f"INFO \t Computing all metrics over a given period. This might take some time...")
# Convert start and end dates from MM-DD to YYYY-MM-DD format
start_date = f'1901-{start_date}'
end_date = f'1901-{end_date}'
# Generate a list of dates from start to end date in MM-DD format
date_range = pd.date_range(start=start_date, end=end_date)
filtered_days = [date.strftime('%m-%d') for date in date_range if date.strftime('%m-%d') in self.pv_capacity_factor]
# Initialize lists to hold results
results = []
for day in filtered_days:
print(f"\t > Day: {day}", end='\r')
# Determine if we should recompute EV demand based on probability
if np.random.rand() < recompute_probability:
print("Recomputing the charging profile to add randomness...", end='') # Optional: You may keep this line to indicate recomputing without verbose output
# Suppress output during recomputing
with open(os.devnull, 'w') as f, contextlib.redirect_stdout(f):
self.ev_calculator.compute_temporal_demand(0.1) # Recalculate demand profile in ev_calculator
self.ev_charging_demand_MW = self.ev_calculator # Update the interpolation function
# Calculate metrics
spearman_coef, p_value = self.spearman_correlation(day, n_points)
pv_prod = self.pv_production(day)
ev_dmd = self.ev_demand()
energy_cov_ratio = self.energy_coverage_ratio(day)
# Precomputed coincident power
coincident_power = lambda x: min(self.pv_power_MW(day)(x), self.ev_charging_demand_MW(x))
result, error = integrate.quad(coincident_power, 0, 24)
self_suf_ratio = self.self_sufficiency_ratio(day, result)
self_cons_ratio = self.self_consumption_ratio(day, result)
excess_pv_rat = self.excess_pv_ratio(day, result)
results.append({
'Day': f'1901-{day}',
'PV Production (MWh)': pv_prod,
'EV Demand (MWh)': ev_dmd,
'Spearman Coefficient': spearman_coef,
'P-Value': p_value,
'Energy Coverage Ratio': energy_cov_ratio,
'Self Sufficiency Ratio': self_suf_ratio,
'Self Consumption Ratio': self_cons_ratio,
'Excess PV Ratio': excess_pv_rat
})
print("")
# Create a DataFrame from the results
df = pd.DataFrame(results)
return df
[docs]
def calculate_fully_solar_charging_vehicles(self, day: str, recompute_probability: float = 0.0) -> int:
"""
Calculate the number of vehicles that can be fully charged using only solar power on a given day.
Args:
day (str): The day for which to perform the calculation, in 'MM-DD' format.
recompute_probability (float): Probability (between 0 and 1) of recomputing EV demand to introduce randomness. Defaults to 0.0.
Returns:
int: The number of vehicles fully charged using solar power.
"""
# Determine if we should recompute EV demand based on probability
if np.random.rand() < recompute_probability:
print("Recomputing the charging profile to add randomness...", end='') # Optional
# Suppress output during recomputing
with open(os.devnull, 'w') as f, contextlib.redirect_stdout(f):
self.ev_calculator.compute_temporal_demand(0.1) # Recompute demand profile
self.ev_charging_demand_MW = self.ev_calculator # Update the interpolation function
# Initialize counter for vehicles fully charged using solar energy
fully_charged_vehicles = 0
# Determine the number of time steps in the demand profile
num_rows = len(self.ev_calculator.temporal_demand_profile)
# Calculate the time step duration (in hours)
time_step = 24 / num_rows # Assuming evenly spaced intervals over 24 hours
# Define the time range for PV availability (6 AM to 9 PM)
pv_start_time = int(6 / time_step) # Convert 6 AM to index
pv_end_time = int(21 / time_step) # Convert 9 PM to index
# Compute the solar power profile for the day in kW
pv_power_profile = [
self.pv_power_MW(day)(t * time_step) * 1000 for t in range(num_rows)
] # Convert MW to kW
# Initialize a copy of the PV energy profile for tracking remaining power
remaining_pv = pv_power_profile[:]
# Iterate over each vehicle in the temporal demand profile
for vehicle in self.ev_calculator.temporal_demand_profile.columns:
# Retrieve the charging demand profile for the vehicle
charging_demand = self.ev_calculator.temporal_demand_profile[vehicle]
# Identify the time indices where charging demand is non-zero
non_zero_indices = charging_demand[charging_demand > 0].index
if non_zero_indices.empty:
continue # Skip vehicles with no charging demand
# Get the start and end times of charging demand
start_time = non_zero_indices[0]
end_time = non_zero_indices[-1]
# Skip vehicles charging outside the PV time range
if start_time < pv_start_time or end_time >= pv_end_time:
continue
# Calculate the total energy needed by the vehicle during the PV time range
energy_need = sum(charging_demand[t] * time_step for t in range(start_time, end_time + 1))
# Calculate the total available PV energy in the same time range
pv_energy_available = sum(remaining_pv[t] * time_step for t in range(start_time, end_time + 1))
# Check if the vehicle can be fully charged using the available PV energy
if pv_energy_available >= energy_need:
# Initialize a flag to check if full charging is possible
fully_charged = True
# Deduct the used PV power from the remaining PV profile
for t in range(start_time, end_time + 1):
required_power = charging_demand[t]
allocated_power = min(remaining_pv[t], required_power)
# Check if allocated power matches the required power
if allocated_power < required_power:
fully_charged = False # Vehicle cannot be fully charged
break
# Increment the counter only if the vehicle was fully charged
if fully_charged:
fully_charged_vehicles += 1
remaining_pv[t] -= allocated_power
return fully_charged_vehicles