DeepTrackNode#

class deeptrack.backend.core.DeepTrackNode(action: Callable[[...], Any] | Any | None = None, node_name: str | None = None, **kwargs: Any)#

Bases: object

Node in a DeepTrack2 computation graph, supporting operator overloading.

DeepTrackNode represents a node within a DeepTrack2 computation graph. Each node can store data and compute new values based on its dependencies. The value of a node is computed by calling its action.

DeepTrackNode supports operator overloading, enabling intuitive construction of computation graphs using standard Python operators. For example, nodes can be added, multiplied, subtracted, or compared directly (e.g., node1 + node2, node1 * 3, node1 > node2), and the resulting node will represent the composed operation.

Parameters#

action: Callable or Any, optional: Action to compute this node’s value. If not provided, uses a no-op action (lambda: None).
node_name: str or None, optional: Optional name assigned to the node. Defaults to None.
**kwargs: Any: Additional arguments for subclasses or extended functionality.

Attributes#

node_name: str or None: Name assigned to the node. Defaults to None.
data: DeepTrackDataDict: Dictionary-like object for storing data, indexed by tuples of integers.
children: WeakSet[DeepTrackNode]: Read-only property exposing the internal weak set ._children containing the nodes that depend on this node (its children). This is a weakref.WeakSet, so references are weak and do not prevent garbage collection of nodes that are no longer used.
dependencies: WeakSet[DeepTrackNode]: Read-only property exposing the internal weak set ._dependencies containing the nodes on which this node depends (its ancestors). This is a weakref.WeakSet, for efficient memory management.
_action: Callable[…, Any]: The function or lambda-function to compute the node value.
_accepts_ID: bool: Whether action accepts an input _ID.
_all_children: WeakSet[DeepTrackNode]: All nodes in the subtree rooted at the node, including the node itself. This is a weakref.WeakSet, for efficient memory management.
_all_dependencies: WeakSet[DeepTrackNode]: All the dependencies for this node, including the node itself. This is a weakref.WeakSet, for efficient memory management.
_citations: list[str]: Citations associated with this node.

Methods#

action: property: Get or set the computation function for the node (stored as _action).
add_child(child) -> DeepTrackNode: Add a child node that depends on this node. Also add the dependency on this node in the child node.
add_dependency(parent) -> DeepTrackNode: Add a dependency, making this node depend on the parent node. Also set this node as a child of the parent node.
store(data, _ID) -> DeepTrackNode: Store computed data for the given _ID.
is_valid(_ID) -> bool: Check whether the data for the given _ID is valid.
valid_index(_ID) -> bool: Check whether the given _ID is valid for this node.
invalidate(_ID) -> DeepTrackNode: Invalidate the data for the given _ID (exact, trimmed, or prefix slice) and all child nodes.
validate(_ID) -> DeepTrackNode: Validate the data for the given _ID (exact, trimmed, or prefix slice), marking it as up-to-date, but not its children.
update() -> DeepTrackNode: Reset the data.
set_value(value, _ID) -> DeepTrackNode: Set a value for the given _ID. If the new value differs from the current value, the node is invalidated to ensure dependencies are recomputed.
print_children_tree(indent) -> None: Print a tree of all child nodes (recursively) for inspection.
recurse_children() -> set[DeepTrackNode]: Return all child nodes in the dependency tree rooted at this node.
print_dependencies_tree(indent) -> None: Print a tree of all parent nodes (recursively) for inspection.
recurse_dependencies() -> Iterator[DeepTrackNode]: Yield all nodes that this node depends on, traversing dependencies.
get_citations() -> set[str]: Return a set of citations for this node and its dependencies.
__call__(_ID) -> Any: Evaluate the node’s computation for the given _ID, recomputing if necessary.
current_value(_ID) -> Any: Return the currently stored value for the given _ID without recomputation.
new(_ID) -> Any: Reset and recompute the value of this node at the given _ID.
__hash__() -> int: Return a unique hash for this node.
__getitem__(idx) -> DeepTrackNode: Creates a new node that indexes into this node’s computed data.
__repr__(self) -> str:: Return a string representation of the node.

Supported Operators#

DeepTrackNode supports the following Python operators:

Arithmetic:

Addition (__add__, __radd__)

Subtraction (__sub__, __rsub__)

Multiplication (__mul__, __rmul__)

/ True division (__truediv__, __rtruediv__) // Floor division (__floordiv__, __rfloordiv__)

Comparison:

< Less than (__lt__, __gt__) > Greater than (__gt__, __lt__) <= Less than or equal (__le__, __ge__) >= Greater than or equal (__ge__, __le__)

Each operation returns a new DeepTrackNode representing the result of the corresponding operation in the computation graph.

Examples#

>>> from deeptrack import DeepTrackNode

Create three DeepTrackNode objects, as parent, child, and grandchild:

>>> parent = DeepTrackNode(
...     node_name="parent",
...     action=lambda: 10,
... )
>>> child = DeepTrackNode(
...     node_name="child",
...     action=lambda _ID=None: parent(_ID) * 2,
... )
>>> grandchild = DeepTrackNode(
...     node_name="grandchild",
...     action=lambda _ID=None: child(_ID) * 3,
... )
>>> parent.add_child(child)
>>> child.add_child(grandchild)

Check all children of parent (includes parent itself):

>>> for node in parent.recurse_children():
...     print(node)
DeepTrackNode(name='parent', len=0, action=<lambda>)
DeepTrackNode(name='child', len=0, action=<lambda>)
DeepTrackNode(name='grandchild', len=0, action=<lambda>)

Print the children tree:

>>> parent.print_children_tree()
- DeepTrackNode 'parent' at 0x334202650
    - DeepTrackNode 'child' at 0x334201cf0
        - DeepTrackNode 'grandchild' at 0x334201ea0

Check all dependencies of grandchild (includes grandchild itself):

>>> for node in grandchild.recurse_dependencies():
...     print(node)
DeepTrackNode(name='grandchild', len=0, action=<lambda>)
DeepTrackNode(name='child', len=0, action=<lambda>)
DeepTrackNode(name='parent', len=0, action=<lambda>)

Print the dependency tree:

>>> grandchild.print_dependencies_tree()
- DeepTrackNode 'grandchild' at 0x334201ea0
    - DeepTrackNode 'child' at 0x334201cf0
        - DeepTrackNode 'parent' at 0x334202650

Store and retrieve data for specific _IDs:

>>> parent.store(15, _ID=(0,))
>>> parent.store(20, _ID=(1,))
>>> parent.current_value((0,))
15
>>> parent.current_value((1,))
20

Compute and retrieve the value for the child and grandchild node:

>>> child(_ID=(0,))
30
>>> child(_ID=(1,))
40
>>> grandchild(_ID=(0,))
90
>>> grandchild(_ID=(1,))
120

Validation and invalidation:

>>> parent.is_valid((0,))
True
>>> child.is_valid((0,))
True
>>> grandchild.is_valid((0,))
True

>>> parent.invalidate((0,))  # Also invalidate child and grandchild
>>> parent.is_valid((0,))
False
>>> child.is_valid((0,))
False
>>> grandchild.is_valid((0,))
False

>>> child.validate((0,))
>>> parent.is_valid((0,))
False
>>> child.is_valid((0,))
True
>>> grandchild.is_valid((0,))
False

Setting a value and automatic invalidation:

>>> parent.current_value((0,))
15
>>> grandchild((0,))  # Computes and stores the value in grandchild
>>> grandchild.current_value((0,))
90

>>> parent.set_value(42, _ID=(0,))
>>> parent.current_value((0,))
42
>>> grandchild((0,))  # Recomputes and stores the value in grandchild
>>> grandchild.current_value((0,))
252

Resetting all data in the dependency tree (recomputation required):

>>> grandchild.update()
>>> grandchild()
60

This is equivalent to: >>> grandchild.new() 60

Operator overloading—arithmetic and comparison:

>>> node_a = DeepTrackNode(lambda: 5)
>>> node_b = DeepTrackNode(lambda: 3)

>>> sum_node = node_a + node_b
>>> sum_node()
8

>>> diff_node = node_a - node_b
>>> diff_node()
2

>>> prod_node = node_a * 2
>>> prod_node()
10

>>> div_node = node_a / node_b
>>> div_node()
1.666...

>>> floordiv_node = node_a // node_b
>>> floordiv_node()
1

>>> lt_node = node_a < node_b
>>> lt_node()
False

>>> ge_node = node_a >= node_b
>>> ge_node()
True

Indexing into computed data:

>>> vector_node = DeepTrackNode(lambda: [10, 20, 30])
>>> first_element = vector_node[0]
>>> first_element()
10

Accessing a value before computing it raises an error:

>>> new_node = DeepTrackNode(lambda: 123)
>>> new_node.is_valid((42,))
False

>>> new_node.current_value((42,))
KeyError: 'Attempting to index an empty dict.'

Working with nested _ID slicing:

>>> parent = DeepTrackNode(lambda: 5)
>>> child = DeepTrackNode(lambda _ID=None: parent(_ID[:1]) + _ID[1])
>>> parent.add_child(child)
>>> child((0, 3))  # Equivalent to parent((0,)) + 3
8

Citations for a node and its dependencies:

>>> parent.get_citations()  # Get of citation strings
{...}

Attributes Summary

`action`	Get the function used to compute this node's value.
`children`	Access the children of the node (read-only).
`dependencies`	Access the dependencies of the node (read-only).

Methods Summary

`__call__`([_ID])	Evaluate this node at _ID.
`add_child`(child)	Add a child node to the current node.
`add_dependency`(parent)	Add a dependency, making this node depend on a parent node.
`current_value`([_ID])	Retrieve the value currently stored at _ID.
`get_citations`()	Get citations from this node and all its dependencies.
`invalidate`([_ID])	Mark this node's data and all its children's data as invalid.
`is_valid`([_ID])	Check whether data for the given _ID is valid.
`new`([_ID])	Reset and recompute the value of this node at the given _ID.
`old_recurse_children`([memory])	Legacy recursive method for traversing children.
`old_recurse_dependencies`([memory])	Legacy recursive method for traversing all dependencies.
`print_children_tree`([indent])	Print a tree of all child nodes (recursively) for debugging.
`print_dependencies_tree`([indent])	Print a tree of all parent nodes (recursively) for debugging.
`recurse_children`()	Return all children of this node.
`recurse_dependencies`()	Return all dependencies of this node.
`set_value`(value[, _ID])	Set a value for this node's data at _ID.
`store`(data[, _ID])	Store computed data in this node.
`update`()	Reset data in all children.
`valid_index`(_ID)	Check if _ID is a valid index for this node's data.
`validate`([_ID])	Mark this node's data as valid.

Attributes Documentation

action#

Get the function used to compute this node’s value.

When accessed, it returns the current action. This is often a function or lambda-function that takes _ID as an optional parameter if _accepts_ID is True.

Returns#

Callable[…, Any]: The function used to compute this node’s value.

children#

Access the children of the node (read-only).

This property exposes the internal _children attribute as a public read-only interface.

Returns#

WeakSet[DeepTrackNode]: A weak set with the children of this node.

dependencies#

Access the dependencies of the node (read-only).

This property exposes the internal _dependencies attribute as a public read-only interface.

Returns#

WeakSet[DeepTrackNode]: A weak set with the dependencies of this node.

Methods Documentation

__call__(_ID: tuple[int, ...] = ()) → Any#

Evaluate this node at _ID.

If valid data is already stored at _ID, it is returned. Otherwise, the node’s action function is called to compute the value, which is then stored and returned. The _ID is passed to action only if it is declared to accept it.

Parameters#

_ID: tuple[int, …], optional: The _ID at which to evaluate the node’s action. Defaults to ().

Returns#

Any: The computed or retrieved data for the given _ID.

add_child(child: DeepTrackNode) → DeepTrackNode#

Add a child node to the current node.

Adds child to self._children, and self to child._dependencies. Also updates _all_children for self and its dependencies, as well as _all_dependencies for self and its children.

Parameters#

child: DeepTrackNode: The child node that depends on this node.

Returns#

self: DeepTrackNode: Return the current node for chaining.

Raises#

ValueError: If adding this child would introduce a cycle in the dependency graph.

add_dependency(parent: DeepTrackNode) → DeepTrackNode#

Add a dependency, making this node depend on a parent node.

Adds parent to self._dependencies and self to parent._children. Also updates _all_children for parent and its dependencies, as well as _all_dependencies for self and its children.

Parameters#

parent: DeepTrackNode: The parent node that this node depends on. If parent changes, this node’s data becomes invalid.

Returns#

self: DeepTrackNode: Return the current node for chaining.

Raises#

ValueError: If adding this parent would introduce a cycle in the dependency graph.

current_value(_ID: tuple[int, ...] = ()) → Any#

Retrieve the value currently stored at _ID.

Parameters#

_ID: tuple[int, …], optional: The _ID at which to retrieve the current value. Defaults to ().

Returns#

Any: The currently stored value for _ID.

get_citations() → set[str]#

Get citations from this node and all its dependencies.

Gathers citations from this node and all nodes that it depends on. Citations are stored as the class attribute _citations.

Returns#

set[str]: Set of all citations relevant to this node and its dependency tree.

invalidate(_ID: tuple[int, ...] = ()) → DeepTrackNode#

Mark this node’s data and all its children’s data as invalid.

Parameters#

_ID: tuple[int, …], optional: The _ID to invalidate. Default is empty tuple, invalidating all cached entries. If _ID is shorter than keylength, invalidates entries matching prefix; if longer, trims.

Returns#

self: DeepTrackNode: Return the current node for chaining.

is_valid(_ID: tuple[int, ...] = ()) → bool#

Check whether data for the given _ID is valid.

Parameters#

_ID: tuple[int, …], optional: The _ID to check validity for.

Returns#

bool: True if data at _ID is valid, otherwise False.

new(_ID: tuple[int, ...] = ()) → Any#

Reset and recompute the value of this node at the given _ID.

Clears the stored data in this node and its dependencies, then immediately computes and returns the new value for the given _ID.

Parameters#

_ID: tuple[int, …], optional: The identifier for which the value should be recomputed. Defaults to an empty tuple.

Returns#

Any: The newly computed value at the given _ID.

old_recurse_children(memory: list[DeepTrackNode] | None = None) → Iterator[DeepTrackNode]#

Legacy recursive method for traversing children.

Parameters#

memory: list, optional: A list to remember visited nodes, ensuring that each node is yielded only once.

Yields#

DeepTrackNode: Yields each node in a depth-first traversal.

Notes#

This method is kept for backward compatibility or debugging purposes.

old_recurse_dependencies(memory: list[DeepTrackNode] | None = None) → Iterator[DeepTrackNode]#

Legacy recursive method for traversing all dependencies.

Parameters#

memory: list, optional: A list of visited nodes to avoid repeated visits or infinite loops.

Yields#

DeepTrackNode: Yields this node and all nodes it depends on.

Notes#

This method is kept for backward compatibility or debugging purposes.

print_children_tree(indent: int = 0) → None#

Print a tree of all child nodes (recursively) for debugging.

Parameters#

indent: int, optional: The indentation level (used internally during recursion).

print_dependencies_tree(indent: int = 0) → None#

Print a tree of all parent nodes (recursively) for debugging.

Parameters#

indent: int, optional: The indentation level (used internally during recursion).

recurse_children() → WeakSet[DeepTrackNode]#

Return all children of this node.

Returns#

WeakSet[DeepTrackNode]: All nodes in the subtree rooted at this node, including itself.

recurse_dependencies() → WeakSet[DeepTrackNode]#

Return all dependencies of this node.

Returns#

WeakSet[DeepTrackNode]: All the dependencies of this node, including itself.

set_value(value: Any, _ID: tuple[int, ...] = ()) → DeepTrackNode#

Set a value for this node’s data at _ID.

If the value is different from the currently stored one (or if it is invalid), it will invalidate the old data before storing the new one.

Parameters#

value: Any: The value to store.
_ID: tuple[int, …], optional: The _ID at which to store the value. Defaults to ().

Returns#

self: DeepTrackNode: Return the current node for chaining.

store(data: Any, _ID: tuple[int, ...] = ()) → DeepTrackNode#

Store computed data in this node.

Parameters#

data: Any: The data to be stored.
_ID: tuple[int, …], optional: The index for this data. If _ID does not exist, it creates it. Defaults to (), indicating a root-level entry.

Returns#

self: DeepTrackNode: Return the current node for chaining.

update() → DeepTrackNode#

Reset data in all children.

This method resets data for all children of each dependency, effectively clearing cached values to force a recomputation on the next evaluation.

Returns#

self: DeepTrackNode: Return the current node for chaining.

valid_index(_ID: tuple[int, ...]) → bool#

Check if _ID is a valid index for this node’s data.

Parameters#

_ID: tuple[int, …]: The _ID to validate.

Returns#

bool: True if _ID is valid, otherwise False.

validate(_ID: tuple[int, ...] = ()) → DeepTrackNode#

Mark this node’s data as valid.

Parameters#

_ID: tuple[int, …], optional: The _ID to validate. Defaults to empty tuple, validating all cached entries. Validation is applied only to this node, not its children.

Returns#

self: DeepTrackNode