Source code for foxes.core.model

import numpy as np
from abc import ABCMeta
from itertools import count

import foxes.constants as FC
from .data import Data

[docs]class Model(metaclass=ABCMeta): """ Base class for all models. Attributes ---------- name: str The model name :group: core """ _ids = {}
[docs] def __init__(self): """ Constructor. """ t = type(self).__name__ if t not in self._ids: self._ids[t] = count(0) self._id = next(self._ids[t]) self._store = {} ext = "" if self._id == 0 else f"{self._id}" = f"{type(self).__name__}{ext}" self.__initialized = False
[docs] def __repr__(self): t = type(self).__name__ return f"{} ({t})"
[docs] def data_to_store(self, name, algo, data): """ Adds data from mdata to the local store, intended for iterative runs. Parameters ---------- name: str The data name algo: foxes.core.Algorithm The algorithm data: foxes.utils.Data The mdata, fdata or pdata object """ i0 = data.states_i0(counter=True, algo=algo) if i0 not in self._store: self._store[i0] = Data( data={}, dims={}, loop_dims=data.loop_dims, name=f"{}_{i0}" ) self._store[i0][name] = data[name] self._store[i0].dims[name] = data.dims[name] if name in data.dims else None
[docs] def from_data_or_store(self, name, algo, data, ret_dims=False, safe=False): """ Get data from mdata or local store Parameters ---------- name: str The data name algo: foxes.core.Algorithm The algorithm data: foxes.utils.Data The mdata, fdata or pdata object ret_dims: bool Return dimensions safe: bool Return None instead of error if not found Returns ------- data: numpy.ndarray The data dims: tuple of dims, optional The data dimensions """ if name in data: return (data[name], data.dims[name]) if ret_dims else data[name] i0 = data.states_i0(counter=True, algo=algo) if not safe or (i0 in self._store and name in self._store[i0]): if ret_dims: return self._store[i0][name], self._store[i0].dims[name] else: return self._store[i0][name] else: return (None, None) if ret_dims else None
[docs] def keep(self, algo): """ Add model and all sub models to the keep_models list Parameters ---------- algo: foxes.core.Algorithm The algorithm """ algo.keep_models.add(
@property def model_id(self): """ Unique id based on the model type. Returns ------- int Unique id of the model object """ return self._id
[docs] def var(self, v): """ Creates a model specific variable name. Parameters ---------- v: str The variable name Returns ------- str Model specific variable name """ return f"{}_{v}"
@property def initialized(self): """ Initialization flag. Returns ------- bool : True if the model has been initialized. """ return self.__initialized
[docs] def initialize(self, algo, verbosity=0): """ Initializes the model. This includes loading all required data from files. The model should return all array type data as part of the idata return dictionary (and not store it under self, for memory reasons). This data will then be chunked and provided as part of the mdata object during calculations. Parameters ---------- algo: foxes.core.Algorithm The calculation algorithm verbosity: int The verbosity level, 0 = silent Returns ------- idata: dict The dict has exactly two entries: `data_vars`, a dict with entries `name_str -> (dim_tuple, data_ndarray)`; and `coords`, a dict with entries `dim_name_str -> dim_array` """ if self.initialized: raise ValueError( f"Model '{}': initialize called for already initialized object" ) self.__initialized = True return {"coords": {}, "data_vars": {}}
[docs] def finalize(self, algo, verbosity=0): """ Finalizes the model. Parameters ---------- algo: foxes.core.Algorithm The calculation algorithm verbosity: int The verbosity level, 0 = silent """ if not self.initialized: raise ValueError( f"Model '{}': Finalization called for uninitialized object" ) self._store = {} self.__initialized = False
[docs] def get_data( self, variable, target, lookup="smfp", mdata=None, fdata=None, pdata=None, states_source_turbine=None, upcast=False, accept_none=False, algo=None, ): """ Getter for a data entry in the model object or provided data sources Parameters ---------- variable: str The variable, serves as data key target: str, optional The dimensions identifier for the output, e.g FC.STATE_TURBINE, FC.STATE_POINT lookup: str The order of data sources. Combination of: 's' for self, 'm' for mdata, 'f' for fdata, 'p' for pdata mdata: foxes.core.Data, optional The model data fdata: foxes.core.Data, optional The farm data pdata: foxes.core.Data, optional The evaluation point data states_source_turbine: numpy.ndarray, optional For each state, one turbine index for the wake causing turbine. Shape: (n_states,) upcast: bool, optional Upcast array to dims if data is scalar data_prio: bool First search the data source, then the object accept_none: bool Do not throw an error if data entry is None or np.nan algo: foxes.core.Algorithm, optional The algorithm, needed for data from previous iteration """ def _geta(a): sources = [s for s in [mdata, fdata, pdata, algo, self] if s is not None] for s in sources: if a == "states_i0": out = s.states_i0(counter=True, algo=algo) if out is not None: return out else: try: out = getattr(s, a) if out is not None: return out except AttributeError: pass raise KeyError( f"Model '{}': Failed to determine '{a}'. Maybe add to arguments of get_data: mdata, fdata, pdata, algo?" ) n_states = _geta("n_states") if target == FC.STATE_TURBINE: n_turbines = _geta("n_turbines") dims = (FC.STATE, FC.TURBINE) elif target == FC.STATE_POINT: n_points = _geta("n_points") dims = (FC.STATE, FC.POINT) else: raise KeyError( f"Model '{}': Wrong parameter 'target = {target}'. Choices: {FC.STATE_TURBINE}, {FC.STATE_POINT}" ) out = None for s in lookup: # lookup self: if s == "s" and hasattr(self, variable): a = getattr(self, variable) if a is not None and upcast: if target == FC.STATE_TURBINE: out = np.full((n_states, n_turbines), np.nan, dtype=FC.DTYPE) out[:] = a elif target == FC.STATE_POINT: out = np.full((n_states, n_points), np.nan, dtype=FC.DTYPE) out[:] = a else: raise KeyError( f"Model '{}': Wrong parameter 'target = {target}' for 'upcast = True' in get_data. Choose: FC.STATE_TURBINE, FC.STATE_POINT" ) else: out = a # lookup mdata: elif ( s == "m" and mdata is not None and variable in mdata and len(mdata.dims[variable]) > 1 and tuple(mdata.dims[variable][:2]) == dims ): out = mdata[variable] # lookup fdata: elif ( s == "f" and fdata is not None and variable in fdata and len(fdata.dims[variable]) > 1 and tuple(fdata.dims[variable][:2]) == (FC.STATE, FC.TURBINE) ): # direct fdata: if target == FC.STATE_TURBINE: out = fdata[variable] # translate state-turbine to state-point data: elif target == FC.STATE_POINT and states_source_turbine is not None: # from fdata, uniform for points: st_sel = (np.arange(n_states), states_source_turbine) out = np.zeros((n_states, n_points), dtype=FC.DTYPE) out[:] = fdata[variable][st_sel][:, None] # from previous iteration, if requested: if pdata is not None and FC.STATES_SEL in pdata: if not np.all( states_source_turbine == pdata[FC.STATE_SOURCE_TURBINE] ): raise ValueError( f"Model '{}': Mismatch of 'states_source_turbine'. Expected {list(pdata[FC.STATE_SOURCE_TURBINE])}, got {list(states_source_turbine)}" ) i0 = _geta("states_i0") sp = pdata[FC.STATES_SEL] sel = sp < i0 if np.any(sel): if algo is None or not hasattr(algo, "prev_farm_results"): raise KeyError( f"Model '{}': Argument algo is either not given, or not an iterative algorithm" ) prev_fdata = getattr(algo, "prev_farm_results") if prev_fdata is None: out[sel] = 0 else: st = np.zeros_like(sp) st[:] = states_source_turbine[:, None] out[sel] = prev_fdata[variable].to_numpy()[ sp[sel], st[sel] ] del st # lookup pdata: elif ( s == "p" and pdata is not None and variable in pdata and len(pdata.dims[variable]) > 1 and tuple(pdata.dims[variable][:2]) == dims ): out = pdata[variable] if out is not None: break # check for None: if not accept_none: try: if out is None or np.all(np.isnan(np.atleast_1d(out))): raise ValueError( f"Model '{}': Variable '{variable}' requested but not provided." ) except TypeError: pass return out
[docs] @classmethod def reduce_states(cls, sel_states, objs): """ Modifies the given objects by selecting a subset of states. Parameters ---------- sel_states: list of int The states selection objs: list of foxes.core.Data The objects, e.g. [mdata, fdata, pdata] Returns ------- mobjs: list of foxes.core.Data The modified objects with reduced states dimension """ out = [] for o in objs: data = { v: d[sel_states] if o.dims[v][0] == FC.STATE else d for v, d in o.items() } out.append(Data(data, o.dims, loop_dims=o.loop_dims, return out