API Reference¶

`loman` ¶

Loman: A Python library for building computation graphs.

Loman provides tools for creating and managing dependency-aware computation graphs where nodes represent data or calculations, and edges represent dependencies.

`Computation` ¶

A computation graph that manages dependencies and calculations.

The Computation class provides a framework for building and executing computation graphs where nodes represent data or calculations, and edges represent dependencies between them.

`init(*, default_executor: Executor | None = None, executor_map: dict[str, Executor] | None = None, metadata: dict[str, Any] | None = None) -> None` ¶

Initialize a new Computation.

:param default_executor: An executor :type default_executor: concurrent.futures.Executor, default ThreadPoolExecutor(max_workers=1)

`get_attribute_view_for_path(nodekey: NodeKey, get_one_func: Callable[[Name], Any], get_many_func: Callable[[Name | Names], Any]) -> AttributeView` ¶

Create an attribute view for a specific node path.

`add_node(name: Name, func: Callable[..., Any] | None = None, *, args: list[Any] | None = None, kwds: dict[str, Any] | None = None, value: Any = _MISSING_VALUE_SENTINEL, converter: Callable[[Any], Any] | None = None, serialize: bool = True, inspect: bool = True, group: str | None = None, tags: Iterable[str] | None = None, style: str | None = None, executor: str | None = None, metadata: dict[str, Any] | None = None) -> None` ¶

Adds or updates a node in a computation.

:param name: Name of the node to add. This may be any hashable object. :param func: Function to use to calculate the node if the node is a calculation node. By default, the input nodes to the function will be implied from the names of the function parameters. For example, a parameter called a would be taken from the node called a. This can be modified with the kwds parameter. :type func: Function, default None :param args: Specifies a list of nodes that will be used to populate arguments of the function positionally for a calculation node. e.g. If args is ['a', 'b', 'c'] then the function would be called with three parameters, taken from the nodes 'a', 'b' and 'c' respectively. :type args: List, default None :param kwds: Specifies a mapping from parameter name to the node that should be used to populate that parameter when calling the function for a calculation node. e.g. If args is {'x': 'a', 'y': 'b'} then the function would be called with parameters named 'x' and 'y', and their values would be taken from nodes 'a' and 'b' respectively. Each entry in the dictionary can be read as "take parameter [key] from node [value]". :type kwds: Dictionary, default None :param value: If given, the value is inserted into the node, and the node state set to UPTODATE. :type value: default None :param serialize: Whether the node should be serialized. Some objects cannot be serialized, in which case, set serialize to False :type serialize: boolean, default True :param inspect: Whether to use introspection to determine the arguments of the function, which can be slow. If this is not set, kwds and args must be set for the function to obtain parameters. :type inspect: boolean, default True :param group: Subgraph to render node in :type group: default None :param tags: Set of tags to apply to node :type tags: Iterable :param styles: Style to apply to node :type styles: String, default None :param executor: Name of executor to run node on :type executor: string

`set_tag(name: Name | Names, tag: str | Iterable[str]) -> None` ¶

Set tags on a node or nodes. Ignored if tags are already set.

:param name: Node or nodes to set tag for :param tag: Tag to set

`clear_tag(name: Name | Names, tag: str | Iterable[str]) -> None` ¶

Clear tag on a node or nodes. Ignored if tags are not set.

:param name: Node or nodes to clear tags for :param tag: Tag to clear

`set_style(name: Name | Names, style: str) -> None` ¶

Set styles on a node or nodes.

:param name: Node or nodes to set style for :param style: Style to set

`clear_style(name: Name | Names) -> None` ¶

Clear style on a node or nodes.

:param name: Node or nodes to clear styles for

`metadata(name: Name) -> dict[str, Any]` ¶

Get metadata for a node.

`delete_node(name: Name) -> None` ¶

Delete a node from a computation.

When nodes are explicitly deleted with delete_node, but are still depended on by other nodes, then they will be set to PLACEHOLDER status. In this case, if the nodes that depend on a PLACEHOLDER node are deleted, then the PLACEHOLDER node will also be deleted.

:param name: Name of the node to delete. If the node does not exist, a NonExistentNodeException will be raised.

`rename_node(old_name: Name | Mapping[Name, Name], new_name: Name | None = None) -> None` ¶

Rename a node in a computation.

:param old_name: Node to rename, or a dictionary of nodes to rename, with existing names as keys, and new names as values :param new_name: New name for node.

`repoint(old_name: Name, new_name: Name) -> None` ¶

Changes all nodes that use old_name as an input to use new_name instead.

Note that if old_name is an input to new_name, then that will not be changed, to try to avoid introducing circular dependencies, but other circular dependencies will not be checked.

If new_name does not exist, then it will be created as a PLACEHOLDER node.

:param old_name: :param new_name: :return:

`insert(name: Name, value: Any, force: bool = False) -> None` ¶

Insert a value into a node of a computation.

Following insertation, the node will have state UPTODATE, and all its descendents will be COMPUTABLE or STALE.

If an attempt is made to insert a value into a node that does not exist, a NonExistentNodeException will be raised.

:param name: Name of the node to add. :param value: The value to be inserted into the node. :param force: Whether to force recalculation of descendents if node value and state would not be changed

`insert_many(name_value_pairs: Iterable[tuple[Name, object]]) -> None` ¶

Insert values into many nodes of a computation simultaneously.

Following insertation, the nodes will have state UPTODATE, and all their descendents will be COMPUTABLE or STALE. In the case of inserting many nodes, some of which are descendents of others, this ensures that the inserted nodes have correct status, rather than being set as STALE when their ancestors are inserted.

If an attempt is made to insert a value into a node that does not exist, a NonExistentNodeException will be raised, and none of the nodes will be inserted.

:param name_value_pairs: Each tuple should be a pair (name, value), where name is the name of the node to insert the value into. :type name_value_pairs: List of tuples

`insert_from(other: Computation, nodes: Iterable[Name] | None = None) -> None` ¶

Insert values into another Computation object into this Computation object.

:param other: The computation object to take values from :type Computation: :param nodes: Only populate the nodes with the names provided in this list. By default, all nodes from the other Computation object that have corresponding nodes in this Computation object will be inserted :type nodes: List, default None

`set_stale(name: Name) -> None` ¶

Set the state of a node and all its dependencies to STALE.

:param name: Name of the node to set as STALE.

`pin(name: Name, value: Any = None) -> None` ¶

Set the state of a node to PINNED.

:param name: Name of the node to set as PINNED. :param value: Value to pin to the node, if provided. :type value: default None

`unpin(name: Name) -> None` ¶

Unpin a node (state of node and all descendents will be set to STALE).

:param name: Name of the node to set as PINNED.

`get_definition_args_kwds(name: Name) -> tuple[list[Any], dict[str, Any]]` ¶

Get the arguments and keyword arguments for a node's function definition.

`compute(name: Name | Iterable[Name], raise_exceptions: bool = False) -> None` ¶

Compute a node and all necessary predecessors.

Following the computation, if successful, the target node, and all necessary ancestors that were not already UPTODATE will have been calculated and set to UPTODATE. Any node that did not need to be calculated will not have been recalculated.

If any nodes raises an exception, then the state of that node will be set to ERROR, and its value set to an object containing the exception object, as well as a traceback. This will not halt the computation, which will proceed as far as it can, until no more nodes that would be required to calculate the target are COMPUTABLE.

:param name: Name of the node to compute :param raise_exceptions: Whether to pass exceptions raised by node computations back to the caller :type raise_exceptions: Boolean, default False

`compute_all(raise_exceptions: bool = False) -> None` ¶

Compute all nodes of a computation that can be computed.

Nodes that are already UPTODATE will not be recalculated. Following the computation, if successful, all nodes will have state UPTODATE, except UNINITIALIZED input nodes and PLACEHOLDER nodes.

If any nodes raises an exception, then the state of that node will be set to ERROR, and its value set to an object containing the exception object, as well as a traceback. This will not halt the computation, which will proceed as far as it can, until no more nodes are COMPUTABLE.

:param raise_exceptions: Whether to pass exceptions raised by node computations back to the caller :type raise_exceptions: Boolean, default False

`nodes() -> list[Name]` ¶

Get a list of nodes in this computation.

:return: List of nodes.

`get_tree_list_children(name: Name) -> set[Name]` ¶

Get a list of nodes in this computation.

:return: List of nodes.

`has_node(name: Name) -> bool` ¶

Check if a node with the given name exists in the computation.

`tree_has_path(name: Name) -> bool` ¶

Check if a hierarchical path exists in the computation tree.

`get_tree_descendents(name: Name | None = None, *, include_stem: bool = True, graph_nodes_only: bool = False) -> set[Name]` ¶

Get a list of descendent blocks and nodes.

Returns blocks and nodes that are descendents of the input node, e.g. for node 'foo', might return ['foo/bar', 'foo/baz'].

:param name: Name of node to get descendents for :return: List of descendent node names

`state(name: Name | Names) -> States | list[States]` ¶

state(name: Name) -> States

state(name: Names) -> list[States]

Get the state of a node.

This can also be accessed using the attribute-style accessor s if name is a valid Python attribute name::

>>> comp = Computation()
>>> comp.add_node('foo', value=1)
>>> comp.state('foo')
<States.UPTODATE: 4>
>>> comp.s.foo
<States.UPTODATE: 4>

:param name: Name or names of the node to get state for :type name: Name or Names

`value(name: Name | Names) -> Any | list[Any]` ¶

value(name: Name) -> Any

value(name: Names) -> list[Any]

Get the current value of a node.

This can also be accessed using the attribute-style accessor v if name is a valid Python attribute name::

>>> comp = Computation()
>>> comp.add_node('foo', value=1)
>>> comp.value('foo')
1
>>> comp.v.foo
1

:param name: Name or names of the node to get the value of :type name: Name or Names

`compute_and_get_value(name: Name) -> Any` ¶

Get the current value of a node.

This can also be accessed using the attribute-style accessor v if name is a valid Python attribute name::

>>> comp = Computation()
>>> comp.add_node('foo', value=1)
>>> comp.add_node('bar', lambda foo: foo + 1)
>>> comp.compute_and_get_value('bar')
2
>>> comp.x.bar
2

:param name: Name or names of the node to get the value of :type name: Name

`tags(name: Name | Names) -> set[str] | list[set[str]]` ¶

tags(name: Name) -> set[str]

tags(name: Names) -> list[set[str]]

Get the tags associated with a node.

>>> comp = Computation()
>>> comp.add_node('a', tags=['foo', 'bar'])
>>> sorted(comp.t.a)
['__serialize__', 'bar', 'foo']

:param name: Name or names of the node to get the tags of :return:

`nodes_by_tag(tag: str | Iterable[str]) -> set[Name]` ¶

Get the names of nodes with a particular tag or tags.

:param tag: Tag or tags for which to retrieve nodes :return: Names of the nodes with those tags

`styles(name: Name | Names) -> str | None | list[str | None]` ¶

styles(name: Name) -> str | None

styles(name: Names) -> list[str | None]

Get the tags associated with a node.

>>> comp = Computation()
>>> comp.add_node('a', style='dot')
>>> comp.style.a
'dot'

:param name: Name or names of the node to get the tags of :return:

`getitem(name: Name | Names) -> NodeData | list[NodeData]` ¶

__getitem__(name: Name) -> NodeData

__getitem__(name: Names) -> list[NodeData]

Get the state and current value of a node.

:param name: Name of the node to get the state and value of

`get_timing(name: Name | Names) -> TimingData | None | list[TimingData | None]` ¶

get_timing(name: Name) -> TimingData | None

get_timing(name: Names) -> list[TimingData | None]

Get the timing information for a node.

:param name: Name or names of the node to get the timing information of :return:

`to_df() -> pd.DataFrame` ¶

Get a dataframe containing the states and value of all nodes of computation.

::

>>> import loman
>>> comp = loman.Computation()
>>> comp.add_node('foo', value=1)
>>> comp.add_node('bar', value=2)
>>> comp.to_df()  # doctest: +NORMALIZE_WHITESPACE
               state  value
foo  States.UPTODATE      1
bar  States.UPTODATE      2

`to_dict() -> dict[NodeKey, Any]` ¶

Get a dictionary containing the values of all nodes of a computation.

::

>>> import loman
>>> comp = loman.Computation()
>>> comp.add_node('foo', value=1)
>>> comp.add_node('bar', value=2)
>>> comp.to_dict()  # doctest: +ELLIPSIS
{NodeKey('foo'): 1, NodeKey('bar'): 2}

`get_inputs(name: Name | Names) -> Names | list[Names]` ¶

get_inputs(name: Name) -> Names

get_inputs(name: Names) -> list[Names]

Get a list of the inputs for a node or set of nodes.

:param name: Name or names of nodes to get inputs for :return: If name is scalar, return a list of upstream nodes used as input. If name is a list, return a list of list of inputs.

`get_ancestors(names: Name | Names, include_self: bool = True) -> Names` ¶

Get all ancestor nodes of the specified nodes.

`get_original_inputs(names: Name | Names | None = None) -> Names` ¶

Get a list of the original non-computed inputs for a node or set of nodes.

:param names: Name or names of nodes to get inputs for :return: Return a list of original non-computed inputs that are ancestors of the input nodes

`get_outputs(name: Name | Names) -> Names | list[Names]` ¶

get_outputs(name: Name) -> Names

get_outputs(name: Names) -> list[Names]

Get a list of the outputs for a node or set of nodes.

:param name: Name or names of nodes to get outputs for :return: If name is scalar, return a list of downstream nodes used as output. If name is a list, return a list of list of outputs.

`get_descendents(names: Name | Names, include_self: bool = True) -> Names` ¶

Get all descendent nodes of the specified nodes.

`get_final_outputs(names: Name | Names | None = None) -> Names` ¶

Get final output nodes (nodes with no descendants) from the specified nodes.

`get_source(name: Name) -> str` ¶

Get the source code for a node.

`print_source(name: Name) -> None` ¶

Print the source code for a computation node.

`restrict(output_names: Name | Names, input_names: Name | Names | None = None) -> None` ¶

Restrict a computation to the ancestors of a set of output nodes.

Excludes ancestors of a set of input nodes.

If the set of input_nodes that is specified is not sufficient for the set of output_nodes then additional nodes that are ancestors of the output_nodes will be included, but the input nodes specified will be input nodes of the modified Computation.

:param output_nodes: :param input_nodes: :return: None - modifies existing computation in place

`getstate() -> dict[str, Any]` ¶

Prepare computation for serialization by removing non-serializable nodes.

`setstate(state: dict[str, Any]) -> None` ¶

Restore computation from serialized state.

`write_dill_old(file_: str | BinaryIO) -> None` ¶

Serialize a computation to a file or file-like object.

:param file_: If string, writes to a file :type file_: File-like object, or string

`write_dill(file_: str | BinaryIO) -> None` ¶

Serialize a computation to a file or file-like object.

.. deprecated:: Use :meth:write_json instead. dill-based serialization will be removed in a future release.

:param file_: If string, writes to a file :type file_: File-like object, or string

`read_dill(file_: str | BinaryIO) -> Computation` `staticmethod` ¶

Deserialize a computation from a file or file-like object.

.. deprecated:: Use :meth:read_json instead. dill-based serialization will be removed in a future release.

.. warning:: This method uses dill.load() which can execute arbitrary code. Only load files from trusted sources. Never load data from untrusted or unauthenticated sources as it may lead to arbitrary code execution.

:param file_: If string, writes to a file :type file_: File-like object, or string

`write_json(file_: str | TextIO, *, serializer: ComputationSerializer | None = None) -> None` ¶

Serialize a computation to a JSON file or file-like object.

Custom types can be supported by passing a custom serializer — either a :class:~loman.serialization.computation.ComputationSerializer instance with extra transformers registered, or a subclass that overrides the transformer factory.

:param file_: Destination file path (str) or text-mode file-like object. :param serializer: Optional custom serializer. If None the default :class:~loman.serialization.computation.ComputationSerializer is used.

`read_json(file_: str | TextIO, *, serializer: ComputationSerializer | None = None) -> Computation` `staticmethod` ¶

Deserialize a computation from a JSON file or file-like object.

:param file_: Source file path (str) or text-mode file-like object. :param serializer: Optional custom serializer. If None the default :class:~loman.serialization.computation.ComputationSerializer is used. :rtype: Computation

`copy() -> Computation` ¶

Create a copy of a computation.

The copy is shallow. Any values in the new Computation's DAG will be the same object as this Computation's DAG. As new objects will be created by any further computations, this should not be an issue.

:rtype: Computation

`add_named_tuple_expansion(name: Name, namedtuple_type: type, group: str | None = None) -> None` ¶

Automatically add nodes to extract each element of a named tuple type.

It is often convenient for a calculation to return multiple values, and it is polite to do this a namedtuple rather than a regular tuple, so that later users have same name to identify elements of the tuple. It can also help make a computation clearer if a downstream computation depends on one element of such a tuple, rather than the entire tuple. This does not affect the computation per se, but it does make the intention clearer.

To avoid having to create many boiler-plate node definitions to expand namedtuples, the add_named_tuple_expansion method automatically creates new nodes for each element of a tuple. The convention is that an element called 'element', in a node called 'node' will be expanded into a new node called 'node.element', and that this will be applied for each element.

Example::

>>> from collections import namedtuple
>>> Coordinate = namedtuple('Coordinate', ['x', 'y'])
>>> comp = Computation()
>>> comp.add_node('c', value=Coordinate(1, 2))
>>> comp.add_named_tuple_expansion('c', Coordinate)
>>> comp.compute_all()
>>> comp.value('c.x')
1
>>> comp.value('c.y')
2

:param name: Node to cera :param namedtuple_type: Expected type of the node :type namedtuple_type: namedtuple class

`add_map_node(result_node: Name, input_node: Name, subgraph: Computation, subgraph_input_node: Name, subgraph_output_node: Name) -> None` ¶

Apply a graph to each element of iterable.

In turn, each element in the input_node of this graph will be inserted in turn into the subgraph's subgraph_input_node, then the subgraph's subgraph_output_node calculated. The resultant list, with an element or each element in input_node, will be inserted into result_node of this graph. In this way add_map_node is similar to map in functional programming.

:param result_node: The node to place a list of results in this graph :param input_node: The node to get a list input values from this graph :param subgraph: The graph to use to perform calculation for each element :param subgraph_input_node: The node in subgraph to insert each element in turn :param subgraph_output_node: The node in subgraph to read the result for each element

`prepend_path(path: Name | ConstantValue, prefix_path: NodeKey) -> NodeKey | ConstantValue` ¶

Prepend a prefix path to a node path.

`add_block(base_path: Name, block: Computation, *, keep_values: bool | None = True, links: dict[str, Name] | None = None, metadata: dict[str, Any] | None = None) -> None` ¶

Add a computation block as a subgraph to this computation.

`link(target: Name, source: Name) -> None` ¶

Create a link between two nodes in the computation graph.

`draw(root: NodeKey | None = None, *, node_transformations: dict[Name, str] | None = None, cmap: Any = None, colors: str = 'state', shapes: str | None = None, graph_attr: dict[str, Any] | None = None, node_attr: dict[str, Any] | None = None, edge_attr: dict[str, Any] | None = None, show_expansion: bool = False, collapse_all: bool = True) -> GraphView` ¶

Draw a computation's current state using the GraphViz utility.

:param root: Optional PathType. Sub-block to draw :param cmap: Default: None :param colors: 'state' - colors indicate state. 'timing' - colors indicate execution time. Default: 'state'. :param shapes: None - ovals. 'type' - shapes indicate type. Default: None. :param graph_attr: Mapping of (attribute, value) pairs for the graph. For example graph_attr={'size': '"10,8"'} can control the size of the output graph :param node_attr: Mapping of (attribute, value) pairs set for all nodes. :param edge_attr: Mapping of (attribute, value) pairs set for all edges. :param collapse_all: Whether to collapse all blocks that aren't explicitly expanded.

`view(cmap: Any = None, colors: str = 'state', shapes: str | None = None) -> None` ¶

Create and display a visualization of the computation graph.

`print_errors() -> None` ¶

Print tracebacks for every node with state "ERROR" in a Computation.

`from_class(definition_class: type, ignore_self: bool = True) -> Computation` `classmethod` ¶

Create a computation from a class with decorated methods.

`inject_dependencies(dependencies: dict[Name, Any], *, force: bool = False) -> None` ¶

Injects dependencies into the nodes of the current computation where nodes are in a placeholder state.

(or all possible nodes when the 'force' parameter is set to True), using values provided in the 'dependencies' dictionary.

Each key in the 'dependencies' dictionary corresponds to a node identifier, and the associated value is the dependency object to inject. If the value is a callable, it will be added as a calc node.

:param dependencies: A dictionary where each key-value pair consists of a node identifier and its corresponding dependency object or a callable that returns the dependency object. :param force: A boolean flag that, when set to True, forces the replacement of existing node values with the ones provided in 'dependencies', regardless of their current state. Defaults to False. :return: None

`NodeTransformations` ¶

Node transformation types for visualization.

`States` ¶

Bases: Enum

Possible states for a computation node.

`CannotInsertToPlaceholderNodeError` ¶

Bases: ComputationError

Exception raised when trying to insert into a placeholder node.

`FittingError` ¶

Bases: ComputationError

Exception raised when curve fitting exceeds error tolerance.

`InvalidBlockTypeError` ¶

Bases: TypeError, ComputationError

Exception raised when a block is not callable or a Computation.

`LoopDetectedError` ¶

Bases: ComputationError

Exception raised when a dependency loop is detected.

`MapError` ¶

Bases: ComputationError

Exception raised during map operations with partial results.

`init(message: str, results: list[object]) -> None` ¶

Initialize MapError with message and partial results.

`NonExistentNodeError` ¶

Bases: ComputationError

Exception raised when trying to access a non-existent node.

`SerializationError` ¶

Bases: ComputationError

Exception raised during serialization/deserialization.

`ValidationError` ¶

Bases: ComputationError

Exception raised during computation validation.

`NodeKey` `dataclass` ¶

Immutable key for identifying nodes in the computation graph hierarchy.

`name: Name` `property` ¶

Get the name of this node (last part of the path).

`label: str` `property` ¶

Get the label for this node (for display purposes).

`parent: NodeKey` `property` ¶

Get the parent node key.

`is_root: bool` `property` ¶

Check if this is the root node key.

`str() -> str` ¶

Return string representation using path notation.

`drop_root(root: Optional[Name]) -> Optional[NodeKey]` ¶

Remove a root prefix from this node key if it matches.

`join(*others: Name) -> NodeKey` ¶

Join this node key with other names to create a new node key.

`join_parts(*parts: Hashable) -> NodeKey` ¶

Join this node key with raw parts to create a new node key.

`truediv(other: Name) -> NodeKey` ¶

Join this node key with other to create a new node key.

`is_descendent_of(other: NodeKey) -> bool` ¶

Check if this node key is a descendant of another node key.

`prepend(nk: NodeKey) -> NodeKey` ¶

Prepend another node key to this one.

`repr() -> str` ¶

Return string representation for debugging.

`eq(other: object) -> bool` ¶

Check equality with another NodeKey.

`root() -> NodeKey` `classmethod` ¶

Get the root node key.

`common_parent(nodekey1: Name, nodekey2: Name) -> NodeKey` `staticmethod` ¶

Find the common parent of two node keys.

`ancestors() -> list[NodeKey]` ¶

Get all ancestor node keys from root to parent.

`ComputationSerializer` ¶

Serialize and deserialize a :class:~loman.computeengine.Computation graph to JSON.

The serialized format is a JSON object with the following top-level keys:

version: integer format version
nodes: list of node objects
edges: list of edge objects

Each node object has:

key: string representation of the NodeKey
state: name of the :class:~loman.consts.States enum member (or null)
value: transformer-encoded value (or null when absent / not serialized)
has_value: bool — false when the node has no meaningful value to restore
func: transformer-encoded callable (or null)
serialize: bool — whether the node has the __serialize__ tag
tags: list of non-system tags

Each edge object has:

src: string key of the source node
dst: string key of the destination node
param_type: "arg" or "kwd"
param: positional index (int) for args, parameter name (str) for kwds

Parameters¶

transformer: Custom :class:~loman.serialization.transformer.Transformer instance. If None, a default transformer is built based on use_dill_for_functions. use_dill_for_functions: When True, lambdas and closures are serialized as base64-encoded dill blobs rather than raising :class:~loman.exception.SerializationError. Has no effect when a custom transformer is supplied. Defaults to False.

`init(transformer: Transformer | None = None, *, use_dill_for_functions: bool = False) -> None` ¶

Initialise with an optional custom transformer.

`dump(comp: Any, fp: TextIO) -> None` ¶

Serialize comp to fp (a text-mode file-like object).

`dumps(comp: Any) -> str` ¶

Serialize comp and return a JSON string.

`load(fp: TextIO) -> Any` ¶

Deserialize a Computation from fp (a text-mode file-like object).

`loads(s: str) -> Any` ¶

Deserialize a Computation from a JSON string.

`GraphView` `dataclass` ¶

A view for visualizing computation graphs as graphical diagrams.

`__post_init__() -> None` ¶

Initialize the graph view after dataclass construction.

`get_sub_block(dag: nx.DiGraph, root: Name | None, node_transformations: dict[NodeKey, str]) -> tuple[nx.DiGraph, defaultdict[NodeKey, list[NodeKey]], set[NodeKey]]` `staticmethod` ¶

Extract a subgraph with node transformations for visualization.

`refresh() -> None` ¶

Refresh the visualization by rebuilding the graph structure.

`svg() -> str | None` ¶

Generate SVG representation of the visualization.

`view() -> None` ¶

Open the visualization in a PDF viewer.

`calc_node(f: F | None = None, **kwds: Any) -> F | Callable[[F], F]` ¶

calc_node(f: F, **kwds: Any) -> F

calc_node(f: None = None, **kwds: Any) -> Callable[[F], F]

Decorator to mark a function as a calculation node.

`computation_factory(maybe_cls: type | None = None, *, ignore_self: bool = True) -> Callable[..., Computation] | Callable[[type], Callable[..., Computation]]` ¶

Factory function to create computations from class definitions.

`node(comp: Computation, name: Name | None = None, *args: Any, **kw: Any) -> Callable[[F], F]` ¶

Decorator to add a function as a node to a computation graph.

`to_nodekey(name: Name) -> NodeKey` ¶

Convert a name to a NodeKey object.

API Reference¶

loman ¶

Computation ¶

__init__(*, default_executor: Executor | None = None, executor_map: dict[str, Executor] | None = None, metadata: dict[str, Any] | None = None) -> None ¶

get_attribute_view_for_path(nodekey: NodeKey, get_one_func: Callable[[Name], Any], get_many_func: Callable[[Name | Names], Any]) -> AttributeView ¶

set_tag(name: Name | Names, tag: str | Iterable[str]) -> None ¶

clear_tag(name: Name | Names, tag: str | Iterable[str]) -> None ¶

set_style(name: Name | Names, style: str) -> None ¶

clear_style(name: Name | Names) -> None ¶

metadata(name: Name) -> dict[str, Any] ¶

delete_node(name: Name) -> None ¶

rename_node(old_name: Name | Mapping[Name, Name], new_name: Name | None = None) -> None ¶

repoint(old_name: Name, new_name: Name) -> None ¶

insert(name: Name, value: Any, force: bool = False) -> None ¶

insert_many(name_value_pairs: Iterable[tuple[Name, object]]) -> None ¶

insert_from(other: Computation, nodes: Iterable[Name] | None = None) -> None ¶

set_stale(name: Name) -> None ¶

pin(name: Name, value: Any = None) -> None ¶

unpin(name: Name) -> None ¶

get_definition_args_kwds(name: Name) -> tuple[list[Any], dict[str, Any]] ¶

compute(name: Name | Iterable[Name], raise_exceptions: bool = False) -> None ¶

compute_all(raise_exceptions: bool = False) -> None ¶

nodes() -> list[Name] ¶

get_tree_list_children(name: Name) -> set[Name] ¶

has_node(name: Name) -> bool ¶

tree_has_path(name: Name) -> bool ¶

get_tree_descendents(name: Name | None = None, *, include_stem: bool = True, graph_nodes_only: bool = False) -> set[Name] ¶

state(name: Name | Names) -> States | list[States] ¶

value(name: Name | Names) -> Any | list[Any] ¶

compute_and_get_value(name: Name) -> Any ¶

tags(name: Name | Names) -> set[str] | list[set[str]] ¶

nodes_by_tag(tag: str | Iterable[str]) -> set[Name] ¶

styles(name: Name | Names) -> str | None | list[str | None] ¶

__getitem__(name: Name | Names) -> NodeData | list[NodeData] ¶

get_timing(name: Name | Names) -> TimingData | None | list[TimingData | None] ¶

to_df() -> pd.DataFrame ¶

to_dict() -> dict[NodeKey, Any] ¶

get_inputs(name: Name | Names) -> Names | list[Names] ¶

get_ancestors(names: Name | Names, include_self: bool = True) -> Names ¶

get_original_inputs(names: Name | Names | None = None) -> Names ¶

get_outputs(name: Name | Names) -> Names | list[Names] ¶

get_descendents(names: Name | Names, include_self: bool = True) -> Names ¶

get_final_outputs(names: Name | Names | None = None) -> Names ¶

get_source(name: Name) -> str ¶

print_source(name: Name) -> None ¶

restrict(output_names: Name | Names, input_names: Name | Names | None = None) -> None ¶

__getstate__() -> dict[str, Any] ¶

__setstate__(state: dict[str, Any]) -> None ¶

write_dill_old(file_: str | BinaryIO) -> None ¶

write_dill(file_: str | BinaryIO) -> None ¶

read_dill(file_: str | BinaryIO) -> Computation staticmethod ¶

write_json(file_: str | TextIO, *, serializer: ComputationSerializer | None = None) -> None ¶

read_json(file_: str | TextIO, *, serializer: ComputationSerializer | None = None) -> Computation staticmethod ¶

copy() -> Computation ¶

add_named_tuple_expansion(name: Name, namedtuple_type: type, group: str | None = None) -> None ¶

add_map_node(result_node: Name, input_node: Name, subgraph: Computation, subgraph_input_node: Name, subgraph_output_node: Name) -> None ¶

prepend_path(path: Name | ConstantValue, prefix_path: NodeKey) -> NodeKey | ConstantValue ¶

add_block(base_path: Name, block: Computation, *, keep_values: bool | None = True, links: dict[str, Name] | None = None, metadata: dict[str, Any] | None = None) -> None ¶

link(target: Name, source: Name) -> None ¶

view(cmap: Any = None, colors: str = 'state', shapes: str | None = None) -> None ¶

print_errors() -> None ¶

from_class(definition_class: type, ignore_self: bool = True) -> Computation classmethod ¶

inject_dependencies(dependencies: dict[Name, Any], *, force: bool = False) -> None ¶

NodeTransformations ¶

States ¶

CannotInsertToPlaceholderNodeError ¶

FittingError ¶

InvalidBlockTypeError ¶

LoopDetectedError ¶

MapError ¶

__init__(message: str, results: list[object]) -> None ¶

NonExistentNodeError ¶

SerializationError ¶

ValidationError ¶

NodeKey dataclass ¶

name: Name property ¶

label: str property ¶

parent: NodeKey property ¶

is_root: bool property ¶

__str__() -> str ¶

`loman` ¶

`Computation` ¶

`init(*, default_executor: Executor | None = None, executor_map: dict[str, Executor] | None = None, metadata: dict[str, Any] | None = None) -> None` ¶

`get_attribute_view_for_path(nodekey: NodeKey, get_one_func: Callable[[Name], Any], get_many_func: Callable[[Name | Names], Any]) -> AttributeView` ¶

`set_tag(name: Name | Names, tag: str | Iterable[str]) -> None` ¶

`clear_tag(name: Name | Names, tag: str | Iterable[str]) -> None` ¶

`set_style(name: Name | Names, style: str) -> None` ¶

`clear_style(name: Name | Names) -> None` ¶

`metadata(name: Name) -> dict[str, Any]` ¶

`delete_node(name: Name) -> None` ¶

`rename_node(old_name: Name | Mapping[Name, Name], new_name: Name | None = None) -> None` ¶

`repoint(old_name: Name, new_name: Name) -> None` ¶

`insert(name: Name, value: Any, force: bool = False) -> None` ¶

`insert_many(name_value_pairs: Iterable[tuple[Name, object]]) -> None` ¶

`insert_from(other: Computation, nodes: Iterable[Name] | None = None) -> None` ¶

`set_stale(name: Name) -> None` ¶

`pin(name: Name, value: Any = None) -> None` ¶

`unpin(name: Name) -> None` ¶

`get_definition_args_kwds(name: Name) -> tuple[list[Any], dict[str, Any]]` ¶

`compute(name: Name | Iterable[Name], raise_exceptions: bool = False) -> None` ¶

`compute_all(raise_exceptions: bool = False) -> None` ¶

`nodes() -> list[Name]` ¶

`get_tree_list_children(name: Name) -> set[Name]` ¶

`has_node(name: Name) -> bool` ¶

`tree_has_path(name: Name) -> bool` ¶

`get_tree_descendents(name: Name | None = None, *, include_stem: bool = True, graph_nodes_only: bool = False) -> set[Name]` ¶

`state(name: Name | Names) -> States | list[States]` ¶

`value(name: Name | Names) -> Any | list[Any]` ¶

`compute_and_get_value(name: Name) -> Any` ¶

`tags(name: Name | Names) -> set[str] | list[set[str]]` ¶

`nodes_by_tag(tag: str | Iterable[str]) -> set[Name]` ¶

`styles(name: Name | Names) -> str | None | list[str | None]` ¶

`getitem(name: Name | Names) -> NodeData | list[NodeData]` ¶

`get_timing(name: Name | Names) -> TimingData | None | list[TimingData | None]` ¶

`to_df() -> pd.DataFrame` ¶

`to_dict() -> dict[NodeKey, Any]` ¶

`get_inputs(name: Name | Names) -> Names | list[Names]` ¶

`get_ancestors(names: Name | Names, include_self: bool = True) -> Names` ¶

`get_original_inputs(names: Name | Names | None = None) -> Names` ¶

`get_outputs(name: Name | Names) -> Names | list[Names]` ¶

`get_descendents(names: Name | Names, include_self: bool = True) -> Names` ¶

`get_final_outputs(names: Name | Names | None = None) -> Names` ¶

`get_source(name: Name) -> str` ¶

`print_source(name: Name) -> None` ¶

`restrict(output_names: Name | Names, input_names: Name | Names | None = None) -> None` ¶

`getstate() -> dict[str, Any]` ¶

`setstate(state: dict[str, Any]) -> None` ¶

`write_dill_old(file_: str | BinaryIO) -> None` ¶

`write_dill(file_: str | BinaryIO) -> None` ¶

`read_dill(file_: str | BinaryIO) -> Computation` `staticmethod` ¶

`write_json(file_: str | TextIO, *, serializer: ComputationSerializer | None = None) -> None` ¶

`read_json(file_: str | TextIO, *, serializer: ComputationSerializer | None = None) -> Computation` `staticmethod` ¶

`copy() -> Computation` ¶

`add_named_tuple_expansion(name: Name, namedtuple_type: type, group: str | None = None) -> None` ¶

`add_map_node(result_node: Name, input_node: Name, subgraph: Computation, subgraph_input_node: Name, subgraph_output_node: Name) -> None` ¶

`prepend_path(path: Name | ConstantValue, prefix_path: NodeKey) -> NodeKey | ConstantValue` ¶

`add_block(base_path: Name, block: Computation, *, keep_values: bool | None = True, links: dict[str, Name] | None = None, metadata: dict[str, Any] | None = None) -> None` ¶

`link(target: Name, source: Name) -> None` ¶

`view(cmap: Any = None, colors: str = 'state', shapes: str | None = None) -> None` ¶

`print_errors() -> None` ¶

`from_class(definition_class: type, ignore_self: bool = True) -> Computation` `classmethod` ¶

`inject_dependencies(dependencies: dict[Name, Any], *, force: bool = False) -> None` ¶

`NodeTransformations` ¶

`States` ¶

`CannotInsertToPlaceholderNodeError` ¶

`FittingError` ¶

`InvalidBlockTypeError` ¶

`LoopDetectedError` ¶

`MapError` ¶

`init(message: str, results: list[object]) -> None` ¶

`NonExistentNodeError` ¶

`SerializationError` ¶

`ValidationError` ¶

`NodeKey` `dataclass` ¶

`name: Name` `property` ¶

`label: str` `property` ¶

`parent: NodeKey` `property` ¶

`is_root: bool` `property` ¶

`str() -> str` ¶

`drop_root(root: Optional[Name]) -> Optional[NodeKey]` ¶