DeepTrackNode#

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

Bases: object

Object corresponding to a node in a computation graph.

DeepTrackNode represents a node within a DeepTrack2 computation graph. In the 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 method.

Attributes#

dataDeepTrackDataDict

Dictionary-like object for storing data, indexed by tuples of integers.

childrenWeakSet[DeepTrackNode]

Nodes that depend on this node (its children, grandchildren, etc.).

dependenciesWeakSet[DeepTrackNode]

Nodes on which this node depends (its parents, grandparents, etc.).

_actionCallable

The function or lambda-function to compute the node value.

_accepts_IDbool

Whether action accepts an input ID.

_all_childrenSet[DeepTrackNode]

All nodes in the subtree rooted at the node, including the node itself.

_citationsList[str]

Citations associated with this node.

Methods#

actionproperty

Gets or sets the computation function for the node.

add_child(child: DeepTrackNode) -> DeepTrackNode

Adds a child node that depends on this node. Also adds the dependency in the child node on this node.

add_dependency(parent: DeepTrackNode) -> DeepTrackNode

Adds a dependency, making this node depend on the parent node. It also sets this node as a child of the parent node.

store(data: Any, _ID: Tuple[int, …] = ()) -> DeepTrackNode

Stores computed data for the given _ID.

is_valid(_ID: Tuple[int, …] = ()) -> bool

Checks if the data for the given _ID is valid.

valid_index(_ID: Tuple[int, …]) -> bool

Checks if the given _ID is valid for this node.

invalidate(_ID: Tuple[int, …] = ()) -> DeepTrackNode

Invalidates the data for the given _ID and all child nodes.

validate(_ID: Tuple[int, …] = ()) -> DeepTrackNode

Validates the data for the given _ID, marking it as up-to-date, but not its children.

update() -> DeepTrackNode

Resets the data.

set_value(value: Any, _ID: Tuple[int, …] = ()) -> DeepTrackNode

Sets a value for the given _ID. If the new value differs from the current value, the node is invalidated to ensure dependencies are recomputed.

previous(_ID: Tuple[int, …] = ()) -> Any

Returns the previously stored value for the given _ID without recomputing it.

recurse_children(memory: Optional[Set[DeepTrackNode]] = None) -> Set[DeepTrackNode]

Returns all child nodes in the dependency tree rooted at this node.

recurse_dependencies(memory: Optional[List[DeepTrackNode]] = None) -> Iterator[DeepTrackNode]

Yields all nodes that this node depends on, traversing dependencies.

get_citations() -> Set[str]

Returns a set of citations for this node and its dependencies.

__call__(_ID: Tuple[int, …] = ()) -> Any

Evaluates the node’s computation for the given _ID, recomputing if necessary.

current_value(_ID: Tuple[int, …] = ()) -> Any

Returns the currently stored value for the given _ID without recomputation.

__hash__() -> int

Returns a unique hash for this node.

__getitem__(idx: Any) -> DeepTrackNode

Creates a new node that indexes into this node’s computed data.

Example#

Create two DeepTrackNode objects:

>>> parent = DeepTrackNode(action=lambda: 10)
>>> child = DeepTrackNode(action=lambda _ID=None: parent(_ID) * 2)

First, establish the dependency between parent and child:

>>> parent.add_child(child)

Store values in the parent node for specific IDs:

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

Compute the values for the child node based on these parent values:

>>> child_value_0 = child(_ID=(0,))
>>> child_value_1 = child(_ID=(1,))
>>> print(child_value_0, child_value_1)
30 40

Invalidate the parent data for a specific ID:

>>> parent.invalidate((0,))
>>> print(parent.is_valid((0,)))
False
>>> print(child.is_valid((0,)))
False

Update the parent value and recompute the child value:

>>> parent.store(25, _ID=(0,))
>>> child_value_recomputed = child(_ID=(0,))
>>> print(child_value_recomputed)
50

Attributes Summary

action

Callable: The function that computes this node’s value.

Methods Summary

__call__([_ID])

Evaluate this node at ID.

add_child(child)

Add a child node to the current node.

add_dependency(parent)

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

current_value([_ID])

Retrieve the currently stored value 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 if data for the given ID is valid.

old_recurse_children([memory])

Legacy recursive method for traversing children.

previous([_ID])

Retrieve the previously stored value at ID without recomputing.

recurse_children([memory])

Return all children of this node.

recurse_dependencies([memory])

Yield all dependencies of this node, ensuring each is visited once.

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#

Callable: The function that computes this node’s value.

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

Methods Documentation

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

Evaluate this node at ID.

If the data at _ID is valid, it returns the stored value. Otherwise, it calls action to compute a new value, stores it, and returns it.

Parameters#

_IDTuple[int, …], optional

The ID at which to evaluate the node’s action.

Returns#

Any

The computed or retrieved data for the given _ID.

add_child(child: DeepTrackNode) → DeepTrackNode#

Add a child node to the current node.

Adding a child also updates _all_children for this node and all its dependencies. It also ensures that dependency and child relationships remain consistent.

Parameters#

childDeepTrackNode

The child node that depends on this node.

Returns#

selfDeepTrackNode

Returns the current node for chaining.

add_dependency(parent: DeepTrackNode) → DeepTrackNode#

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

Parameters#

parentDeepTrackNode

The parent node that this node depends on. If parent changes, this node’s data may become invalid.

Returns#

selfDeepTrackNode

Returns the current node for chaining.

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

Retrieve the currently stored value at ID.

Parameters#

_IDTuple[int, …], optional

The ID at which to retrieve the current value.

Returns#

Any

The currently stored value for _ID.

get_citations() → Set[str]#

Get citations from this node and all its dependencies.

It 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#

_IDTuple[int, …], optional

The ID to invalidate. Default is empty tuple, indicating potentially the full dataset.

Returns#

selfDeepTrackNode

Returns the current node for chaining.

Note#

At the moment, the code to invalidate specific IDs is not implemented, so the _ID parameter is not effectively used.

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

Check if data for the given ID is valid.

Parameters#

_IDTuple[int, …], optional

The ID to check validity for.

Returns#

bool

True if data at _ID is valid, otherwise False.

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

Legacy recursive method for traversing children.

Parameters#

memorylist, 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.

previous(_ID: Tuple[int, ...] = ()) → Any#

Retrieve the previously stored value at ID without recomputing.

Parameters#

_IDTuple[int, …], optional

The ID for which to retrieve the previous value.

Returns#

Any

The previously stored value if _ID is valid. Returns [] if _ID is not a valid index.

recurse_children(memory: Set[DeepTrackNode] | None = None) → Set[DeepTrackNode]#

Return all children of this node.

Parameters#

memoryset, optional

Memory set to track visited nodes (not used directly here).

Returns#

set

All nodes in the subtree rooted at this node, including itself.

recurse_dependencies(memory: List[DeepTrackNode] | None = None) → Iterator[DeepTrackNode]#

Yield all dependencies of this node, ensuring each is visited once.

Parameters#

memorylist, optional

A list of visited nodes to avoid repeated visits or infinite loops.

Yields#

DeepTrackNode

Yields this node and all nodes it depends on.

set_value(value, _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#

valueAny

The value to store.

_IDTuple[int, …], optional

The ID at which to store the value.

Returns#

selfDeepTrackNode

Returns the current node for chaining.

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

Store computed data in this node.

Parameters#

dataAny

The data to be stored.

_IDTuple[int, …], optional

The index for this data. Default is the empty tuple (), indicating a root-level entry.

Returns#

selfDeepTrackNode

Returns 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#

selfDeepTrackNode

Returns the current node for chaining.

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

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

Parameters#

_IDTuple[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#

_IDTuple[int, …], optional

The ID to validate. Default is empty tuple.

Returns#

self : DeepTrackNode