How to use the pywr.parameters.ConstantParameter function in pywr

def test_constant_level(self, simple_storage_model, efficiency):
        """ Test HydropowerRecorder """
        m = simple_storage_model

        strg = m.nodes['Storage']
        otpt = m.nodes['Output']

        elevation = ConstantParameter(m, 100)
        rec = HydropowerRecorder(m, otpt, elevation, efficiency=efficiency)
        rec_total = TotalHydroEnergyRecorder(m, otpt, elevation, efficiency=efficiency)

        m.setup()
        m.step()

        # First step
        # Head: 100m
        # Flow: 8 m3/day
        # Power: 1000 * 9.81 * 8 * 100
        # Energy: power * 1 day = power
        np.testing.assert_allclose(rec.data[0, 0], 1000 * 9.81 * 8 * 100 * 1e-6 * efficiency)
        # Second step has the same answer in this model
        m.step()
        np.testing.assert_allclose(rec.data[1, 0], 1000 * 9.81 * 8 * 100 * 1e-6 * efficiency)
        np.testing.assert_allclose(rec_total.values()[0], 2 * 1000 * 9.81 * 8 * 100 * 1e-6 * efficiency)

def test_piecewise_with_parameters_json():
    """Test using parameters with piecewise link."""
    model = load_model("piecewise1_with_parameters.json")
    sublinks = model.nodes["link1"].sublinks

    assert isinstance(sublinks[0].max_flow, ConstantParameter)
    assert np.isinf(sublinks[1].max_flow)
    assert isinstance(sublinks[0].cost, ConstantParameter)
    assert isinstance(sublinks[1].cost, ConstantParameter)

    model.run()

    assert_allclose(model.nodes["demand1"].flow, 20)

def test_storage_max_volume_param():
    """Test a that an max_volume with a Parameter results in the correct current_pc

    """

    model = Model(
        start=pandas.to_datetime('2016-01-01'),
        end=pandas.to_datetime('2016-01-01')
    )

    storage = Storage(model, 'storage', num_inputs=1, num_outputs=0)
    otpt = Output(model, 'output', max_flow=99999, cost=-99999)
    storage.connect(otpt)

    p = ConstantParameter(model, 20.0)
    storage.max_volume = p
    storage.initial_volume = 10.0
    storage.initial_volume_pc = 0.5

    model.setup()
    np.testing.assert_allclose(storage.current_pc, 0.5)

    model.run()

    p.set_double_variables(np.asarray([40.0, ]))
    model.reset()

    # This is a demonstration of the issue describe in #470
    #   https://github.com/pywr/pywr/issues/470
    # The initial storage is defined in both absolute and relative terms
    # but these are now not consistent with one another and the updated max_volume

def test_single_control_curve(self, simple_storage_model):
        """Test `ControlCurvePiecewiseInterpolatedParameter` with one control curve. """
        model = simple_storage_model
        storage_node = model.nodes["Storage"]
        input_node = model.nodes["Input"]
        output_node = model.nodes["Output"]

        control_curves = [
            ConstantParameter(model, 0.5),
        ]

        parameter = ControlCurvePiecewiseInterpolatedParameter(model, storage_node, control_curves,
                                                               [(500, 200), (100, 50)], name="CCPIP")
        assert parameter.minimum == 0.0
        assert parameter.maximum == 1.0

        input_node.max_flow = 1.0
        input_node.cost = 0
        output_node.max_flow = 0.0
        storage_node.initial_volume = 0.0
        storage_node.max_volume = 100.0
        storage_node.cost = -10

        model.timestepper.start = "1920-01-01"
        model.timestepper.delta = 1

def test_control_curve_interpolated(model, use_parameters):
    m = model
    m.timestepper.delta = 200

    s = m.nodes['Storage']
    o = m.nodes['Output']
    s.connect(o)

    cc = ConstantParameter(model, 0.8)
    values = [20.0, 5.0, 0.0]

    if use_parameters:
        # Create the parameter using parameters for the values
        parameters = [ConstantParameter(model, v) for v in values]
        s.cost = p = ControlCurveInterpolatedParameter(model, s, cc, parameters=parameters)
    else:
        # Create the parameter using a list of values
        s.cost = p = ControlCurveInterpolatedParameter(model, s, cc, values)

    @assert_rec(model, p)
    def expected_func(timestep, scenario_index):
        v = s.initial_volume
        c = cc.value(timestep, scenario_index)
        if c == 1.0 and v == 100.0:
            expected = values[1]

def test_start_end_dates(self, model):
        """ Test the assignment of earliest_date and latest_date keywords """
        date = '2015-02-01'
        before = ConstantParameter(model, 0)
        after = ConstantParameter(model, 1)

        # Default to model start and end dates
        p = TransientDecisionParameter(model, date, before, after)

        assert p.earliest_date == model.timestepper.start
        assert p.latest_date == model.timestepper.end

        # Test with user defined start and end dates
        p = TransientDecisionParameter(model, date, before, after, earliest_date='2015-02-01',
                                       latest_date='2020-02-03')

        assert p.earliest_date == pandas.to_datetime('2015-02-01')
        assert p.latest_date == pandas.to_datetime('2020-02-03')

def test_orphaned_components(simple_linear_model):
    model = simple_linear_model
    model.nodes["Input"].max_flow = ConstantParameter(model, 10.0)

    result = model.find_orphaned_parameters()
    assert(not result)
    # assert that warning not raised by check
    with pytest.warns(None) as record:
        model.check()
    for w in record:
        if isinstance(w, OrphanedParameterWarning):
            pytest.fail("OrphanedParameterWarning raised unexpectedly!")

    # add some orphans
    orphan1 = ConstantParameter(model, 5.0)
    orphan2 = ConstantParameter(model, 10.0)
    orphans = {orphan1, orphan2}
    result = model.find_orphaned_parameters()
    assert(orphans == result)

def test_two_control_curves(self, simple_storage_model):
        """Test `ControlCurvePiecewiseInterpolatedParameter` with two control curves. """
        model = simple_storage_model
        storage_node = model.nodes["Storage"]
        input_node = model.nodes["Input"]
        output_node = model.nodes["Output"]

        control_curves = [
            ConstantParameter(model, 0.75),
            ConstantParameter(model, 0.25),
        ]

        parameter = ControlCurvePiecewiseInterpolatedParameter(model, storage_node, control_curves,
                                                               [(500, 200), (100, 50), (0, -100)], name="CCPIP")
        assert parameter.minimum == 0.0
        assert parameter.maximum == 1.0

        input_node.max_flow = 1.0
        input_node.cost = 0
        output_node.max_flow = 0.0
        storage_node.initial_volume = 0.0
        storage_node.max_volume = 100.0
        storage_node.cost = -10

        model.timestepper.start = "1920-01-01"
        model.timestepper.delta = 1

def fset(self, parameter):
            old_parameter = getattr(self, private_name)
            if old_parameter:
                self.children.remove(old_parameter)
            if wrap_constants and isinstance(parameter, (float, int)):
                from pywr.parameters import ConstantParameter
                parameter = ConstantParameter(self.model, value=parameter)
            self.children.add(parameter)
            setattr(self, private_name, parameter)
        return {"fget": fget, "fset": fset}

"""

        Keywords:
            control_curve - A Parameter object that can return the control curve position,
                as a percentage of fill, for the given timestep.
        """
        control_curve = pop_kwarg_parameter(kwargs, 'control_curve', None)
        above_curve_cost = kwargs.pop('above_curve_cost', None)
        cost = kwargs.pop('cost', 0.0)
        if above_curve_cost is not None:
            if control_curve is None:
                # Make a default control curve at 100% capacity
                control_curve = ConstantParameter(model, 1.0)
            elif not isinstance(control_curve, Parameter):
                # Assume parameter is some kind of constant and coerce to ConstantParameter
                control_curve = ConstantParameter(model, control_curve)

            if not isinstance(cost, Parameter):
                # In the case where an above_curve_cost is given and cost is not a Parameter
                # a default cost Parameter is created.
                kwargs['cost'] = ControlCurveParameter(model, self, control_curve, [above_curve_cost, cost])
            else:
                raise ValueError('If an above_curve_cost is given cost must not be a Parameter.')
        else:
            # reinstate the given cost parameter to pass to the parent constructors
            kwargs['cost'] = cost
        super(Reservoir, self).__init__(model, *args, **kwargs)