refactor: use check module

postreise/analyze/generation/binding.py

@@ -1,29 +1,4 @@
-from powersimdata.scenario.scenario import Scenario
-from powersimdata.scenario.analyze import Analyze
-
-
-def _check_scenario(scenario):
-    """Private function used only for type-checking for public functions.
-    :param powersimdata.scenario.scenario.Scenario scenario: scenario instance.
-    :raises TypeError: if scenario is not a Scenario object.
-    :raises ValueError: if scenario is not in Analyze state.
-    """
-    if not isinstance(scenario, Scenario):
-        raise TypeError("scenario must be a Scenario object")
-    if not isinstance(scenario.state, Analyze):
-        raise ValueError("scenario.state must be Analyze")
-
-
-def _check_epsilon(epsilon):
-    """Private function used only for type-checking for public functions.
-    :param float/int epsilon: precision for binding constraints.
-    :raises TypeError: if epsilon is not a float or an int.
-    :raises ValueError: if epsilon is negative.
-    """
-    if not isinstance(epsilon, (float, int)):
-        raise TypeError("epsilon must be numeric")
-    if epsilon < 0:
-        raise ValueError("epsilon must be non-negative")
+from postreise.analyze.check import _check_scenario_is_in_analyze_state, _check_epsilon
 
 
 def pmin_constraints(scenario, epsilon=1e-3):
@@ -33,7 +8,7 @@ def pmin_constraints(scenario, epsilon=1e-3):
     :param float epsilon: allowable 'fuzz' for whether constraint is binding.
     :return: (*pandas.DataFrame*) -- Boolean dataframe of same shape as PG.
     """
-    _check_scenario(scenario)
+    _check_scenario_is_in_analyze_state(scenario)
     _check_epsilon(epsilon)
 
     pg = scenario.state.get_pg()
@@ -51,7 +26,7 @@ def pmax_constraints(scenario, epsilon=1e-3):
     :param float epsilon: allowable 'fuzz' for whether constraint is binding.
     :return: (*pandas.DataFrame*) -- Boolean dataframe of same shape as PG.
     """
-    _check_scenario(scenario)
+    _check_scenario_is_in_analyze_state(scenario)
     _check_epsilon(epsilon)
 
     pg = scenario.state.get_pg()
@@ -70,7 +45,7 @@ def ramp_constraints(scenario, epsilon=1e-3):
     :param float epsilon: allowable 'fuzz' for whether constraint is binding.
     :return: (*pandas.DataFrame*) -- Boolean dataframe of same shape as PG.
     """
-    _check_scenario(scenario)
+    _check_scenario_is_in_analyze_state(scenario)
     _check_epsilon(epsilon)
 
     pg = scenario.state.get_pg()

postreise/analyze/generation/capacity_value.py

@@ -1,48 +1,9 @@
-from powersimdata.scenario.scenario import Scenario
-from powersimdata.scenario.analyze import Analyze
-
-
-def check_scenario_resources_hours(scenario, resources, hours):
-    """
-    :param powersimdata.scenario.scenario.Scenario scenario: analyzed scenario.
-    :param (str/list/tuple/set) resources: one or more resources to analyze.
-    :param int hours: number of hours to analyze.
-    :return: (*set*) -- set of valid resources.
-    :raises TypeError: if scenario is not a Scenario, resources is not one of
-        str or list/tuple/set of str's, or hours is not an int.
-    :raises ValueError: if scenario is not in Analyze state, if hours is
-        non-positive or greater than the length of the scenario, if resources
-        is empty, or not all resources are present in the grid.
-    """
-    # Check scenario
-    if not isinstance(scenario, Scenario):
-        raise TypeError("scenario must be a Scenario object")
-    if not isinstance(scenario.state, Analyze):
-        raise ValueError("scenario must be in Analyze state")
-    # Check resources
-    if isinstance(resources, str):
-        resources = {resources}
-    elif isinstance(resources, (list, set, tuple)):
-        if not all([isinstance(r, str) for r in resources]):
-            raise TypeError("all resources must be str")
-        resources = set(resources)
-    else:
-        raise TypeError("resources must be str or list/tuple/set of str")
-    if len(resources) == 0:
-        raise ValueError("resources must be nonempty")
-    valid_resources = set(scenario.state.get_grid().plant.type.unique())
-    if not resources <= valid_resources:
-        difference = resources - valid_resources
-        raise ValueError("Invalid resource(s): %s" % " | ".join(difference))
-    # Check hours
-    if not isinstance(hours, int):
-        raise TypeError("hours must be an int")
-    if hours < 1:
-        raise ValueError("hours must be positive")
-    if hours > len(scenario.state.get_demand()):
-        raise ValueError("hours must not be greater than simulation length")
-    # Finally, return the set of resources
-    return resources
+from postreise.analyze.check import (
+    _check_scenario_is_in_analyze_state,
+    _check_resources,
+    _check_resources_are_in_scenario,
+    _check_number_hours_to_analyze,
+)
 
 
 def calculate_NLDC(scenario, resources, hours=100):
@@ -51,12 +12,15 @@ def calculate_NLDC(scenario, resources, hours=100):
     net demand. NLDC = 'Net Load Duration Curve'.
 
     :param powersimdata.scenario.scenario.Scenario scenario: analyzed scenario.
-    :param (str/list/tuple/set) resources: one or more resources to analyze.
+    :param list/tuple/set resources: one or more resources to analyze.
     :param int hours: number of hours to analyze.
     :return: (*float*) -- difference between peak demand and peak net demand.
     """
-    # Check inputs
-    resources = check_scenario_resources_hours(scenario, resources, hours)
+    _check_scenario_is_in_analyze_state(scenario)
+    _check_resources(resources)
+    _check_resources_are_in_scenario(resources, scenario)
+    _check_number_hours_to_analyze(scenario, hours)
+
     # Then calculate capacity value
     total_demand = scenario.state.get_demand().sum(axis=1)
     prev_peak = total_demand.sort_values(ascending=False).head(hours).mean()
@@ -75,12 +39,15 @@ def calculate_net_load_peak(scenario, resources, hours=100):
     power generated in the top N hours of net load peak.
 
     :param powersimdata.scenario.scenario.Scenario scenario: analyzed scenario.
-    :param (str/list/tuple/set) resources: one or more resources to analyze.
+    :param list/tuple/set resources: one or more resources to analyze.
     :param int hours: number of hours to analyze.
     :return: (*float*) -- resource capacity during hours of peak net demand.
     """
-    # Check inputs
-    resources = check_scenario_resources_hours(scenario, resources, hours)
+    _check_scenario_is_in_analyze_state(scenario)
+    _check_resources(resources)
+    _check_resources_are_in_scenario(resources, scenario)
+    _check_number_hours_to_analyze(scenario, hours)
+
     # Then calculate capacity value
     total_demand = scenario.state.get_demand().sum(axis=1)
     plant_groupby = scenario.state.get_grid().plant.groupby("type")

postreise/analyze/generation/carbon.py

@@ -4,6 +4,11 @@
 
 from powersimdata.scenario.scenario import Scenario
 from powersimdata.scenario.analyze import Analyze
+from postreise.analyze.check import (
+    _check_scenario_is_in_analyze_state,
+    _check_gencost,
+    _check_time_series,
+)
 
 # For simple methods:
 # MWh to metric tons of CO2
@@ -38,18 +43,14 @@ def generate_carbon_stats(scenario, method="simple"):
     :param str method: selected method to handle no-load fuel consumption.
     :return: (*pandas.DataFrame*) -- carbon data frame.
     """
+    _check_scenario_is_in_analyze_state(scenario)
 
     allowed_methods = ("simple", "always-on", "decommit")
     if not isinstance(method, str):
         raise TypeError("method must be a str")
     if method not in allowed_methods:
         raise ValueError("Unknown method for generate_carbon_stats()")
 
-    if not isinstance(scenario, Scenario):
-        raise TypeError("scenario must be a Scenario object")
-    if not isinstance(scenario.state, Analyze):
-        raise ValueError("scenario.state must be Analyze")
-
     pg = scenario.state.get_pg()
     grid = scenario.state.get_grid()
     carbon = pd.DataFrame(np.zeros_like(pg), index=pg.index, columns=pg.columns)
@@ -138,71 +139,3 @@ def calc_costs(pg, gencost, decommit=False):
         costs = np.where(pg.to_numpy() < decommit_threshold, 0, costs)
 
     return costs
-
-
-def _check_gencost(gencost):
-    """Checks that gencost is specified properly.
-
-    :param pandas.DataFrame gencost: cost curve polynomials.
-    """
-
-    # check for nonempty dataframe
-    if not isinstance(gencost, pd.DataFrame):
-        raise TypeError("gencost must be a pandas.DataFrame")
-    if not gencost.shape[0] > 0:
-        raise ValueError("gencost must have at least one row")
-
-    # check for proper columns
-    required_columns = ("type", "n")
-    for r in required_columns:
-        if r not in gencost.columns:
-            raise ValueError("gencost must have column " + r)
-
-    # check that gencosts are all specified as type 2 (polynomial)
-    cost_type = gencost["type"]
-    if not cost_type.where(cost_type == 2).equals(cost_type):
-        raise ValueError("each gencost must be type 2 (polynomial)")
-
-    # check that all gencosts are specified as same order polynomial
-    if not (gencost["n"].nunique() == 1):
-        raise ValueError("all polynomials must be of same order")
-
-    # check that this order is an integer > 0
-    n = gencost["n"].iloc[0]
-    if not isinstance(n, (int, np.integer)):
-        print(type(n))
-        raise TypeError("polynomial degree must be specified as an int")
-    if n < 1:
-        raise ValueError("polynomial must be at least of order 1 (constant)")
-
-    # check that the right columns are here for this dataframe
-    coef_columns = ["c" + str(i) for i in range(n)]
-    for c in coef_columns:
-        if c not in gencost.columns:
-            err_msg = "gencost of order {0} must have column {1}".format(n, c)
-            raise ValueError(err_msg)
-
-
-def _check_time_series(df, label, tolerance=1e-3):
-    """Checks that a time series dataframe is specified properly.
-
-    :param pandas.DataFrame df: dataframe to check.
-    :param str label: Name of dataframe (used for error messages).
-    :param float tolerance: tolerance value for checking non-negativity.
-    :raises TypeError: if df is not a dataframe or label is not a str.
-    :raises ValueError: if df does not have at least one row and one column, or
-        if it contains values that are more negative than the tolerance allows.
-    """
-    if not isinstance(label, str):
-        raise TypeError("label must be a str")
-
-    # check for nonempty dataframe
-    if not isinstance(df, pd.DataFrame):
-        raise TypeError(label + " must be a pandas.DataFrame")
-    if not df.shape[0] > 0:
-        raise ValueError(label + " must have at least one row")
-    if not df.shape[1] > 0:
-        raise ValueError(label + " must have at least one column")
-    # check to ensure that all values are non-negative
-    if any((df < -1 * tolerance).to_numpy().ravel()):
-        raise ValueError(label + " must be non-negative")

postreise/analyze/generation/curtailment.py

@@ -1,11 +1,15 @@
 import pandas as pd
 
+from postreise.analyze.check import (
+    _check_scenario_is_in_analyze_state,
+    _check_resources_are_renewables,
+    _check_resources_are_in_scenario,
+)
 from postreise.analyze.helpers import (
     summarize_plant_to_bus,
     summarize_plant_to_location,
 )
-from powersimdata.scenario.scenario import Scenario
-from powersimdata.scenario.analyze import Analyze
+from powersimdata.network.usa_tamu.constants.plants import renewable_resources
 
 
 # What is the name of the function in scenario.state to get the profiles?
@@ -17,52 +21,12 @@
 }
 
 
-def _check_scenario(scenario):
-    """Ensure that the input is a Scenario in Analyze state.
-    :param powersimdata.scenario.scenario.Scenario scenario: scenario instance.
-    """
-    if not isinstance(scenario, Scenario):
-        raise TypeError("scenario must be a Scenario object")
-    if not isinstance(scenario.state, Analyze):
-        raise ValueError("scenario.state must be Analyze")
-
-
-def _check_resources(resources):
-    """Ensure that the input is a tuple/list/set of strs in _resource_func.
-    :param tuple/list/set resources: list of resources to analyze.
-    """
-    if not isinstance(resources, (tuple, list, set)):
-        raise TypeError("resources must be iterable (tuple, list, set)")
-    for r in resources:
-        if not isinstance(r, str):
-            raise TypeError("each resource must be a str")
-        if r not in _resource_func.keys():
-            err_msg = "resource {0} not found in list of resource functions."
-            err_msg += " Allowable: " + ", ".join(_resource_func.keys())
-            raise ValueError(err_msg)
-
-
-def _check_resource_in_scenario(resources, scenario):
-    """Ensure that each item in resources is represented in at least one
-    generator in scenario grid.
-    :param tuple/list/set resources: list of resources to analyze.
-    :param powersimdata.scenario.scenario.Scenario scenario: scenario instance.
-    :return: (*None*).
-    """
-    gentypes_in_grid = set(scenario.state.get_grid().plant["type"].unique())
-    if not set(resources) <= gentypes_in_grid:
-        err_msg = "Curtailment requested for resources not in scenario."
-        err_msg += " Requested: " + ", ".join(resources)
-        err_msg += ". Scenario: " + ", ".join(gentypes_in_grid)
-        raise ValueError(err_msg)
-
-
 def _check_curtailment_in_grid(curtailment, grid):
-    """Ensure that curtailment is a dict of dataframes, and that each key is
+    """Ensures that curtailment is a dict of dataframes, and that each key is
     represented in at least one generator in grid.
+
     :param dict curtailment: keys are resources, values are pandas.DataFrame.
-    :param powersimdata.input.grid.Grid grid: Grid instance.
-    :return: (*None*).
+    :param powersimdata.input.grid.Grid grid: a Grid object.
     """
     if not isinstance(curtailment, dict):
         raise TypeError("curtailment must be a dict")
@@ -80,18 +44,19 @@ def _check_curtailment_in_grid(curtailment, grid):
 
 
 def calculate_curtailment_time_series(scenario, resources=None):
-    """Calculate a time series of curtailment for a set of valid resources.
+    """Calculates a time series of curtailment for a set of valid resources.
+
     :param powersimdata.scenario.scenario.Scenario scenario: scenario instance.
     :param tuple/list/set resources: names of resources to analyze. Default is
         all resources which can be curtailed, defined in _resource_func.
     :return: (*dict*) -- keys are resources, values are pandas.DataFrames
     indexed by (datetime, plant) where plant is only plants of matching type.
     """
     if resources is None:
-        resources = tuple(_resource_func.keys())
-    _check_scenario(scenario)
-    _check_resources(resources)
-    _check_resource_in_scenario(resources, scenario)
+        resources = renewable_resources
+    _check_scenario_is_in_analyze_state(scenario)
+    _check_resources_are_renewables(resources)
+    _check_resources_are_in_scenario(resources, scenario)
 
     # Get input dataframes from scenario object
     pg = scenario.state.get_pg()
@@ -111,17 +76,18 @@ def calculate_curtailment_time_series(scenario, resources=None):
 
 
 def calculate_curtailment_percentage(scenario, resources=None):
-    """Calculate scenario-long average curtailment for selected resources.
+    """Calculates scenario-long average curtailment for selected resources.
+
     :param powersimdata.scenario.scenario.Scenario scenario: scenario instance.
     :param tuple/list/set resources: names of resources to analyze. Default is
         all resources which can be curtailed, defined in _resource_func.
     :return: (*float*) -- Average curtailment fraction over the scenario.
     """
     if resources is None:
-        resources = list(_resource_func.keys())
-    _check_scenario(scenario)
-    _check_resources(resources)
-    _check_resource_in_scenario(resources, scenario)
+        resources = renewable_resources
+    _check_scenario_is_in_analyze_state(scenario)
+    _check_resources_are_renewables(resources)
+    _check_resources_are_in_scenario(resources, scenario)
 
     plant = scenario.state.get_grid().plant
     curtailment = calculate_curtailment_time_series(scenario, resources)

postreise/analyze/generation/summarize.py

@@ -11,6 +11,7 @@
     loadzone2interconnect,
 )
 from powersimdata.network.usa_tamu.constants.plants import label2type
+from postreise.analyze.check import _check_scenario_is_in_analyze_state
 
 
 def sum_generation_by_type_zone(scenario: Scenario) -> pd.DataFrame:
@@ -19,10 +20,7 @@ def sum_generation_by_type_zone(scenario: Scenario) -> pd.DataFrame:
     :return: (*pandas.DataFrame*) -- total generation, indexed by {type, zone}.
     :raise Exception: if scenario is not a Scenario object in Analyze state.
     """
-    if not isinstance(scenario, Scenario):
-        raise TypeError("scenario must be a Scenario object")
-    if not isinstance(scenario.state, Analyze):
-        raise ValueError("scenario.state must be Analyze")
+    _check_scenario_is_in_analyze_state(scenario)
 
     pg = scenario.state.get_pg()
     grid = scenario.state.get_grid()