Source code for hydpy.core.selectiontools

# -*- coding: utf-8 -*-
"""This module implements tools for defining subsets of |Node| and |Element| objects of
large *HydPy* projects, called "selections"."""
# import...
# ...from standard library
from __future__ import annotations
import collections
import copy
import itertools
import types

# ...from site-packages
import black
import networkx

# ...from HydPy
import hydpy
from hydpy.core import devicetools
from hydpy.core import hydpytools
from hydpy.core import importtools
from hydpy.core import modeltools
from hydpy.core import objecttools
from hydpy.core.typingtools import *

ModelTypesArg = Union[modeltools.Model, types.ModuleType, str]


[docs] class Selections: """Collection class for |Selection| objects. You can pass an arbitrary number of |Selection| objects to the constructor of class |Selections|: >>> sel1 = Selection("sel1", ["node1", "node2"], ["element1"]) >>> sel2 = Selection("sel2", ["node1", "node3"], ["element2"]) >>> selections = Selections(sel1, sel2) >>> selections Selections("sel1", "sel2") Also, you can query, add, and remove |Selection| objects via attribute access: >>> selections.sel3 # doctest: +ELLIPSIS Traceback (most recent call last): ... AttributeError: The actual Selections object handles neither a normal attribute \ nor a Selection object called `sel3`... >>> sel3 = Selection("sel3", ["node1", "node4"], ["element3"]) >>> selections.sel3 = sel3 >>> selections.sel3 Selection("sel3", nodes=("node1", "node4"), elements="element3") >>> "sel3" in dir(selections) True >>> del selections.sel3 >>> "sel3" in dir(selections) False >>> del selections.sel3 Traceback (most recent call last): ... AttributeError: The actual Selections object handles neither a normal attribute \ nor a Selection object called `sel3` that could be deleted. Attribute names must be consistent with the `name` attribute of the respective |Selection| object: >>> selections.sel4 = sel3 Traceback (most recent call last): ... ValueError: To avoid inconsistencies when handling Selection objects as \ attributes of a Selections object, attribute name and Selection name must be \ identical. However, for selection `sel3` the given attribute name is `sel4`. You can use item access alternatively: >>> selections["sel4"] Traceback (most recent call last): ... KeyError: 'The actual Selections object does not handle a Selection object called \ `sel4`.' >>> selections["sel4"] = Selection("sel4") >>> selections["sel4"] Selection("sel4", nodes=(), elements=()) >>> del selections["sel4"] >>> del selections["sel4"] Traceback (most recent call last): ... KeyError: 'The actual Selections object does not handle a Selection object called \ `sel4` that could be deleted.' You can ask for the existence of specific |Selection| objects within a |Selections| object both via its name and via the object itself: >>> sel1 in selections True >>> "sel1" in selections True >>> sel3 in selections False >>> "sel3" in selections False Class |Selections| supports both the |iter| and |len| operators: >>> for selection in selections: ... print(selection.name) sel1 sel2 >>> len(selections) 2 For convenience, use the "+", "-", "+=", and "-=" operators to compare and modify |Selections| objects either based on single |Selection| objects or collections of |Selection| objects: >>> larger = selections + sel3 >>> smaller = selections - sel2 >>> sorted(selections.names) ['sel1', 'sel2'] >>> sorted(larger.names) ['sel1', 'sel2', 'sel3'] >>> smaller.names ('sel1',) >>> smaller += larger >>> sorted(smaller.names) ['sel1', 'sel2', 'sel3'] >>> smaller -= sel1, sel2 >>> smaller.names ('sel3',) Note that trying to remove non-existing |Selection| objects does not raise errors: >>> smaller -= sel2 >>> smaller.names ('sel3',) >>> smaller - (sel1, sel2, sel3) Selections() The binary operators do not support types other than the mentioned ones: >>> smaller -= "sel3" Traceback (most recent call last): ... TypeError: Binary operations on Selections objects are defined for other \ Selections objects, single Selection objects, or iterables containing `Selection` \ objects, but the type of the given argument is `str`. >>> smaller -= 1 # doctest: +ELLIPSIS Traceback (most recent call last): ... TypeError: ... is `int`. Use the "==" operator to compare two |Selections| objects: >>> larger == smaller False >>> larger == (smaller + selections) True >>> larger == (sel1, sel2, sel3) False """ def __init__(self, *selections: Selection) -> None: self.__selections: dict[str, Selection] = {} self.add_selections(*selections) @property def names(self) -> tuple[str, ...]: """The names of the actual |Selection| objects. >>> from hydpy import Selection, Selections >>> selections = Selections( ... Selection("sel1", ["node1", "node2"], ["element1"]), ... Selection("sel2", ["node1", "node3"], ["element2"])) >>> sorted(selections.names) ['sel1', 'sel2'] """ return tuple(self.__selections.keys()) @property def nodes(self) -> devicetools.Nodes: """The |Node| objects of all handled |Selection| objects. >>> from hydpy import Selection, Selections >>> selections = Selections( ... Selection("sel1", ["node1", "node2"], ["element1"]), ... Selection("sel2", ["node1", "node3"], ["element2"])) >>> selections.nodes Nodes("node1", "node2", "node3") """ nodes = devicetools.Nodes() for selection in self: nodes += selection.nodes return nodes @property def elements(self) -> devicetools.Elements: """The |Element| objects of all handled |Selection| objects. >>> from hydpy import Selection, Selections >>> selections = Selections( ... Selection("sel1", ["node1"], ["element1"]), ... Selection("sel2", ["node1"], ["element2", "element3"])) >>> selections.elements Elements("element1", "element2", "element3") """ elements = devicetools.Elements() for selection in self: elements += selection.elements return elements @property def complete(self) -> Selection: """An automatically created selection that comprises the nodes and elements of all currently available, user-defined selections. >>> from hydpy import Selection, Selections >>> selections = Selections( ... Selection("sel1", ["node1"], ["element1"]), ... Selection("sel2", ["node1"], ["element2", "element3"])) >>> selections.complete Selection("complete", nodes="node1", elements=("element1", "element2", "element3")) The selection |Selections.complete| is always freshly created and so reflects the current state |Selections| instance: >>> selections.sel1.nodes.add_device("node2") >>> selections.complete Selection("complete", nodes=("node1", "node2"), elements=("element1", "element2", "element3")) Therefore, changing the |Selection| object returned by property |Selections.complete| does neither change the |Selections| object nor subsequentially returned |Selections.complete| selections: >>> selections.complete.nodes.add_device("node3") >>> assert "node3" not in selections.nodes >>> selections.complete Selection("complete", nodes=("node1", "node2"), elements=("element1", "element2", "element3")) Item access is provided: >>> assert selections["complete"] == selections.complete Selection |Selections.complete| is ignored when iterating through the user-defined selections: >>> assert len(selections) == 2 """ return Selection("complete", nodes=self.nodes, elements=self.elements)
[docs] def add_selections(self, *selections: Selection) -> None: """Add the given |Selection| object(s) to the current |Selections| object. >>> from hydpy import Selection, Selections >>> selections = Selections(Selection("sel1", ["node1"], ["element1"])) >>> selections.add_selections( ... Selection("sel2", ["node1"], ["element2", "element3"]), ... Selection("sel3", ["node2"], [])) >>> selections Selections("sel1", "sel2", "sel3") >>> selections.nodes Nodes("node1", "node2") >>> selections.elements Elements("element1", "element2", "element3") |Selections| rejects any |Selection| objects named `complete` to avoid conflicts with the automatically created |Selections.complete| selection: >>> selections.add_selections(Selection("complete")) Traceback (most recent call last): ... ValueError: You cannot assign a selection with the name `complete` to a \ `Selections` object because it would conflict with the selection automatically \ created by its property `Selections.complete`. """ for selection in selections: if selection.name == "complete": raise ValueError( "You cannot assign a selection with the name `complete` to a " "`Selections` object because it would conflict with the selection " "automatically created by its property `Selections.complete`." ) self.__selections[selection.name] = selection
[docs] def remove_selections(self, *selections: Selection) -> None: """Remove the given |Selection| object(s) from the current |Selections| object. >>> from hydpy import Selection, Selections >>> selections = Selections( ... Selection("sel1", ["node1"], ["element1"]), ... Selection("sel2", ["node1"], ["element2", "element3"])) >>> selections.remove_selections( ... Selection("sel3", ["node2"], []), selections["sel1"]) >>> selections Selections("sel2") >>> selections.nodes Nodes("node1") >>> selections.elements Elements("element2", "element3") """ for selection in selections: try: del self[selection.name] except KeyError: pass
[docs] def find(self, device: devicetools.NodeOrElement) -> Selections: """Return all |Selection| objects containing the given |Node| or |Element| object. >>> from hydpy import Elements, Nodes, Selection, Selections >>> nodes = Nodes("n1", "n2", "n3") >>> elements = Elements("e1", "e2") >>> selections = Selections( ... Selection("s1", ["n1", "n2"], ["e1"]), ... Selection("s2", ["n1"])) >>> selections.find(nodes.n1) Selections("s1", "s2") >>> selections.find(nodes.n2) Selections("s1") >>> selections.find(nodes.n3) Selections() >>> selections.find(elements.e1) Selections("s1") >>> selections.find(elements.e2) Selections() """ attr = "nodes" if isinstance(device, devicetools.Node) else "elements" selections = ( selection for selection in self if device in getattr(selection, attr) ) return Selections(*selections)
@overload def query_intersections( self, selection2element: Literal[True] = ... ) -> dict[Selection, dict[Selection, devicetools.Elements]]: ... @overload def query_intersections( self, selection2element: Literal[False] ) -> dict[devicetools.Element, Selections]: ...
[docs] def query_intersections( self, selection2element: bool = True ) -> Union[ dict[Selection, dict[Selection, devicetools.Elements]], dict[devicetools.Element, Selections], ]: """A dictionary covering all cases where one |Element| object is a member of multiple |Selection| objects. The dictionary's structure depends on the value of the optional argument `selection2element`. See method |Selections.print_intersections| for an example. """ if selection2element: intersections: dict[Selection, dict[Selection, devicetools.Elements]] = ( collections.defaultdict(dict) ) for selection1, selection2 in itertools.combinations(self, 2): intersection = selection1.elements.intersection(*selection2.elements) if intersection: intersections[selection1][selection2] = intersection intersections[selection2][selection1] = intersection return dict(intersections) intersections_: dict[devicetools.Element, Selections] = {} for element in self.elements: selections = self.find(element) if len(selections) > 1: intersections_[element] = selections return intersections_
[docs] def print_intersections(self, selection2element: bool = True) -> None: """Print the result of method |Selections.query_intersections|. We use method |Selections.print_intersections| to check if any combination of the following selections handles the same elements. >>> from hydpy import Selection, Selections >>> selections = Selections( ... Selection("s1", nodes="n1",elements=("e1", "e2", "e3")), ... Selection("s2", nodes="n1", elements=("e2", "e3", "e4")), ... Selection("s3", nodes="n1", elements="e3"), ... Selection("s4", nodes="n1", elements=("e5", "e6")), ... ) If we call method |Selections.print_intersections| with argument `selection2element` |True|, we find out which selection intersects with which other and which elements are affected: >>> selections.print_intersections() selection s1 intersects with... ...selection s2 due to the following elements: e2 and e3 ...selection s3 due to the following elements: e3 selection s2 intersects with... ...selection s1 due to the following elements: e2 and e3 ...selection s3 due to the following elements: e3 selection s3 intersects with... ...selection s1 due to the following elements: e3 ...selection s2 due to the following elements: e3 If we call method |Selections.print_intersections| with argument `selection2element` |False|, we find out which element occurs multiple times in which selections: >>> selections.print_intersections(selection2element=False) element e2 is a member of multiple selections: s1 and s2 element e3 is a member of multiple selections: s1, s2, and s3 """ if selection2element: intersections = self.query_intersections(True) for selection1, selection2elements in intersections.items(): print("selection", selection1, "intersects with...") for selection2, elements in selection2elements.items(): print( " ...selection", selection2, "due to the following elements:", objecttools.enumeration(elements.names), ) else: intersections_ = self.query_intersections(False) for element, selections in intersections_.items(): print( "element", element.name, "is a member of multiple selections:", objecttools.enumeration(selections.names), )
def __getattr__(self, key: str) -> Selection: try: return self.__selections[key] except KeyError: raise AttributeError( f"The actual Selections object handles neither a normal attribute nor " f"a Selection object called `{key}`." ) from None def __setattr__(self, name: str, value: object) -> None: if isinstance(value, Selection): self[name] = value else: super().__setattr__(name, value) def __delattr__(self, key: str) -> None: try: del self.__selections[key] except KeyError: raise AttributeError( f"The actual Selections object handles neither a normal attribute nor " f"a Selection object called `{key}` that could be deleted." ) from None def __getitem__(self, key: str) -> Selection: if key == "complete": return self.complete try: return self.__selections[key] except KeyError: raise KeyError( f"The actual Selections object does not handle a Selection object " f"called `{key}`." ) from None def __setitem__(self, key: str, value: Selection) -> None: if key != value.name: raise ValueError( f"To avoid inconsistencies when handling Selection objects as " f"attributes of a Selections object, attribute name and Selection " f"name must be identical. However, for selection `{value.name}` the " f"given attribute name is `{key}`." ) self.add_selections(value) def __delitem__(self, key: str) -> None: try: del self.__selections[key] except KeyError: raise KeyError( f"The actual Selections object does not handle a Selection object " f"called `{key}` that could be deleted." ) from None def __contains__(self, value: Union[str, Selection]) -> bool: if isinstance(value, str): return value in self.names return value in self.__selections.values() def __iter__(self) -> Iterator[Selection]: return iter(self.__selections.values()) def __len__(self) -> int: return len(self.__selections) @staticmethod def __getiterable(value: Mayberable1[Selection]) -> list[Selection]: """Try to convert the given argument to a |list| of |Selection| objects and return it.""" try: return list(objecttools.extract(value, (Selection,))) except TypeError: raise TypeError( f"Binary operations on Selections objects are defined for other " f"Selections objects, single Selection objects, or iterables " f"containing `Selection` objects, but the type of the given argument " f"is `{type(value).__name__}`." ) from None def __add__(self, other: Mayberable1[Selection]) -> Selections: new = copy.copy(self) new.add_selections(*self.__getiterable(other)) return new def __iadd__(self, other: Mayberable1[Selection]) -> Selections: self.add_selections(*self.__getiterable(other)) return self def __sub__(self, other: Mayberable1[Selection]) -> Selections: selections = self.__getiterable(other) new = copy.copy(self) for selection in selections: try: del new[selection.name] except KeyError: pass return new def __isub__(self, other: Mayberable1[Selection]) -> Selections: selections = self.__getiterable(other) for selection in selections: try: del self[selection.name] except KeyError: pass return self def __eq__(self, other: object) -> bool: if isinstance(other, (Selection, Selections, hydpytools.HydPy)): return (self.nodes == self.nodes) and (self.elements == other.elements) return False def __copy__(self) -> Selections: return type(self)(*self.__selections.values()) def __repr__(self) -> str: return self.assignrepr("")
[docs] def assignrepr(self, prefix: str = "") -> str: """Return a |repr| string with a prefixed assignment.""" with objecttools.repr_.preserve_strings(True): options = hydpy.pub.options with options.ellipsis(2, optional=True): prefix = f"{prefix}{type(self).__name__}(" return ( f"{objecttools.assignrepr_values(sorted(self.names), prefix, 70)})" )
def __dir__(self) -> list[str]: return cast(list[str], super().__dir__()) + list(self.names)
[docs] class Selection: """Handles and modifies combinations of |Node| and |Element| objects. In *HydPy*, |Node|, and |Element| objects are the fundamental means to structure projects. However, keeping the overview of huge projects involving thousands of nodes and elements requires additional strategies. One such strategy is to define different instances of class |Selection| for different aspects of a project. Often, a selection contains all nodes and elements of a certain subcatchment or all elements handling certain model types. Selections can be overlapping, meaning, for example, that an element can be part of a subcatchment selection and of model-type selection at the same time. Selections can be written to and read from individual network files, as explained in the documentation on class |NetworkManager|. Read selections are available via the |pub| module. In most application scripts (e.g. for parameter calibration), one performs different operations on the nodes and elements of the different selections (e.g. change parameter "a" and "b" for models of selection "x" and "y", respectively). However, class |Selection| also provides features for creating combinations of |Node| and |Element| objects suitable for different tasks, as explained in the documentation of the respective methods. Here we only show its basic usage with the help of the `HydPy-H-Lahn` example project prepared by function |prepare_full_example_2|: >>> from hydpy.core.testtools import prepare_full_example_2 >>> _, pub, _ = prepare_full_example_2() For example, `HydPy-H-Lahn` defines a `headwaters` selection: >>> pub.selections.headwaters Selection("headwaters", nodes=("dill_assl", "lahn_marb"), elements=("land_dill_assl", "land_lahn_marb")) You can compare this selection with other new or already available selections, with "headwaters < complete" returning |True| meaning that all nodes and elements of the headwater catchments are also part of the entire catchment: >>> from hydpy import Selection >>> test = Selection("test", ... elements=("land_dill_assl", "land_lahn_marb"), ... nodes=("dill_assl", "lahn_marb")) >>> pub.selections.headwaters < test False >>> pub.selections.headwaters <= test True >>> pub.selections.headwaters == test True >>> pub.selections.headwaters != test False >>> pub.selections.headwaters >= test True >>> pub.selections.headwaters > test False >>> pub.selections.headwaters < pub.selections.complete True >>> pub.selections.headwaters <= pub.selections.complete True >>> pub.selections.headwaters == pub.selections.complete False >>> pub.selections.headwaters != pub.selections.complete True >>> pub.selections.headwaters >= pub.selections.complete False >>> pub.selections.headwaters > pub.selections.complete False The |len| operator returns the total number of handled node and element objects: >>> len(test) 4 Use the "+=" and "-=" operators to add or remove nodes and elements: >>> test += pub.selections.complete >>> len(test) 11 >>> test -= pub.selections.complete >>> len(test) 0 Passing a wrong argument to the binary operators results in errors like the following: >>> test += 1 Traceback (most recent call last): ... AttributeError: While trying to add selection `test` with object `1` of type \ `int`, the following error occurred: 'int' object has no attribute 'nodes' >>> test -= pub.selections.complete.nodes.dill_assl Traceback (most recent call last): ... AttributeError: While trying to subtract selection `test` with object `dill_assl` \ of type `Node`, the following error occurred: 'Node' object has no attribute 'nodes' >>> test < "wrong" Traceback (most recent call last): ... AttributeError: While trying to compare selection `test` with object `wrong` of \ type `str`, the following error occurred: 'str' object has no attribute 'nodes' But as usual, checking for equality or inequality returns |False| and |True| for uncomparable objects: >>> test == "wrong" False >>> test != "wrong" True Applying the |str| function only returns the selection name: >>> str(test) 'test' """ name: str """The selection's name.""" nodes: devicetools.Nodes """The explicitly handled |Node| objects (m). |Selection.nodes| does not necessarily contain all nodes to which the elements in |Selection.elements| are linked. """ elements: devicetools.Elements """The explicitly handled |Element| objects. |Selection.elements| does not necessarily contain all nodes to which the elements in |Selection.nodes| are linked. """ def __init__( self, name: str, nodes: devicetools.NodesConstrArg = None, elements: devicetools.ElementsConstrArg = None, ) -> None: self.name = str(name) self.nodes = devicetools.Nodes(nodes).copy() self.elements = devicetools.Elements(elements).copy() def _check_device( self, device: devicetools.TypeNodeElement, type_of_device: str ) -> devicetools.TypeNodeElement: if isinstance(device, devicetools.Node): return self.nodes[device.name] if isinstance(device, devicetools.Element): return self.elements[device.name] raise TypeError( f"Either a `Node` or an `Element` object is required as the " f'"{type_of_device} device", but the given `device` value is of type ' f"`{type(device).__name__}`." )
[docs] def search_upstream( self, device: devicetools.NodeOrElement, name: str = "upstream", inclusive: bool = True, ) -> Selection: """Return the network upstream of the given starting point, including the starting point itself. >>> from hydpy.core.testtools import prepare_full_example_2 >>> hp, pub, _ = prepare_full_example_2() You can pass both |Node| and |Element| objects and, optionally, the name of the newly created |Selection| object: >>> test = pub.selections.complete.copy("test") >>> test.search_upstream(hp.nodes.lahn_leun) Selection("upstream", nodes=("dill_assl", "lahn_leun", "lahn_marb"), elements=("land_dill_assl", "land_lahn_leun", "land_lahn_marb", "stream_dill_assl_lahn_leun", "stream_lahn_marb_lahn_leun")) >>> test.search_upstream(hp.elements.stream_lahn_marb_lahn_leun, "UPSTREAM") Selection("UPSTREAM", nodes=("lahn_leun", "lahn_marb"), elements=("land_lahn_marb", "stream_lahn_marb_lahn_leun")) Method |Selection.search_upstream| generally selects all |Node| objects directly connected to any upstream |Element| object. Set the `inclusive` argument to |False| to circumvent this: >>> test.search_upstream(hp.elements.stream_lahn_marb_lahn_leun, "UPSTREAM", ... False) Selection("UPSTREAM", nodes="lahn_marb", elements=("land_lahn_marb", "stream_lahn_marb_lahn_leun")) Wrong device specifications result in errors like the following: >>> test.search_upstream(1) Traceback (most recent call last): ... TypeError: While trying to determine an upstream network of selection `test`, \ the following error occurred: Either a `Node` or an `Element` object is required as \ the "outlet device", but the given `device` value is of type `int`. >>> pub.selections.headwaters.search_upstream(hp.nodes.lahn_kalk) Traceback (most recent call last): ... KeyError: "While trying to determine an upstream network of selection \ `headwaters`, the following error occurred: 'No node named `lahn_kalk` available.'" Method |Selection.select_upstream| restricts the current selection to the one determined with the method |Selection.search_upstream|: >>> test.select_upstream(hp.nodes.lahn_leun) Selection("test", nodes=("dill_assl", "lahn_leun", "lahn_marb"), elements=("land_dill_assl", "land_lahn_leun", "land_lahn_marb", "stream_dill_assl_lahn_leun", "stream_lahn_marb_lahn_leun")) On the contrary, the method |Selection.deselect_upstream| restricts the current selection to all devices not determined by method |Selection.search_upstream|: >>> complete = pub.selections.complete.deselect_upstream(hp.nodes.lahn_leun) >>> complete Selection("complete", nodes="lahn_kalk", elements=("land_lahn_kalk", "stream_lahn_leun_lahn_kalk")) If necessary, include the "outlet device" manually afterwards: >>> complete.nodes.add_device(hp.nodes.lahn_leun) >>> complete Selection("complete", nodes=("lahn_kalk", "lahn_leun"), elements=("land_lahn_kalk", "stream_lahn_leun_lahn_kalk")) Method |Selection.search_downstream| generally selects all |Node| objects directly connected to any upstream |Element| object. Set the `inclusive` argument to |False| to circumvent this: >>> from hydpy import Element, Nodes, Selection >>> nodes = Nodes( ... "inlet", "outlet1", "outlet2", "input_", "output", "receiver", "sender") >>> upper = Element("upper", ... inlets=nodes.inlet, outlets=(nodes.outlet1, nodes.outlet2), ... inputs=nodes.input_, outputs=nodes.output, ... receivers=nodes.receiver, senders=nodes.sender) >>> test = Selection("test", nodes=nodes, elements=upper) >>> test.search_upstream(nodes.outlet1, inclusive=True) Selection("upstream", nodes=("inlet", "input_", "outlet1", "outlet2", "output", "receiver", "sender"), elements="upper") >>> test.search_upstream(nodes.outlet1, inclusive=False) Selection("upstream", nodes=("inlet", "input_", "outlet1"), elements="upper") """ try: device = self._check_device(device, "outlet") graph = hydpytools.create_directedgraph(self.nodes, self.elements) devices = networkx.ancestors(graph, source=device) devices.add(device) selection = Selection( name=name, nodes=[d for d in devices if isinstance(d, devicetools.Node)], elements=[d for d in devices if isinstance(d, devicetools.Element)], ) if inclusive: add_device = selection.nodes.add_device for element in selection.elements: for nodes in ( element.outlets, element.outputs, element.receivers, element.senders, ): for node in nodes: add_device(node) return selection except BaseException: objecttools.augment_excmessage( f"While trying to determine an upstream network of selection " f"`{self.name}`" )
[docs] def select_upstream( self, device: devicetools.NodeOrElement, inclusive: bool = True ) -> Selection: """Restrict the current selection to the network upstream of the given starting point, including the starting point itself. See the documentation on method |Selection.search_upstream| for additional information. """ upstream = self.search_upstream(device, inclusive=inclusive) self.nodes = upstream.nodes self.elements = upstream.elements return self
[docs] def deselect_upstream( self, device: devicetools.NodeOrElement, inclusive: bool = True ) -> Selection: """Remove the network upstream of the given starting point from the current selection, including the starting point itself. See the documentation on method |Selection.search_upstream| for additional information. """ upstream = self.search_upstream(device, inclusive=inclusive) self.nodes -= upstream.nodes self.elements -= upstream.elements return self
[docs] def search_downstream( self, device: devicetools.NodeOrElement, name: str = "downstream", inclusive: bool = True, ) -> Selection: """Return the network downstream of the given starting point, including the starting point itself. >>> from hydpy.core.testtools import prepare_full_example_2 >>> hp, pub, _ = prepare_full_example_2() You can pass both |Node| and |Element| objects and, optionally, the name of the newly created |Selection| object: >>> test = pub.selections.complete.copy("test") >>> test.search_downstream(hp.nodes.lahn_marb) Selection("downstream", nodes=("lahn_kalk", "lahn_leun", "lahn_marb"), elements=("stream_lahn_leun_lahn_kalk", "stream_lahn_marb_lahn_leun")) >>> test.search_downstream(hp.elements.land_lahn_marb, "DOWNSTREAM") Selection("DOWNSTREAM", nodes=("lahn_kalk", "lahn_leun", "lahn_marb"), elements=("land_lahn_marb", "stream_lahn_leun_lahn_kalk", "stream_lahn_marb_lahn_leun")) Wrong device specifications result in errors like the following: >>> test.search_downstream(1) Traceback (most recent call last): ... TypeError: While trying to determine a downstream network of selection \ `test`, the following error occurred: Either a `Node` or an `Element` object is \ required as the "inlet device", but the given `device` value is of type `int`. >>> pub.selections.headwaters.search_downstream(hp.nodes.lahn_kalk) Traceback (most recent call last): ... KeyError: "While trying to determine a downstream network of selection \ `headwaters`, the following error occurred: 'No node named `lahn_kalk` available.'" Method |Selection.select_downstream| restricts the current selection to the one determined with the method |Selection.search_upstream|: >>> test.select_downstream(hp.nodes.lahn_marb) Selection("test", nodes=("lahn_kalk", "lahn_leun", "lahn_marb"), elements=("stream_lahn_leun_lahn_kalk", "stream_lahn_marb_lahn_leun")) On the contrary, the method |Selection.deselect_downstream| restricts the current selection to all devices not determined by method |Selection.search_downstream|: >>> complete = pub.selections.complete.deselect_downstream( ... hp.nodes.lahn_marb) >>> complete Selection("complete", nodes="dill_assl", elements=("land_dill_assl", "land_lahn_kalk", "land_lahn_leun", "land_lahn_marb", "stream_dill_assl_lahn_leun")) If necessary, include the "inlet device" manually afterwards: >>> complete.nodes.add_device(hp.nodes.lahn_marb) >>> complete Selection("complete", nodes=("dill_assl", "lahn_marb"), elements=("land_dill_assl", "land_lahn_kalk", "land_lahn_leun", "land_lahn_marb", "stream_dill_assl_lahn_leun")) Method |Selection.search_downstream| generally selects all |Node| objects directly connected to any upstream |Element| object. Set the `inclusive` argument to |False| to circumvent this: >>> from hydpy import Element, Nodes, Selection >>> nodes = Nodes( ... "inlet1", "inlet2", "outlet", "input_", "output", "receiver", "sender") >>> lower = Element("lower", ... inlets=(nodes.inlet1, nodes.inlet2), outlets=nodes.outlet, ... inputs=nodes.input_, outputs=nodes.output, ... receivers=nodes.receiver, senders=nodes.sender) >>> test = Selection("test", nodes=nodes, elements=lower) >>> test.search_downstream(nodes.inlet1, inclusive=True) Selection("downstream", nodes=("inlet1", "inlet2", "input_", "outlet", "output", "receiver", "sender"), elements="lower") >>> test.search_downstream(nodes.inlet1, inclusive=False) Selection("downstream", nodes=("inlet1", "outlet", "output"), elements="lower") """ try: device = self._check_device(device, "inlet") graph = hydpytools.create_directedgraph(self.nodes, self.elements) devices = networkx.descendants(graph, source=device) devices.add(device) selection = Selection( name=name, nodes=[d for d in devices if isinstance(d, devicetools.Node)], elements=[d for d in devices if isinstance(d, devicetools.Element)], ) if inclusive: add_device = selection.nodes.add_device for element in selection.elements: for nodes in ( element.inlets, element.inputs, element.receivers, element.senders, ): for node in nodes: add_device(node) return selection except BaseException: objecttools.augment_excmessage( f"While trying to determine a downstream network of selection " f"`{self.name}`" )
[docs] def select_downstream( self, device: devicetools.NodeOrElement, inclusive: bool = True ) -> Selection: """Restrict the current selection to the network downstream of the given starting point, including the starting point itself. See the documentation on method |Selection.search_downstream| for additional information. """ downstream = self.search_downstream(device, inclusive=inclusive) self.nodes = downstream.nodes self.elements = downstream.elements return self
[docs] def deselect_downstream( self, device: devicetools.NodeOrElement, inclusive: bool = True ) -> Selection: """Remove the network downstream of the given starting point from the current selection, including the starting point itself. See the documentation on method |Selection.search_downstream| for additional information. """ downstream = self.search_downstream(device, inclusive=inclusive) self.nodes -= downstream.nodes self.elements -= downstream.elements return self
[docs] def search_modeltypes( self, *models: ModelTypesArg, name: str = "modeltypes" ) -> Selection: """Return a |Selection| object containing only the elements currently handling models of the given types. >>> from hydpy.core.testtools import prepare_full_example_2 >>> hp, pub, _ = prepare_full_example_2() You can pass both |Model| objects and names and, as a keyword argument, the name of the newly created |Selection| object: >>> test = pub.selections.complete.copy("test") >>> from hydpy import prepare_model >>> hland_96 = prepare_model("hland_96") >>> test.search_modeltypes(hland_96) Selection("modeltypes", nodes=(), elements=("land_dill_assl", "land_lahn_kalk", "land_lahn_leun", "land_lahn_marb")) >>> test.search_modeltypes( ... hland_96, "musk_classic", "lland_dd", name="MODELTYPES") Selection("MODELTYPES", nodes=(), elements=("land_dill_assl", "land_lahn_kalk", "land_lahn_leun", "land_lahn_marb", "stream_dill_assl_lahn_leun", "stream_lahn_leun_lahn_kalk", "stream_lahn_marb_lahn_leun")) Wrong model specifications result in errors like the following: >>> test.search_modeltypes("wrong") Traceback (most recent call last): ... ModuleNotFoundError: While trying to determine the elements of selection \ `test` handling the model defined by the argument(s) `wrong` of type(s) `str`, the \ following error occurred: No module named 'hydpy.models.wrong' Method |Selection.select_modeltypes| restricts the current selection to the one determined with the method the |Selection.search_modeltypes|: >>> test.select_modeltypes(hland_96) Selection("test", nodes=(), elements=("land_dill_assl", "land_lahn_kalk", "land_lahn_leun", "land_lahn_marb")) On the contrary, the method |Selection.deselect_upstream| restricts the current selection to all devices not determined by method the |Selection.search_upstream|: >>> pub.selections.complete.deselect_modeltypes(hland_96) Selection("complete", nodes=(), elements=("stream_dill_assl_lahn_leun", "stream_lahn_leun_lahn_kalk", "stream_lahn_marb_lahn_leun")) """ try: typelist = [] for model in models: if not isinstance(model, modeltools.Model): model = importtools.prepare_model(model) typelist.append(type(model)) typetuple = tuple(typelist) selection = Selection(name) for element in self.elements: if isinstance(element.model, typetuple): selection.elements += element except BaseException: values = objecttools.enumeration(models) classes = objecttools.enumeration(type(model).__name__ for model in models) objecttools.augment_excmessage( f"While trying to determine the elements of selection `{self.name}` " f"handling the model defined by the argument(s) `{values}` of type(s) " f"`{classes}`" ) return selection
[docs] def select_modeltypes(self, *models: ModelTypesArg) -> Selection: """Restrict the current |Selection| object to all elements containing the given model types (removes all nodes). See the documentation on method |Selection.search_modeltypes| for additional information. """ self.nodes = devicetools.Nodes() self.elements = self.search_modeltypes(*models).elements return self
[docs] def deselect_modeltypes(self, *models: ModelTypesArg) -> Selection: """Restrict the current selection to all elements not containing the given model types (removes all nodes). See the documentation on method |Selection.search_modeltypes| for additional information. """ self.nodes = devicetools.Nodes() self.elements -= self.search_modeltypes(*models).elements return self
[docs] def search_nodenames(self, *substrings: str, name: str = "nodenames") -> Selection: """Return a new selection containing all nodes of the current selection with a name containing at least one of the given substrings. >>> from hydpy.core.testtools import prepare_full_example_2 >>> hp, pub, _ = prepare_full_example_2() Pass the (sub)strings as positional arguments and, optionally, the name of the newly created |Selection| object as a keyword argument: >>> test = pub.selections.complete.copy("test") >>> from hydpy import prepare_model >>> test.search_nodenames("dill_assl", "lahn_marb") Selection("nodenames", nodes=("dill_assl", "lahn_marb"), elements=()) Wrong string specifications result in errors like the following: >>> test.search_nodenames(["dill_assl", "lahn_marb"]) Traceback (most recent call last): ... TypeError: While trying to determine the nodes of selection `test` with names \ containing at least one of the given substrings `['dill_assl', 'lahn_marb']`, the \ following error occurred: 'in <string>' requires string as left operand, not list Method |Selection.select_nodenames| restricts the current selection to the one determined with the the method |Selection.search_nodenames|: >>> test.select_nodenames("dill_assl", "lahn_marb") Selection("test", nodes=("dill_assl", "lahn_marb"), elements=("land_dill_assl", "land_lahn_kalk", "land_lahn_leun", "land_lahn_marb", "stream_dill_assl_lahn_leun", "stream_lahn_leun_lahn_kalk", "stream_lahn_marb_lahn_leun")) On the contrary, the method |Selection.deselect_nodenames| restricts the current selection to all devices not determined by the method |Selection.search_nodenames|: >>> pub.selections.complete.deselect_nodenames("dill_assl", "lahn_marb") Selection("complete", nodes=("lahn_kalk", "lahn_leun"), elements=("land_dill_assl", "land_lahn_kalk", "land_lahn_leun", "land_lahn_marb", "stream_dill_assl_lahn_leun", "stream_lahn_leun_lahn_kalk", "stream_lahn_marb_lahn_leun")) """ try: selection = Selection(name) for node in self.nodes: for substring in substrings: if substring in node.name: selection.nodes += node break except BaseException: values = objecttools.enumeration(substrings) objecttools.augment_excmessage( f"While trying to determine the nodes of selection `{self.name}` with " f"names containing at least one of the given substrings `{values}`" ) return selection
[docs] def select_nodenames(self, *substrings: str) -> Selection: """Restrict the current selection to all nodes with a name containing at least one of the given substrings (does not affect any elements). See the documentation on method |Selection.search_nodenames| for additional information. """ self.nodes = self.search_nodenames(*substrings).nodes return self
[docs] def deselect_nodenames(self, *substrings: str) -> Selection: """Restrict the current selection to all nodes with a name not containing at least one of the given substrings (does not affect any elements). See the documentation on method |Selection.search_nodenames| for additional information. """ self.nodes -= self.search_nodenames(*substrings).nodes return self
[docs] def search_elementnames( self, *substrings: str, name: str = "elementnames" ) -> Selection: """Return a new selection containing all elements of the current selection with a name containing at least one of the given substrings. >>> from hydpy.core.testtools import prepare_full_example_2 >>> hp, pub, _ = prepare_full_example_2() Pass the (sub)strings as positional arguments and, optionally, the name of the newly created |Selection| object as a keyword argument: >>> test = pub.selections.complete.copy("test") >>> from hydpy import prepare_model >>> test.search_elementnames("dill", "lahn_marb") Selection("elementnames", nodes=(), elements=("land_dill_assl", "land_lahn_marb", "stream_dill_assl_lahn_leun", "stream_lahn_marb_lahn_leun")) Wrong string specifications result in errors like the following: >>> test.search_elementnames(["dill", "lahn_marb"]) Traceback (most recent call last): ... TypeError: While trying to determine the elements of selection `test` with \ names containing at least one of the given substrings `['dill', 'lahn_marb']`, the \ following error occurred: 'in <string>' requires string as left operand, not list Method |Selection.select_elementnames| restricts the current selection to the one determined with the method |Selection.search_elementnames|: >>> test.select_elementnames("dill", "lahn_marb") Selection("test", nodes=("dill_assl", "lahn_kalk", "lahn_leun", "lahn_marb"), elements=("land_dill_assl", "land_lahn_marb", "stream_dill_assl_lahn_leun", "stream_lahn_marb_lahn_leun")) On the contrary, the method |Selection.deselect_elementnames| restricts the current selection to all devices not determined by the method |Selection.search_elementnames|: >>> pub.selections.complete.deselect_elementnames("dill", "lahn_marb") Selection("complete", nodes=("dill_assl", "lahn_kalk", "lahn_leun", "lahn_marb"), elements=("land_lahn_kalk", "land_lahn_leun", "stream_lahn_leun_lahn_kalk")) """ try: selection = Selection(name) for element in self.elements: for substring in substrings: if substring in element.name: selection.elements += element break except BaseException: values = objecttools.enumeration(substrings) objecttools.augment_excmessage( f"While trying to determine the elements of selection `{self.name}` " f"with names containing at least one of the given substrings `{values}`" ) return selection
[docs] def select_elementnames(self, *substrings: str) -> Selection: """Restrict the current selection to all elements with a name containing at least one of the given substrings (does not affect any nodes). See the documentation on method |Selection.search_elementnames| for additional information. """ self.elements = self.search_elementnames(*substrings).elements return self
[docs] def deselect_elementnames(self, *substrings: str) -> Selection: """Restrict the current selection to all elements with a name not containing at least one of the given substrings. (does not affect any nodes). See the documentation on method |Selection.search_elementnames| for additional information. """ self.elements -= self.search_elementnames(*substrings).elements return self
[docs] def copy(self, name: str) -> Selection: """Return a new |Selection| object with the given name and copies of the handled |Nodes| and |Elements| objects based on method |Devices.copy|.""" return type(self)(name, copy.copy(self.nodes), copy.copy(self.elements))
[docs] def add_remotes(self) -> None: """Add all remote nodes linked to at least one of the currently handled elements. One often encounters the situation (for example, after calling method |Selection.select_upstream|), when a selection does not explicitly include all relevant remote nodes, like in the following example: >>> from hydpy import Element, Selection >>> dam = Element("dam", inlets="inflow", outlets="outflow", ... receivers="discharge_downstream", senders="water_level") >>> sel = Selection("Dam", elements=dam, nodes=("inflow", "outflow")) >>> sel Selection("Dam", nodes=("inflow", "outflow"), elements="dam") The method |Selection.add_remotes| is a small auxiliary function that takes care of this: >>> sel.add_remotes() >>> sel Selection("Dam", nodes=("discharge_downstream", "inflow", "outflow", "water_level"), elements="dam") """ nodes = self.nodes for element in self.elements: for node in itertools.chain(element.receivers, element.senders): nodes.add_device(node)
[docs] def save_networkfile( self, filepath: Union[str, None] = None, write_defaultnodes: bool = True ) -> None: """Save the selection as a network file. >>> from hydpy.core.testtools import prepare_full_example_2 >>> _, pub, TestIO = prepare_full_example_2() In most cases, one should conveniently write network files via method |NetworkManager.save_files| of class |NetworkManager|. However, using the method |Selection.save_networkfile| allows for additional configuration via the arguments `filepath` and `write_defaultnodes`: >>> with TestIO(): ... pub.selections.headwaters.save_networkfile() ... with open("headwaters.py") as networkfile: ... print(networkfile.read()) # -*- coding: utf-8 -*- <BLANKLINE> from hydpy import Element, Node <BLANKLINE> <BLANKLINE> Node("dill_assl", variable="Q", keywords="gauge") <BLANKLINE> Node("lahn_marb", variable="Q", keywords="gauge") <BLANKLINE> <BLANKLINE> Element("land_dill_assl", outlets="dill_assl", keywords="catchment") <BLANKLINE> Element("land_lahn_marb", outlets="lahn_marb", keywords="catchment") <BLANKLINE> >>> with TestIO(): ... pub.selections.headwaters.save_networkfile( ... "test.py", write_defaultnodes=False) ... with open("test.py") as networkfile: ... print(networkfile.read()) # -*- coding: utf-8 -*- <BLANKLINE> from hydpy import Element, Node <BLANKLINE> <BLANKLINE> Element("land_dill_assl", outlets="dill_assl", keywords="catchment") <BLANKLINE> Element("land_lahn_marb", outlets="lahn_marb", keywords="catchment") <BLANKLINE> The `write_defaultnodes` argument does only affect nodes handling the default variable `Q`: >>> from hydpy import FusedVariable, Node >>> from hydpy.aliases import ( ... hland_inputs_P, hland_inputs_T, lland_inputs_Nied, dam_receivers_OWL, ... hland_fluxes_Perc, hland_fluxes_Q0, hland_fluxes_Q1, ... dam_factors_WaterLevel) >>> Precip = FusedVariable("Precip", hland_inputs_P, lland_inputs_Nied) >>> Runoff = FusedVariable("Runoff", hland_fluxes_Q0, hland_fluxes_Q1) >>> Level = FusedVariable("Level", dam_receivers_OWL, dam_factors_WaterLevel) >>> nodes = pub.selections.headwaters.nodes >>> nodes.add_device(Node("test1", variable="X")) >>> nodes.add_device(Node("test2", variable=hland_inputs_T)) >>> nodes.add_device(Node("test3", variable=Precip)) >>> nodes.add_device(Node("test4", variable=hland_fluxes_Perc)) >>> nodes.add_device(Node("test5", variable=Runoff)) >>> nodes.add_device(Node("test6", variable=Level)) >>> with TestIO(): ... pub.selections.headwaters.save_networkfile( ... "test.py", write_defaultnodes=False) ... with open("test.py") as networkfile: ... print(networkfile.read()) # -*- coding: utf-8 -*- <BLANKLINE> from hydpy import Element, FusedVariable, Node from hydpy.aliases import ( dam_factors_WaterLevel, dam_receivers_OWL, hland_fluxes_Perc, hland_fluxes_Q0, hland_fluxes_Q1, hland_inputs_P, hland_inputs_T, lland_inputs_Nied, ) <BLANKLINE> <BLANKLINE> Level = FusedVariable("Level", dam_factors_WaterLevel, dam_receivers_OWL) Precip = FusedVariable("Precip", hland_inputs_P, lland_inputs_Nied) Runoff = FusedVariable("Runoff", hland_fluxes_Q0, hland_fluxes_Q1) <BLANKLINE> <BLANKLINE> Node("test1", variable="X") <BLANKLINE> Node("test2", variable=hland_inputs_T) <BLANKLINE> Node("test3", variable=Precip) <BLANKLINE> Node("test4", variable=hland_fluxes_Perc) <BLANKLINE> Node("test5", variable=Runoff) <BLANKLINE> Node("test6", variable=Level) <BLANKLINE> <BLANKLINE> Element("land_dill_assl", outlets="dill_assl", keywords="catchment") <BLANKLINE> Element("land_lahn_marb", outlets="lahn_marb", keywords="catchment") <BLANKLINE> """ aliases: set[str] = set() fusedvariables: set[devicetools.FusedVariable] = set() for variable in self.nodes.variables: if isinstance(variable, str): continue if isinstance(variable, devicetools.FusedVariable): fusedvariables.add(variable) else: aliases.add(hydpy.sequence2alias[variable]) for fusedvariable in fusedvariables: for sequence in fusedvariable: aliases.add(hydpy.sequence2alias[sequence]) if filepath is None: filepath = self.name + ".py" with open(filepath, "w", encoding="utf-8") as file_: file_.write("# -*- coding: utf-8 -*-\n") if fusedvariables: file_.write("\nfrom hydpy import Element, FusedVariable, Node") else: file_.write("\nfrom hydpy import Element, Node") if aliases: import_aliases = ", ".join(sorted(aliases)) import_aliases = f"from hydpy.aliases import {import_aliases}" import_aliases = black.format_str(import_aliases, mode=black.FileMode()) file_.write(f"\n{import_aliases}") file_.write("\n\n") for fusedvariable in sorted(fusedvariables, key=str): file_.write(f"{fusedvariable} = {repr(fusedvariable)}\n") if fusedvariables: file_.write("\n") written = False for node in self.nodes: if write_defaultnodes or (node.variable != "Q"): file_.write("\n" + repr(node) + "\n") written = True if written: file_.write("\n") for element in self.elements: file_.write("\n" + repr(element) + "\n")
def __len__(self) -> int: return len(self.nodes) + len(self.elements) _ERRORMESSAGE = ( "selection `{self.name}` with object `{other}` of type `{classname(other)}`" ) @objecttools.excmessage_decorator(f"add {_ERRORMESSAGE}") def __iadd__(self, other: Selection) -> Selection: self.nodes += other.nodes self.elements += other.elements return self @objecttools.excmessage_decorator(f"subtract {_ERRORMESSAGE}") def __isub__(self, other: Selection) -> Selection: self.nodes -= other.nodes self.elements -= other.elements return self @objecttools.excmessage_decorator(f"compare {_ERRORMESSAGE}") def __lt__(self, other: Selection) -> bool: # type: ignore[has-type] return (self.nodes < other.nodes) and (self.elements < other.elements) @objecttools.excmessage_decorator(f"compare {_ERRORMESSAGE}") def __le__(self, other: Selection) -> bool: # type: ignore[has-type] return (self.nodes <= other.nodes) and (self.elements <= other.elements) def __eq__(self, other: object) -> bool: if isinstance(other, (hydpytools.HydPy, Selection)): return (self.nodes == other.nodes) and (self.elements == other.elements) return False def __ne__(self, other: object) -> bool: if isinstance(other, (hydpytools.HydPy, Selection)): return (self.nodes != other.nodes) or (self.elements != other.elements) return True @objecttools.excmessage_decorator(f"compare {_ERRORMESSAGE}") def __ge__(self, other: Selection) -> bool: return (self.nodes >= other.nodes) and (self.elements >= other.elements) @objecttools.excmessage_decorator(f"compare {_ERRORMESSAGE}") def __gt__(self, other: Selection) -> bool: return (self.nodes > other.nodes) and (self.elements >= other.elements) def __hash__(self) -> int: return id(self) def __str__(self) -> str: return self.name def __repr__(self) -> str: return self.assignrepr("")
[docs] def assignrepr(self, prefix: str = "") -> str: """Return a |repr| string with a prefixed assignment.""" with objecttools.repr_.preserve_strings(True): options = hydpy.pub.options with options.ellipsis(2, optional=True): with objecttools.assignrepr_tuple.always_bracketed(False): classname = type(self).__name__ blanks = " " * (len(prefix + classname) + 1) nodestr = objecttools.assignrepr_tuple( self.nodes.names, blanks + "nodes=", 70 ) elementstr = objecttools.assignrepr_tuple( self.elements.names, blanks + "elements=", 70 ) return ( f'{prefix}{classname}("{self.name}",\n' f"{nodestr},\n" f"{elementstr})" )