Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from scipy.integrate import quad
import pandas
class FunctionParameter(Parameter):
def __init__(self, model, parent, func, *args, **kwargs):
super(FunctionParameter, self).__init__(model, *args, **kwargs)
self._parent = parent
self._func = func
def value(self, ts, scenario_index):
return self._func(self._parent, ts, scenario_index)
FunctionParameter.register()
class ScaledProfileParameter(Parameter):
def __init__(self, model, scale, profile, *args, **kwargs):
super(ScaledProfileParameter, self).__init__(model, *args, **kwargs)
self.scale = scale
profile.parents.add(self)
self.profile = profile
@classmethod
def load(cls, model, data):
scale = float(data.pop("scale"))
profile = load_parameter(model, data.pop("profile"))
return cls(model, scale, profile, **data)
def value(self, ts, si):
p = self.profile.get_value(si)
return self.scale * p
def __new__(cls, *args, **kwargs):
if cls is License:
raise TypeError('License cannot be instantiated directly')
else:
return BaseParameter.__new__(cls)
def end_date():
def fget(self):
if self._latest_date is not None:
return self._latest_date
else:
return self.model.timestepper.end
def fset(self, value):
if isinstance(value, pandas.Timestamp):
self._latest_date = value
else:
self._latest_date = pandas.to_datetime(value)
return locals()
end_date = property(**end_date())
class ScenarioTreeDecisionParameter(Parameter):
def __init__(self, model, root, parameter_factory, **kwargs):
super(ScenarioTreeDecisionParameter, self).__init__(model, **kwargs)
self.root = root
self.parameter_factory = parameter_factory
self.path_index = None
self._cached_paths = None
self.parameters = None
# Setup the parameters associated with the tree
self._create_scenario_parameters()
def _create_scenario_parameters(self):
parameters = {}
def make_parameter(scenario):
p = self.parameter_factory(self.model, scenario)
parameters[scenario] = p
def __init__(self, model, decision_date, before_parameter, after_parameter,
earliest_date=None, latest_date=None, decision_freq='AS', **kwargs):
super(TransientDecisionParameter, self).__init__(model, **kwargs)
self._decision_date = None
self.decision_date = decision_date
if not isinstance(before_parameter, Parameter):
raise ValueError('The `before` value should be a Parameter instance.')
before_parameter.parents.add(self)
self.before_parameter = before_parameter
if not isinstance(after_parameter, Parameter):
raise ValueError('The `after` value should be a Parameter instance.')
after_parameter.parents.add(self)
self.after_parameter = after_parameter
# These parameters are mostly used if this class is used as variable.
self._earliest_date = None
self.earliest_date = earliest_date
self._latest_date = None
self.latest_date = latest_date
cost, err = quad(self.interp, a, b)
return cost
@classmethod
def load(cls, model, data):
upper_parameter = load_parameter(model, data.pop("upper_parameter"))
lower_parameter = load_parameter(model, data.pop("lower_parameter", None))
x = np.array(data.pop("x"))
y = np.array(data.pop("y"))
kind = data.pop("kind", "linear")
return cls(model, upper_parameter, x, y, lower_parameter=lower_parameter,
interp_kwargs={'kind': kind})
InterpolatedQuadratureParameter.register()
class ScenarioWrapperParameter(Parameter):
"""Parameter that utilises a different child parameter in each scenario ensemble.
This parameter is used to switch between different child parameters based on different
ensembles in a given `Scenario`. It can be used to vary data in a non-scenario aware
parameter type across multiple scenario ensembles. For example, many of control curve or
interpolation parameters do not explicitly support scenarios. This parameter can be used
to test multiple control curve definitions as part of a single simulation.
Parameters
----------
scenario : Scenario
The scenario instance which is used to select the parameters.
parameters : iterable of Parameter instances
The child parameters that are used in each of `scenario`'s ensembles. The number
of parameters must equal the size of the given scenario.
profile.parents.add(self)
self.profile = profile
@classmethod
def load(cls, model, data):
scale = float(data.pop("scale"))
profile = load_parameter(model, data.pop("profile"))
return cls(model, scale, profile, **data)
def value(self, ts, si):
p = self.profile.get_value(si)
return self.scale * p
ScaledProfileParameter.register()
class AbstractInterpolatedParameter(Parameter):
def __init__(self, model, x, y, interp_kwargs=None, **kwargs):
super(AbstractInterpolatedParameter, self).__init__(model, **kwargs)
self.x = x
self.y = y
self.interp = None
default_interp_kwargs = dict(kind='linear', bounds_error=True)
if interp_kwargs is not None:
# Overwrite or add to defaults with given values
default_interp_kwargs.update(interp_kwargs)
self.interp_kwargs = default_interp_kwargs
def _value_to_interpolate(self, ts, scenario_index):
raise NotImplementedError()
def setup(self):
super(AbstractInterpolatedParameter, self).setup()
""" This module contains `Parameter` subclasses for modelling transient changes.
Examples include the modelling of a decision at a fixed point during a simulation.
"""
from .._component import Component
from ._parameters import Parameter, ConstantParameter, BinaryVariableParameter
import numpy as np
import pandas
class TransientDecisionParameter(Parameter):
""" Return one of two values depending on the current time-step
This `Parameter` can be used to model a discrete decision event
that happens at a given date. Prior to this date the `before`
value is returned, and post this date the `after` value is returned.
Parameters
----------
decision_date : string or pandas.Timestamp
The trigger date for the decision.
before_parameter : Parameter
The value to use before the decision date.
after_parameter : Parameter
The value to use after the decision date.
earliest_date : string or pandas.Timestamp or None
Earliest date that the variable can be set to. Defaults to `model.timestepper.start`
#!/usr/bin/env python
import calendar, datetime
from ._parameters import Parameter as BaseParameter
import numpy as np
inf = float('inf')
class License(BaseParameter):
"""Base license class from which others inherit
This class should not be instantiated directly. Instead, use one of the
subclasses (e.g. DailyLicense).
"""
def __new__(cls, *args, **kwargs):
if cls is License:
raise TypeError('License cannot be instantiated directly')
else:
return BaseParameter.__new__(cls)
def __init__(self, model, node, **kwargs):
super(License, self).__init__(model, **kwargs)
self._node = node
def resource_state(self, timestep):
def pop_kwarg_parameter(kwargs, key, default):
"""Pop a parameter from the keyword arguments dictionary
Parameters
----------
kwargs : dict
A keyword arguments dictionary
key : string
The argument name, e.g. 'flow'
default : object
The default value to use if the dictionary does not have that key
Returns a Parameter
"""
value = kwargs.pop(key, default)
if isinstance(value, Parameter):
return value
elif callable(value):
# TODO this is broken?
return FunctionParameter(self, value)
else:
return value
ArrayIndexedScenarioParameter, ScenarioMonthlyProfileParameter, ScenarioDailyProfileParameter,
ScenarioWeeklyProfileParameter, align_and_resample_dataframe, DataFrameParameter,
IndexParameter, AggregatedParameter, AggregatedIndexParameter, PiecewiseIntegralParameter,
NegativeParameter, MaxParameter, NegativeMaxParameter, MinParameter, NegativeMinParameter,
DeficitParameter, DivisionParameter, load_parameter, load_parameter_values, load_dataframe)
from . import licenses
from ._polynomial import Polynomial1DParameter, Polynomial2DStorageParameter
from ._thresholds import StorageThresholdParameter, RecorderThresholdParameter
from ._hydropower import HydropowerTargetParameter
import numpy as np
from scipy.interpolate import interp1d
from scipy.integrate import quad
import pandas
class FunctionParameter(Parameter):
def __init__(self, model, parent, func, *args, **kwargs):
super(FunctionParameter, self).__init__(model, *args, **kwargs)
self._parent = parent
self._func = func
def value(self, ts, scenario_index):
return self._func(self._parent, ts, scenario_index)
FunctionParameter.register()
class ScaledProfileParameter(Parameter):
def __init__(self, model, scale, profile, *args, **kwargs):
super(ScaledProfileParameter, self).__init__(model, *args, **kwargs)
self.scale = scale
profile.parents.add(self)