nndeploy.dag.graph

Functions

deserialize(json_str)

Deserialize graph from JSON string

get_dag_json()

Get DAG JSON representation containing graph, nodes and edge types

get_graph_json()

Get graph JSON representation

load_file(path)

Load graph from file

save_file(graph, path)

Save graph to file

serialize(graph)

Serialize graph to JSON string

Classes

Graph(name[, inputs, outputs])

GraphCreator()

Graph creator for creating graph nodes
class nndeploy.dag.graph.Graph(name: str, inputs: Union[Edge, List[Edge]] = None, outputs: Union[Edge, List[Edge]] = None)[源代码]

基类：Graph

__init__(name: str, inputs: Union[Edge, List[Edge]] = None, outputs: Union[Edge, List[Edge]] = None)[源代码]

Initialize Graph object

参数:

  • name – Graph name

  • inputs – Input edge or list of input edges

  • outputs – Output edge or list of output edges

add_image_url(url: str)[源代码]

Add image URL

参数:

url – URL address

返回:

Status code

remove_image_url(url: str)[源代码]

Remove image URL

参数:

url – URL address

返回:

Status code

add_video_url(url: str)[源代码]

Add video URL

参数:

url – URL address

返回:

Status code

remove_video_url(url: str)[源代码]

Remove video URL

参数:

url – URL address

返回:

Status code

add_audio_url(url: str)[源代码]

Add audio URL

参数:

url – URL address

返回:

Status code

remove_audio_url(url: str)[源代码]

Remove audio URL

参数:

url – URL address

返回:

Status code

add_model_url(url: str)[源代码]

Add model URL

参数:

url – URL address

返回:

Status code

remove_model_url(url: str)[源代码]

Remove model URL

参数:

url – URL address

返回:

Status code

add_other_url(url: str)[源代码]

Add other URL

参数:

url – URL address

返回:

Status code

remove_other_url(url: str)[源代码]

Remove other URL

参数:

url – URL address

返回:

Status code

get_image_url()[源代码]

Get image URL list

返回:

URL address list

get_video_url()[源代码]

Get video URL list

返回:

URL address list

get_audio_url()[源代码]

Get audio URL list

返回:

URL address list

get_model_url()[源代码]

Get model URL list

返回:

URL address list

get_other_url()[源代码]

Get other URL list

返回:

URL address list

set_input_type(input_type: type, desc: str = '')[源代码]

Set input type

参数:

input_type – Input type

返回:

Status code

set_output_type(output_type: type, desc: str = '')[源代码]

Set output type

参数:

output_type – Output type

返回:

Status code

set_edge_queue_max_size(queue_max_size: int)[源代码]

Set edge queue maximum size

参数:

queue_max_size – Maximum queue size

get_edge_queue_max_size() → int[源代码]

Get edge queue maximum size

返回:

Maximum queue size

set_input(input: Edge, index: int = -1)[源代码]

Set input edge

参数:

  • input – Input edge

  • index – Input edge index, default is -1

set_output(output: Edge, index: int = -1)[源代码]

Set output edge

参数:

  • output – Output edge

  • index – Output edge index, default is -1

set_inputs(inputs: List[Edge])[源代码]

Set input edge list

参数:

inputs – List of input edges

set_outputs(outputs: List[Edge])[源代码]

Set output edge list

参数:

outputs – List of output edges

create_edge(name: str) → Edge[源代码]

Create an edge

参数:

name – Edge name

返回:

Edge object

add_edge(edge: Edge)[源代码]

Add an edge

参数:

edge – Edge object

delete_edge(edge: Edge)[源代码]

Delete an edge

参数:

edge – Edge object

update_edge(edge_wrapper: Edge, edge: Edge, is_external: bool = True)[源代码]

Update edge

参数:

  • edge_wrapper – Edge wrapper

  • edge – Edge object

  • is_external – Whether it’s an external edge

get_edge(name: str) → Edge[源代码]

Get edge by name

参数:

name – Edge name

返回:

Edge object

create_node(key_or_desc: Union[str, NodeDesc], name: str = '') → Node[源代码]

Create node

参数:

  • key_or_desc – Node key or node description object

  • name – Node name, only valid when key_or_desc is str, default is empty string

返回:

Node object

set_node_desc(node: Node, desc: NodeDesc)[源代码]

Set node description

参数:

  • node – Node object

  • desc – Node description object

add_node(node: Node)[源代码]

Add node

参数:

node – Node object

delete_node(node: Node)[源代码]

Delete node

参数:

node – Node object

get_node(name_or_index: Union[str, int]) → Node[源代码]

Get node by name or index

参数:

name_or_index – Node name or index

返回:

Node object

get_node_by_key(key: str) → Node[源代码]

Get node by key

参数:

key – Node key

返回:

Node object

get_nodes_by_key(key: str) → List[Node][源代码]

Get all matching nodes by key

参数:

key – Node key

返回:

List of Node objects

get_node_count() → int[源代码]

Get node count in the graph

返回:

Number of nodes

get_nodes() → List[Node][源代码]

Get all nodes in the graph

返回:

List of all nodes

set_node_param(node_name: str, param: Param)[源代码]

Set node parameter

参数:

  • node_name – Node name

  • param – Parameter object

get_node_param(node_name: str) → Param[源代码]

Get node parameter

参数:

node_name – Node name

返回:

Parameter object

set_external_param(key: str, param: Param)[源代码]

Set external parameter

参数:

  • key – Parameter key

  • param – Parameter object

get_external_param(key: str) → Param[源代码]

Get external parameter

参数:

key – Parameter key

返回:

Parameter object

set_node_parallel_type(node_name: str, parallel_type: ParallelType)[源代码]

Set node parallel type

参数:

  • node_name – Node name

  • parallel_type – Parallel type

set_graph_node_share_stream(flag: bool)[源代码]

Set graph node stream sharing flag

参数:

flag – Flag value

get_graph_node_share_stream() → bool[源代码]

Get graph node stream sharing flag

返回:

Stream sharing flag

update_node_io(node: Node, inputs: List[Edge], outputs: List[str])[源代码]

Update node inputs and outputs

参数:

  • node – Node object

  • inputs – List of input edges

  • outputs – List of output edge names

mark_input_edge(inputs: List[Edge])[源代码]

Mark edges as input edges

参数:

inputs – List of input edges

mark_output_edge(outputs: List[Edge])[源代码]

Mark edges as output edges

参数:

outputs – List of output edges

default_param()[源代码]

Initialize graph with default parameters

返回:

Status code

init()[源代码]

Initialize graph

deinit()[源代码]

Deinitialize graph

run()[源代码]

Run graph execution

synchronize()[源代码]

Synchronize graph execution

interrupt()[源代码]

Interrupt graph execution

dump()[源代码]

Dump graph information to standard output

set_trace_flag(flag: bool)[源代码]

Set trace flag

参数:

flag – Trace flag

trace(inputs: Optional[Union[List[Edge], Edge]] = None) → List[Edge][源代码]

Trace graph execution flow

参数:

inputs – List of input edges, single edge or None. If None, use default inputs

返回:

List of traced edges

to_static_graph()[源代码]

Convert to static graph representation

返回:

Static graph object

get_edge_wrapper(edge: Union[Edge, str]) → EdgeWrapper[源代码]

Get edge wrapper

参数:

edge – Edge object or edge name

返回:

Edge wrapper object

get_node_wrapper(node: Union[Node, str]) → NodeWrapper[源代码]

Get node wrapper

参数:

node – Node object or node name

返回:

Node wrapper object

serialize() → str[源代码]

Serialize graph to JSON string

返回:

JSON string representation of the graph

save_file(path: str)[源代码]

Save graph to file

参数:

path – File path

deserialize(json_str: str)[源代码]

Deserialize graph from JSON string

参数:

json_str – JSON string to deserialize

load_file(path: str)[源代码]

Load graph from file

参数:

path – File path

set_node_value(*args)[源代码]

Set node value

Supports multiple calling methods: 1. set_node_value(node_value_str) - Set through JSON string 2. set_node_value(node_name, key, value) - Set single node key-value pair 3. set_node_value(node_value_map) - Set multiple node values through dictionary

get_node_value()[源代码]

Get all node values

返回:

Nested dictionary of node values, format: {node_name: {key: value}}

返回类型:

dict

set_unused_node_names(*args)[源代码]

Set unused node names

参数:

node_name – Node name

remove_unused_node_names(*args)[源代码]

Remove unused node names

参数:

node_name – Node name

get_unused_node_names()[源代码]

Get unused node names

返回:

List of unused node names

返回类型:

list

add_dropdown_param(self: nndeploy._nndeploy_internal.dag.Node, dropdown_param: str, dropdown_values: list[str]) → nndeploy._nndeploy_internal.base.Status
add_io_param(self: nndeploy._nndeploy_internal.dag.Node, io_param: str) → nndeploy._nndeploy_internal.base.Status
add_node_input_and_output(self: nndeploy._nndeploy_internal.dag.Graph, node_wrapper: nndeploy::dag::NodeWrapper, inputs: list[nndeploy::dag::Edge], outputs: list[nndeploy::dag::Edge]) → nndeploy._nndeploy_internal.base.Status
add_required_param(self: nndeploy._nndeploy_internal.dag.Node, required_param: str) → nndeploy._nndeploy_internal.base.Status
add_resource_with_state(self: nndeploy._nndeploy_internal.dag.Graph, key: str, edge: nndeploy::dag::Edge) → nndeploy._nndeploy_internal.base.Status
add_resource_without_state(self: nndeploy._nndeploy_internal.dag.Graph, key: str, value: nndeploy::base::Any) → nndeploy._nndeploy_internal.base.Status
add_ui_param(self: nndeploy._nndeploy_internal.dag.Node, ui_param: str) → nndeploy._nndeploy_internal.base.Status
check_inputs(self: nndeploy._nndeploy_internal.dag.Node, inputs: list[nndeploy::dag::Edge]) → bool
check_outputs(*args, **kwargs)

Overloaded function.

  1. check_outputs(self: nndeploy._nndeploy_internal.dag.Node, outputs_name: list[str]) -> bool

  2. check_outputs(self: nndeploy._nndeploy_internal.dag.Node, outputs: list[nndeploy::dag::Edge]) -> bool

clear_dropdown_params(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.Status
clear_io_params(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.Status
clear_required_params(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.Status
clear_ui_params(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.Status
connect(self: nndeploy._nndeploy_internal.dag.Graph, predecessor: nndeploy._nndeploy_internal.dag.Node, successor: nndeploy._nndeploy_internal.dag.Node, predecessor_port: int = 0, successor_port: int = 0) → nndeploy._nndeploy_internal.base.Status
create_internal_output_edge(self: nndeploy._nndeploy_internal.dag.Node, name: str) → nndeploy::dag::Edge
create_resource_with_state(self: nndeploy._nndeploy_internal.dag.Node, key: str) → nndeploy::dag::Edge
disconnect(self: nndeploy._nndeploy_internal.dag.Graph, predecessor: nndeploy._nndeploy_internal.dag.Node, successor: nndeploy._nndeploy_internal.dag.Node, predecessor_port: int = 0, successor_port: int = 0) → nndeploy._nndeploy_internal.base.Status
forward(*args, **kwargs)

Overloaded function.

  1. forward(self: nndeploy._nndeploy_internal.dag.Graph, inputs: list[nndeploy::dag::Edge]) -> list[nndeploy::dag::Edge]

  2. forward(self: nndeploy._nndeploy_internal.dag.Graph) -> list[nndeploy::dag::Edge]

  3. forward(self: nndeploy._nndeploy_internal.dag.Graph, input: nndeploy::dag::Edge) -> list[nndeploy::dag::Edge]

get_all_input(self: nndeploy._nndeploy_internal.dag.Node) → list[nndeploy::dag::Edge]
get_all_output(self: nndeploy._nndeploy_internal.dag.Node) → list[nndeploy::dag::Edge]
get_completed_size(self: nndeploy._nndeploy_internal.dag.Node) → int
get_composite_node(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy::dag::CompositeNode
get_constructed(self: nndeploy._nndeploy_internal.dag.Node) → bool
get_debug_flag(self: nndeploy._nndeploy_internal.dag.Node) → bool
get_desc(self: nndeploy._nndeploy_internal.dag.Node) → str
get_developer(self: nndeploy._nndeploy_internal.dag.Node) → str
get_device_type(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.DeviceType
get_dropdown_params(self: nndeploy._nndeploy_internal.dag.Node) → dict[str, list[str]]
get_edge_queue_drop_count(self: nndeploy._nndeploy_internal.dag.Graph) → int
get_edge_queue_overflow_policy(self: nndeploy._nndeploy_internal.dag.Graph) → nndeploy._nndeploy_internal.base.QueueOverflowPolicy
get_graph(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy::dag::Graph
get_graph_flag(self: nndeploy._nndeploy_internal.dag.Node) → bool
get_infer_node(self: nndeploy._nndeploy_internal.dag.Graph, index: int) → nndeploy._nndeploy_internal.dag.Node
get_initialized(self: nndeploy._nndeploy_internal.dag.Node) → bool
get_input(self: nndeploy._nndeploy_internal.dag.Node, index: int = 0) → nndeploy::dag::Edge
get_input_count(self: nndeploy._nndeploy_internal.dag.Node) → int
get_input_data(self: nndeploy._nndeploy_internal.dag.Node, index: int = 0) → object
get_input_index(self: nndeploy._nndeploy_internal.dag.Node, name: str) → int
get_input_name(self: nndeploy._nndeploy_internal.dag.Node, index: int = 0) → str
get_input_names(self: nndeploy._nndeploy_internal.dag.Node) → list[str]
get_input_node(self: nndeploy._nndeploy_internal.dag.Graph, index: int) → nndeploy._nndeploy_internal.dag.Node
get_input_type_info(self: nndeploy._nndeploy_internal.dag.Node) → list[nndeploy::dag::EdgeTypeInfo]
get_io_params(self: nndeploy._nndeploy_internal.dag.Node) → list[str]
get_io_type(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy::dag::IOType
get_key(self: nndeploy._nndeploy_internal.dag.Node) → str
get_loop_count(self: nndeploy._nndeploy_internal.dag.Graph) → int
get_loop_count_map(self: nndeploy._nndeploy_internal.dag.Graph) → dict[str, int]
get_memory_size(self: nndeploy._nndeploy_internal.dag.Node) → int
get_name(self: nndeploy._nndeploy_internal.dag.Node) → str
get_node_type(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy::dag::NodeType
get_nodes_name(self: nndeploy._nndeploy_internal.dag.Graph) → list[str]
get_nodes_name_recursive(self: nndeploy._nndeploy_internal.dag.Graph) → list[str]
get_nodes_recursive(self: nndeploy._nndeploy_internal.dag.Graph) → list[nndeploy._nndeploy_internal.dag.Node]
get_nodes_run_status(self: nndeploy._nndeploy_internal.dag.Graph) → dict[str, nndeploy::dag::RunStatus]
get_nodes_run_status_recursive(self: nndeploy._nndeploy_internal.dag.Graph) → dict[str, nndeploy::dag::RunStatus]
get_output(self: nndeploy._nndeploy_internal.dag.Node, index: int = 0) → nndeploy::dag::Edge
get_output_count(self: nndeploy._nndeploy_internal.dag.Node) → int
get_output_index(self: nndeploy._nndeploy_internal.dag.Node, name: str) → int
get_output_name(self: nndeploy._nndeploy_internal.dag.Node, index: int = 0) → str
get_output_names(self: nndeploy._nndeploy_internal.dag.Node) → list[str]
get_output_node(self: nndeploy._nndeploy_internal.dag.Graph, index: int) → nndeploy._nndeploy_internal.dag.Node
get_output_type_info(self: nndeploy._nndeploy_internal.dag.Node) → list[nndeploy::dag::EdgeTypeInfo]
get_parallel_type(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.ParallelType
get_param(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.Param
get_real_outputs_name(self: nndeploy._nndeploy_internal.dag.Node) → list[str]
get_required_params(self: nndeploy._nndeploy_internal.dag.Node) → list[str]
get_resource_with_state(self: nndeploy._nndeploy_internal.dag.Graph, key: str) → nndeploy::dag::Edge
get_resource_without_state(self: nndeploy._nndeploy_internal.dag.Graph, key: str) → nndeploy::base::Any
get_run_size(self: nndeploy._nndeploy_internal.dag.Node) → int
get_run_status(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy::dag::RunStatus
get_source(self: nndeploy._nndeploy_internal.dag.Node) → str
get_stream(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.device.Stream
get_time_profile_flag(self: nndeploy._nndeploy_internal.dag.Node) → bool
get_trace_flag(self: nndeploy._nndeploy_internal.dag.Node) → bool
get_ui_params(self: nndeploy._nndeploy_internal.dag.Node) → list[str]
get_version(self: nndeploy._nndeploy_internal.dag.Node) → str
is_dynamic_input(self: nndeploy._nndeploy_internal.dag.Node) → bool
is_dynamic_output(self: nndeploy._nndeploy_internal.dag.Node) → bool
is_inputs_changed(self: nndeploy._nndeploy_internal.dag.Node, inputs: list[nndeploy::dag::Edge]) → bool
is_running(self: nndeploy._nndeploy_internal.dag.Node) → bool
remove_dropdown_param(self: nndeploy._nndeploy_internal.dag.Node, dropdown_param: str) → nndeploy._nndeploy_internal.base.Status
remove_in_out_node(self: nndeploy._nndeploy_internal.dag.Graph) → None
remove_io_param(self: nndeploy._nndeploy_internal.dag.Node, io_param: str) → nndeploy._nndeploy_internal.base.Status
remove_required_param(self: nndeploy._nndeploy_internal.dag.Node, required_param: str) → nndeploy._nndeploy_internal.base.Status
remove_ui_param(self: nndeploy._nndeploy_internal.dag.Node, ui_param: str) → nndeploy._nndeploy_internal.base.Status
set_composite_node(self: nndeploy._nndeploy_internal.dag.Node, composite_node: nndeploy::dag::CompositeNode) → nndeploy._nndeploy_internal.base.Status
set_debug_flag(self: nndeploy._nndeploy_internal.dag.Node, flag: bool) → None
set_desc(self: nndeploy._nndeploy_internal.dag.Node, desc: str) → None
set_developer(self: nndeploy._nndeploy_internal.dag.Node, developer: str) → None
set_device_type(self: nndeploy._nndeploy_internal.dag.Node, device_type: nndeploy._nndeploy_internal.base.DeviceType) → nndeploy._nndeploy_internal.base.Status
set_dropdown_params(self: nndeploy._nndeploy_internal.dag.Node, dropdown_params: dict[str, list[str]]) → nndeploy._nndeploy_internal.base.Status
set_dynamic_input(self: nndeploy._nndeploy_internal.dag.Node, is_dynamic_input: bool) → None
set_dynamic_output(self: nndeploy._nndeploy_internal.dag.Node, is_dynamic_output: bool) → None
set_edge_queue_overflow_policy(self: nndeploy._nndeploy_internal.dag.Graph, policy: nndeploy._nndeploy_internal.base.QueueOverflowPolicy, drop_count: int = 1) → nndeploy._nndeploy_internal.base.Status
set_graph(self: nndeploy._nndeploy_internal.dag.Node, graph: nndeploy::dag::Graph) → nndeploy._nndeploy_internal.base.Status
set_graph_flag(self: nndeploy._nndeploy_internal.dag.Node, flag: bool) → None
set_initialized_flag(self: nndeploy._nndeploy_internal.dag.Node, flag: bool) → None
set_inner_flag(self: nndeploy._nndeploy_internal.dag.Node, flag: bool) → None
set_input_name(self: nndeploy._nndeploy_internal.dag.Node, name: str, index: int = 0) → nndeploy._nndeploy_internal.base.Status
set_input_names(self: nndeploy._nndeploy_internal.dag.Node, names: list[str]) → nndeploy._nndeploy_internal.base.Status
set_input_shared_ptr(self: nndeploy._nndeploy_internal.dag.Node, input: nndeploy::dag::Edge, index: int = -1) → nndeploy._nndeploy_internal.base.Status
set_input_type_info(self: nndeploy._nndeploy_internal.dag.Node, input_type_info: nndeploy::dag::EdgeTypeInfo, desc: str = '') → nndeploy._nndeploy_internal.base.Status
set_inputs_shared_ptr(self: nndeploy._nndeploy_internal.dag.Node, inputs: list[nndeploy::dag::Edge]) → nndeploy._nndeploy_internal.base.Status
set_io_params(self: nndeploy._nndeploy_internal.dag.Node, io_params: list[str]) → nndeploy._nndeploy_internal.base.Status
set_io_type(self: nndeploy._nndeploy_internal.dag.Node, io_type: nndeploy::dag::IOType) → None
set_iter_input(self: nndeploy._nndeploy_internal.dag.Node, input: nndeploy::dag::Edge, index: int = -1) → nndeploy._nndeploy_internal.base.Status
set_key(self: nndeploy._nndeploy_internal.dag.Node, key: str) → None
set_loop_count(self: nndeploy._nndeploy_internal.dag.Graph, loop_count: int) → None
set_memory(self: nndeploy._nndeploy_internal.dag.Node, buffer: nndeploy._nndeploy_internal.device.Buffer) → nndeploy._nndeploy_internal.base.Status
set_name(self: nndeploy._nndeploy_internal.dag.Node, arg0: str) → None
set_node_type(self: nndeploy._nndeploy_internal.dag.Node, node_type: nndeploy::dag::NodeType) → None
set_output_data(self: nndeploy._nndeploy_internal.dag.Node, obj: object, index: int = 0, is_external: bool = True) → nndeploy._nndeploy_internal.base.Status
set_output_name(self: nndeploy._nndeploy_internal.dag.Node, name: str, index: int = 0) → nndeploy._nndeploy_internal.base.Status
set_output_names(self: nndeploy._nndeploy_internal.dag.Node, names: list[str]) → nndeploy._nndeploy_internal.base.Status
set_output_shared_ptr(self: nndeploy._nndeploy_internal.dag.Node, output: nndeploy::dag::Edge, index: int = -1) → nndeploy._nndeploy_internal.base.Status
set_output_type_info(self: nndeploy._nndeploy_internal.dag.Node, output_type_info: nndeploy::dag::EdgeTypeInfo, desc: str = '') → nndeploy._nndeploy_internal.base.Status
set_outputs_shared_ptr(self: nndeploy._nndeploy_internal.dag.Node, outputs: list[nndeploy::dag::Edge]) → nndeploy._nndeploy_internal.base.Status
set_parallel_type(self: nndeploy._nndeploy_internal.dag.Node, parallel_type: nndeploy._nndeploy_internal.base.ParallelType) → nndeploy._nndeploy_internal.base.Status
set_param(*args, **kwargs)

Overloaded function.

  1. set_param(self: nndeploy._nndeploy_internal.dag.Node, param: nndeploy._nndeploy_internal.base.Param) -> nndeploy._nndeploy_internal.base.Status

  2. set_param(self: nndeploy._nndeploy_internal.dag.Node, key: str, value: str) -> nndeploy._nndeploy_internal.base.Status

set_required_params(self: nndeploy._nndeploy_internal.dag.Node, required_params: list[str]) → nndeploy._nndeploy_internal.base.Status
set_running_flag(self: nndeploy._nndeploy_internal.dag.Node, flag: bool) → None
set_source(self: nndeploy._nndeploy_internal.dag.Node, source: str) → None
set_stream(self: nndeploy._nndeploy_internal.dag.Node, stream: nndeploy._nndeploy_internal.device.Stream) → None
set_time_profile_flag(self: nndeploy._nndeploy_internal.dag.Node, flag: bool) → None
set_ui_params(self: nndeploy._nndeploy_internal.dag.Node, ui_params: list[str]) → nndeploy._nndeploy_internal.base.Status
set_version(self: nndeploy._nndeploy_internal.dag.Node, version: str) → nndeploy._nndeploy_internal.base.Status
update_input(self: nndeploy._nndeploy_internal.dag.Node) → nndeploy._nndeploy_internal.base.EdgeUpdateFlag
nndeploy.dag.graph.serialize(graph: Graph) → str[源代码]

Serialize graph to JSON string

参数:

graph – Graph object to serialize

返回:

JSON string representation

nndeploy.dag.graph.save_file(graph: Graph, path: str)[源代码]

Save graph to file

参数:

  • graph – Graph object to save

  • path – File path

nndeploy.dag.graph.deserialize(json_str: str) → Graph[源代码]

Deserialize graph from JSON string

参数:

json_str – JSON string to deserialize

返回:

Graph object

nndeploy.dag.graph.load_file(path: str) → Graph[源代码]

Load graph from file

参数:

path – File path

返回:

Graph object

class nndeploy.dag.graph.GraphCreator[源代码]

基类：NodeCreator

Graph creator for creating graph nodes

__init__(self: nndeploy._nndeploy_internal.dag.NodeCreator) → None[源代码]
create_node(name: str, inputs: list[nndeploy.dag.edge.Edge], outputs: list[nndeploy.dag.edge.Edge])[源代码]

Create a graph node

参数:

  • name – Node name

  • inputs – Input edges

  • outputs – Output edges

返回:

Graph node

create_node_shared_ptr(self: nndeploy._nndeploy_internal.dag.NodeCreator, arg0: str, arg1: list[nndeploy::dag::Edge], arg2: list[nndeploy::dag::Edge]) → nndeploy._nndeploy_internal.dag.Node
nndeploy.dag.graph.get_graph_json()[源代码]

Get graph JSON representation

返回:

JSON string containing graph information

nndeploy.dag.graph.get_dag_json()[源代码]

Get DAG JSON representation containing graph, nodes and edge types

返回:

JSON string containing complete DAG information