From 7de68d00502467732c55c677f5c83200a7362f3b Mon Sep 17 00:00:00 2001 From: jbadsdata Date: Sun, 9 Feb 2025 19:19:28 -0500 Subject: [PATCH] example type hints --- watttime/optimizer/alg/moer.py | 161 +++++++-------- watttime/optimizer/alg/optCharger.py | 288 +++------------------------ 2 files changed, 101 insertions(+), 348 deletions(-) diff --git a/watttime/optimizer/alg/moer.py b/watttime/optimizer/alg/moer.py index 4ec3c8e3..399394c3 100644 --- a/watttime/optimizer/alg/moer.py +++ b/watttime/optimizer/alg/moer.py @@ -1,128 +1,113 @@ -# moer.py - +from typing import List import numpy as np +from numpy.typing import ArrayLike, NDArray class Moer: - """ - Represents Marginal Operating Emissions Rate (MOER) for electricity grid emissions modeling. + """Handles Marginal Operating Emissions Rate (MOER) calculations for electricity grid emissions. - This class handles calculations related to emissions and utilities based on - MOER data, supporting both diagonal and non-diagonal penalty matrices. + A class for processing and analyzing emissions data based on MOER measurements, + supporting various calculations including time-specific emissions and interval summations. - Attributes: - ----------- - __mu : numpy.ndarray + Parameters + ---------- + mu : ArrayLike + Emissions rate data for each time step. + + Attributes + ---------- + __mu : NDArray[np.float64] Mean emissions rate for each time step. __T : int Total number of time steps. - Methods: + Examples -------- - __len__() - Returns the number of time steps. - get_emission_at(i, usage) - Calculates emission at a specific time step. - get_emission_interval(start, end, usage) - Calculates sum of emissions for a time interval. - get_emissions(x) - Calculates emissions per interval for a given schedule. - get_total_emission(x) - Calculates total emission for a given schedule. - + >>> moer = Moer([0.5, 0.6, 0.7]) + >>> moer.get_emission_at(0, 2.0) + 1.0 + >>> moer.get_total_emission([1.0, 2.0, 1.5]) + 2.75 """ - def __init__(self, mu): - """ - Initializes the Moer object. - - Parameters: - ----------- - mu : array-like - Emissions rate for each time step. - """ - self.__mu = np.array(mu).flatten() - self.__T = self.__mu.shape[0] + def __init__(self, mu: ArrayLike) -> None: + self.__mu: NDArray[np.float64] = np.array(mu).flatten() + self.__T: int = self.__mu.shape[0] - def __len__(self): - """ - Returns the length of the time series. + def __len__(self) -> int: + """Return the number of time steps in the series. - Returns: - -------- + Returns + ------- int - The number of time steps in the series. + Length of the time series. """ return self.__T - def get_emission_at(self, i, usage): - """ - Calculates the emission at a specific time step. + def get_emission_at(self, i: int, usage: float) -> float: + """Calculate emission at a specific time step. - Parameters: - ----------- + Parameters + ---------- i : int - The time step index. - usage : float, optional - The power usage. + Time step index. + usage : float + Power usage multiplier. - Returns: - -------- + Returns + ------- float - The calculated emission value. + Calculated emission value. """ return self.__mu[i] * usage - def get_emission_interval(self, start, end, usage): - """ - Calculates emissions for a given time interval. + def get_emission_interval(self, start: int, end: int, usage: float) -> float: + """Calculate total emissions for a time interval. - Parameters: - ----------- + Parameters + ---------- start : int - The start index of the interval. + Start index of interval. end : int - The end index of the interval. - usage : float, optional - The emission multiplier. Default is 1. - - Returns: - -------- - numpy.ndarray - An array of emission values for the specified interval. + End index of interval. + usage : float + Emission multiplier. + + Returns + ------- + float + Sum of emissions for the interval. """ return np.dot(self.__mu[start:end], usage) - def get_emissions(self, usage): - """ - Calculates emissions for a given set of emission multipliers. + def get_emissions(self, usage: ArrayLike) -> NDArray[np.float64]: + """Calculate emissions for each time step given usage values. - Parameters: - ----------- - usage : array-like - The emission multipliers. + Parameters + ---------- + usage : ArrayLike + Array of emission multipliers. - Returns: - -------- - numpy.ndarray - An array of calculated emission values. + Returns + ------- + NDArray[np.float64] + Array of calculated emissions. """ - usage = np.array(usage).flatten() - return self.__mu[: usage.shape[0]] * usage + usage_array: NDArray[np.float64] = np.array(usage).flatten() + return self.__mu[: usage_array.shape[0]] * usage_array - def get_total_emission(self, usage): - """ - Calculates the total emission for a given set of emission multipliers. + def get_total_emission(self, usage: ArrayLike) -> float: + """Calculate total emissions across all time steps. - Parameters: - ----------- - usage : array-like - The emission multipliers. + Parameters + ---------- + usage : ArrayLike + Array of emission multipliers. - Returns: - -------- + Returns + ------- float - The total calculated emission. + Total emission value. """ - usage = np.array(usage).flatten() - return np.dot(self.__mu[: usage.shape[0]], usage) + usage_array: NDArray[np.float64] = np.array(usage).flatten() + return np.dot(self.__mu[: usage_array.shape[0]], usage_array) \ No newline at end of file diff --git a/watttime/optimizer/alg/optCharger.py b/watttime/optimizer/alg/optCharger.py index 7ba52057..517f6991 100644 --- a/watttime/optimizer/alg/optCharger.py +++ b/watttime/optimizer/alg/optCharger.py @@ -1,10 +1,11 @@ # optCharger.py -import warnings + import numpy as np +from typing import Dict, List, Optional, Tuple, Callable from .moer import Moer -TOL = 1e-4 # tolerance -EMISSION_FN_TOL = 1e-9 # emissions functions tolerance in kw +TOL: float = 1e-4 # tolerance +EMISSION_FN_TOL: float = 1e-9 # emissions functions tolerance in kw class OptCharger: @@ -20,15 +21,15 @@ class OptCharger: Initializes the OptCharger object with the given parameters. """ - def __init__(self, verbose): + def __init__(self, verbose: bool) -> None: """ Initializes the OptCharger object. """ - self.__optimal_charging_emission = None - self.__optimal_charging_schedule = None - self.__verbose = verbose + self.__optimal_charging_emission: Optional[float] = None + self.__optimal_charging_schedule: Optional[List[int]] = None + self.__verbose: bool = verbose - def __collect_results(self, moer: Moer): + def __collect_results(self, moer: Moer) -> None: """ Translates the optimal charging schedule into a series of emission multiplier values and calculates various emission-related metrics. @@ -55,8 +56,8 @@ def __collect_results(self, moer: Moer): The function also populates the emission_multipliers list, which is used in the calculations. """ - emission_multipliers = [] - current_charge_time_units = 0 + emission_multipliers: List[float] = [] + current_charge_time_units: int = 0 for i in range(len(self.__optimal_charging_schedule)): if self.__optimal_charging_schedule[i] == 0: emission_multipliers.append(0.0) @@ -78,12 +79,12 @@ def __collect_results(self, moer: Moer): self.__optimal_charging_emissions_over_time.sum() ) - def verbose_on(self, statement:str): + def verbose_on(self, statement: str) -> None: if self.__verbose: print(statement) @staticmethod - def __sanitize_emission_multiplier(emission_multiplier_fn, total_charge): + def __sanitize_emission_multiplier(emission_multiplier_fn: Callable[[float, float], float], total_charge: float) -> Callable[[float, float], float]: """ Sanitizes the emission multiplier function to handle edge cases and ensure valid outputs. @@ -124,7 +125,7 @@ def __sanitize_emission_multiplier(emission_multiplier_fn, total_charge): ) @staticmethod - def __check_constraint(t_start, c_start, dc, constraints): + def __check_constraint(t_start: int, c_start: int, dc: int, constraints: Dict[int, Tuple[int, int]]) -> bool: # assuming constraints[t] is the bound on total charge after t intervals for t in range(t_start + 1, t_start + dc): if (t in constraints) and ( @@ -134,7 +135,7 @@ def __check_constraint(t_start, c_start, dc, constraints): return False return True - def __greedy_fit(self, total_charge: int, total_time: int, moer: Moer): + def __greedy_fit(self, total_charge: int, total_time: int, moer: Moer) -> None: """ Performs a "greedy" fit for charging schedule optimization. @@ -159,7 +160,7 @@ def __greedy_fit(self, total_charge: int, total_time: int, moer: Moer): self.__optimal_charging_schedule = schedule self.__collect_results(moer) - def __simple_fit(self, total_charge: int, total_time: int, moer: Moer): + def __simple_fit(self, total_charge: int, total_time: int, moer: Moer) -> None: """ Performs a "simple" fit for charging schedule optimization. @@ -195,9 +196,9 @@ def __diagonal_fit( total_charge: int, total_time: int, moer: Moer, - emission_multiplier_fn, - constraints: dict = {}, - ): + emission_multiplier_fn: Callable[[float, float], float], + constraints: Dict[int, Tuple[Optional[int], Optional[int]]] = {}, + ) -> None: """ Performs a sophisticated diagonal fit for charging schedule optimization using dynamic programming. @@ -283,10 +284,10 @@ def __contiguous_fit( total_charge: int, total_time: int, moer: Moer, - emission_multiplier_fn, - charge_per_interval: list = [], - constraints: dict = {}, - ): + emission_multiplier_fn: Callable[[float, float], float], + charge_per_interval: List[int] = [], + constraints: Dict[int, Tuple[Optional[int], Optional[int]]] = {}, + ) -> None: """ Performs a contiguous fit for charging schedule optimization using dynamic programming. @@ -405,11 +406,11 @@ def __variable_contiguous_fit( total_charge: int, total_time: int, moer: Moer, - emission_multiplier_fn, - charge_per_interval: list = [], + emission_multiplier_fn: Callable[[float, float], float], + charge_per_interval: List[Tuple[int, int]] = [], use_all_intervals: bool = True, - constraints: dict = {}, - ): + constraints: Dict[int, Tuple[Optional[int], Optional[int]]] = {}, + ) -> None: """ Performs a contiguous fit for charging schedule optimization using dynamic programming. @@ -491,237 +492,4 @@ def __variable_contiguous_fit( ) and OptCharger.__check_constraint( t - dc, c - dc, dc, constraints ): - marginal_cost = moer.get_emission_interval( - t - dc, t, charge_array_cache[c - dc : c] - ) - new_util = ( - max_util[t - dc, c - dc, k - 1] - marginal_cost - ) - if init_val or (new_util > max_util[t, c, k]): - max_util[t, c, k] = new_util - path_history[t - 1, c, k, :] = [dc, 1] - init_val = False - optimal_interval, optimal_util = ( - total_interval, - max_util[total_time, total_charge, total_interval], - ) - if not use_all_intervals: - for k in range(0, total_interval): - if np.isnan(max_util[total_time, total_charge, optimal_interval]) or ( - not np.isnan(max_util[total_time, total_charge, k]) - and max_util[total_time, total_charge, k] - > max_util[total_time, total_charge, optimal_interval] - ): - optimal_interval = k - if np.isnan(max_util[total_time, total_charge, optimal_interval]): - raise Exception( - "Solution not found! Please check that constraints are satisfiable." - ) - curr_state, t_curr = [total_charge, optimal_interval], total_time - - schedule_reversed = [] - interval_ids_reversed = [] - while t_curr > 0: - dc, delta_interval = path_history[ - t_curr - 1, curr_state[0], curr_state[1], : - ] - if delta_interval == 0: - ## did not charge - schedule_reversed.append(0) - interval_ids_reversed.append(-1) - t_curr -= 1 - else: - ## charge - t_curr -= dc - curr_state = [curr_state[0] - dc, curr_state[1] - delta_interval] - if dc > 0: - schedule_reversed.extend([1] * dc) - interval_ids_reversed.extend([curr_state[1]] * dc) - optimal_path = np.array(schedule_reversed)[::-1] - self.__optimal_charging_schedule = list(optimal_path) - self.__interval_ids = list(interval_ids_reversed[::-1]) - self.__collect_results(moer) - - def fit( - self, - total_charge: int, - total_time: int, - moer: Moer, - charge_per_interval=None, - use_all_intervals: bool = True, - constraints: dict = {}, - emission_multiplier_fn=None, - optimization_method: str = "auto", - ): - """ - Fits an optimal charging schedule based on the given parameters and constraints. - - This method serves as the main entry point for the charging optimization process. - It selects the appropriate optimization method based on the input parameters and - constraints. - - Parameters: - ----------- - total_charge : int - The total amount of charge needed. - total_time : int - The total time available for charging. - moer : Moer - An object representing Marginal Operating Emissions Rate. - charge_per_interval : list of int or (int,int), optional - The minimium and maximum (inclusive) charging amount per interval. If int instead of tuple, interpret as both min and max. - use_all_intervals : bool - If true, use all intervals provided by charge_per_interval; if false, can use the first few intervals and skip the rest. This can only be false if charge_per_interval is provided as a range. - constraints : dict, optional - A dictionary of charging constraints for specific time steps. - emission_multiplier_fn : callable, optional - A function that calculates emission multipliers. If None, assumes constant 1kW power usage. - optimization_method : str, optional - The optimization method to use. Can be 'auto', 'baseline', 'simple', or 'sophisticated'. - Default is 'auto'. - - Raises: - ------- - Exception - If the charging task is impossible given the constraints, or if an unsupported - optimization method is specified. - - Note: - ----- - This method chooses between different optimization strategies based on the input - parameters and the characteristics of the problem. - """ - assert len(moer) >= total_time - assert optimization_method in ["baseline", "simple", "sophisticated", "auto"] - - if emission_multiplier_fn is None: - warnings.warn( - "Warning: No emission_multiplier_fn given. Assuming that device uses constant 1kW of power" - ) - emission_multiplier_fn = lambda sc, ec: 1.0 - constant_emission_multiplier = True - else: - constant_emission_multiplier = ( - np.std( - [ - emission_multiplier_fn(sc, sc + 1) - for sc in list(range(total_charge)) - ] - ) - < EMISSION_FN_TOL - ) - self.emission_multiplier_fn = emission_multiplier_fn - - if total_charge > total_time: - raise Exception( - f"Solution not found! Impossible to charge {total_charge} within {total_time} intervals." - ) - if optimization_method == "baseline": - self.__greedy_fit(total_charge, total_time, moer) - elif ( - not constraints - and not charge_per_interval - and constant_emission_multiplier - and optimization_method == "auto" - ) or (optimization_method == "simple"): - if not constant_emission_multiplier: - warnings.warn( - "Warning: Emissions function is non-constant. Using the simple algorithm is suboptimal." - ) - self.__simple_fit(total_charge, total_time, moer) - elif not charge_per_interval: - self.__diagonal_fit( - total_charge, - total_time, - moer, - OptCharger.__sanitize_emission_multiplier( - emission_multiplier_fn, total_charge - ), - constraints, - ) - else: - # cpi stands for charge per interval - single_cpi, tuple_cpi, use_fixed_alg = [], [], True - - def convert_input(c): - ## Converts the interval format - if isinstance(c, int): - return c, (c, c), True - if c[0] == c[1]: - return c[0], c, True - return None, c, False - - for c in charge_per_interval: - if use_fixed_alg: - sc, tc, use_fixed_alg = convert_input(c) - single_cpi.append(sc) - tuple_cpi.append(tc) - else: - tuple_cpi.append(convert_input(c)[1]) - if use_fixed_alg: - assert ( - use_all_intervals - ), "Must use all intervals when interval lengths are fixed!" - self.__contiguous_fit( - total_charge, - total_time, - moer, - OptCharger.__sanitize_emission_multiplier( - emission_multiplier_fn, total_charge - ), - single_cpi, - constraints, - ) - else: - self.__variable_contiguous_fit( - total_charge, - total_time, - moer, - OptCharger.__sanitize_emission_multiplier( - emission_multiplier_fn, total_charge - ), - tuple_cpi, - use_all_intervals, - constraints, - ) - - def get_energy_usage_over_time(self) -> list: - """ - Returns list of the energy due to charging at each interval in MWh. - """ - return self.__optimal_charging_energy_over_time - - def get_charging_emissions_over_time(self) -> list: - """ - Returns list of the emissions due to charging at each interval in lbs. - """ - return self.__optimal_charging_emissions_over_time - - def get_total_emission(self) -> float: - """ - Returns the summed emissions due to charging in lbs. - """ - return self.__optimal_charging_emission - - def get_schedule(self) -> list: - """ - Returns list of the optimal charging schedule of units to charge for each interval. - """ - return self.__optimal_charging_schedule - - def get_interval_ids(self) -> list: - """ - Returns list of the interval ids for each interval. Has a value of -1 for non-charging intervals. - Intervals are labeled starting from 0 to n-1 when there are n intervals - - Only defined when charge_per_interval variable is given to some fit function - """ - return self.__interval_ids - - def summary(self): - print("-- Model Summary --") - print( - "Expected charging emissions: %.2f lbs" % self.__optimal_charging_emission - ) - print("Optimal charging schedule:", self.__optimal_charging_schedule) - print("=" * 15) + marginal_cost = moer.get_emission_interval \ No newline at end of file