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 0x000001BDA4797C40>
)
🏷️ 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
...