Skip to content

Module streamgen.nodes

🪢 different node implementations using anytree NodeMixin.

View Source
"""🪢 different node implementations using [anytree](https://anytree.readthedocs.io/en/stable/) `NodeMixin`."""

from collections import deque

from collections.abc import Callable

from typing import Any, Protocol, runtime_checkable

import anytree

import IPython

import matplotlib.pyplot as plt

import seaborn as sns

from beartype import beartype

from loguru import logger

from matplotlib import animation

from streamgen.enums import ArgumentPassingStrategy, ArgumentPassingStrategyLit

from streamgen.parameter import Parameter

from streamgen.parameter.store import ParameterStore

from streamgen.transforms import noop

@runtime_checkable

class Traverse(Protocol):

    """🏃 transform-node traversal protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

    If a node has children, return the next node to traverse. Otherwise return None and stop traversal.

    """

    def traverse(input: Any) -> tuple[Any, anytree.NodeMixin | None]:  # noqa: D102, N805, A002

        ...

class TransformNode(anytree.NodeMixin):

    """🪢 parametric transform node base class using [anytree](https://anytree.readthedocs.io/en/stable/) `NodeMixin`.

    A node can be linked dynamically to other nodes via the `parent` and `children` attributes.

    The main idea of a `TransformNode` is that is contains a transformation, which is called when the node is traversed.

    The parameters of the node can be updated.

    Args:

        transform (Callable): any callable function

        params (Parameter | ParameterStore | None, optional): parameters/arguments for the `transform`.

            Can by fetched dynamically from a `ParameterStore` using `TransformNode.fetch_params`. Defaults to None.

        argument_strategy (ArgumentPassingStrategy | ArgumentPassingStrategyLit, optional):

            strategy which defines how parameters are passed to functions. Defaults to "unpack", where params are passed as kwargs.

        name (str | None, optional): node name. If none, use `transform.__name__`. Defaults to None.

        emoji (str, optional): emoji for string representation. Defaults to "➡️".

    """

    @beartype()

    def __init__(  # noqa: D107

        self,

        transform: Callable,

        params: Parameter | ParameterStore | None = None,

        argument_strategy: ArgumentPassingStrategy | ArgumentPassingStrategyLit = "unpack",

        name: str | None = None,

        emoji: str = "➡️",

    ) -> None:

        super().__init__()

        self.transform = transform

        self.params: ParameterStore | None = ParameterStore([params]) if isinstance(params, Parameter) else params

        self.argument_strategy = argument_strategy

        self.name = name if name else transform.__name__

        self.emoji = emoji

        self.parent = None

    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin | None]:  # noqa: A002, ANN401

        """🏃 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        If `self` has children, return the next node to traverse. Otherwise return None and stop traversal.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and potential next node to traverse

        """

        match (self.params, self.argument_strategy):

            case (None, _):

                output = self.transform(input)

            case (_, ArgumentPassingStrategy.DICT):

                output = self.transform(input, self.params)

            case (_, ArgumentPassingStrategy.UNPACK):

                output = self.transform(input, **self.params.get_params())

        if self.children == ():

            return output, None

        if len(self.children) != 1:

            logger.warning(f"Node {self} contains more than one child. Only the first one is traversed.")

        return output, self.children[0]

    def update(self) -> None:

        """🆙 updates every parameter."""

        if self.params:

            self.params.update()

    def set_update_step(self, idx: int) -> None:

        """🕐 updates every parameter of `self.params` to a certain update step using `self.params.set_update_step`.

        Args:

            idx (int): parameter update step

        Returns:

            None: this function mutates `self`

        """

        if self.params:

            self.params.set_update_step(idx)

    def fetch_params(self, params: ParameterStore) -> None:

        """⚙️ fetches params from a ParameterStore.

        If the node was explicitly parameterized, use those params.

        Args:

            params (ParameterStore): _description_

        """

        if self.params:

            return

        if self.name in params.scopes:

            self.params = params.get_scope(self.name)

    def get_params(self) -> ParameterStore | None:

        """⚙️ returns current parameters.

        Returns:

            ParameterStore | None: parameters. None is there are no parameters.

        """

        if self.params is None:

            return None

        store = ParameterStore([])

        store.scopes.add(self.name)

        store.parameters[self.name] = {}

        for name, param in self.params.parameters.items():

            store.parameters[self.name][name] = param

            store.parameter_names.add(f"{self.name}.{name}")

        return store

    def __repr__(self) -> str:

        """🏷️ Returns the string representation `str(self)`.

        Returns:

            str: string representation of self

        """

        return f"{self.emoji} `{self.name}({repr(self.params)[1:-1] if self.params else ''})`"

class ClassLabelNode(TransformNode):

    """🏷️ node which sets the class label.

    Args:

        label (str | int): class label or index

        label_func (Callable[[Any, str | int], Any], optional): function which adds the label information.

            Defaults to `lambda input, label: (input, label)`.

    """

    def __init__(  # noqa: D107

        self,

        label: str | int,

        label_func: Callable[[Any, str | int], Any] = lambda input, label: (input, label),  # noqa: A002

    ) -> None:

        self.label = label

        self.label_func = label_func

        super().__init__(transform=noop, name=f"label={self.label}", emoji="🏷️")

    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin]:  # noqa: A002, ANN401

        """🏃🎲 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        During traversal, a label node sets the label in `input` with `self.label_func`.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and next node to traverse

        """

        output = self.label_func(input, self.label)

        return super().traverse(output)

    def __repr__(self) -> str:

        """🏷️ Returns the string representation `str(self)`.

        Returns:

            str: string representation of self

        """

        return f"🏷️ `{self.label}`"

class SampleBufferNode(TransformNode):

    """🗃️ node which remembers the last samples.

    Args:

        name (str | None, optional): name of the buffer. Defaults to "sample buffer".

        num_samples (int, optional): maximum number of samples to store. Defaults to 4.

    """

    def __init__(  # noqa: D107

        self,

        name: str | None = None,

        num_samples: int = 4,

    ) -> None:

        name = name if name is not None else "sample buffer"

        self.samples = deque(maxlen=num_samples)

        super().__init__(transform=noop, name=name, emoji="🗃️")

    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin]:  # noqa: A002, ANN401

        """🏃🎲 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        During traversal, a sample buffer node adds samples to `self.samples`.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and next node to traverse

        """

        self.samples.append(input)

        return super().traverse(input)

    def _plotting_func_wrapper(self, idx: int, ax: plt.Axes, plotting_func: Callable[[Any, plt.Axes], plt.Axes]) -> None:

        """🖼️ private wrapper for the `plotting_func` argument in `self.plot`.

        Args:

            idx (int): index of the frame. Passed into this function by matplotlib.

            ax (plt.Axes): artist to draw to

            plotting_func (Callable[[Any, plt.Axes], plt.Axes]): function to visualize a single sample.

                The function should take a sample and a `plt.Axes` as arguments

        """

        labeled = isinstance(self.samples[0], tuple)

        if labeled:

            sample, target = self.samples[idx]

        else:

            sample = self.samples[idx]

        ax.clear()

        plotting_func(sample, ax)

        ax.get_xaxis().set_visible(False)

        ax.get_yaxis().set_visible(False)

    def plot(

        self,

        plotting_func: Callable[[Any, plt.Axes], plt.Axes],

        display: bool = True,  # noqa: FBT001, FBT002

    ) -> IPython.display.HTML | animation.FuncAnimation | None:

        """📹 visualizes the samples in the buffer as an animation.

        Args:

            plotting_func (Callable[[Any, plt.Axes], plt.Axes]): function to visualize a single sample.

                The function should take a sample and a `plt.Axes` as arguments.

            display (bool, optional): If true, wraps the animation object in an `IPython.display.HTML`. Defaults to True.

        Returns:

            IPython.display.HTML | animation.FuncAnimation | None: matplotlib animation object

        """

        if len(self.samples) == 0:

            return None

        sns.set_theme()

        fig, ax = plt.subplots(figsize=(3.2, 2.4))

        anim = animation.FuncAnimation(fig, self._plotting_func_wrapper, frames=len(self.samples), fargs=(ax, plotting_func))

        return IPython.display.HTML(anim.to_jshtml()) if display else anim

Classes

ClassLabelNode

class ClassLabelNode(
    label: str | int,
    label_func: collections.abc.Callable[[typing.Any, str | int], typing.Any] = <function ClassLabelNode.<lambda> at 0x0000027EAF8CC540>
)

🏷️ node which sets the class label.

Attributes

Name Type Description Default
label str int class label or index
label_func Callable[[Any, str int], Any] function which adds the label information.
Defaults to lambda input, label: (input, label).
View Source
class ClassLabelNode(TransformNode):

    """🏷️ node which sets the class label.

    Args:

        label (str | int): class label or index

        label_func (Callable[[Any, str | int], Any], optional): function which adds the label information.

            Defaults to `lambda input, label: (input, label)`.

    """

    def __init__(  # noqa: D107

        self,

        label: str | int,

        label_func: Callable[[Any, str | int], Any] = lambda input, label: (input, label),  # noqa: A002

    ) -> None:

        self.label = label

        self.label_func = label_func

        super().__init__(transform=noop, name=f"label={self.label}", emoji="🏷️")

    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin]:  # noqa: A002, ANN401

        """🏃🎲 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        During traversal, a label node sets the label in `input` with `self.label_func`.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and next node to traverse

        """

        output = self.label_func(input, self.label)

        return super().traverse(output)

    def __repr__(self) -> str:

        """🏷️ Returns the string representation `str(self)`.

        Returns:

            str: string representation of self

        """

        return f"🏷️ `{self.label}`"

Ancestors (in MRO)

  • streamgen.nodes.TransformNode
  • anytree.node.nodemixin.NodeMixin

Class variables

separator

Instance variables

ancestors

All parent nodes and their parent nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.ancestors () marc.ancestors (Node('/Udo'),) lian.ancestors (Node('/Udo'), Node('/Udo/Marc'))

anchestors

All parent nodes and their parent nodes - see :any:ancestors.

The attribute anchestors is just a typo of ancestors. Please use ancestors. This attribute will be removed in the 3.0.0 release.

children

All child nodes.

from anytree import Node n = Node("n") a = Node("a", parent=n) b = Node("b", parent=n) c = Node("c", parent=n) n.children (Node('/n/a'), Node('/n/b'), Node('/n/c'))

Modifying the children attribute modifies the tree.

Detach

The children attribute can be updated by setting to an iterable.

n.children = [a, b] n.children (Node('/n/a'), Node('/n/b'))

Node c is removed from the tree. In case of an existing reference, the node c does not vanish and is the root of its own tree.

c Node('/c')

Attach

d = Node("d") d Node('/d') n.children = [a, b, d] n.children (Node('/n/a'), Node('/n/b'), Node('/n/d')) d Node('/n/d')

Duplicate

A node can just be the children once. Duplicates cause a :any:TreeError:

n.children = [a, b, d, a] Traceback (most recent call last): ... anytree.node.exceptions.TreeError: Cannot add node Node('/n/a') multiple times as child.

depth

Number of edges to the root Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.depth 0 marc.depth 1 lian.depth 2

descendants

All child nodes and all their child nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) soe = Node("Soe", parent=lian) udo.descendants (Node('/Udo/Marc'), Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lian/Soe'), Node('/Udo/Marc/Loui')) marc.descendants (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lian/Soe'), Node('/Udo/Marc/Loui')) lian.descendants (Node('/Udo/Marc/Lian/Soe'),)

height

Number of edges on the longest path to a leaf Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.height 2 marc.height 1 lian.height 0

is_leaf

Node has no children (External Node).

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.is_leaf False marc.is_leaf False lian.is_leaf True

is_root

Node is tree root.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.is_root True marc.is_root False lian.is_root False

leaves

Tuple of all leaf nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) lazy = Node("Lazy", parent=marc) udo.leaves (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy')) marc.leaves (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy'))

parent

Parent Node.

On set, the node is detached from any previous parent node and attached to the new node.

from anytree import Node, RenderTree udo = Node("Udo") marc = Node("Marc") lian = Node("Lian", parent=marc) print(RenderTree(udo)) Node('/Udo') print(RenderTree(marc)) Node('/Marc') └── Node('/Marc/Lian')

Attach

marc.parent = udo print(RenderTree(udo)) Node('/Udo') └── Node('/Udo/Marc') └── Node('/Udo/Marc/Lian')

Detach

To make a node to a root node, just set this attribute to None.

marc.is_root False marc.parent = None marc.is_root True

path

Path from root node down to this Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.path (Node('/Udo'),) marc.path (Node('/Udo'), Node('/Udo/Marc')) lian.path (Node('/Udo'), Node('/Udo/Marc'), Node('/Udo/Marc/Lian'))

root

Tree Root Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.root Node('/Udo') marc.root Node('/Udo') lian.root Node('/Udo')

siblings

Tuple of nodes with the same parent.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) lazy = Node("Lazy", parent=marc) udo.siblings () marc.siblings () lian.siblings (Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy')) loui.siblings (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lazy'))

size

Tree size --- the number of nodes in tree starting at this node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) soe = Node("Soe", parent=lian) udo.size 5 marc.size 4 lian.size 2 loui.size 1

Methods

fetch_params

def fetch_params(
    self,
    params: streamgen.parameter.store.ParameterStore
) -> None

⚙️ fetches params from a ParameterStore.

If the node was explicitly parameterized, use those params.

Parameters:

Name Type Description Default
params ParameterStore description None
View Source
    def fetch_params(self, params: ParameterStore) -> None:

        """⚙️ fetches params from a ParameterStore.

        If the node was explicitly parameterized, use those params.

        Args:

            params (ParameterStore): _description_

        """

        if self.params:

            return

        if self.name in params.scopes:

            self.params = params.get_scope(self.name)

get_params

def get_params(
    self
) -> streamgen.parameter.store.ParameterStore | None

⚙️ returns current parameters.

Returns:

Type Description
None ParameterStore
View Source
    def get_params(self) -> ParameterStore | None:

        """⚙️ returns current parameters.

        Returns:

            ParameterStore | None: parameters. None is there are no parameters.

        """

        if self.params is None:

            return None

        store = ParameterStore([])

        store.scopes.add(self.name)

        store.parameters[self.name] = {}

        for name, param in self.params.parameters.items():

            store.parameters[self.name][name] = param

            store.parameter_names.add(f"{self.name}.{name}")

        return store

iter_path_reverse

def iter_path_reverse(
    self
)

Iterate up the tree from the current node to the root node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) for node in udo.iter_path_reverse(): ... print(node) Node('/Udo') for node in marc.iter_path_reverse(): ... print(node) Node('/Udo/Marc') Node('/Udo') for node in lian.iter_path_reverse(): ... print(node) Node('/Udo/Marc/Lian') Node('/Udo/Marc') Node('/Udo')

View Source
    def iter_path_reverse(self):

        """

        Iterate up the tree from the current node to the root node.

        >>> from anytree import Node

        >>> udo = Node("Udo")

        >>> marc = Node("Marc", parent=udo)

        >>> lian = Node("Lian", parent=marc)

        >>> for node in udo.iter_path_reverse():

        ...     print(node)

        Node('/Udo')

        >>> for node in marc.iter_path_reverse():

        ...     print(node)

        Node('/Udo/Marc')

        Node('/Udo')

        >>> for node in lian.iter_path_reverse():

        ...     print(node)

        Node('/Udo/Marc/Lian')

        Node('/Udo/Marc')

        Node('/Udo')

        """

        node = self

        while node is not None:

            yield node

            node = node.parent

set_update_step

def set_update_step(
    self,
    idx: int
) -> None

🕐 updates every parameter of self.params to a certain update step using self.params.set_update_step.

Parameters:

Name Type Description Default
idx int parameter update step None

Returns:

Type Description
None this function mutates self
View Source
    def set_update_step(self, idx: int) -> None:

        """🕐 updates every parameter of `self.params` to a certain update step using `self.params.set_update_step`.

        Args:

            idx (int): parameter update step

        Returns:

            None: this function mutates `self`

        """

        if self.params:

            self.params.set_update_step(idx)

traverse

def traverse(
    self,
    input: Any
) -> tuple[typing.Any, anytree.node.nodemixin.NodeMixin]

🏃🎲 streamgen.transforms.Traverse protocol (input: Any) -> (output, anytree.NodeMixin | None).

During traversal, a label node sets the label in input with self.label_func.

Parameters:

Name Type Description Default
input Any any input None

Returns:

Type Description
tuple[Any, anytree.NodeMixin None]
View Source
    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin]:  # noqa: A002, ANN401

        """🏃🎲 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        During traversal, a label node sets the label in `input` with `self.label_func`.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and next node to traverse

        """

        output = self.label_func(input, self.label)

        return super().traverse(output)

update

def update(
    self
) -> None

🆙 updates every parameter.

View Source
    def update(self) -> None:

        """🆙 updates every parameter."""

        if self.params:

            self.params.update()

SampleBufferNode

class SampleBufferNode(
    name: str | None = None,
    num_samples: int = 4
)

🗃️ node which remembers the last samples.

Attributes

Name Type Description Default
name str None name of the buffer. Defaults to "sample buffer".
num_samples int maximum number of samples to store. Defaults to 4. 4
View Source
class SampleBufferNode(TransformNode):

    """🗃️ node which remembers the last samples.

    Args:

        name (str | None, optional): name of the buffer. Defaults to "sample buffer".

        num_samples (int, optional): maximum number of samples to store. Defaults to 4.

    """

    def __init__(  # noqa: D107

        self,

        name: str | None = None,

        num_samples: int = 4,

    ) -> None:

        name = name if name is not None else "sample buffer"

        self.samples = deque(maxlen=num_samples)

        super().__init__(transform=noop, name=name, emoji="🗃️")

    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin]:  # noqa: A002, ANN401

        """🏃🎲 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        During traversal, a sample buffer node adds samples to `self.samples`.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and next node to traverse

        """

        self.samples.append(input)

        return super().traverse(input)

    def _plotting_func_wrapper(self, idx: int, ax: plt.Axes, plotting_func: Callable[[Any, plt.Axes], plt.Axes]) -> None:

        """🖼️ private wrapper for the `plotting_func` argument in `self.plot`.

        Args:

            idx (int): index of the frame. Passed into this function by matplotlib.

            ax (plt.Axes): artist to draw to

            plotting_func (Callable[[Any, plt.Axes], plt.Axes]): function to visualize a single sample.

                The function should take a sample and a `plt.Axes` as arguments

        """

        labeled = isinstance(self.samples[0], tuple)

        if labeled:

            sample, target = self.samples[idx]

        else:

            sample = self.samples[idx]

        ax.clear()

        plotting_func(sample, ax)

        ax.get_xaxis().set_visible(False)

        ax.get_yaxis().set_visible(False)

    def plot(

        self,

        plotting_func: Callable[[Any, plt.Axes], plt.Axes],

        display: bool = True,  # noqa: FBT001, FBT002

    ) -> IPython.display.HTML | animation.FuncAnimation | None:

        """📹 visualizes the samples in the buffer as an animation.

        Args:

            plotting_func (Callable[[Any, plt.Axes], plt.Axes]): function to visualize a single sample.

                The function should take a sample and a `plt.Axes` as arguments.

            display (bool, optional): If true, wraps the animation object in an `IPython.display.HTML`. Defaults to True.

        Returns:

            IPython.display.HTML | animation.FuncAnimation | None: matplotlib animation object

        """

        if len(self.samples) == 0:

            return None

        sns.set_theme()

        fig, ax = plt.subplots(figsize=(3.2, 2.4))

        anim = animation.FuncAnimation(fig, self._plotting_func_wrapper, frames=len(self.samples), fargs=(ax, plotting_func))

        return IPython.display.HTML(anim.to_jshtml()) if display else anim

Ancestors (in MRO)

  • streamgen.nodes.TransformNode
  • anytree.node.nodemixin.NodeMixin

Class variables

separator

Instance variables

ancestors

All parent nodes and their parent nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.ancestors () marc.ancestors (Node('/Udo'),) lian.ancestors (Node('/Udo'), Node('/Udo/Marc'))

anchestors

All parent nodes and their parent nodes - see :any:ancestors.

The attribute anchestors is just a typo of ancestors. Please use ancestors. This attribute will be removed in the 3.0.0 release.

children

All child nodes.

from anytree import Node n = Node("n") a = Node("a", parent=n) b = Node("b", parent=n) c = Node("c", parent=n) n.children (Node('/n/a'), Node('/n/b'), Node('/n/c'))

Modifying the children attribute modifies the tree.

Detach

The children attribute can be updated by setting to an iterable.

n.children = [a, b] n.children (Node('/n/a'), Node('/n/b'))

Node c is removed from the tree. In case of an existing reference, the node c does not vanish and is the root of its own tree.

c Node('/c')

Attach

d = Node("d") d Node('/d') n.children = [a, b, d] n.children (Node('/n/a'), Node('/n/b'), Node('/n/d')) d Node('/n/d')

Duplicate

A node can just be the children once. Duplicates cause a :any:TreeError:

n.children = [a, b, d, a] Traceback (most recent call last): ... anytree.node.exceptions.TreeError: Cannot add node Node('/n/a') multiple times as child.

depth

Number of edges to the root Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.depth 0 marc.depth 1 lian.depth 2

descendants

All child nodes and all their child nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) soe = Node("Soe", parent=lian) udo.descendants (Node('/Udo/Marc'), Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lian/Soe'), Node('/Udo/Marc/Loui')) marc.descendants (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lian/Soe'), Node('/Udo/Marc/Loui')) lian.descendants (Node('/Udo/Marc/Lian/Soe'),)

height

Number of edges on the longest path to a leaf Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.height 2 marc.height 1 lian.height 0

is_leaf

Node has no children (External Node).

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.is_leaf False marc.is_leaf False lian.is_leaf True

is_root

Node is tree root.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.is_root True marc.is_root False lian.is_root False

leaves

Tuple of all leaf nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) lazy = Node("Lazy", parent=marc) udo.leaves (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy')) marc.leaves (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy'))

parent

Parent Node.

On set, the node is detached from any previous parent node and attached to the new node.

from anytree import Node, RenderTree udo = Node("Udo") marc = Node("Marc") lian = Node("Lian", parent=marc) print(RenderTree(udo)) Node('/Udo') print(RenderTree(marc)) Node('/Marc') └── Node('/Marc/Lian')

Attach

marc.parent = udo print(RenderTree(udo)) Node('/Udo') └── Node('/Udo/Marc') └── Node('/Udo/Marc/Lian')

Detach

To make a node to a root node, just set this attribute to None.

marc.is_root False marc.parent = None marc.is_root True

path

Path from root node down to this Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.path (Node('/Udo'),) marc.path (Node('/Udo'), Node('/Udo/Marc')) lian.path (Node('/Udo'), Node('/Udo/Marc'), Node('/Udo/Marc/Lian'))

root

Tree Root Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.root Node('/Udo') marc.root Node('/Udo') lian.root Node('/Udo')

siblings

Tuple of nodes with the same parent.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) lazy = Node("Lazy", parent=marc) udo.siblings () marc.siblings () lian.siblings (Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy')) loui.siblings (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lazy'))

size

Tree size --- the number of nodes in tree starting at this node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) soe = Node("Soe", parent=lian) udo.size 5 marc.size 4 lian.size 2 loui.size 1

Methods

fetch_params

def fetch_params(
    self,
    params: streamgen.parameter.store.ParameterStore
) -> None

⚙️ fetches params from a ParameterStore.

If the node was explicitly parameterized, use those params.

Parameters:

Name Type Description Default
params ParameterStore description None
View Source
    def fetch_params(self, params: ParameterStore) -> None:

        """⚙️ fetches params from a ParameterStore.

        If the node was explicitly parameterized, use those params.

        Args:

            params (ParameterStore): _description_

        """

        if self.params:

            return

        if self.name in params.scopes:

            self.params = params.get_scope(self.name)

get_params

def get_params(
    self
) -> streamgen.parameter.store.ParameterStore | None

⚙️ returns current parameters.

Returns:

Type Description
None ParameterStore
View Source
    def get_params(self) -> ParameterStore | None:

        """⚙️ returns current parameters.

        Returns:

            ParameterStore | None: parameters. None is there are no parameters.

        """

        if self.params is None:

            return None

        store = ParameterStore([])

        store.scopes.add(self.name)

        store.parameters[self.name] = {}

        for name, param in self.params.parameters.items():

            store.parameters[self.name][name] = param

            store.parameter_names.add(f"{self.name}.{name}")

        return store

iter_path_reverse

def iter_path_reverse(
    self
)

Iterate up the tree from the current node to the root node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) for node in udo.iter_path_reverse(): ... print(node) Node('/Udo') for node in marc.iter_path_reverse(): ... print(node) Node('/Udo/Marc') Node('/Udo') for node in lian.iter_path_reverse(): ... print(node) Node('/Udo/Marc/Lian') Node('/Udo/Marc') Node('/Udo')

View Source
    def iter_path_reverse(self):

        """

        Iterate up the tree from the current node to the root node.

        >>> from anytree import Node

        >>> udo = Node("Udo")

        >>> marc = Node("Marc", parent=udo)

        >>> lian = Node("Lian", parent=marc)

        >>> for node in udo.iter_path_reverse():

        ...     print(node)

        Node('/Udo')

        >>> for node in marc.iter_path_reverse():

        ...     print(node)

        Node('/Udo/Marc')

        Node('/Udo')

        >>> for node in lian.iter_path_reverse():

        ...     print(node)

        Node('/Udo/Marc/Lian')

        Node('/Udo/Marc')

        Node('/Udo')

        """

        node = self

        while node is not None:

            yield node

            node = node.parent

plot

def plot(
    self,
    plotting_func: collections.abc.Callable[[typing.Any, matplotlib.axes._axes.Axes], matplotlib.axes._axes.Axes],
    display: bool = True
) -> IPython.core.display.HTML | matplotlib.animation.FuncAnimation | None

📹 visualizes the samples in the buffer as an animation.

Parameters:

Name Type Description Default
plotting_func Callable[[Any, plt.Axes], plt.Axes] function to visualize a single sample.
The function should take a sample and a plt.Axes as arguments.
None
display bool If true, wraps the animation object in an IPython.display.HTML. Defaults to True. True

Returns:

Type Description
None IPython.display.HTML
View Source
    def plot(

        self,

        plotting_func: Callable[[Any, plt.Axes], plt.Axes],

        display: bool = True,  # noqa: FBT001, FBT002

    ) -> IPython.display.HTML | animation.FuncAnimation | None:

        """📹 visualizes the samples in the buffer as an animation.

        Args:

            plotting_func (Callable[[Any, plt.Axes], plt.Axes]): function to visualize a single sample.

                The function should take a sample and a `plt.Axes` as arguments.

            display (bool, optional): If true, wraps the animation object in an `IPython.display.HTML`. Defaults to True.

        Returns:

            IPython.display.HTML | animation.FuncAnimation | None: matplotlib animation object

        """

        if len(self.samples) == 0:

            return None

        sns.set_theme()

        fig, ax = plt.subplots(figsize=(3.2, 2.4))

        anim = animation.FuncAnimation(fig, self._plotting_func_wrapper, frames=len(self.samples), fargs=(ax, plotting_func))

        return IPython.display.HTML(anim.to_jshtml()) if display else anim

set_update_step

def set_update_step(
    self,
    idx: int
) -> None

🕐 updates every parameter of self.params to a certain update step using self.params.set_update_step.

Parameters:

Name Type Description Default
idx int parameter update step None

Returns:

Type Description
None this function mutates self
View Source
    def set_update_step(self, idx: int) -> None:

        """🕐 updates every parameter of `self.params` to a certain update step using `self.params.set_update_step`.

        Args:

            idx (int): parameter update step

        Returns:

            None: this function mutates `self`

        """

        if self.params:

            self.params.set_update_step(idx)

traverse

def traverse(
    self,
    input: Any
) -> tuple[typing.Any, anytree.node.nodemixin.NodeMixin]

🏃🎲 streamgen.transforms.Traverse protocol (input: Any) -> (output, anytree.NodeMixin | None).

During traversal, a sample buffer node adds samples to self.samples.

Parameters:

Name Type Description Default
input Any any input None

Returns:

Type Description
tuple[Any, anytree.NodeMixin None]
View Source
    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin]:  # noqa: A002, ANN401

        """🏃🎲 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        During traversal, a sample buffer node adds samples to `self.samples`.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and next node to traverse

        """

        self.samples.append(input)

        return super().traverse(input)

update

def update(
    self
) -> None

🆙 updates every parameter.

View Source
    def update(self) -> None:

        """🆙 updates every parameter."""

        if self.params:

            self.params.update()

TransformNode

class TransformNode(
    transform: collections.abc.Callable,
    params: streamgen.parameter.Parameter | streamgen.parameter.store.ParameterStore | None = None,
    argument_strategy: Union[streamgen.enums.ArgumentPassingStrategy, Literal['dict', 'unpack']] = 'unpack',
    name: str | None = None,
    emoji: str = '➡️'
)

🪢 parametric transform node base class using anytree NodeMixin.

A node can be linked dynamically to other nodes via the parent and children attributes. The main idea of a TransformNode is that is contains a transformation, which is called when the node is traversed.

The parameters of the node can be updated.

Attributes

Name Type Description Default
transform Callable any callable function None
params Parameter ParameterStore None
argument_strategy ArgumentPassingStrategy ArgumentPassingStrategyLit strategy which defines how parameters are passed to functions. Defaults to "unpack", where params are passed as kwargs.
name str None node name. If none, use transform.__name__. Defaults to None.
emoji str emoji for string representation. Defaults to "➡️". "➡️"
View Source
class TransformNode(anytree.NodeMixin):

    """🪢 parametric transform node base class using [anytree](https://anytree.readthedocs.io/en/stable/) `NodeMixin`.

    A node can be linked dynamically to other nodes via the `parent` and `children` attributes.

    The main idea of a `TransformNode` is that is contains a transformation, which is called when the node is traversed.

    The parameters of the node can be updated.

    Args:

        transform (Callable): any callable function

        params (Parameter | ParameterStore | None, optional): parameters/arguments for the `transform`.

            Can by fetched dynamically from a `ParameterStore` using `TransformNode.fetch_params`. Defaults to None.

        argument_strategy (ArgumentPassingStrategy | ArgumentPassingStrategyLit, optional):

            strategy which defines how parameters are passed to functions. Defaults to "unpack", where params are passed as kwargs.

        name (str | None, optional): node name. If none, use `transform.__name__`. Defaults to None.

        emoji (str, optional): emoji for string representation. Defaults to "➡️".

    """

    @beartype()

    def __init__(  # noqa: D107

        self,

        transform: Callable,

        params: Parameter | ParameterStore | None = None,

        argument_strategy: ArgumentPassingStrategy | ArgumentPassingStrategyLit = "unpack",

        name: str | None = None,

        emoji: str = "➡️",

    ) -> None:

        super().__init__()

        self.transform = transform

        self.params: ParameterStore | None = ParameterStore([params]) if isinstance(params, Parameter) else params

        self.argument_strategy = argument_strategy

        self.name = name if name else transform.__name__

        self.emoji = emoji

        self.parent = None

    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin | None]:  # noqa: A002, ANN401

        """🏃 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        If `self` has children, return the next node to traverse. Otherwise return None and stop traversal.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and potential next node to traverse

        """

        match (self.params, self.argument_strategy):

            case (None, _):

                output = self.transform(input)

            case (_, ArgumentPassingStrategy.DICT):

                output = self.transform(input, self.params)

            case (_, ArgumentPassingStrategy.UNPACK):

                output = self.transform(input, **self.params.get_params())

        if self.children == ():

            return output, None

        if len(self.children) != 1:

            logger.warning(f"Node {self} contains more than one child. Only the first one is traversed.")

        return output, self.children[0]

    def update(self) -> None:

        """🆙 updates every parameter."""

        if self.params:

            self.params.update()

    def set_update_step(self, idx: int) -> None:

        """🕐 updates every parameter of `self.params` to a certain update step using `self.params.set_update_step`.

        Args:

            idx (int): parameter update step

        Returns:

            None: this function mutates `self`

        """

        if self.params:

            self.params.set_update_step(idx)

    def fetch_params(self, params: ParameterStore) -> None:

        """⚙️ fetches params from a ParameterStore.

        If the node was explicitly parameterized, use those params.

        Args:

            params (ParameterStore): _description_

        """

        if self.params:

            return

        if self.name in params.scopes:

            self.params = params.get_scope(self.name)

    def get_params(self) -> ParameterStore | None:

        """⚙️ returns current parameters.

        Returns:

            ParameterStore | None: parameters. None is there are no parameters.

        """

        if self.params is None:

            return None

        store = ParameterStore([])

        store.scopes.add(self.name)

        store.parameters[self.name] = {}

        for name, param in self.params.parameters.items():

            store.parameters[self.name][name] = param

            store.parameter_names.add(f"{self.name}.{name}")

        return store

    def __repr__(self) -> str:

        """🏷️ Returns the string representation `str(self)`.

        Returns:

            str: string representation of self

        """

        return f"{self.emoji} `{self.name}({repr(self.params)[1:-1] if self.params else ''})`"

Ancestors (in MRO)

  • anytree.node.nodemixin.NodeMixin

Descendants

  • streamgen.nodes.ClassLabelNode
  • streamgen.nodes.SampleBufferNode
  • streamgen.samplers.tree.BranchingNode

Class variables

separator

Instance variables

ancestors

All parent nodes and their parent nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.ancestors () marc.ancestors (Node('/Udo'),) lian.ancestors (Node('/Udo'), Node('/Udo/Marc'))

anchestors

All parent nodes and their parent nodes - see :any:ancestors.

The attribute anchestors is just a typo of ancestors. Please use ancestors. This attribute will be removed in the 3.0.0 release.

children

All child nodes.

from anytree import Node n = Node("n") a = Node("a", parent=n) b = Node("b", parent=n) c = Node("c", parent=n) n.children (Node('/n/a'), Node('/n/b'), Node('/n/c'))

Modifying the children attribute modifies the tree.

Detach

The children attribute can be updated by setting to an iterable.

n.children = [a, b] n.children (Node('/n/a'), Node('/n/b'))

Node c is removed from the tree. In case of an existing reference, the node c does not vanish and is the root of its own tree.

c Node('/c')

Attach

d = Node("d") d Node('/d') n.children = [a, b, d] n.children (Node('/n/a'), Node('/n/b'), Node('/n/d')) d Node('/n/d')

Duplicate

A node can just be the children once. Duplicates cause a :any:TreeError:

n.children = [a, b, d, a] Traceback (most recent call last): ... anytree.node.exceptions.TreeError: Cannot add node Node('/n/a') multiple times as child.

depth

Number of edges to the root Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.depth 0 marc.depth 1 lian.depth 2

descendants

All child nodes and all their child nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) soe = Node("Soe", parent=lian) udo.descendants (Node('/Udo/Marc'), Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lian/Soe'), Node('/Udo/Marc/Loui')) marc.descendants (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lian/Soe'), Node('/Udo/Marc/Loui')) lian.descendants (Node('/Udo/Marc/Lian/Soe'),)

height

Number of edges on the longest path to a leaf Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.height 2 marc.height 1 lian.height 0

is_leaf

Node has no children (External Node).

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.is_leaf False marc.is_leaf False lian.is_leaf True

is_root

Node is tree root.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.is_root True marc.is_root False lian.is_root False

leaves

Tuple of all leaf nodes.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) lazy = Node("Lazy", parent=marc) udo.leaves (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy')) marc.leaves (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy'))

parent

Parent Node.

On set, the node is detached from any previous parent node and attached to the new node.

from anytree import Node, RenderTree udo = Node("Udo") marc = Node("Marc") lian = Node("Lian", parent=marc) print(RenderTree(udo)) Node('/Udo') print(RenderTree(marc)) Node('/Marc') └── Node('/Marc/Lian')

Attach

marc.parent = udo print(RenderTree(udo)) Node('/Udo') └── Node('/Udo/Marc') └── Node('/Udo/Marc/Lian')

Detach

To make a node to a root node, just set this attribute to None.

marc.is_root False marc.parent = None marc.is_root True

path

Path from root node down to this Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.path (Node('/Udo'),) marc.path (Node('/Udo'), Node('/Udo/Marc')) lian.path (Node('/Udo'), Node('/Udo/Marc'), Node('/Udo/Marc/Lian'))

root

Tree Root Node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) udo.root Node('/Udo') marc.root Node('/Udo') lian.root Node('/Udo')

siblings

Tuple of nodes with the same parent.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) lazy = Node("Lazy", parent=marc) udo.siblings () marc.siblings () lian.siblings (Node('/Udo/Marc/Loui'), Node('/Udo/Marc/Lazy')) loui.siblings (Node('/Udo/Marc/Lian'), Node('/Udo/Marc/Lazy'))

size

Tree size --- the number of nodes in tree starting at this node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) loui = Node("Loui", parent=marc) soe = Node("Soe", parent=lian) udo.size 5 marc.size 4 lian.size 2 loui.size 1

Methods

fetch_params

def fetch_params(
    self,
    params: streamgen.parameter.store.ParameterStore
) -> None

⚙️ fetches params from a ParameterStore.

If the node was explicitly parameterized, use those params.

Parameters:

Name Type Description Default
params ParameterStore description None
View Source
    def fetch_params(self, params: ParameterStore) -> None:

        """⚙️ fetches params from a ParameterStore.

        If the node was explicitly parameterized, use those params.

        Args:

            params (ParameterStore): _description_

        """

        if self.params:

            return

        if self.name in params.scopes:

            self.params = params.get_scope(self.name)

get_params

def get_params(
    self
) -> streamgen.parameter.store.ParameterStore | None

⚙️ returns current parameters.

Returns:

Type Description
None ParameterStore
View Source
    def get_params(self) -> ParameterStore | None:

        """⚙️ returns current parameters.

        Returns:

            ParameterStore | None: parameters. None is there are no parameters.

        """

        if self.params is None:

            return None

        store = ParameterStore([])

        store.scopes.add(self.name)

        store.parameters[self.name] = {}

        for name, param in self.params.parameters.items():

            store.parameters[self.name][name] = param

            store.parameter_names.add(f"{self.name}.{name}")

        return store

iter_path_reverse

def iter_path_reverse(
    self
)

Iterate up the tree from the current node to the root node.

from anytree import Node udo = Node("Udo") marc = Node("Marc", parent=udo) lian = Node("Lian", parent=marc) for node in udo.iter_path_reverse(): ... print(node) Node('/Udo') for node in marc.iter_path_reverse(): ... print(node) Node('/Udo/Marc') Node('/Udo') for node in lian.iter_path_reverse(): ... print(node) Node('/Udo/Marc/Lian') Node('/Udo/Marc') Node('/Udo')

View Source
    def iter_path_reverse(self):

        """

        Iterate up the tree from the current node to the root node.

        >>> from anytree import Node

        >>> udo = Node("Udo")

        >>> marc = Node("Marc", parent=udo)

        >>> lian = Node("Lian", parent=marc)

        >>> for node in udo.iter_path_reverse():

        ...     print(node)

        Node('/Udo')

        >>> for node in marc.iter_path_reverse():

        ...     print(node)

        Node('/Udo/Marc')

        Node('/Udo')

        >>> for node in lian.iter_path_reverse():

        ...     print(node)

        Node('/Udo/Marc/Lian')

        Node('/Udo/Marc')

        Node('/Udo')

        """

        node = self

        while node is not None:

            yield node

            node = node.parent

set_update_step

def set_update_step(
    self,
    idx: int
) -> None

🕐 updates every parameter of self.params to a certain update step using self.params.set_update_step.

Parameters:

Name Type Description Default
idx int parameter update step None

Returns:

Type Description
None this function mutates self
View Source
    def set_update_step(self, idx: int) -> None:

        """🕐 updates every parameter of `self.params` to a certain update step using `self.params.set_update_step`.

        Args:

            idx (int): parameter update step

        Returns:

            None: this function mutates `self`

        """

        if self.params:

            self.params.set_update_step(idx)

traverse

def traverse(
    self,
    input: Any
) -> tuple[typing.Any, anytree.node.nodemixin.NodeMixin | None]

🏃 streamgen.transforms.Traverse protocol (input: Any) -> (output, anytree.NodeMixin | None).

If self has children, return the next node to traverse. Otherwise return None and stop traversal.

Parameters:

Name Type Description Default
input Any any input None

Returns:

Type Description
tuple[Any, anytree.NodeMixin None]
View Source
    def traverse(self, input: Any) -> tuple[Any, anytree.NodeMixin | None]:  # noqa: A002, ANN401

        """🏃 `streamgen.transforms.Traverse` protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

        If `self` has children, return the next node to traverse. Otherwise return None and stop traversal.

        Args:

            input (Any): any input

        Returns:

            tuple[Any, anytree.NodeMixin | None]: output and potential next node to traverse

        """

        match (self.params, self.argument_strategy):

            case (None, _):

                output = self.transform(input)

            case (_, ArgumentPassingStrategy.DICT):

                output = self.transform(input, self.params)

            case (_, ArgumentPassingStrategy.UNPACK):

                output = self.transform(input, **self.params.get_params())

        if self.children == ():

            return output, None

        if len(self.children) != 1:

            logger.warning(f"Node {self} contains more than one child. Only the first one is traversed.")

        return output, self.children[0]

update

def update(
    self
) -> None

🆙 updates every parameter.

View Source
    def update(self) -> None:

        """🆙 updates every parameter."""

        if self.params:

            self.params.update()

Traverse

class Traverse(
    *args,
    **kwargs
)

🏃 transform-node traversal protocol (input: Any) -> (output, anytree.NodeMixin | None).

If a node has children, return the next node to traverse. Otherwise return None and stop traversal.

View Source
@runtime_checkable

class Traverse(Protocol):

    """🏃 transform-node traversal protocol `(input: Any) -> (output, anytree.NodeMixin | None)`.

    If a node has children, return the next node to traverse. Otherwise return None and stop traversal.

    """

    def traverse(input: Any) -> tuple[Any, anytree.NodeMixin | None]:  # noqa: D102, N805, A002

        ...

Ancestors (in MRO)

  • typing.Protocol
  • typing.Generic

Methods

traverse

def traverse(
    input: Any
) -> tuple[typing.Any, anytree.node.nodemixin.NodeMixin | None]
View Source
    def traverse(input: Any) -> tuple[Any, anytree.NodeMixin | None]:  # noqa: D102, N805, A002

        ...