feat: create modules with check functions

postreise/analyze/check.py

@@ -0,0 +1,207 @@
+import numpy as np
+import pandas as pd
+
+from powersimdata.scenario.scenario import Scenario
+from powersimdata.scenario.analyze import Analyze
+from powersimdata.input.grid import Grid
+from powersimdata.network.usa_tamu.constants.plants import (
+    all_resources,
+    renewable_resources,
+)
+
+
+def _check_data_frame(df):
+    """Ensures that input parameter is a data frame.
+
+    :param pandas.DataFrame df: a data frame.
+    :raises TypeError: if input is not a data frame.
+    """
+    if not isinstance(df, pd.DataFrame):
+        raise TypeError("df must be pandas.DataFrame")
+
+
+def _check_grid(grid):
+    """Ensures that input is a Grid instance.
+
+    :param powersimdata.input.grid.Grid grid: a Grid instance.
+    :raises TypeError: if input is not Grid instance.
+    """
+    if not isinstance(grid, Grid):
+        raise TypeError("grid must be powersimdata.input.grid.Grid object")
+
+
+def _check_scenario_is_in_analyze_state(scenario):
+    """Ensures that input parameter is a Scenario object in the analyze state.
+
+    :param powersimdata.scenario.scenario.Scenario scenario: scenario instance.
+    :raises TypeError: if scenario is not a Scenario 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_resources(resources):
+    """Ensures that input parameter are valid resources.
+
+    :param list/tuple/set resources: list of resources to analyze.
+    :raises TypeError: if resources is not a list/tuple/set of str.
+    :raises ValueError: if resources is empty or not valid.
+    """
+    if isinstance(resources, (list, set, tuple)):
+        if not all([isinstance(r, str) for r in resources]):
+            raise TypeError("all resources must be str")
+    else:
+        raise TypeError("resources must be a list/tuple/set of str")
+    if len(resources) == 0:
+        raise ValueError("resources must be nonempty")
+    if not set(resources) <= set(all_resources):
+        diff = set(resources) - set(all_resources)
+        raise ValueError("Invalid resource(s): %s" % " | ".join(diff))
+
+
+def _check_resources_are_renewables(resources):
+    """Ensures that input parameter are valid renewables.
+
+    :param list/tuple/set resources: list of resources to analyze.
+    :raises ValueError: if resources iare not renewables.
+    """
+    _check_resources(resources)
+    if not set(resources) <= set(renewable_resources):
+        diff = set(resources) - set(all_resources)
+        raise ValueError("Invalid renewable resource(s): %s" % " | ".join(diff))
+
+
+def _check_resources_are_in_scenario(resources, scenario):
+    """Ensures that each item in resources is represented in at least one generator in
+    the grid used for the scenario.
+
+    :param tuple/list/set resources: list of resources to analyze.
+    :param powersimdata.scenario.scenario.Scenario scenario: scenario instance.
+    :raises ValueError: if resources is not used in scenario.
+    """
+    if isinstance(resources, str):
+        resources = {resources}
+    else:
+        resources = set(resources)
+    valid_resources = set(scenario.state.get_grid().plant["type"].unique())
+    if not resources <= valid_resources:
+        raise ValueError("Resources in scenario: %s" % " | ".join(valid_resources))
+
+
+def _check_plants_are_in_grid(plant_id, grid):
+    """Ensures that dataframe and grid are of proper type for processing.
+
+    :param list plant_id: list of plant_id.
+    :param powersimdata.input.grid.Grid grid: Grid instance.
+    :raises TypeError: if plant_id is not a list and grid is not a Grid object.
+    :raises ValueError: if plant id is not a subset of the index of the plant
+        data frame.
+    """
+    if not isinstance(plant_id, list):
+        raise TypeError("plant_id must be a list")
+    if not isinstance(grid, Grid):
+        raise TypeError("grid must be powersimdata.input.grid.Grid object")
+    if not set(plant_id) <= set(grid.plant.index):
+        raise ValueError("plant_id must be subset of plant index")
+
+
+def _check_number_hours_to_analyze(scenario, hours):
+    """Ensures that input parameter of proper type and values for analysis.
+
+    :param powersimdata.scenario.scenario.Scenario scenario: scenario instance.
+    :param int hours: number of hours to analyze.
+    :raises TypeError: if hours is not int.
+    :raises ValueError: if hours is negative or greater than simulation length
+    """
+    start_date = pd.Timestamp(scenario.info["start_date"])
+    end_date = pd.Timestamp(scenario.info["end_date"])
+    if not isinstance(hours, int):
+        raise TypeError("hours must be an int")
+    if hours < 1:
+        raise ValueError("hours must be positive")
+    if hours > (end_date - start_date).total_seconds() / 3600 + 1:
+        raise ValueError("hours must not be greater than simulation length")
+
+
+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")
+
+
+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")