network.energy_system.EnergySystem.restore() - Code Metrics - oemof/oemof-network - Measure and Improve Code Quality continuously with Scrutinizer

network.energy_system.EnergySystem.restore() A
last analyzed 2022-04-13 19:53 UTC

↳ Parent: network.energy_system

Complexity

Conditions

Size

Total Lines	18
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	12
nop	3
dl	0
loc	18
rs	9.8
c	0
b	0
f	0

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

"""Basic EnergySystem class

This file is part of project oemof (github.com/oemof/oemof). It's copyrighted
by the contributors recorded in the version control history of the file,
available from its original location oemof/oemof/energy_system.py

SPDX-FileCopyrightText: Stephan Günther <>
SPDX-FileCopyrightText: Uwe Krien <[email protected]>
SPDX-FileCopyrightText: Simon Hilpert <>
SPDX-FileCopyrightText: Cord Kaldemeyer <>

SPDX-License-Identifier: MIT
"""

import logging
import os
from collections import deque

import blinker
import dill as pickle

from oemof.network.groupings import DEFAULT as BY_UID
from oemof.network.groupings import Grouping
from oemof.network.groupings import Nodes


class EnergySystem:
    r"""Defining an energy supply system to use oemof's solver libraries.

    Note
    ----
    The list of regions is not necessary to use the energy system with solph.

    Parameters
    ----------
    entities : list of :class:`Entity <oemof.core.network.Entity>`, optional
        A list containing the already existing :class:`Entities
        <oemof.core.network.Entity>` that should be part of the energy system.
        Stored in the :attr:`entities` attribute.
        Defaults to `[]` if not supplied.
    timeindex : pandas.datetimeindex
        Defines the time range and, if equidistant, the timeindex for the
        energy system
    timeincrement : numeric (sequence)
        Define the timeincrement for the energy system
    groupings : list
        The elements of this list are used to construct :class:`Groupings
        <oemof.core.energy_system.Grouping>` or they are used directly if they
        are instances of :class:`Grouping <oemof.core.energy_system.Grouping>`.
        These groupings are then used to aggregate the entities added to this
        energy system into :attr:`groups`.
        By default, there'll always be one group for each :attr:`uid
        <oemof.core.network.Entity.uid>` containing exactly the entity with the
        given :attr:`uid <oemof.core.network.Entity.uid>`.
        See the :ref:`examples <energy-system-examples>` for more information.

    Attributes
    ----------
    entities : list of :class:`Entity <oemof.core.network.Entity>`
        A list containing the :class:`Entities <oemof.core.network.Entity>`
        that comprise the energy system. If this :class:`EnergySystem` is
        set as the :attr:`registry <oemof.core.network.Entity.registry>`
        attribute, which is done automatically on :class:`EnergySystem`
        construction, newly created :class:`Entities
        <oemof.core.network.Entity>` are automatically added to this list on
        construction.
    groups : dict
    results : dictionary
        A dictionary holding the results produced by the energy system.
        Is `None` while no results are produced.
        Currently only set after a call to :meth:`optimize` after which it
        holds the return value of :meth:`om.results()
        <oemof.solph.optimization_model.OptimizationModel.results>`.
        See the documentation of that method for a detailed description of the
        structure of the results dictionary.
    timeindex : pandas.index, optional
        Define the time range and increment for the energy system. This is an
        optional attribute but might be import for other functions/methods that
        use the EnergySystem class as an input parameter.


    .. _energy-system-examples:
    Examples
    --------

    Regardles of additional groupings, :class:`entities
    <oemof.core.network.Entity>` will always be grouped by their :attr:`uid
    <oemof.core.network.Entity.uid>`:

    >>> from oemof.network.network import Bus, Sink
    >>> es = EnergySystem()
    >>> bus = Bus(label='electricity')
    >>> es.add(bus)
    >>> bus is es.groups['electricity']
    True
    >>> es.dump()  # doctest: +ELLIPSIS
    'Attributes dumped to:...
    >>> es = EnergySystem()
    >>> es.restore()  # doctest: +ELLIPSIS
    'Attributes restored from:...
    >>> bus is es.groups['electricity']
    False
    >>> es.groups['electricity']
    "<oemof.network.network.Bus: 'electricity'>"

    For simple user defined groupings, you can just supply a function that
    computes a key from an :class:`entity <oemof.core.network.Entity>` and the
    resulting groups will be sets of :class:`entities
    <oemof.core.network.Entity>` stored under the returned keys, like in this
    example, where :class:`entities <oemof.core.network.Entity>` are grouped by
    their `type`:

    >>> es = EnergySystem(groupings=[type])
    >>> buses = set(Bus(label="Bus {}".format(i)) for i in range(9))
    >>> es.add(*buses)
    >>> components = set(Sink(label="Component {}".format(i))
    ...                   for i in range(9))
    >>> es.add(*components)
    >>> buses == es.groups[Bus]
    True
    >>> components == es.groups[Sink]
    True

    """

    signals = {}
    """A dictionary of blinker_ signals emitted by energy systems.

    Currently only one signal is supported. This signal is emitted whenever a
    `Node <oemof.network.Node>` is `add`ed to an energy system. The signal's
    `sender` is set to the `node <oemof.network.Node>` that got added to the
    energy system so that `nodes <oemof.network.Node>` have an easy way to only
    receive signals for when they themselves get added to an energy system.

    .. _blinker: https://pythonhosted.org/blinker/
    """

    def __init__(self, **kwargs):
        self._first_ungrouped_node_index_ = 0
        self._groups = {}
        self._groupings = [BY_UID] + [
            g if isinstance(g, Grouping) else Nodes(g)
            for g in kwargs.get("groupings", [])
        ]
        self.entities = []

        self.results = kwargs.get("results")

        self.timeindex = kwargs.get("timeindex")

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

        self.temporal = kwargs.get("temporal")

        self.add(*kwargs.get("entities", ()))

    def add(self, *nodes):
        """Add :class:`nodes <oemof.network.Node>` to this energy system."""
        self.nodes.extend(nodes)
        for n in nodes:
            self.signals[type(self).add].send(n, EnergySystem=self)

    signals[add] = blinker.signal(add)

    @property
    def groups(self):
        gs = self._groups
        deque(
            (
                g(n, gs)

                for g in self._groupings
                for n in self.nodes[self._first_ungrouped_node_index_ :]
            ),
            maxlen=0,
        )
        self._first_ungrouped_node_index_ = len(self.nodes)
        return self._groups

    @property
    def nodes(self):
        return self.entities

    @nodes.setter
    def nodes(self, value):
        self.entities = value

    def flows(self):
        return {
            (source, target): source.outputs[target]
            for source in self.nodes
            for target in source.outputs
        }

    def dump(self, dpath=None, filename=None):
        r"""Dump an EnergySystem instance."""
        if dpath is None:
            bpath = os.path.join(os.path.expanduser("~"), ".oemof")
            if not os.path.isdir(bpath):
                os.mkdir(bpath)
            dpath = os.path.join(bpath, "dumps")
            if not os.path.isdir(dpath):
                os.mkdir(dpath)

        if filename is None:
            filename = "es_dump.oemof"

        pickle.dump(self.__dict__, open(os.path.join(dpath, filename), "wb"))

        msg = "Attributes dumped to: {0}".format(os.path.join(dpath, filename))
        logging.debug(msg)
        return msg

    def restore(self, dpath=None, filename=None):
        r"""Restore an EnergySystem instance."""
        logging.info(
            "Restoring attributes will overwrite existing attributes."
        )
        if dpath is None:
            dpath = os.path.join(os.path.expanduser("~"), ".oemof", "dumps")

        if filename is None:
            filename = "es_dump.oemof"

        self.__dict__ = pickle.load(open(os.path.join(dpath, filename), "rb"))

        msg = "Attributes restored from: {0}".format(
            os.path.join(dpath, filename)
        )
        logging.debug(msg)
        return msg


1			# -- coding: utf-8 --
2
3			"""Basic EnergySystem class
4
5			This file is part of project oemof (github.com/oemof/oemof). It's copyrighted
6			by the contributors recorded in the version control history of the file,
7			available from its original location oemof/oemof/energy_system.py
8
9			SPDX-FileCopyrightText: Stephan Günther <>
10			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
11			SPDX-FileCopyrightText: Simon Hilpert <>
12			SPDX-FileCopyrightText: Cord Kaldemeyer <>
13
14			SPDX-License-Identifier: MIT
15			"""
16
17			import logging
18			import os
19			from collections import deque
20
21			import blinker
22			import dill as pickle
23
24			from oemof.network.groupings import DEFAULT as BY_UID
25			from oemof.network.groupings import Grouping
26			from oemof.network.groupings import Nodes
27
28
29			class EnergySystem:
30			r"""Defining an energy supply system to use oemof's solver libraries.
31
32			Note
33			----
34			The list of regions is not necessary to use the energy system with solph.
35
36			Parameters
37			----------
38			entities : list of :class:`Entity <oemof.core.network.Entity>`, optional
39			A list containing the already existing :class:`Entities
40			<oemof.core.network.Entity>` that should be part of the energy system.
41			Stored in the :attr:`entities` attribute.
42			Defaults to `[]` if not supplied.
43			timeindex : pandas.datetimeindex
44			Defines the time range and, if equidistant, the timeindex for the
45			energy system
46			timeincrement : numeric (sequence)
47			Define the timeincrement for the energy system
48			groupings : list
49			The elements of this list are used to construct :class:`Groupings
50			<oemof.core.energy_system.Grouping>` or they are used directly if they
51			are instances of :class:`Grouping <oemof.core.energy_system.Grouping>`.
52			These groupings are then used to aggregate the entities added to this
53			energy system into :attr:`groups`.
54			By default, there'll always be one group for each :attr:`uid
55			<oemof.core.network.Entity.uid>` containing exactly the entity with the
56			given :attr:`uid <oemof.core.network.Entity.uid>`.
57			See the :ref:`examples <energy-system-examples>` for more information.
58
59			Attributes
60			----------
61			entities : list of :class:`Entity <oemof.core.network.Entity>`
62			A list containing the :class:`Entities <oemof.core.network.Entity>`
63			that comprise the energy system. If this :class:`EnergySystem` is
64			set as the :attr:`registry <oemof.core.network.Entity.registry>`
65			attribute, which is done automatically on :class:`EnergySystem`
66			construction, newly created :class:`Entities
67			<oemof.core.network.Entity>` are automatically added to this list on
68			construction.
69			groups : dict
70			results : dictionary
71			A dictionary holding the results produced by the energy system.
72			Is `None` while no results are produced.
73			Currently only set after a call to :meth:`optimize` after which it
74			holds the return value of :meth:`om.results()
75			<oemof.solph.optimization_model.OptimizationModel.results>`.
76			See the documentation of that method for a detailed description of the
77			structure of the results dictionary.
78			timeindex : pandas.index, optional
79			Define the time range and increment for the energy system. This is an
80			optional attribute but might be import for other functions/methods that
81			use the EnergySystem class as an input parameter.
82
83
84			.. _energy-system-examples:
85			Examples
86			--------
87
88			Regardles of additional groupings, :class:`entities
89			<oemof.core.network.Entity>` will always be grouped by their :attr:`uid
90			<oemof.core.network.Entity.uid>`:
91
92			>>> from oemof.network.network import Bus, Sink
93			>>> es = EnergySystem()
94			>>> bus = Bus(label='electricity')
95			>>> es.add(bus)
96			>>> bus is es.groups['electricity']
97			True
98			>>> es.dump() # doctest: +ELLIPSIS
99			'Attributes dumped to:...
100			>>> es = EnergySystem()
101			>>> es.restore() # doctest: +ELLIPSIS
102			'Attributes restored from:...
103			>>> bus is es.groups['electricity']
104			False
105			>>> es.groups['electricity']
106			"<oemof.network.network.Bus: 'electricity'>"
107
108			For simple user defined groupings, you can just supply a function that
109			computes a key from an :class:`entity <oemof.core.network.Entity>` and the
110			resulting groups will be sets of :class:`entities
111			<oemof.core.network.Entity>` stored under the returned keys, like in this
112			example, where :class:`entities <oemof.core.network.Entity>` are grouped by
113			their `type`:
114
115			>>> es = EnergySystem(groupings=[type])
116			>>> buses = set(Bus(label="Bus {}".format(i)) for i in range(9))
117			>>> es.add(*buses)
118			>>> components = set(Sink(label="Component {}".format(i))
119			... for i in range(9))
120			>>> es.add(*components)
121			>>> buses == es.groups[Bus]
122			True
123			>>> components == es.groups[Sink]
124			True
125
126			"""
127
128			signals = {}
129			"""A dictionary of blinker_ signals emitted by energy systems.
130
131			Currently only one signal is supported. This signal is emitted whenever a
132			`Node <oemof.network.Node>` is `add`ed to an energy system. The signal's
133			`sender` is set to the `node <oemof.network.Node>` that got added to the
134			energy system so that `nodes <oemof.network.Node>` have an easy way to only
135			receive signals for when they themselves get added to an energy system.
136
137			.. _blinker: https://pythonhosted.org/blinker/
138			"""
139
140			def __init__(self, **kwargs):
141			self._first_ungrouped_node_index_ = 0
142			self._groups = {}
143			self._groupings = [BY_UID] + [
144			g if isinstance(g, Grouping) else Nodes(g)
145			for g in kwargs.get("groupings", [])
146			]
147			self.entities = []
148
149			self.results = kwargs.get("results")
150
151			self.timeindex = kwargs.get("timeindex")
152
153			self.timeincrement = kwargs.get("timeincrement", None)
154
155			self.temporal = kwargs.get("temporal")
156
157			self.add(*kwargs.get("entities", ()))
158
159			def add(self, *nodes):
160			"""Add :class:`nodes <oemof.network.Node>` to this energy system."""
161			self.nodes.extend(nodes)
162			for n in nodes:
163			self.signals[type(self).add].send(n, EnergySystem=self)
164
165			signals[add] = blinker.signal(add)
166
167			@property
168			def groups(self):
169			gs = self._groups
170			deque(
171			(
172			g(n, gs)
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-12-16 12:53 UTC by Report Bug Copy Issue Report The variable `n` does not seem to be defined. Loading history... Comprehensibility Best Practice introduced 2021-12-16 12:53 UTC by Report Bug Copy Issue Report The variable `g` does not seem to be defined. Loading history...
173			for g in self._groupings
174			for n in self.nodes[self._first_ungrouped_node_index_ :]
175			),
176			maxlen=0,
177			)
178			self._first_ungrouped_node_index_ = len(self.nodes)
179			return self._groups
180
181			@property
182			def nodes(self):
183			return self.entities
184
185			@nodes.setter
186			def nodes(self, value):
187			self.entities = value
188
189			def flows(self):
190			return {
191			(source, target): source.outputs[target]
192			for source in self.nodes
193			for target in source.outputs
194			}
195
196			def dump(self, dpath=None, filename=None):
197			r"""Dump an EnergySystem instance."""
198			if dpath is None:
199			bpath = os.path.join(os.path.expanduser("~"), ".oemof")
200			if not os.path.isdir(bpath):
201			os.mkdir(bpath)
202			dpath = os.path.join(bpath, "dumps")
203			if not os.path.isdir(dpath):
204			os.mkdir(dpath)
205
206			if filename is None:
207			filename = "es_dump.oemof"
208
209			pickle.dump(self.__dict__, open(os.path.join(dpath, filename), "wb"))
210
211			msg = "Attributes dumped to: {0}".format(os.path.join(dpath, filename))
212			logging.debug(msg)
213			return msg
214
215			def restore(self, dpath=None, filename=None):
216			r"""Restore an EnergySystem instance."""
217			logging.info(
218			"Restoring attributes will overwrite existing attributes."
219			)
220			if dpath is None:
221			dpath = os.path.join(os.path.expanduser("~"), ".oemof", "dumps")
222
223			if filename is None:
224			filename = "es_dump.oemof"
225
226			self.__dict__ = pickle.load(open(os.path.join(dpath, filename), "rb"))
227
228			msg = "Attributes restored from: {0}".format(
229			os.path.join(dpath, filename)
230			)
231			logging.debug(msg)
232			return msg
233

oemof / oemof-network

network.energy_system.EnergySystem.restore() A last analyzed 2022-04-13 19:53 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

network.energy_system.EnergySystem.restore() A
last analyzed 2022-04-13 19:53 UTC