network.network.Transformer.__init__() - Code Metrics - oemof/oemof-network - Measure and Improve Code Quality continuously with Scrutinizer

network.network.Transformer.init() A
last analyzed 2022-04-13 19:53 UTC

↳ Parent: network.network

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	3
dl	0
loc	2
rs	10
c	0
b	0
f	0

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

"""This package (along with its subpackages) contains the classes used to model
energy systems. An energy system is modelled as a graph/network of entities
with very specific constraints on which types of entities are allowed to be
connected.

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/network.py

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

SPDX-License-Identifier: MIT
"""

import warnings
from collections import UserDict as UD
from collections import namedtuple as NT
from collections.abc import Mapping
from collections.abc import MutableMapping as MM
from contextlib import contextmanager
from functools import total_ordering

# TODO:
#
#   * Only allow setting a Node's label if `_delay_registration_` is active
#     and/or the node is not yet registered.
#   * Only allow setting an Edge's input/output if it is None
#   * Document the `register` method. Maybe also document the
#     `_delay_registration_` attribute and make it official. This could also be
#     a good chance to finally use `blinker` to put an event on
#     `_delay_registration_` for deletion/assignment to trigger registration.
#     I always had the hunch that using blinker could help to straighten out
#     that delayed auto registration hack via partial functions. Maybe this
#     could be a good starting point for this.
#   * Finally get rid of `Entity`.
#


class Inputs(MM):
    """A special helper to map `n1.inputs[n2]` to `n2.outputs[n1]`."""

    def __init__(self, target):
        self.target = target

    def __getitem__(self, key):
        return key.outputs.__getitem__(self.target)

    def __delitem__(self, key):
        return key.outputs.__delitem__(self.target)

    def __setitem__(self, key, value):
        return key.outputs.__setitem__(self.target, value)

    def __iter__(self):
        return iter(self.target._in_edges)

    def __len__(self):
        return self.target._in_edges.__len__()

    def __repr__(self):
        return repr(
            "<{0.__module__}.{0.__name__}: {1!r}>".format(
                type(self), dict(self)
            )
        )


class Outputs(UD):
    """
    Helper that intercepts modifications to update `Inputs` symmetrically.
    """

    def __init__(self, source):
        self.source = source
        super().__init__()

    def __delitem__(self, key):
        key._in_edges.remove(self.source)
        return super().__delitem__(key)

    def __setitem__(self, key, value):
        key._in_edges.add(self.source)
        return super().__setitem__(key, value)


class Metaclass(type):
    """The metaclass for objects in an oemof energy system."""

    @property
    def registry(cls):
        warnings.warn(cls.registry_warning)
        return cls._registry

    @registry.setter
    def registry(cls, registry):
        warnings.warn(cls.registry_warning)
        cls._registry = registry


@total_ordering
class Node(metaclass=Metaclass):
    """Represents a Node in an energy system graph.

    Abstract superclass of the two general types of nodes of an energy system
    graph, collecting attributes and operations common to all types of nodes.
    Users should neither instantiate nor subclass this, but use
    :class:`Component`, :class:`Bus`, :class:`Edge` or one of their subclasses
    instead.

    .. role:: python(code)
      :language: python

    Parameters
    ----------
    label: `hashable`, optional
        Used as the string representation of this node. If this parameter is
        not an instance of :class:`str` it will be converted to a string and
        the result will be used as this node's :attr:`label`, which should be
        unique with respect to the other nodes in the energy system graph this
        node belongs to. If this parameter is not supplied, the string
        representation of this node will instead be generated based on this
        nodes `class` and `id`.
    inputs: list or dict, optional
        Either a list of this nodes' input nodes or a dictionary mapping input
        nodes to corresponding inflows (i.e. input values).
    outputs: list or dict, optional
        Either a list of this nodes' output nodes or a dictionary mapping
        output nodes to corresponding outflows (i.e. output values).

    Attributes
    ----------
    __slots__: str or iterable of str
        See the Python documentation on `__slots__
        <https://docs.python.org/3/reference/datamodel.html#slots>`_ for more
        information.
    """

    registry_warning = FutureWarning(
        "\nAutomatic registration of `Node`s is deprecated in favour of\n"
        "explicitly adding `Node`s to an `EnergySystem` via "
        "`EnergySystem.add`.\n"
        "This feature, i.e. the `Node.registry` attribute and functionality\n"
        "pertaining to it, will be removed in future versions.\n"
    )

    _registry = None
    __slots__ = ["_label", "_in_edges", "_inputs", "_outputs"]

    def __init__(self, *args, **kwargs):
        args = list(args)
        args.reverse
        self._inputs = Inputs(self)
        self._outputs = Outputs(self)
        for optional in ["label"]:
            if optional in kwargs:
                if args:
                    raise (
                        TypeError(
                            (
                                "{}.__init__()\n"
                                "  got multiple values for argument '{}'"
                            ).format(type(self), optional)
                        )
                    )
                setattr(self, "_" + optional, kwargs[optional])
            else:
                if args:
                    setattr(self, "_" + optional, args.pop())
        self._in_edges = set()
        for i in kwargs.get("inputs", {}):
            assert isinstance(i, Node), (
                "\n\nInput\n\n  {!r}\n\nof\n\n  {!r}\n\n"
                "not an instance of Node, but of {}."
            ).format(i, self, type(i))
            self._in_edges.add(i)
            try:
                flow = kwargs["inputs"].get(i)
            except AttributeError:
                flow = None
            edge = globals()["Edge"].from_object(flow)
            edge.input = i
            edge.output = self
        for o in kwargs.get("outputs", {}):
            assert isinstance(o, Node), (
                "\n\nOutput\n\n  {!r}\n\nof\n\n  {!r}\n\n"
                "not an instance of Node, but of {}."
            ).format(o, self, type(o))
            try:
                flow = kwargs["outputs"].get(o)
            except AttributeError:
                flow = None
            edge = globals()["Edge"].from_object(flow)
            edge.input = self
            edge.output = o

        self.register()
        """
        This could be slightly more efficient than the loops above, but doesn't
        play well with the assertions:

        inputs = kwargs.get('inputs', {})
        self.in_edges = {
                Edge(input=i, output=self,
                    flow=None if not isinstance(inputs, MM) else inputs[i])
                for i in inputs}

        outputs = kwargs.get('outputs', {})
        self.out_edges = {
                Edge(input=self, output=o,
                    flow=None if not isinstance(outputs, MM) else outputs[o])
                for o in outputs}
        self.edges = self.in_edges.union(self.out_edges)
        """

    def register(self):
        with warnings.catch_warnings():
            warnings.simplefilter("ignore")
            registry = __class__.registry


        if registry is not None and not getattr(
            self, "_delay_registration_", False
        ):
            __class__.registry.add(self)

    def __eq__(self, other):
        return id(self) == id(other)

    def __lt__(self, other):
        return str(self) < str(other)

    def __hash__(self):
        return hash(self.label)

    def __str__(self):
        return str(self.label)

    def __repr__(self):
        return repr(
            "<{0.__module__}.{0.__name__}: {1!r}>".format(
                type(self), self.label
            )
        )

    @property
    def label(self):
        """object :
        If this node was given a `label` on construction, this
        attribute holds the actual object passed as a parameter. Otherwise
        :py:`node.label` is a synonym for :py:`str(node)`.
        """
        return (
            self._label
            if hasattr(self, "_label")
            else "<{} #0x{:x}>".format(type(self).__name__, id(self))
        )

    @label.setter
    def label(self, label):
        self._label = label

    @property
    def inputs(self):
        """dict:
        Dictionary mapping input :class:`Nodes <Node>` :obj:`n` to
        :class:`Edge`s from :obj:`n` into :obj:`self`.
        If :obj:`self` is an :class:`Edge`, returns a dict containing the
        :class:`Edge`'s single input node as the key and the flow as the value.
        """
        return self._inputs

    @property
    def outputs(self):
        """dict:
        Dictionary mapping output :class:`Nodes <Node>` :obj:`n` to
        :class:`Edges` from :obj:`self` into :obj:`n`.
        If :obj:`self` is an :class:`Edge`, returns a dict containing the
        :class:`Edge`'s single output node as the key and the flow as the
        value.
        """
        return self._outputs


EdgeLabel = NT("EdgeLabel", ["input", "output"])


class Edge(Node):
    """
    :class:`Bus`es/:class:`Component`s are always connected by an
    :class:`Edge`.

    :class:`Edge`s connect a single non-:class:`Edge` Node with another. They
    are directed and have a (sequence of) value(s) attached to them so they can
    be used to represent a flow from a source/an input to a target/an output.

    Parameters
    ----------
    input, output: :class:`Bus` or :class:`Component`, optional
    flow, values: object, optional
        The (list of) object(s) representing the values flowing from this
        edge's input into its output. Note that these two names are aliases of
        each other, so `flow` and `values` are mutually exclusive.

    Note that all of these parameters are also set as attributes with the same
    name.
    """

    Label = EdgeLabel

    def __init__(
        self, input=None, output=None, flow=None, values=None, **kwargs
    ):
        if flow is not None and values is not None:
            raise ValueError(
                "\n\n`Edge`'s `flow` and `values` keyword arguments are "
                "aliases of each other,\nso they're mutually exclusive.\n"
                "You supplied:\n"
                + "    `flow`  : {}\n".format(flow)
                + "    `values`: {}\n".format(values)
                + "Choose one."
            )
        if input is None or output is None:
            self._delay_registration_ = True
        super().__init__(label=Edge.Label(input, output))
        self.values = values if values is not None else flow
        if input is not None and output is not None:
            input.outputs[output] = self

    @classmethod
    def from_object(cls, o):
        """Creates an `Edge` instance from a single object.

        This method inspects its argument and does something different
        depending on various cases:

          * If `o` is an instance of `Edge`, `o` is returned unchanged.
          * If `o` is a `Mapping`, the instance is created by calling
            `cls(**o)`,
          * In all other cases, `o` will be used as the `values` keyword
            argument to `Edge`s constructor.
        """
        if isinstance(o, Edge):
            return o
        elif isinstance(o, Mapping):
            return cls(**o)
        else:
            return Edge(values=o)

    @property
    def flow(self):
        return self.values

    @flow.setter
    def flow(self, values):
        self.values = values

    @property
    def input(self):
        return self.label.input

    @input.setter
    def input(self, i):
        old_input = self.input
        self.label = Edge.Label(i, self.label.output)
        if old_input is None and i is not None and self.output is not None:
            del self._delay_registration_
            self.register()
            i.outputs[self.output] = self

    @property
    def output(self):
        return self.label.output

    @output.setter
    def output(self, o):
        old_output = self.output
        self.label = Edge.Label(self.label.input, o)
        if old_output is None and o is not None and self.input is not None:
            del self._delay_registration_
            self.register()
            o.inputs[self.input] = self


class Bus(Node):
    pass


class Component(Node):
    pass


class Sink(Component):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)


class Source(Component):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)


class Transformer(Component):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)


@contextmanager
def registry_changed_to(r):
    """
    Override registry during execution of a block and restore it afterwards.
    """
    backup = Node.registry
    Node.registry = r
    yield
    Node.registry = backup


def temporarily_modifies_registry(f):
    """Decorator that disables `Node` registration during `f`'s execution.

    It does so by setting `Node.registry` to `None` while `f` is executing, so
    `f` can freely set `Node.registry` to something else. The registration's
    original value is restored afterwards.
    """

    def result(*xs, **ks):
        with registry_changed_to(None):
            return f(*xs, **ks)

    return result


1			# -- coding: utf-8 --
2
3			"""This package (along with its subpackages) contains the classes used to model
4			energy systems. An energy system is modelled as a graph/network of entities
5			with very specific constraints on which types of entities are allowed to be
6			connected.
7
8			This file is part of project oemof (github.com/oemof/oemof). It's copyrighted
9			by the contributors recorded in the version control history of the file,
10			available from its original location oemof/oemof/network.py
11
12			SPDX-FileCopyrightText: Stephan Günther <>
13			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
14			SPDX-FileCopyrightText: Simon Hilpert <>
15			SPDX-FileCopyrightText: Cord Kaldemeyer <>
16			SPDX-FileCopyrightText: Patrik Schönfeldt <[email protected]>
17
18			SPDX-License-Identifier: MIT
19			"""
20
21			import warnings
22			from collections import UserDict as UD
23			from collections import namedtuple as NT
24			from collections.abc import Mapping
25			from collections.abc import MutableMapping as MM
26			from contextlib import contextmanager
27			from functools import total_ordering
28
29			# TODO:
30			#
31			# * Only allow setting a Node's label if `_delay_registration_` is active
32			# and/or the node is not yet registered.
33			# * Only allow setting an Edge's input/output if it is None
34			# * Document the `register` method. Maybe also document the
35			# `_delay_registration_` attribute and make it official. This could also be
36			# a good chance to finally use `blinker` to put an event on
37			# `_delay_registration_` for deletion/assignment to trigger registration.
38			# I always had the hunch that using blinker could help to straighten out
39			# that delayed auto registration hack via partial functions. Maybe this
40			# could be a good starting point for this.
41			# * Finally get rid of `Entity`.
42			#
43
44
45			class Inputs(MM):
46			"""A special helper to map `n1.inputs[n2]` to `n2.outputs[n1]`."""
47
48			def __init__(self, target):
49			self.target = target
50
51			def __getitem__(self, key):
52			return key.outputs.__getitem__(self.target)
53
54			def __delitem__(self, key):
55			return key.outputs.__delitem__(self.target)
56
57			def __setitem__(self, key, value):
58			return key.outputs.__setitem__(self.target, value)
59
60			def __iter__(self):
61			return iter(self.target._in_edges)
62
63			def __len__(self):
64			return self.target._in_edges.__len__()
65
66			def __repr__(self):
67			return repr(
68			"<{0.__module__}.{0.__name__}: {1!r}>".format(
69			type(self), dict(self)
70			)
71			)
72
73
74			class Outputs(UD):
75			"""
76			Helper that intercepts modifications to update `Inputs` symmetrically.
77			"""
78
79			def __init__(self, source):
80			self.source = source
81			super().__init__()
82
83			def __delitem__(self, key):
84			key._in_edges.remove(self.source)
85			return super().__delitem__(key)
86
87			def __setitem__(self, key, value):
88			key._in_edges.add(self.source)
89			return super().__setitem__(key, value)
90
91
92			class Metaclass(type):
93			"""The metaclass for objects in an oemof energy system."""
94
95			@property
96			def registry(cls):
97			warnings.warn(cls.registry_warning)
98			return cls._registry
99
100			@registry.setter
101			def registry(cls, registry):
102			warnings.warn(cls.registry_warning)
103			cls._registry = registry
104
105
106			@total_ordering
107			class Node(metaclass=Metaclass):
108			"""Represents a Node in an energy system graph.
109
110			Abstract superclass of the two general types of nodes of an energy system
111			graph, collecting attributes and operations common to all types of nodes.
112			Users should neither instantiate nor subclass this, but use
113			:class:`Component`, :class:`Bus`, :class:`Edge` or one of their subclasses
114			instead.
115
116			.. role:: python(code)
117			:language: python
118
119			Parameters
120			----------
121			label: `hashable`, optional
122			Used as the string representation of this node. If this parameter is
123			not an instance of :class:`str` it will be converted to a string and
124			the result will be used as this node's :attr:`label`, which should be
125			unique with respect to the other nodes in the energy system graph this
126			node belongs to. If this parameter is not supplied, the string
127			representation of this node will instead be generated based on this
128			nodes `class` and `id`.
129			inputs: list or dict, optional
130			Either a list of this nodes' input nodes or a dictionary mapping input
131			nodes to corresponding inflows (i.e. input values).
132			outputs: list or dict, optional
133			Either a list of this nodes' output nodes or a dictionary mapping
134			output nodes to corresponding outflows (i.e. output values).
135
136			Attributes
137			----------
138			__slots__: str or iterable of str
139			See the Python documentation on `__slots__
140			<https://docs.python.org/3/reference/datamodel.html#slots>`_ for more
141			information.
142			"""
143
144			registry_warning = FutureWarning(
145			"\nAutomatic registration of `Node`s is deprecated in favour of\n"
146			"explicitly adding `Node`s to an `EnergySystem` via "
147			"`EnergySystem.add`.\n"
148			"This feature, i.e. the `Node.registry` attribute and functionality\n"
149			"pertaining to it, will be removed in future versions.\n"
150			)
151
152			_registry = None
153			__slots__ = ["_label", "_in_edges", "_inputs", "_outputs"]
154
155			def __init__(self, args, *kwargs):
156			args = list(args)
157			args.reverse
158			self._inputs = Inputs(self)
159			self._outputs = Outputs(self)
160			for optional in ["label"]:
161			if optional in kwargs:
162			if args:
163			raise (
164			TypeError(
165			(
166			"{}.__init__()\n"
167			" got multiple values for argument '{}'"
168			).format(type(self), optional)
169			)
170			)
171			setattr(self, "_" + optional, kwargs[optional])
172			else:
173			if args:
174			setattr(self, "_" + optional, args.pop())
175			self._in_edges = set()
176			for i in kwargs.get("inputs", {}):
177			assert isinstance(i, Node), (
178			"\n\nInput\n\n {!r}\n\nof\n\n {!r}\n\n"
179			"not an instance of Node, but of {}."
180			).format(i, self, type(i))
181			self._in_edges.add(i)
182			try:
183			flow = kwargs["inputs"].get(i)
184			except AttributeError:
185			flow = None
186			edge = globals()["Edge"].from_object(flow)
187			edge.input = i
188			edge.output = self
189			for o in kwargs.get("outputs", {}):
190			assert isinstance(o, Node), (
191			"\n\nOutput\n\n {!r}\n\nof\n\n {!r}\n\n"
192			"not an instance of Node, but of {}."
193			).format(o, self, type(o))
194			try:
195			flow = kwargs["outputs"].get(o)
196			except AttributeError:
197			flow = None
198			edge = globals()["Edge"].from_object(flow)
199			edge.input = self
200			edge.output = o
201
202			self.register()
203			"""
204			This could be slightly more efficient than the loops above, but doesn't
205			play well with the assertions:
206
207			inputs = kwargs.get('inputs', {})
208			self.in_edges = {
209			Edge(input=i, output=self,
210			flow=None if not isinstance(inputs, MM) else inputs[i])
211			for i in inputs}
212
213			outputs = kwargs.get('outputs', {})
214			self.out_edges = {
215			Edge(input=self, output=o,
216			flow=None if not isinstance(outputs, MM) else outputs[o])
217			for o in outputs}
218			self.edges = self.in_edges.union(self.out_edges)
219			"""
220
221			def register(self):
222			with warnings.catch_warnings():
223			warnings.simplefilter("ignore")
224			registry = __class__.registry
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-12-16 12:53 UTC by Report Bug Copy Issue Report The variable `__class__` does not seem to be defined. Loading history...
225
226			if registry is not None and not getattr(
227			self, "_delay_registration_", False
228			):
229			__class__.registry.add(self)
230
231			def __eq__(self, other):
232			return id(self) == id(other)
233
234			def __lt__(self, other):
235			return str(self) < str(other)
236
237			def __hash__(self):
238			return hash(self.label)
239
240			def __str__(self):
241			return str(self.label)
242
243			def __repr__(self):
244			return repr(
245			"<{0.__module__}.{0.__name__}: {1!r}>".format(
246			type(self), self.label
247			)
248			)
249
250			@property
251			def label(self):
252			"""object :
253			If this node was given a `label` on construction, this
254			attribute holds the actual object passed as a parameter. Otherwise
255			:py:`node.label` is a synonym for :py:`str(node)`.
256			"""
257			return (
258			self._label
259			if hasattr(self, "_label")
260			else "<{} #0x{:x}>".format(type(self).__name__, id(self))
261			)
262
263			@label.setter
264			def label(self, label):
265			self._label = label
266
267			@property
268			def inputs(self):
269			"""dict:
270			Dictionary mapping input :class:`Nodes <Node>` :obj:`n` to
271			:class:`Edge`s from :obj:`n` into :obj:`self`.
272			If :obj:`self` is an :class:`Edge`, returns a dict containing the
273			:class:`Edge`'s single input node as the key and the flow as the value.
274			"""
275			return self._inputs
276
277			@property
278			def outputs(self):
279			"""dict:
280			Dictionary mapping output :class:`Nodes <Node>` :obj:`n` to
281			:class:`Edges` from :obj:`self` into :obj:`n`.
282			If :obj:`self` is an :class:`Edge`, returns a dict containing the
283			:class:`Edge`'s single output node as the key and the flow as the
284			value.
285			"""
286			return self._outputs
287
288
289			EdgeLabel = NT("EdgeLabel", ["input", "output"])
290
291
292			class Edge(Node):
293			"""
294			:class:`Bus`es/:class:`Component`s are always connected by an
295			:class:`Edge`.
296
297			:class:`Edge`s connect a single non-:class:`Edge` Node with another. They
298			are directed and have a (sequence of) value(s) attached to them so they can
299			be used to represent a flow from a source/an input to a target/an output.
300
301			Parameters
302			----------
303			input, output: :class:`Bus` or :class:`Component`, optional
304			flow, values: object, optional
305			The (list of) object(s) representing the values flowing from this
306			edge's input into its output. Note that these two names are aliases of
307			each other, so `flow` and `values` are mutually exclusive.
308
309			Note that all of these parameters are also set as attributes with the same
310			name.
311			"""
312
313			Label = EdgeLabel
314
315			def __init__(
316			self, input=None, output=None, flow=None, values=None, **kwargs
317			):
318			if flow is not None and values is not None:
319			raise ValueError(
320			"\n\n`Edge`'s `flow` and `values` keyword arguments are "
321			"aliases of each other,\nso they're mutually exclusive.\n"
322			"You supplied:\n"
323			+ " `flow` : {}\n".format(flow)
324			+ " `values`: {}\n".format(values)
325			+ "Choose one."
326			)
327			if input is None or output is None:
328			self._delay_registration_ = True
329			super().__init__(label=Edge.Label(input, output))
330			self.values = values if values is not None else flow
331			if input is not None and output is not None:
332			input.outputs[output] = self
333
334			@classmethod
335			def from_object(cls, o):
336			"""Creates an `Edge` instance from a single object.
337
338			This method inspects its argument and does something different
339			depending on various cases:
340
341			* If `o` is an instance of `Edge`, `o` is returned unchanged.
342			* If `o` is a `Mapping`, the instance is created by calling
343			`cls(**o)`,
344			* In all other cases, `o` will be used as the `values` keyword
345			argument to `Edge`s constructor.
346			"""
347			if isinstance(o, Edge):
348			return o
349			elif isinstance(o, Mapping):
350			return cls(**o)
351			else:
352			return Edge(values=o)
353
354			@property
355			def flow(self):
356			return self.values
357
358			@flow.setter
359			def flow(self, values):
360			self.values = values
361
362			@property
363			def input(self):
364			return self.label.input
365
366			@input.setter
367			def input(self, i):
368			old_input = self.input
369			self.label = Edge.Label(i, self.label.output)
370			if old_input is None and i is not None and self.output is not None:
371			del self._delay_registration_
372			self.register()
373			i.outputs[self.output] = self
374
375			@property
376			def output(self):
377			return self.label.output
378
379			@output.setter
380			def output(self, o):
381			old_output = self.output
382			self.label = Edge.Label(self.label.input, o)
383			if old_output is None and o is not None and self.input is not None:
384			del self._delay_registration_
385			self.register()
386			o.inputs[self.input] = self
387
388
389			class Bus(Node):
390			pass
391
392
393			class Component(Node):
394			pass
395
396
397			class Sink(Component):
398			def __init__(self, args, *kwargs):
399			super().__init__(args, *kwargs)
400
401
402			class Source(Component):
403			def __init__(self, args, *kwargs):
404			super().__init__(args, *kwargs)
405
406
407			class Transformer(Component):
408			def __init__(self, args, *kwargs):
409			super().__init__(args, *kwargs)
410
411
412			@contextmanager
413			def registry_changed_to(r):
414			"""
415			Override registry during execution of a block and restore it afterwards.
416			"""
417			backup = Node.registry
418			Node.registry = r
419			yield
420			Node.registry = backup
421
422
423			def temporarily_modifies_registry(f):
424			"""Decorator that disables `Node` registration during `f`'s execution.
425
426			It does so by setting `Node.registry` to `None` while `f` is executing, so
427			`f` can freely set `Node.registry` to something else. The registration's
428			original value is restored afterwards.
429			"""
430
431			def result(xs, *ks):
432			with registry_changed_to(None):
433			return f(xs, *ks)
434
435			return result
436

oemof / oemof-network

network.network.Transformer.__init__() A last analyzed 2022-04-13 19:53 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

network.network.Transformer.init() A
last analyzed 2022-04-13 19:53 UTC