network.groupings.Nodes.merge() - Code Metrics - Inspection of "Adhere to graph language" - oemof/oemof-network - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — dev (#14)

by Patrik

created 2022-02-17 15:55 UTC

network.groupings.Nodes.merge() A

↳ Parent: network.groupings

Complexity

Conditions

Size

Total Lines	6
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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

""" All you need to create groups of stuff in your energy system.

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

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

SPDX-License-Identifier: MIT
"""

from collections.abc import Hashable
from collections.abc import Iterable
from collections.abc import Mapping
from collections.abc import MutableMapping as MuMa
from itertools import chain
from itertools import filterfalse

from oemof.network.network import Edge

# TODO: Update docstrings.
#
#   * Make them easier to understand.
#   * Update them to use nodes instead of entities.
#


class Grouping:
    """
    Used to aggregate :class:`entities <oemof.core.network.Entity>` in an
    :class:`energy system <oemof.core.energy_system.EnergySystem>` into
    :attr:`groups <oemof.core.energy_system.EnergySystem.groups>`.

    The way :class:`Groupings <Grouping>` work is that each :class:`Grouping`
    :obj:`g` of an energy system is called whenever an :class:`entity
    <oemof.core.network.Entity>` is added to the energy system (and for each
    :class:`entity <oemof.core.network.Entity>` already present, if the energy
    system is created with existing enties).
    The call :obj:`g(e, groups)`, where :obj:`e` is an :class:`entity
    <oemof.core.network.Entity>` and :attr:`groups
    <oemof.core.energy_system.EnergySystem.groups>` is a dictionary mapping
    group keys to groups, then uses the three functions :meth:`key
    <Grouping.key>`, :meth:`value <Grouping.value>` and :meth:`merge
    <Grouping.merge>` in the following way:

        - :meth:`key(e) <Grouping.key>` is called to obtain a key :obj:`k`
          under which the group should be stored,
        - :meth:`value(e) <Grouping.value>` is called to obtain a value
          :obj:`v` (the actual group) to store under :obj:`k`,
        - if you supplied a :func:`filter` argument, :obj:`v` is
          :func:`filtered <builtins.filter>` using that function,
        - otherwise, if there is not yet anything stored under
          :obj:`groups[k]`, :obj:`groups[k]` is set to :obj:`v`. Otherwise
          :meth:`merge <Grouping.merge>` is used to figure out how to merge
          :obj:`v` into the old value of :obj:`groups[k]`, i.e.
          :obj:`groups[k]` is set to :meth:`merge(v, groups[k])
          <Grouping.merge>`.

    Instead of trying to use this class directly, have a look at its
    subclasses, like :class:`Entities`, which should cater for most use cases.

    Notes
    -----

    When overriding methods using any of the constructor parameters, you don't
    have access to :obj:`self` in the corresponding function. If you need
    access to :obj:`self`, subclass :class:`Grouping` and override the methods
    in the subclass.

    A :class:`Grouping` may be called more than once on the same object
    :obj:`e`, so one should make sure that user defined :class:`Grouping`
    :obj:`g` is idempotent, i.e. :obj:`g(e, g(e, d)) == g(e, d)`.

    Parameters
    ----------

    key: callable or hashable

        Specifies (if not callable) or extracts (if callable) a :meth:`key
        <Grouping.key>` for each :class:`entity <oemof.core.network.Entity>` of
        the :class:`energy system <oemof.core.energy_system.EnergySystem>`.

    constant_key: hashable (optional)

        Specifies a constant :meth:`key <Grouping.key>`. Keys specified using
        this parameter are not called but taken as is.

    value: callable, optional

        Overrides the default behaviour of :meth:`value <Grouping.value>`.

    filter: callable, optional

        If supplied, whatever is returned by :meth:`value` is :func:`filtered
        <builtins.filter>` through this. Mostly useful in conjunction with
        static (i.e. non-callable) :meth:`keys <key>`.
        See :meth:`filter` for more details.

    merge: callable, optional

        Overrides the default behaviour of :meth:`merge <Grouping.merge>`.

    """

    def __init__(self, key=None, constant_key=None, filter=None, **kwargs):
        if key and constant_key:
            raise TypeError(
                "Grouping arguments `key` and `constant_key` are "
                + " mutually exclusive."
            )
        if constant_key:
            self.key = lambda _: constant_key
        elif key:
            self.key = key
        else:
            raise TypeError(
                "Grouping constructor missing required argument: "
                + "one of `key` or `constant_key`."
            )
        self.filter = filter
        for kw in ["value", "merge", "filter"]:
            if kw in kwargs:
                setattr(self, kw, kwargs[kw])

    def key(self, node):
        """Obtain a key under which to store the group.

        You have to supply this method yourself using the :obj:`key` parameter
        when creating :class:`Grouping` instances.

        Called for every :class:`node <oemof.core.network.Entity>` of the energy
        system. Expected to return the key (i.e. a valid :class:`hashable`)
        under which the group :meth:`value(node) <Grouping.value>` will be
        stored. If it should be added to more than one group, return a
        :class:`list` (or any other non-:class:`hashable <Hashable>`,
        :class:`iterable`) containing the group keys.

        Return :obj:`None` if you don't want to store :obj:`e` in a group.
        """
        raise NotImplementedError(
            "\n\n"
            "There is no default implementation for `Groupings.key`.\n"
            "Congratulations, you managed to execute supposedly "
            "unreachable code.\n"
            "Please let us know by filing a bug at:\n\n    "
            "https://github.com/oemof/oemof/issues\n"
        )

    def value(self, e):
        """Generate the group obtained from :obj:`e`.

        This methd returns the actual group obtained from :obj:`e`. Like
        :meth:`key <Grouping.key>`, it is called for every :obj:`e` in the
        energy system. If there is no group stored under :meth:`key(e)
        <Grouping.key>`, :obj:`groups[key(e)]` is set to :meth:`value(e)
        <Grouping.value>`. Otherwise :meth:`merge(value(e), groups[key(e)])
        <Grouping.merge>` is called.

        The default returns the :class:`entity <oemof.core.network.Entity>`
        itself.
        """
        return e

    def merge(self, new, old):
        """Merge a known :obj:`old` group with a :obj:`new` one.

        This method is called if there is already a value stored under
        :obj:`group[key(e)]`. In that case, :meth:`merge(value(e),
        group[key(e)]) <Grouping.merge>` is called and should return the new
        group to store under :meth:`key(e) <Grouping.key>`.

        The default behaviour is to raise an error if :obj:`new` and :obj:`old`
        are not identical.
        """
        if old is new:
            return old
        raise ValueError(
            "\nGrouping \n  "
            "{}:{}\nand\n  {}:{}\ncollides.\n".format(
                id(old), old, id(new), new
            )
            + "Possibly duplicate uids/labels?"
        )

    def filter(self, group):
        """
        :func:`Filter <builtins.filter>` the group returned by :meth:`value`
        before storing it.

        Should return a boolean value. If the :obj:`group` returned by
        :meth:`value` is :class:`iterable <collections.abc.Iterable>`, this
        function is used (via Python's :func:`builtin filter
        <builtins.filter>`) to select the values which should be retained in
        :obj:`group`. If :obj:`group` is not :class:`iterable
        <collections.abc.Iterable>`, it is simply called on :obj:`group` itself
        and the return value decides whether :obj:`group` is stored
        (:obj:`True`) or not (:obj:`False`).

        """
        raise NotImplementedError(
            "\n\n"
            "`Groupings.filter` called without being overridden.\n"
            "Congratulations, you managed to execute supposedly "
            "unreachable code.\n"
            "Please let us know by filing a bug at:\n\n    "
            "https://github.com/oemof/oemof/issues\n"
        )

    def __call__(self, e, d):
        k = self.key(e) if callable(self.key) else self.key
        if k is None:
            return
        v = self.value(e)
        if isinstance(v, MuMa):
            for k in list(filterfalse(self.filter, v)):
                v.pop(k)
        elif isinstance(v, Mapping):
            v = type(v)(dict((k, v[k]) for k in filter(self.filter, v)))
        elif isinstance(v, Iterable):
            v = type(v)(filter(self.filter, v))
        elif self.filter and not self.filter(v):
            return
        if not v:
            return
        for group in (
            k
            if (isinstance(k, Iterable) and not isinstance(k, Hashable))
            else [k]
        ):
            d[group] = self.merge(v, d[group]) if group in d else v


class Entities(Grouping):
    """
    Specialises :class:`Grouping` to group :class:`nodes <oemof.network.Entity>`
    into :class:`sets <set>`.
    """

    def value(self, e):
        """
        Returns a :class:`set` containing only :obj:`e`, so groups are
        :class:`sets <set>` of :class:`node <oemof.network.Entity>`.
        """
        return {e}

    def merge(self, new, old):
        """
        :meth:`Updates <set.update>` :obj:`old` to be the union of :obj:`old`
        and :obj:`new`.
        """
        return old.union(new)


class Flows(Entities):
    """
    Specialises :class:`Grouping` to group the flows connected to :class:`nodes
    <oemof.network.Entity>` into :class:`sets <set>`.
    Note that this specifically means that the :meth:`key <Flows.key>`, and
    :meth:`value <Flows.value>` functions act on a set of flows.
    """

    def value(self, flows):
        """
        Returns a :class:`set` containing only :obj:`flows`, so groups are
        :class:`sets <set>` of flows.
        """
        return set(flows)

    def __call__(self, n, d):
        flows = (
            {n}
            if isinstance(n, Edge)
            else set(chain(n.outputs.values(), n.inputs.values()))
        )
        super().__call__(flows, d)


class FlowsWithNodes(Entities):
    """
    Specialises :class:`Grouping` to act on the flows connected to
    :class:`nodes <oemof.network.Entity>` and create :class:`sets <set>` of
    :obj:`(source, target, flow)` tuples.
    Note that this specifically means that the :meth:`key <Flows.key>`, and
    :meth:`value <Flows.value>` functions act on sets like these.
    """

    def value(self, tuples):
        """
        Returns a :class:`set` containing only :obj:`tuples`, so groups are
        :class:`sets <set>` of :obj:`tuples`.
        """
        return set(tuples)

    def __call__(self, n, d):
        tuples = (
            {(n.input, n.output, n)}
            if isinstance(n, Edge)
            else set(
                chain(
                    ((n, t, f) for (t, f) in n.outputs.items()),
                    ((s, n, f) for (s, f) in n.inputs.items()),
                )
            )
        )
        super().__call__(tuples, d)


def _uid_or_str(node_or_entity):
    """Helper function to support the transition from `Entitie`s to `Entity`s."""
    return (
        node_or_entity.uid
        if hasattr(node_or_entity, "uid")
        else str(node_or_entity)
    )


DEFAULT = Grouping(_uid_or_str)
""" The default :class:`Grouping`.

This one is always present in an :class:`energy system
<oemof.core.energy_system.EnergySystem>`. It stores every :class:`entity
<oemof.core.network.Entity>` under its :attr:`uid
<oemof.core.network.Entity.uid>` and raises an error if another :class:`entity
<oemof.core.network.Entity>` with the same :attr:`uid
<oemof.core.network.Entity.uid>` get's added to the :class:`energy system
<oemof.core.energy_system.EnergySystem>`.
"""


1			# -- coding: utf-8 --
2
3			""" All you need to create groups of stuff in your energy system.
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/groupings.py
8
9			SPDX-FileCopyrightText: Stephan Günther <>
10			SPDX-FileCopyrightText: Uwe Krien <[email protected]>
11
12			SPDX-License-Identifier: MIT
13			"""
14
15			from collections.abc import Hashable
16			from collections.abc import Iterable
17			from collections.abc import Mapping
18			from collections.abc import MutableMapping as MuMa
19			from itertools import chain
20			from itertools import filterfalse
21
22			from oemof.network.network import Edge
23
24			# TODO: Update docstrings.
25			#
26			# * Make them easier to understand.
27			# * Update them to use nodes instead of entities.
28			#
29
30
31			class Grouping:
32			"""
33			Used to aggregate :class:`entities <oemof.core.network.Entity>` in an
34			:class:`energy system <oemof.core.energy_system.EnergySystem>` into
35			:attr:`groups <oemof.core.energy_system.EnergySystem.groups>`.
36
37			The way :class:`Groupings <Grouping>` work is that each :class:`Grouping`
38			:obj:`g` of an energy system is called whenever an :class:`entity
39			<oemof.core.network.Entity>` is added to the energy system (and for each
40			:class:`entity <oemof.core.network.Entity>` already present, if the energy
41			system is created with existing enties).
42			The call :obj:`g(e, groups)`, where :obj:`e` is an :class:`entity
43			<oemof.core.network.Entity>` and :attr:`groups
44			<oemof.core.energy_system.EnergySystem.groups>` is a dictionary mapping
45			group keys to groups, then uses the three functions :meth:`key
46			<Grouping.key>`, :meth:`value <Grouping.value>` and :meth:`merge
47			<Grouping.merge>` in the following way:
48
49			- :meth:`key(e) <Grouping.key>` is called to obtain a key :obj:`k`
50			under which the group should be stored,
51			- :meth:`value(e) <Grouping.value>` is called to obtain a value
52			:obj:`v` (the actual group) to store under :obj:`k`,
53			- if you supplied a :func:`filter` argument, :obj:`v` is
54			:func:`filtered <builtins.filter>` using that function,
55			- otherwise, if there is not yet anything stored under
56			:obj:`groups[k]`, :obj:`groups[k]` is set to :obj:`v`. Otherwise
57			:meth:`merge <Grouping.merge>` is used to figure out how to merge
58			:obj:`v` into the old value of :obj:`groups[k]`, i.e.
59			:obj:`groups[k]` is set to :meth:`merge(v, groups[k])
60			<Grouping.merge>`.
61
62			Instead of trying to use this class directly, have a look at its
63			subclasses, like :class:`Entities`, which should cater for most use cases.
64
65			Notes
66			-----
67
68			When overriding methods using any of the constructor parameters, you don't
69			have access to :obj:`self` in the corresponding function. If you need
70			access to :obj:`self`, subclass :class:`Grouping` and override the methods
71			in the subclass.
72
73			A :class:`Grouping` may be called more than once on the same object
74			:obj:`e`, so one should make sure that user defined :class:`Grouping`
75			:obj:`g` is idempotent, i.e. :obj:`g(e, g(e, d)) == g(e, d)`.
76
77			Parameters
78			----------
79
80			key: callable or hashable
81
82			Specifies (if not callable) or extracts (if callable) a :meth:`key
83			<Grouping.key>` for each :class:`entity <oemof.core.network.Entity>` of
84			the :class:`energy system <oemof.core.energy_system.EnergySystem>`.
85
86			constant_key: hashable (optional)
87
88			Specifies a constant :meth:`key <Grouping.key>`. Keys specified using
89			this parameter are not called but taken as is.
90
91			value: callable, optional
92
93			Overrides the default behaviour of :meth:`value <Grouping.value>`.
94
95			filter: callable, optional
96
97			If supplied, whatever is returned by :meth:`value` is :func:`filtered
98			<builtins.filter>` through this. Mostly useful in conjunction with
99			static (i.e. non-callable) :meth:`keys <key>`.
100			See :meth:`filter` for more details.
101
102			merge: callable, optional
103
104			Overrides the default behaviour of :meth:`merge <Grouping.merge>`.
105
106			"""
107
108			def __init__(self, key=None, constant_key=None, filter=None, **kwargs):
109			if key and constant_key:
110			raise TypeError(
111			"Grouping arguments `key` and `constant_key` are "
112			+ " mutually exclusive."
113			)
114			if constant_key:
115			self.key = lambda _: constant_key
116			elif key:
117			self.key = key
118			else:
119			raise TypeError(
120			"Grouping constructor missing required argument: "
121			+ "one of `key` or `constant_key`."
122			)
123			self.filter = filter
124			for kw in ["value", "merge", "filter"]:
125			if kw in kwargs:
126			setattr(self, kw, kwargs[kw])
127
128			def key(self, node):
129			"""Obtain a key under which to store the group.
130
131			You have to supply this method yourself using the :obj:`key` parameter
132			when creating :class:`Grouping` instances.
133
134			Called for every :class:`node <oemof.core.network.Entity>` of the energy
135			system. Expected to return the key (i.e. a valid :class:`hashable`)
136			under which the group :meth:`value(node) <Grouping.value>` will be
137			stored. If it should be added to more than one group, return a
138			:class:`list` (or any other non-:class:`hashable <Hashable>`,
139			:class:`iterable`) containing the group keys.
140
141			Return :obj:`None` if you don't want to store :obj:`e` in a group.
142			"""
143			raise NotImplementedError(
144			"\n\n"
145			"There is no default implementation for `Groupings.key`.\n"
146			"Congratulations, you managed to execute supposedly "
147			"unreachable code.\n"
148			"Please let us know by filing a bug at:\n\n "
149			"https://github.com/oemof/oemof/issues\n"
150			)
151
152			def value(self, e):
153			"""Generate the group obtained from :obj:`e`.
154
155			This methd returns the actual group obtained from :obj:`e`. Like
156			:meth:`key <Grouping.key>`, it is called for every :obj:`e` in the
157			energy system. If there is no group stored under :meth:`key(e)
158			<Grouping.key>`, :obj:`groups[key(e)]` is set to :meth:`value(e)
159			<Grouping.value>`. Otherwise :meth:`merge(value(e), groups[key(e)])
160			<Grouping.merge>` is called.
161
162			The default returns the :class:`entity <oemof.core.network.Entity>`
163			itself.
164			"""
165			return e
166
167			def merge(self, new, old):
168			"""Merge a known :obj:`old` group with a :obj:`new` one.
169
170			This method is called if there is already a value stored under
171			:obj:`group[key(e)]`. In that case, :meth:`merge(value(e),
172			group[key(e)]) <Grouping.merge>` is called and should return the new
173			group to store under :meth:`key(e) <Grouping.key>`.
174
175			The default behaviour is to raise an error if :obj:`new` and :obj:`old`
176			are not identical.
177			"""
178			if old is new:
179			return old
180			raise ValueError(
181			"\nGrouping \n "
182			"{}:{}\nand\n {}:{}\ncollides.\n".format(
183			id(old), old, id(new), new
184			)
185			+ "Possibly duplicate uids/labels?"
186			)
187
188			def filter(self, group):
189			"""
190			:func:`Filter <builtins.filter>` the group returned by :meth:`value`
191			before storing it.
192
193			Should return a boolean value. If the :obj:`group` returned by
194			:meth:`value` is :class:`iterable <collections.abc.Iterable>`, this
195			function is used (via Python's :func:`builtin filter
196			<builtins.filter>`) to select the values which should be retained in
197			:obj:`group`. If :obj:`group` is not :class:`iterable
198			<collections.abc.Iterable>`, it is simply called on :obj:`group` itself
199			and the return value decides whether :obj:`group` is stored
200			(:obj:`True`) or not (:obj:`False`).
201
202			"""
203			raise NotImplementedError(
204			"\n\n"
205			"`Groupings.filter` called without being overridden.\n"
206			"Congratulations, you managed to execute supposedly "
207			"unreachable code.\n"
208			"Please let us know by filing a bug at:\n\n "
209			"https://github.com/oemof/oemof/issues\n"
210			)
211
212			def __call__(self, e, d):
213			k = self.key(e) if callable(self.key) else self.key
214			if k is None:
215			return
216			v = self.value(e)
217			if isinstance(v, MuMa):
218			for k in list(filterfalse(self.filter, v)):
219			v.pop(k)
220			elif isinstance(v, Mapping):
221			v = type(v)(dict((k, v[k]) for k in filter(self.filter, v)))
222			elif isinstance(v, Iterable):
223			v = type(v)(filter(self.filter, v))
224			elif self.filter and not self.filter(v):
225			return
226			if not v:
227			return
228			for group in (
229			k
230			if (isinstance(k, Iterable) and not isinstance(k, Hashable))
231			else [k]
232			):
233			d[group] = self.merge(v, d[group]) if group in d else v
234
235
236			class Entities(Grouping):
237			"""
238			Specialises :class:`Grouping` to group :class:`nodes <oemof.network.Entity>`
239			into :class:`sets <set>`.
240			"""
241
242			def value(self, e):
243			"""
244			Returns a :class:`set` containing only :obj:`e`, so groups are
245			:class:`sets <set>` of :class:`node <oemof.network.Entity>`.
246			"""
247			return {e}
248
249			def merge(self, new, old):
250			"""
251			:meth:`Updates <set.update>` :obj:`old` to be the union of :obj:`old`
252			and :obj:`new`.
253			"""
254			return old.union(new)
255
256
257			class Flows(Entities):
258			"""
259			Specialises :class:`Grouping` to group the flows connected to :class:`nodes
260			<oemof.network.Entity>` into :class:`sets <set>`.
261			Note that this specifically means that the :meth:`key <Flows.key>`, and
262			:meth:`value <Flows.value>` functions act on a set of flows.
263			"""
264
265			def value(self, flows):
266			"""
267			Returns a :class:`set` containing only :obj:`flows`, so groups are
268			:class:`sets <set>` of flows.
269			"""
270			return set(flows)
271
272			def __call__(self, n, d):
273			flows = (
274			{n}
275			if isinstance(n, Edge)
276			else set(chain(n.outputs.values(), n.inputs.values()))
277			)
278			super().__call__(flows, d)
279
280
281			class FlowsWithNodes(Entities):
282			"""
283			Specialises :class:`Grouping` to act on the flows connected to
284			:class:`nodes <oemof.network.Entity>` and create :class:`sets <set>` of
285			:obj:`(source, target, flow)` tuples.
286			Note that this specifically means that the :meth:`key <Flows.key>`, and
287			:meth:`value <Flows.value>` functions act on sets like these.
288			"""
289
290			def value(self, tuples):
291			"""
292			Returns a :class:`set` containing only :obj:`tuples`, so groups are
293			:class:`sets <set>` of :obj:`tuples`.
294			"""
295			return set(tuples)
296
297			def __call__(self, n, d):
298			tuples = (
299			{(n.input, n.output, n)}
300			if isinstance(n, Edge)
301			else set(
302			chain(
303			((n, t, f) for (t, f) in n.outputs.items()),
304			((s, n, f) for (s, f) in n.inputs.items()),
305			)
306			)
307			)
308			super().__call__(tuples, d)
309
310
311			def _uid_or_str(node_or_entity):
312			"""Helper function to support the transition from `Entitie`s to `Entity`s."""
313			return (
314			node_or_entity.uid
315			if hasattr(node_or_entity, "uid")
316			else str(node_or_entity)
317			)
318
319
320			DEFAULT = Grouping(_uid_or_str)
321			""" The default :class:`Grouping`.
322
323			This one is always present in an :class:`energy system
324			<oemof.core.energy_system.EnergySystem>`. It stores every :class:`entity
325			<oemof.core.network.Entity>` under its :attr:`uid
326			<oemof.core.network.Entity.uid>` and raises an error if another :class:`entity
327			<oemof.core.network.Entity>` with the same :attr:`uid
328			<oemof.core.network.Entity.uid>` get's added to the :class:`energy system
329			<oemof.core.energy_system.EnergySystem>`.
330			"""
331

oemof / oemof-network

Pull Request — dev (#14)

network.groupings.Nodes.merge() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like