solph._models.Model._add_parent_block_variables() - Code Metrics - Inspection of "Implement Investmentflow wrapper" - oemof/oemof-solph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — dev (#799)

by Uwe

created 2022-06-24 10:01 UTC

solph._models.Model._add_parent_block_variables() D

↳ Parent: solph._models

Complexity

Conditions

Size

Total Lines	33
Code Lines	25

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	25
dl	0
loc	33
rs	4.8
c	0
b	0
f	0
cc	12
nop	1

How to fix Complexity

# -*- coding: utf-8 -*-

"""Solph Optimization Models.

SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert
SPDX-FileCopyrightText: Cord Kaldemeyer
SPDX-FileCopyrightText: gplssm
SPDX-FileCopyrightText: Patrik Schönfeldt

SPDX-License-Identifier: MIT

"""
import logging
import warnings
from logging import getLogger

from pyomo import environ as po
from pyomo.core.plugins.transform.relax_integrality import RelaxIntegrality
from pyomo.opt import SolverFactory

from oemof.solph import processing
from oemof.solph.buses._bus import BusBlock
from oemof.solph.components._transformer import TransformerBlock
from oemof.solph.flows._flow import FlowBlock
from oemof.solph.flows._investment_flow import InvestmentFlowBlock
from oemof.solph.flows._non_convex_flow import NonConvexFlowBlock


class LoggingError(BaseException):
    """Raised when the wrong logging level is used."""

    pass


class BaseModel(po.ConcreteModel):
    """The BaseModel for other solph-models (Model, MultiPeriodModel, etc.)

    Parameters
    ----------
    energysystem : EnergySystem object
        Object that holds the nodes of an oemof energy system graph
    constraint_groups : list (optional)
        Solph looks for these groups in the given energy system and uses them
        to create the constraints of the optimization problem.
        Defaults to `Model.CONSTRAINTS`
    objective_weighting : array like (optional)
        Weights used for temporal objective function
        expressions. If nothing is passed `timeincrement` will be used which
        is calculated from the freq length of the energy system timeindex .
    auto_construct : boolean
        If this value is true, the set, variables, constraints, etc. are added,
        automatically when instantiating the model. For sequential model
        building process set this value to False
        and use methods `_add_parent_block_sets`,
        `_add_parent_block_variables`, `_add_blocks`, `_add_objective`

    Attributes:
    -----------
    timeincrement : sequence
        Time increments.
    flows : dict
        Flows of the model.
    name : str
        Name of the model.
    es : solph.EnergySystem
        Energy system of the model.
    meta : `pyomo.opt.results.results_.SolverResults` or None
        Solver results.
    dual : ... or None
    rc : ... or None

    """

    CONSTRAINT_GROUPS = []

    def __init__(self, energysystem, **kwargs):
        super().__init__()

        # Check root logger. Due to a problem with pyomo the building of the
        # model will take up to a 100 times longer if the root logger is set
        # to DEBUG

        if getLogger().level <= 10 and kwargs.get("debug", False) is False:
            msg = (
                "The root logger level is 'DEBUG'.\nDue to a communication "
                "problem between solph and the pyomo package,\nusing the "
                "DEBUG level will slow down the modelling process by the "
                "factor ~100.\nIf you need the debug-logging you can "
                "initialise the Model with 'debug=True`\nYou should only do "
                "this for small models. To avoid the slow-down use the "
                "logger\nfunction of oemof.tools (read docstring) or "
                "change the level of the root logger:\n\nimport logging\n"
                "logging.getLogger().setLevel(logging.INFO)"
            )
            raise LoggingError(msg)

        # ########################  Arguments #################################

        self.name = kwargs.get("name", type(self).__name__)
        self.es = energysystem
        self.timeincrement = kwargs.get("timeincrement", self.es.timeincrement)

        self.objective_weighting = kwargs.get(
            "objective_weighting", self.timeincrement
        )

        self._constraint_groups = type(self).CONSTRAINT_GROUPS + kwargs.get(
            "constraint_groups", []
        )

        self._constraint_groups += [
            i
            for i in self.es.groups
            if hasattr(i, "CONSTRAINT_GROUP")
            and i not in self._constraint_groups
        ]

        self.flows = self.es.flows()

        self.solver_results = None
        self.dual = None
        self.rc = None

        if kwargs.get("auto_construct", True):
            self._construct()

    def _construct(self):
        """ """
        self._add_parent_block_sets()
        self._add_parent_block_variables()
        self._add_child_blocks()
        self._add_objective()

    def _add_parent_block_sets(self):
        """ " Method to create all sets located at the parent block, i.e. the
        model itself as they are to be shared across all model components.
        """
        pass

    def _add_parent_block_variables(self):
        """ " Method to create all variables located at the parent block,
        i.e. the model itself as these variables  are to be shared across
        all model components.
        """
        pass

    def _add_child_blocks(self):
        """Method to add the defined child blocks for components that have
        been grouped in the defined constraint groups.
        """

        for group in self._constraint_groups:
            # create instance for block
            block = group()
            # Add block to model
            self.add_component(str(block), block)
            # create constraints etc. related with block for all nodes
            # in the group
            block._create(group=self.es.groups.get(group))

    def _add_objective(self, sense=po.minimize, update=False):
        """Method to sum up all objective expressions from the child blocks
        that have been created. This method looks for `_objective_expression`
        attribute in the block definition and will call this method to add
        their return value to the objective function.
        """
        if update:
            self.del_component("objective")

        expr = 0

        for block in self.component_data_objects():
            if hasattr(block, "_objective_expression"):
                expr += block._objective_expression()

        self.objective = po.Objective(sense=sense, expr=expr)

    def receive_duals(self):
        """Method sets solver suffix to extract information about dual
        variables from solver. Shadow prices (duals) and reduced costs (rc) are
        set as attributes of the model.

        """
        # shadow prices
        del self.dual
        self.dual = po.Suffix(direction=po.Suffix.IMPORT)
        # reduced costs
        del self.rc
        self.rc = po.Suffix(direction=po.Suffix.IMPORT)

    def results(self):
        """Returns a nested dictionary of the results of this optimization"""
        return processing.results(self)

    def solve(self, solver="cbc", solver_io="lp", **kwargs):
        r"""Takes care of communication with solver to solve the model.

        Parameters
        ----------
        solver : string
            solver to be used e.g. "glpk","gurobi","cplex"
        solver_io : string
            pyomo solver interface file format: "lp","python","nl", etc.
        \**kwargs : keyword arguments
            Possible keys can be set see below:

        Other Parameters
        ----------------
        solve_kwargs : dict
            Other arguments for the pyomo.opt.SolverFactory.solve() method
            Example : {"tee":True}
        cmdline_options : dict
            Dictionary with command line options for solver e.g.
            {"mipgap":"0.01"} results in "--mipgap 0.01"
            {"interior":" "} results in "--interior"
            Gurobi solver takes numeric parameter values such as
            {"method": 2}

        """
        solve_kwargs = kwargs.get("solve_kwargs", {})
        solver_cmdline_options = kwargs.get("cmdline_options", {})

        opt = SolverFactory(solver, solver_io=solver_io)
        # set command line options
        options = opt.options
        for k in solver_cmdline_options:
            options[k] = solver_cmdline_options[k]

        solver_results = opt.solve(self, **solve_kwargs)

        status = solver_results["Solver"][0]["Status"]
        termination_condition = solver_results["Solver"][0][
            "Termination condition"
        ]

        if status == "ok" and termination_condition == "optimal":
            logging.info("Optimization successful...")
        else:
            msg = (
                "Optimization ended with status {0} and termination "
                "condition {1}"
            )
            warnings.warn(
                msg.format(status, termination_condition), UserWarning
            )
        self.es.results = solver_results
        self.solver_results = solver_results

        return solver_results

    def relax_problem(self):
        """Relaxes integer variables to reals of optimization model self."""
        relaxer = RelaxIntegrality()
        relaxer._apply_to(self)

        return self


class Model(BaseModel):
    """An  energy system model for operational and investment
    optimization.

    Parameters
    ----------
    energysystem : EnergySystem object
        Object that holds the nodes of an oemof energy system graph
    constraint_groups : list
        Solph looks for these groups in the given energy system and uses them
        to create the constraints of the optimization problem.
        Defaults to `Model.CONSTRAINTS`

    **The following basic sets are created**:

    NODES :
        A set with all nodes of the given energy system.

    TIMESTEPS :
        A set with all timesteps of the given time horizon.

    FLOWS :
        A 2 dimensional set with all flows. Index: `(source, target)`

    **The following basic variables are created**:

    flow
        FlowBlock from source to target indexed by FLOWS, TIMESTEPS.
        Note: Bounds of this variable are set depending on attributes of
        the corresponding flow object.

    """

    CONSTRAINT_GROUPS = [
        BusBlock,
        TransformerBlock,
        InvestmentFlowBlock,
        FlowBlock,
        NonConvexFlowBlock,
    ]

    def __init__(self, energysystem, **kwargs):
        super().__init__(energysystem, **kwargs)

    def _add_parent_block_sets(self):
        """ """
        # set with all nodes
        self.NODES = po.Set(initialize=[n for n in self.es.nodes])

        if self.es.timeincrement is None:
            msg = (
                "The EnergySystem needs to have a valid 'timeincrement' "
                "attribute to build a model."
            )
            raise AttributeError(msg)

        # pyomo set for timesteps of optimization problem
        self.TIMESTEPS = po.Set(
            initialize=range(len(self.es.timeincrement)), ordered=True
        )
        self.TIMEPOINTS = po.Set(
            initialize=range(len(self.es.timeincrement) + 1), ordered=True
        )

        # previous timesteps
        previous_timesteps = [x - 1 for x in self.TIMESTEPS]
        previous_timesteps[0] = self.TIMESTEPS.last()

        self.previous_timesteps = dict(zip(self.TIMESTEPS, previous_timesteps))

        # pyomo set for all flows in the energy system graph
        self.FLOWS = po.Set(
            initialize=self.flows.keys(), ordered=True, dimen=2
        )

        self.BIDIRECTIONAL_FLOWS = po.Set(
            initialize=[
                k
                for (k, v) in self.flows.items()
                if hasattr(v, "bidirectional")
            ],
            ordered=True,
            dimen=2,
            within=self.FLOWS,
        )

        self.UNIDIRECTIONAL_FLOWS = po.Set(
            initialize=[
                k
                for (k, v) in self.flows.items()
                if not hasattr(v, "bidirectional")
            ],
            ordered=True,
            dimen=2,
            within=self.FLOWS,
        )

    def _add_parent_block_variables(self):
        """ """
        self.flow = po.Var(self.FLOWS, self.TIMESTEPS, within=po.Reals)

        for (o, i) in self.FLOWS:
            if self.flows[o, i].nominal_value is not None:
                if self.flows[o, i].fix[self.TIMESTEPS.at(1)] is not None:
                    for t in self.TIMESTEPS:
                        self.flow[o, i, t].value = (
                            self.flows[o, i].fix[t]
                            * self.flows[o, i].nominal_value
                        )
                        self.flow[o, i, t].fix()
                else:
                    for t in self.TIMESTEPS:
                        self.flow[o, i, t].setub(
                            self.flows[o, i].max[t]
                            * self.flows[o, i].nominal_value
                        )

                    if not self.flows[o, i].nonconvex:
                        for t in self.TIMESTEPS:
                            self.flow[o, i, t].setlb(
                                self.flows[o, i].min[t]
                                * self.flows[o, i].nominal_value
                            )
                    elif (o, i) in self.UNIDIRECTIONAL_FLOWS:
                        for t in self.TIMESTEPS:
                            self.flow[o, i, t].setlb(0)
            else:
                if (o, i) in self.UNIDIRECTIONAL_FLOWS:
                    for t in self.TIMESTEPS:
                        self.flow[o, i, t].setlb(0)


1			# -- coding: utf-8 --
2
3			"""Solph Optimization Models.
4
5			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
6			SPDX-FileCopyrightText: Simon Hilpert
7			SPDX-FileCopyrightText: Cord Kaldemeyer
8			SPDX-FileCopyrightText: gplssm
9			SPDX-FileCopyrightText: Patrik Schönfeldt
10
11			SPDX-License-Identifier: MIT
12
13			"""
14			import logging
15			import warnings
16			from logging import getLogger
17
18			from pyomo import environ as po
19			from pyomo.core.plugins.transform.relax_integrality import RelaxIntegrality
20			from pyomo.opt import SolverFactory
21
22			from oemof.solph import processing
23			from oemof.solph.buses._bus import BusBlock
24			from oemof.solph.components._transformer import TransformerBlock
25			from oemof.solph.flows._flow import FlowBlock
26			from oemof.solph.flows._investment_flow import InvestmentFlowBlock
27			from oemof.solph.flows._non_convex_flow import NonConvexFlowBlock
28
29
30			class LoggingError(BaseException):
31			"""Raised when the wrong logging level is used."""
32
33			pass
34
35
36			class BaseModel(po.ConcreteModel):
37			"""The BaseModel for other solph-models (Model, MultiPeriodModel, etc.)
38
39			Parameters
40			----------
41			energysystem : EnergySystem object
42			Object that holds the nodes of an oemof energy system graph
43			constraint_groups : list (optional)
44			Solph looks for these groups in the given energy system and uses them
45			to create the constraints of the optimization problem.
46			Defaults to `Model.CONSTRAINTS`
47			objective_weighting : array like (optional)
48			Weights used for temporal objective function
49			expressions. If nothing is passed `timeincrement` will be used which
50			is calculated from the freq length of the energy system timeindex .
51			auto_construct : boolean
52			If this value is true, the set, variables, constraints, etc. are added,
53			automatically when instantiating the model. For sequential model
54			building process set this value to False
55			and use methods `_add_parent_block_sets`,
56			`_add_parent_block_variables`, `_add_blocks`, `_add_objective`
57
58			Attributes:
59			-----------
60			timeincrement : sequence
61			Time increments.
62			flows : dict
63			Flows of the model.
64			name : str
65			Name of the model.
66			es : solph.EnergySystem
67			Energy system of the model.
68			meta : `pyomo.opt.results.results_.SolverResults` or None
69			Solver results.
70			dual : ... or None
71			rc : ... or None
72
73			"""
74
75			CONSTRAINT_GROUPS = []
76
77			def __init__(self, energysystem, **kwargs):
78			super().__init__()
79
80			# Check root logger. Due to a problem with pyomo the building of the
81			# model will take up to a 100 times longer if the root logger is set
82			# to DEBUG
83
84			if getLogger().level <= 10 and kwargs.get("debug", False) is False:
85			msg = (
86			"The root logger level is 'DEBUG'.\nDue to a communication "
87			"problem between solph and the pyomo package,\nusing the "
88			"DEBUG level will slow down the modelling process by the "
89			"factor ~100.\nIf you need the debug-logging you can "
90			"initialise the Model with 'debug=True`\nYou should only do "
91			"this for small models. To avoid the slow-down use the "
92			"logger\nfunction of oemof.tools (read docstring) or "
93			"change the level of the root logger:\n\nimport logging\n"
94			"logging.getLogger().setLevel(logging.INFO)"
95			)
96			raise LoggingError(msg)
97
98			# ######################## Arguments #################################
99
100			self.name = kwargs.get("name", type(self).__name__)
101			self.es = energysystem
102			self.timeincrement = kwargs.get("timeincrement", self.es.timeincrement)
103
104			self.objective_weighting = kwargs.get(
105			"objective_weighting", self.timeincrement
106			)
107
108			self._constraint_groups = type(self).CONSTRAINT_GROUPS + kwargs.get(
109			"constraint_groups", []
110			)
111
112			self._constraint_groups += [
113			i
114			for i in self.es.groups
115			if hasattr(i, "CONSTRAINT_GROUP")
116			and i not in self._constraint_groups
117			]
118
119			self.flows = self.es.flows()
120
121			self.solver_results = None
122			self.dual = None
123			self.rc = None
124
125			if kwargs.get("auto_construct", True):
126			self._construct()
127
128			def _construct(self):
129			""" """
130			self._add_parent_block_sets()
131			self._add_parent_block_variables()
132			self._add_child_blocks()
133			self._add_objective()
134
135			def _add_parent_block_sets(self):
136			""" " Method to create all sets located at the parent block, i.e. the
137			model itself as they are to be shared across all model components.
138			"""
139			pass
140
141			def _add_parent_block_variables(self):
142			""" " Method to create all variables located at the parent block,
143			i.e. the model itself as these variables are to be shared across
144			all model components.
145			"""
146			pass
147
148			def _add_child_blocks(self):
149			"""Method to add the defined child blocks for components that have
150			been grouped in the defined constraint groups.
151			"""
152
153			for group in self._constraint_groups:
154			# create instance for block
155			block = group()
156			# Add block to model
157			self.add_component(str(block), block)
158			# create constraints etc. related with block for all nodes
159			# in the group
160			block._create(group=self.es.groups.get(group))
161
162			def _add_objective(self, sense=po.minimize, update=False):
163			"""Method to sum up all objective expressions from the child blocks
164			that have been created. This method looks for `_objective_expression`
165			attribute in the block definition and will call this method to add
166			their return value to the objective function.
167			"""
168			if update:
169			self.del_component("objective")
170
171			expr = 0
172
173			for block in self.component_data_objects():
174			if hasattr(block, "_objective_expression"):
175			expr += block._objective_expression()
176
177			self.objective = po.Objective(sense=sense, expr=expr)
178
179			def receive_duals(self):
180			"""Method sets solver suffix to extract information about dual
181			variables from solver. Shadow prices (duals) and reduced costs (rc) are
182			set as attributes of the model.
183
184			"""
185			# shadow prices
186			del self.dual
187			self.dual = po.Suffix(direction=po.Suffix.IMPORT)
188			# reduced costs
189			del self.rc
190			self.rc = po.Suffix(direction=po.Suffix.IMPORT)
191
192			def results(self):
193			"""Returns a nested dictionary of the results of this optimization"""
194			return processing.results(self)
195
196			def solve(self, solver="cbc", solver_io="lp", **kwargs):
197			r"""Takes care of communication with solver to solve the model.
198
199			Parameters
200			----------
201			solver : string
202			solver to be used e.g. "glpk","gurobi","cplex"
203			solver_io : string
204			pyomo solver interface file format: "lp","python","nl", etc.
205			\**kwargs : keyword arguments
206			Possible keys can be set see below:
207
208			Other Parameters
209			----------------
210			solve_kwargs : dict
211			Other arguments for the pyomo.opt.SolverFactory.solve() method
212			Example : {"tee":True}
213			cmdline_options : dict
214			Dictionary with command line options for solver e.g.
215			{"mipgap":"0.01"} results in "--mipgap 0.01"
216			{"interior":" "} results in "--interior"
217			Gurobi solver takes numeric parameter values such as
218			{"method": 2}
219
220			"""
221			solve_kwargs = kwargs.get("solve_kwargs", {})
222			solver_cmdline_options = kwargs.get("cmdline_options", {})
223
224			opt = SolverFactory(solver, solver_io=solver_io)
225			# set command line options
226			options = opt.options
227			for k in solver_cmdline_options:
228			options[k] = solver_cmdline_options[k]
229
230			solver_results = opt.solve(self, **solve_kwargs)
231
232			status = solver_results["Solver"][0]["Status"]
233			termination_condition = solver_results["Solver"][0][
234			"Termination condition"
235			]
236
237			if status == "ok" and termination_condition == "optimal":
238			logging.info("Optimization successful...")
239			else:
240			msg = (
241			"Optimization ended with status {0} and termination "
242			"condition {1}"
243			)
244			warnings.warn(
245			msg.format(status, termination_condition), UserWarning
246			)
247			self.es.results = solver_results
248			self.solver_results = solver_results
249
250			return solver_results
251
252			def relax_problem(self):
253			"""Relaxes integer variables to reals of optimization model self."""
254			relaxer = RelaxIntegrality()
255			relaxer._apply_to(self)
256
257			return self
258
259
260			class Model(BaseModel):
261			"""An energy system model for operational and investment
262			optimization.
263
264			Parameters
265			----------
266			energysystem : EnergySystem object
267			Object that holds the nodes of an oemof energy system graph
268			constraint_groups : list
269			Solph looks for these groups in the given energy system and uses them
270			to create the constraints of the optimization problem.
271			Defaults to `Model.CONSTRAINTS`
272
273			The following basic sets are created:
274
275			NODES :
276			A set with all nodes of the given energy system.
277
278			TIMESTEPS :
279			A set with all timesteps of the given time horizon.
280
281			FLOWS :
282			A 2 dimensional set with all flows. Index: `(source, target)`
283
284			The following basic variables are created:
285
286			flow
287			FlowBlock from source to target indexed by FLOWS, TIMESTEPS.
288			Note: Bounds of this variable are set depending on attributes of
289			the corresponding flow object.
290
291			"""
292
293			CONSTRAINT_GROUPS = [
294			BusBlock,
295			TransformerBlock,
296			InvestmentFlowBlock,
297			FlowBlock,
298			NonConvexFlowBlock,
299			]
300
301			def __init__(self, energysystem, **kwargs):
302			super().__init__(energysystem, **kwargs)
303
304			def _add_parent_block_sets(self):
305			""" """
306			# set with all nodes
307			self.NODES = po.Set(initialize=[n for n in self.es.nodes])
308
309			if self.es.timeincrement is None:
310			msg = (
311			"The EnergySystem needs to have a valid 'timeincrement' "
312			"attribute to build a model."
313			)
314			raise AttributeError(msg)
315
316			# pyomo set for timesteps of optimization problem
317			self.TIMESTEPS = po.Set(
318			initialize=range(len(self.es.timeincrement)), ordered=True
319			)
320			self.TIMEPOINTS = po.Set(
321			initialize=range(len(self.es.timeincrement) + 1), ordered=True
322			)
323
324			# previous timesteps
325			previous_timesteps = [x - 1 for x in self.TIMESTEPS]
326			previous_timesteps[0] = self.TIMESTEPS.last()
327
328			self.previous_timesteps = dict(zip(self.TIMESTEPS, previous_timesteps))
329
330			# pyomo set for all flows in the energy system graph
331			self.FLOWS = po.Set(
332			initialize=self.flows.keys(), ordered=True, dimen=2
333			)
334
335			self.BIDIRECTIONAL_FLOWS = po.Set(
336			initialize=[
337			k
338			for (k, v) in self.flows.items()
339			if hasattr(v, "bidirectional")
340			],
341			ordered=True,
342			dimen=2,
343			within=self.FLOWS,
344			)
345
346			self.UNIDIRECTIONAL_FLOWS = po.Set(
347			initialize=[
348			k
349			for (k, v) in self.flows.items()
350			if not hasattr(v, "bidirectional")
351			],
352			ordered=True,
353			dimen=2,
354			within=self.FLOWS,
355			)
356
357			def _add_parent_block_variables(self):
358			""" """
359			self.flow = po.Var(self.FLOWS, self.TIMESTEPS, within=po.Reals)
360
361			for (o, i) in self.FLOWS:
362			if self.flows[o, i].nominal_value is not None:
363			if self.flows[o, i].fix[self.TIMESTEPS.at(1)] is not None:
364			for t in self.TIMESTEPS:
365			self.flow[o, i, t].value = (
366			self.flows[o, i].fix[t]
367			* self.flows[o, i].nominal_value
368			)
369			self.flow[o, i, t].fix()
370			else:
371			for t in self.TIMESTEPS:
372			self.flow[o, i, t].setub(
373			self.flows[o, i].max[t]
374			* self.flows[o, i].nominal_value
375			)
376
377			if not self.flows[o, i].nonconvex:
378			for t in self.TIMESTEPS:
379			self.flow[o, i, t].setlb(
380			self.flows[o, i].min[t]
381			* self.flows[o, i].nominal_value
382			)
383			elif (o, i) in self.UNIDIRECTIONAL_FLOWS:
384			for t in self.TIMESTEPS:
385			self.flow[o, i, t].setlb(0)
386			else:
387			if (o, i) in self.UNIDIRECTIONAL_FLOWS:
388			for t in self.TIMESTEPS:
389			self.flow[o, i, t].setlb(0)
390

oemof / oemof-solph

Pull Request — dev (#799)

solph._models.Model._add_parent_block_variables() D

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like