network.network.Inputs.__delitem__()   A
last analyzed

Complexity

Conditions 1

Size

Total Lines 2
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 2
nop 2
dl 0
loc 2
rs 10
c 0
b 0
f 0
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 by
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