Source code for foxes_opt.problems.opt_farm_vars

import numpy as np
import pandas as pd

from foxes_opt.core import FarmVarsProblem
from foxes.models.turbine_models import SetFarmVars
from foxes.config import config


[docs] class OptFarmVars(FarmVarsProblem): """ Optimize a selection of farm variables. :group: opt.problems """
[docs] def __init__(self, *args, **kwargs): """ Constructor. Parameters ---------- args: tuple, optional Arguments for `FarmVarsProblem` kwargs: dict, optional Keyword arguments for `FarmVarsProblem` """ super().__init__(*args, **kwargs) self._vars = None
[docs] def add_var( self, variable, typ, init, min, max, level="uniform", sel=None, pre_rotor=False, model_key=None, ): """ Add a variable. Parameters ---------- variable: str The foxes farm variable name typ: type The variable type, either float or int init: float or int The initial value min: float or int The min value max: float or int The max value level: str Choices: uniform, state, turbine, state-turbine sel: numpy.ndarray, optional States/turbines/state-turbine selection, depending on the level pre_rotor: bool Apply this variable before rotor model model_key: str, optional Creates sub-model which can then be placed in the turbine model list. Repeated keys are added to the same turbine model """ if typ == "float": typ = float elif float == "int": typ = int elif typ is not float and typ is not int: raise TypeError( f"Problem '{self.name}': Expecting float or int, got type '{type(typ).__name__}'" ) mname = self.name if model_key is None else model_key if mname in self.algo.mbook.turbine_models: m = self.algo.mbook.turbine_models[mname] if not isinstance(m, SetFarmVars): raise KeyError( f"Problem '{self.name}': Turbine model entry '{mname}' already exists in model book, and is not of type SetFarmVars" ) elif m.pre_rotor != pre_rotor: raise ValueError( f"Problem '{self.name}': Turbine model entry '{mname}' exists in model book, and disagrees on pre_rotor = {pre_rotor}" ) else: self.algo.mbook.turbine_models[mname] = SetFarmVars(pre_rotor=pre_rotor) if self._vars is None: i0 = 0 i0i = 0 i0f = 0 else: if variable in self._vars["var"].tolist(): raise ValueError( f"Problem '{self.name}': Attempt to add variable '{variable}' twice" ) i0 = len(self._vars.index) grps = self._vars.groupby("type") i0i = len(grps.get_group("int").index) if "int" in grps.groups.keys() else 0 i0f = ( len(grps.get_group("float").index) if "float" in grps.groups.keys() else 0 ) del grps if level == "uniform": hdata = pd.DataFrame(index=[i0]) hdata.loc[i0, "name"] = variable hdata.loc[i0, "var"] = variable hdata.loc[i0, "type"] = "int" if typ is int else "float" hdata.loc[i0, "index"] = i0i if typ is int else i0f hdata.loc[i0, "level"] = level hdata.loc[i0, "state"] = -1 hdata.loc[i0, "turbine"] = -1 hdata.loc[i0, "sel_turbine"] = -1 hdata.loc[i0, "init"] = np.array([init], dtype=config.dtype_double) hdata.loc[i0, "min"] = np.array([min], dtype=config.dtype_double) hdata.loc[i0, "max"] = np.array([max], dtype=config.dtype_double) hdata.loc[i0, "pre_rotor"] = pre_rotor hdata.loc[i0, "model_key"] = mname elif level == "state": if not self.algo.initialized: self.algo.initialize() states = np.arange(self.algo.n_states) if sel is not None: states = states[sel] inds = i0 + np.arange(len(states)) tinds = inds - i0 + i0i if typ is int else inds - i0 + i0f hdata = pd.DataFrame(index=inds) hdata.loc[inds, "name"] = [ f"{variable}_{i:05d}" for i in range(len(states)) ] hdata.loc[inds, "var"] = variable hdata.loc[inds, "type"] = "int" if typ is int else "float" hdata.loc[inds, "index"] = tinds hdata.loc[inds, "level"] = level hdata.loc[inds, "state"] = states hdata.loc[inds, "turbine"] = -1 hdata.loc[inds, "sel_turbine"] = -1 for c, d in [("init", init), ("min", min), ("max", max)]: data = np.full(len(inds), np.nan, dtype=config.dtype_double) data[:] = d hdata.loc[inds, c] = data hdata.loc[inds, "pre_rotor"] = pre_rotor hdata.loc[inds, "model_key"] = mname elif level == "turbine": if sel is None: turbines = self.sel_turbines else: turbines = np.arange(self.algo.n_turbines)[sel] inds = i0 + np.arange(len(turbines)) tinds = inds - i0 + i0i if typ is int else inds - i0 + i0f hdata = pd.DataFrame(index=inds) hdata.loc[inds, "name"] = [ f"{variable}_{i:04d}" for i in range(len(turbines)) ] hdata.loc[inds, "var"] = variable hdata.loc[inds, "type"] = "int" if typ is int else "float" hdata.loc[inds, "index"] = tinds hdata.loc[inds, "level"] = level hdata.loc[inds, "state"] = -1 hdata.loc[inds, "turbine"] = turbines hdata.loc[inds, "sel_turbine"] = [ self.sel_turbines.index(ti) for ti in turbines ] for c, d in [("init", init), ("min", min), ("max", max)]: data = np.full(len(inds), np.nan, dtype=config.dtype_double) data[:] = d hdata.loc[inds, c] = data hdata.loc[inds, "pre_rotor"] = pre_rotor hdata.loc[inds, "model_key"] = mname elif level == "state-turbine": if not self.algo.initialized: self.algo.initialize() n_states = self.algo.n_states n_turbines = self.algo.n_turbines if sel is None: sel = np.zeros((n_states, n_turbines), dtype=bool) sel[:, self.sel_turbines] = True else: sel = np.array(sel, dtype=bool) st = np.arange(n_states * n_turbines).reshape(n_states, n_turbines)[sel] whr = np.where(sel) n_inds = len(st) inds = i0 + np.arange(n_inds) tinds = inds - i0 + i0i if typ is int else inds - i0 + i0f hdata = pd.DataFrame(index=inds) hdata.loc[inds, "name"] = [ f"{variable}_{whr[0][i]:05d}_{whr[1][i]:04d}" for i in range(len(st)) ] hdata.loc[inds, "var"] = variable hdata.loc[inds, "type"] = "int" if typ is int else "float" hdata.loc[inds, "index"] = tinds hdata.loc[inds, "level"] = level hdata.loc[inds, "state"] = whr[0] hdata.loc[inds, "turbine"] = whr[1] hdata.loc[inds, "sel_turbine"] = [ self.sel_turbines.index(ti) for ti in whr[1] ] for c, d in [("init", init), ("min", min), ("max", max)]: data = np.full(n_inds, np.nan, dtype=config.dtype_double) if isinstance(d, np.ndarray) and len(d.shape) > 1: data[:] = d[sel] else: data[:] = d hdata.loc[inds, c] = data hdata.loc[inds, "pre_rotor"] = pre_rotor hdata.loc[inds, "model_key"] = mname else: raise ValueError( f"Problem '{self.name}': Unknown level '{level}'. Choices: uniform, state, turbine, state-turbine" ) if self._vars is None: self._vars = hdata else: self._vars = pd.concat([self._vars, hdata], axis=0) icols = ["index", "state", "turbine", "sel_turbine"] for c in icols: self._vars[c] = self._vars[c].astype(config.dtype_int)
[docs] def initialize(self, verbosity=1, **kwargs): """ Initialize the object. Parameters ---------- verbosity: int The verbosity level, 0 = silent kwargs: dict, optional Additional parameters for super class init """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) if verbosity > 0: print(f"Problem '{self.name}': Optimization variable list") print() print(self._vars) print() prev = {} postv = {} for (mname, pre), g in self._vars.groupby(["model_key", "pre_rotor"]): if (pre and mname in postv) or (not pre and mname in prev): raise ValueError( f"Problem '{self.name}': Model '{mname}' reveived both pre_rotor and non-pre_rotor variables" ) tg = prev if pre else postv if mname not in tg: tg[mname] = set(g["var"].tolist()) else: tg[mname] = tg[mname].update(g["var"].tolist()) super().initialize( pre_rotor_vars={mname: list(vrs) for mname, vrs in prev.items()}, post_rotor_vars={mname: list(vrs) for mname, vrs in postv.items()}, verbosity=verbosity, **kwargs, )
[docs] def var_names_int(self): """ The names of int variables. Returns ------- names: list of str The names of the int variables """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) grps = self._vars.groupby("type") if "int" not in grps.groups.keys(): return [] else: return grps.get_group("int")["name"].tolist()
[docs] def initial_values_int(self): """ The initial values of the int variables. Returns ------- values: numpy.ndarray Initial int values, shape: (n_vars_int,) """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) grps = self._vars.groupby("type") if "int" not in grps.groups.keys(): return [] else: return grps.get_group("int")["init"].to_numpy(config.dtype_int)
[docs] def min_values_int(self): """ The minimal values of the integer variables. Use -self.INT_INF for unbounded. Returns ------- values: numpy.ndarray Minimal int values, shape: (n_vars_int,) """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) grps = self._vars.groupby("type") if "int" not in grps.groups.keys(): return [] else: return grps.get_group("int")["min"].to_numpy(config.dtype_int)
[docs] def max_values_int(self): """ The maximal values of the integer variables. Use self.INT_INF for unbounded. Returns ------- values: numpy.ndarray Maximal int values, shape: (n_vars_int,) """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) grps = self._vars.groupby("type") if "int" not in grps.groups.keys(): return [] else: return grps.get_group("int")["max"].to_numpy(config.dtype_int)
[docs] def var_names_float(self): """ The names of float variables. Returns ------- names: list of str The names of the float variables """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) grps = self._vars.groupby("type") if "float" not in grps.groups.keys(): return [] else: return grps.get_group("float")["name"].tolist()
[docs] def initial_values_float(self): """ The initial values of the float variables. Returns ------- values: numpy.ndarray Initial float values, shape: (n_vars_float,) """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) grps = self._vars.groupby("type") if "float" not in grps.groups.keys(): return [] else: return grps.get_group("float")["init"].to_numpy(config.dtype_double)
[docs] def min_values_float(self): """ The minimal values of the float variables. Use -numpy.inf for unbounded. Returns ------- values: numpy.ndarray Minimal float values, shape: (n_vars_float,) """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) grps = self._vars.groupby("type") if "float" not in grps.groups.keys(): return [] else: return grps.get_group("float")["min"].to_numpy(config.dtype_double)
[docs] def max_values_float(self): """ The maximal values of the float variables. Use numpy.inf for unbounded. Returns ------- values: numpy.ndarray Maximal float values, shape: (n_vars_float,) """ if self._vars is None: raise ValueError( f"Problem '{self.name}': No variables added for optimization." ) grps = self._vars.groupby("type") if "float" not in grps.groups.keys(): return [] else: return grps.get_group("float")["max"].to_numpy(config.dtype_double)
[docs] def opt2farm_vars_individual(self, vars_int, vars_float): """ Translates optimization variables to farm variables Parameters ---------- vars_int: numpy.ndarray The integer optimization variable values, shape: (n_vars_int,) vars_float: numpy.ndarray The float optimization variable values, shape: (n_vars_float,) Returns ------- farm_vars: dict The foxes farm variables. Key: var name, value: numpy.ndarray with values, shape: (n_states, n_sel_turbines) """ n_states = self.algo.n_states n_sturb = self.n_sel_turbines farm_vars = {} grps = self._vars.groupby(["type", "var", "level"]) for (typ, var, level), g in grps: src = vars_int if typ == "int" else vars_float i0 = g.index[0] i1 = g.index[-1] data = src[np.s_[i0 : i1 + 1]] if level == "uniform": farm_vars[var] = np.full( (n_states, n_sturb), data[0], dtype=config.dtype_double ) elif level == "state": farm_vars[var] = np.full( (n_states, n_sturb), np.nan, dtype=config.dtype_double ) if np.all(g["state"] == np.arange(n_states)): farm_vars[var][:] = data[:, None] else: farm_vars[var][g["state"]] = data[:, None] elif level == "turbine": farm_vars[var] = np.full( (n_states, n_sturb), np.nan, dtype=config.dtype_double ) if np.all(g["sel_turbine"] == np.arange(n_sturb)): farm_vars[var][:] = data[None, :] else: farm_vars[var][:, g["sel_turbine"]] = data[None, :] elif level == "state-turbine": farm_vars[var] = np.full( (n_states, n_sturb), np.nan, dtype=config.dtype_double ) farm_vars[var][g["state"], g["sel_turbine"]] = data else: raise ValueError( f"Problem '{self.name}': Unknown level '{level}' encountered for variable '{var}'. Valid choices: uniform, state, turbine, state-turbine" ) return farm_vars
[docs] def opt2farm_vars_population(self, vars_int, vars_float, n_states): """ Translates optimization variables to farm variables Parameters ---------- vars_int: numpy.ndarray The integer optimization variable values, shape: (n_pop, n_vars_int) vars_float: numpy.ndarray The float optimization variable values, shape: (n_pop, n_vars_float) n_states: int The number of original (non-pop) states Returns ------- farm_vars: dict The foxes farm variables. Key: var name, value: numpy.ndarray with values, shape: (n_pop, n_states, n_sel_turbines) """ n_pop = vars_float.shape[0] n_sturb = self.n_sel_turbines farm_vars = {} grps = self._vars.groupby(["type", "var", "level"]) for (typ, var, level), g in grps: src = vars_int if typ == "int" else vars_float i0 = g.index[0] i1 = g.index[-1] data = src[:, np.s_[i0 : i1 + 1]] if level == "uniform": farm_vars[var] = np.full( (n_pop, n_states, n_sturb), np.nan, dtype=config.dtype_double ) farm_vars[var][:] = data[:, 0, None, None] elif level == "state": farm_vars[var] = np.full( (n_pop, n_states, n_sturb), np.nan, dtype=config.dtype_double ) if np.all(g["state"] == np.arange(n_states)): farm_vars[var][:] = data[:, :, None] else: farm_vars[var][:, g["state"]] = data[:, :, None] elif level == "turbine": farm_vars[var] = np.full( (n_pop, n_states, n_sturb), np.nan, dtype=config.dtype_double ) if np.all(g["sel_turbine"] == np.arange(n_sturb)): farm_vars[var][:] = data[:, None, :] else: farm_vars[var][:, :, g["sel_turbine"]] = data[:, None, :] elif level == "state-turbine": farm_vars[var] = np.full( (n_pop, n_states, n_sturb), np.nan, dtype=config.dtype_double ) farm_vars[var][:, g["state"], g["sel_turbine"]] = data else: raise ValueError( f"Problem '{self.name}': Unknown level '{level}' encountered for variable '{var}'. Valid choices: uniform, state, turbine, state-turbine" ) return farm_vars