nndeploy.dag.edge

Functions

add_accepted_edge_type_map(edge_type_map)

Add acceptable Edge type mapping relationships

get_accepted_edge_type_json()

Get JSON format string of acceptable Edge type mapping relationships

get_accepted_edge_type_map()

Get current acceptable Edge type mapping relationships

sub_accepted_edge_type_map(edge_type_map)

Remove acceptable Edge type mapping relationships

Classes

Edge([name])

Edge class is an edge object used to connect nodes in the nndeploy framework, inheriting from the C++ Edge class.
class nndeploy.dag.edge.Edge(name: str = '')[源代码]

基类：Edge

Edge class is an edge object used to connect nodes in the nndeploy framework, inheriting from the C++ Edge class. Edge is responsible for passing data between nodes, supporting multiple data types and parallel modes.

Main functions: - Data transfer: Pass various types of data between nodes - Type management: Manage and validate data type information - Queue control: Control data queue size and overflow policies - Parallel support: Support different parallel execution modes

__init__(name: str = '')[源代码]

Initialize Edge object

参数:

name (str) – Name of the Edge, default is empty string

get_name() → str[源代码]

Get the name of the Edge

返回:

Name of the Edge

返回类型:

str

set_queue_max_size(queue_max_size: int)[源代码]

Set the maximum capacity of the queue

参数:

queue_max_size (int) – Maximum queue capacity, used to control memory usage

get_queue_max_size() → int[源代码]

Get the maximum capacity of the queue

返回:

Currently set maximum queue capacity

返回类型:

int

set_queue_overflow_policy(policy, drop_count: int = 1)[源代码]

Set queue overflow policy

参数:

  • policy – Processing policy when queue overflows (such as node backpressure, data dropping, etc.)

  • drop_count (int) – Number of data items to drop when overflow occurs, default is 1

get_queue_overflow_policy()[源代码]

Get the current queue overflow policy

返回:

Queue overflow policy object

get_queue_drop_count() → int[源代码]

Get the number of data items dropped when queue overflows

返回:

Number of data items dropped

返回类型:

int

get_parallel_type() → ParallelType[源代码]

Get the parallel type of the Edge

返回:

Current parallel execution type

返回类型:

nndeploy.base.ParallelType

set_parallel_type(parallel_type: ParallelType)[源代码]

Set the parallel type of the Edge

参数:

parallel_type (nndeploy.base.ParallelType) – Parallel execution type to set

empty() → bool[源代码]

Check if the Edge is empty (no data)

返回:

True indicates Edge is empty, False indicates there is data

返回类型:

bool

construct()[源代码]

Construct Edge object, initialize internal data structures

set(data: any)[源代码]

Set data into the Edge

参数:

data (any) – Data to set, can be any type

get(node: Node)[源代码]

Get data from the Edge (for a specific node)

参数:

node (_C.dag.Node) – Node requesting data

返回:

Data retrieved from the Edge

get_graph_output()[源代码]

Get graph output data (when Edge serves as graph output)

返回:

Graph output data

create_buffer(device: Device, desc: BufferDesc)[源代码]

Create Buffer on specified device

参数:
返回:

Created Buffer object

create_tensor(device: Device, desc: TensorDesc, tensor_name: str = '')[源代码]

Create Tensor on specified device

参数:
返回:

Created Tensor object

notify_written(data: Union[Buffer, Tensor])[源代码]

Notify that data has been written

参数:

data (Union[nndeploy.device.Buffer, nndeploy.device.Tensor]) – Data object that has been written

get_index(node: Node) → int[源代码]

Get the index of specified node in the Edge

参数:

node (_C.dag.Node) – Target node

返回:

Index of the node in the Edge

返回类型:

int

reset_index()[源代码]

Reset the index counter of the Edge

get_graph_output_index() → int[源代码]

Get the index of graph output

返回:

Graph output index

返回类型:

int

get_position(node: Node) → int[源代码]

Get the position of specified node in the Edge

参数:

node (_C.dag.Node) – Target node

返回:

Position of the node in the Edge

返回类型:

int

get_graph_output_position() → int[源代码]

Get the position of graph output

返回:

Graph output position

返回类型:

int

update(node: Node) → EdgeUpdateFlag[源代码]

Update Edge status (for a specific node)

参数:

node (_C.dag.Node) – Node triggering the update

返回:

Update flag

返回类型:

nndeploy.base.EdgeUpdateFlag

mark_graph_output() → bool[源代码]

Mark the Edge as graph output

返回:

Whether marking was successful

返回类型:

bool

increase_producers(producers: list[nndeploy._nndeploy_internal.dag.Node])[源代码]

Add producer nodes list

参数:

producers (list[_C.dag.Node]) – List of producer nodes to add

increase_consumers(consumers: list[nndeploy._nndeploy_internal.dag.Node])[源代码]

Add consumer nodes list

参数:

consumers (list[_C.dag.Node]) – List of consumer nodes to add

get_producers() → list[nndeploy._nndeploy_internal.dag.Node][源代码]

Get all producer nodes

返回:

List of producer nodes

返回类型:

list[_C.dag.Node]

get_consumers() → list[nndeploy._nndeploy_internal.dag.Node][源代码]

Get all consumer nodes

返回:

List of consumer nodes

返回类型:

list[_C.dag.Node]

request_terminate() → bool[源代码]

Request termination of Edge execution

返回:

Whether termination request was successful

返回类型:

bool

set_type(type_val)[源代码]

Set the data type of the Edge

参数:

type_val – Data type value, can be Python type or type object

set_type_name(type_name: str)[源代码]

Set the type name of the Edge

参数:

type_name (str) – Type name string

get_type_name() → str[源代码]

Get the type name of the Edge

返回:

Type name string

返回类型:

str

set_type_info(type_info: EdgeTypeInfo)[源代码]

Set the type information object of the Edge

参数:

type_info (EdgeTypeInfo) – Type information object

get_type_info() → EdgeTypeInfo[源代码]

Get the type information object of the Edge

返回:

Type information object

返回类型:

EdgeTypeInfo

check_type_info(type_info: EdgeTypeInfo) → bool[源代码]

Check if type information matches

参数:

type_info (EdgeTypeInfo) – Type information to check

返回:

True indicates type matches, False indicates mismatch

返回类型:

bool

create(*args, **kwargs)

Overloaded function.

  1. create(self: nndeploy._nndeploy_internal.dag.Edge, device: nndeploy._nndeploy_internal.device.Device, desc: nndeploy._nndeploy_internal.device.BufferDesc) -> nndeploy._nndeploy_internal.device.Buffer

  2. create(self: nndeploy._nndeploy_internal.dag.Edge, device: nndeploy._nndeploy_internal.device.Device, desc: nndeploy._nndeploy_internal.device.TensorDesc, tensor_name: str = ‘’) -> nndeploy._nndeploy_internal.device.Tensor

nndeploy.dag.edge.add_accepted_edge_type_map(edge_type_map)[源代码]

Add acceptable Edge type mapping relationships

Supports two input formats: 1. dict[str, list[str]]: One type corresponds to a list of multiple compatible types 2. dict[str, str]: One type corresponds to one compatible type

参数:

edge_type_map – Type mapping dictionary to add - Format 1: {“type1”: [“compatible_type1”, “compatible_type2”]} - Format 2: {“type1”: “compatible_type1”}

示例

# Add custom type mapping add_accepted_edge_type_map({

“MyCustomType”: [“nndeploy.device.Tensor”, “numpy.ndarray”]

})

nndeploy.dag.edge.sub_accepted_edge_type_map(edge_type_map: Union[dict[str, list[str]], dict[str, str]])[源代码]

Remove acceptable Edge type mapping relationships

参数:

edge_type_map – Type mapping dictionary to remove Format: {“type1”: [“compatible_type1”, “compatible_type2”]}

示例

# Remove specific type mapping sub_accepted_edge_type_map({

“MyCustomType”: [“nndeploy.device.Tensor”]

})

nndeploy.dag.edge.get_accepted_edge_type_map()[源代码]

Get current acceptable Edge type mapping relationships

返回:

Current type mapping dictionary

返回类型:

dict

nndeploy.dag.edge.get_accepted_edge_type_json()[源代码]

Get JSON format string of acceptable Edge type mapping relationships

Mainly used for: - Debugging and logging - Interface interaction with other systems - Configuration file export

返回:

JSON format type mapping string containing complete type compatibility information

返回类型:

str

示例

Return format: {

“accepted_edge_types”: {

“nndeploy.device.Buffer”: [“nndeploy::device::Buffer”], “nndeploy::device::Buffer”: [“nndeploy.device.Buffer”], …

}

}