Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/graph/node.py: 89%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ------------------------------------------------------------------- 

4# ----------------------------------------------------------------------------------------------------------- 

5# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

6# This program is free software, you can redistribute it and/or modify it under the terms and conditions of 

7# CANN Open Software License Agreement Version 2.0 (the "License"). 

8# Please refer to the License for details. You may not use this file except in compliance with the License. 

9# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, 

10# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. 

11# See LICENSE in the root of the software repository for the full text of the License. 

12# ----------------------------------------------------------------------------------------------------------- 

13 

14"""Node module for GraphEngine node operations.""" 

15 

16import ctypes 

17from typing import Any, List, Tuple 

18 

19from ge._capi.pygraph_wrapper import graph_lib 

20 

21from ._attr import _AttrValue 

22from .tensor_desc import TensorDesc 

23 

24 

25class Node: 

26 """Node class for GraphEngine node operations. 

27 

28 This class provides a Pythonic interface for node operations 

29 using the GraphEngine C API. 

30 

31 Example: 

32 >>> # Node objects are created internally by Graph operations 

33 >>> nodes = graph.get_all_nodes() 

34 >>> for node in nodes: 

35 ... name = node.name 

36 ... node_type = node.type 

37 ... attr = node.get_attr("some_attr") 

38 ... node.set_attr("new_attr", attr_value) 

39 """ 

40 

41 def __init__(self) -> None: 

42 """Prevent direct instantiation of Node objects.""" 

43 raise RuntimeError("Node objects should not be created directly.") 

44 

45 def __del__(self) -> None: 

46 """Clean up resources.""" 

47 if self._owns_handle: 

48 graph_lib.GeApiWrapper_GNode_DestroyGNode(self._handle) 

49 self._handle = None 

50 

51 def __copy__(self) -> None: 

52 """Copy is not supported.""" 

53 raise RuntimeError("Node does not support copy") 

54 

55 def __deepcopy__(self, memodict) -> None: 

56 """Deep copy is not supported.""" 

57 raise RuntimeError("Node does not support deepcopy") 

58 

59 @classmethod 

60 def _create_from(cls, node_handle: ctypes.c_void_p, owns_handle: bool = True) -> "Node": 

61 """Create Node object from C++ handle.(internal use only by e.g Graph.get_all_nodes() 

62 or TensorHolder._get_node_snapshot(), do not use this method directly) 

63 

64 Args: 

65 node_handle: C++ GNode object handle. 

66 owns_handle: Whether to own the handle. 

67 

68 Returns: 

69 Node object. 

70 

71 Raises: 

72 ValueError: If node_handle is None. 

73 """ 

74 if not node_handle: 

75 raise ValueError("Node handle cannot be None") 

76 instance = cls.__new__(cls) 

77 instance._handle = node_handle 

78 instance._owns_handle = owns_handle 

79 return instance 

80 

81 @property 

82 def name(self) -> str: 

83 """Get node name. 

84 

85 Returns: 

86 Node name. 

87 

88 Raises: 

89 RuntimeError: If name retrieval fails. 

90 """ 

91 c_str = graph_lib.GeApiWrapper_GNode_GetName(self._handle) 

92 if not c_str: 

93 raise RuntimeError("Failed to get Node name") 

94 

95 try: 

96 return ctypes.string_at(c_str).decode("utf-8") 

97 finally: 

98 graph_lib.GeApiWrapper_FreeString(c_str) 

99 

100 @property 

101 def type(self) -> str: 

102 """Get node type. 

103 

104 Returns: 

105 Node type. 

106 

107 Raises: 

108 RuntimeError: If type retrieval fails. 

109 """ 

110 c_str = graph_lib.GeApiWrapper_GNode_GetType(self._handle) 

111 if not c_str: 

112 raise RuntimeError("Failed to get Node type") 

113 

114 try: 

115 return ctypes.string_at(c_str).decode("utf-8") 

116 finally: 

117 graph_lib.GeApiWrapper_FreeString(c_str) 

118 

119 def get_in_control_nodes(self) -> List["Node"]: 

120 """Get input control nodes. 

121 

122 Returns: 

123 List of input control Node objects. 

124 """ 

125 node_num = ctypes.c_size_t() 

126 nodes = graph_lib.GeApiWrapper_GNode_GetInControlNodes(self._handle, ctypes.byref(node_num)) 

127 if not nodes: 

128 return [] 

129 

130 try: 

131 return [Node._create_from(nodes[i]) for i in range(node_num.value)] 

132 finally: 

133 graph_lib.GeApiWrapper_GNode_FreeGNodeArray(nodes) 

134 

135 def get_in_data_nodes_and_port_indexes(self, in_index: int) -> Tuple["Node", int]: 

136 """Get input data node and port index. 

137 

138 Args: 

139 in_index: Input index. 

140 

141 Returns: 

142 Tuple of (input Node, port index). 

143 

144 Raises: 

145 TypeError: If in_index is not an integer. 

146 RuntimeError: If retrieval fails. 

147 """ 

148 if not isinstance(in_index, int): 

149 raise TypeError("Input index must be an integer") 

150 

151 in_node = ctypes.c_void_p() 

152 index = ctypes.c_int32() 

153 

154 ret = graph_lib.GeApiWrapper_GNode_GetInDataNodesAndPortIndexes( 

155 self._handle, in_index, ctypes.byref(in_node), ctypes.byref(index) 

156 ) 

157 if ret != 0: # GRAPH_SUCCESS 

158 raise RuntimeError(f"Failed to get input data node for index {in_index}") 

159 

160 return Node._create_from(in_node), index.value 

161 

162 def get_out_data_nodes_and_port_indexes(self, out_index: int) -> List[Tuple["Node", int]]: 

163 """Get output data nodes and port indexes. 

164 

165 Args: 

166 out_index: Output index. 

167 

168 Returns: 

169 List of Tuple of (output Node, port index). 

170 

171 Raises: 

172 TypeError: If out_index is not an integer. 

173 RuntimeError: If retrieval fails. 

174 """ 

175 if not isinstance(out_index, int): 

176 raise TypeError("Output index must be an integer") 

177 out_nodes = ctypes.POINTER(ctypes.c_void_p)() 

178 out_indexes = ctypes.POINTER(ctypes.c_int32)() 

179 size = ctypes.c_int() 

180 ret = graph_lib.GeApiWrapper_GNode_GetOutDataNodesAndPortIndexes( 

181 self._handle, 

182 out_index, 

183 ctypes.byref(out_nodes), 

184 ctypes.byref(out_indexes), 

185 ctypes.byref(size), 

186 ) 

187 if ret != 0: # GRAPH_SUCCESS 

188 raise RuntimeError(f"Failed to get output data node for index {out_index}") 

189 try: 

190 return [(Node._create_from(out_nodes[i]), out_indexes[i]) for i in range(size.value)] 

191 finally: 

192 graph_lib.GeApiWrapper_GNode_FreeGNodeArray(out_nodes) 

193 graph_lib.GeApiWrapper_GNode_FreeIntArray(out_indexes) 

194 

195 def get_attr(self, key: str) -> Any: 

196 """Get node attribute. 

197 

198 Args: 

199 key: Attribute name. 

200 

201 Returns: 

202 Attribute value. 

203 

204 Raises: 

205 TypeError: If key is not a string. 

206 RuntimeError: If attribute retrieval fails. 

207 """ 

208 if not isinstance(key, str): 

209 raise TypeError("Attribute key must be a string") 

210 

211 attr_value = _AttrValue() 

212 key_bytes = key.encode("utf-8") 

213 

214 ret = graph_lib.GeApiWrapper_GNode_GetAttr(self._handle, key_bytes, attr_value._av_ptr) 

215 if ret != 0: # GRAPH_SUCCESS 

216 raise RuntimeError(f"Failed to get attribute '{key}' from Node") 

217 

218 return attr_value.get_value() 

219 

220 def set_attr(self, key: str, value: Any) -> None: 

221 """Set node attribute. 

222 

223 Args: 

224 key: Attribute name. 

225 value: Attribute value. 

226 

227 Raises: 

228 TypeError: If arguments have wrong types. 

229 RuntimeError: If attribute setting fails. 

230 """ 

231 if not isinstance(key, str): 

232 raise TypeError("Attribute key must be a string") 

233 

234 key_bytes = key.encode("utf-8") 

235 attr_value = _AttrValue() 

236 attr_value.set_value(value) 

237 ret = graph_lib.GeApiWrapper_GNode_SetAttr(self._handle, key_bytes, attr_value._av_ptr) 

238 if ret != 0: # GRAPH_SUCCESS 

239 raise RuntimeError(f"Failed to set attribute '{key}' on Node {self.name}") 

240 

241 def get_input_attr(self, attr_name: str, input_index: int) -> Any: 

242 """Get input attribute. 

243 

244 Args: 

245 attr_name: Attribute name. 

246 input_index: Input index. 

247 

248 Returns: 

249 Attribute value. 

250 

251 Raises: 

252 TypeError: If arguments have wrong types. 

253 RuntimeError: If attribute retrieval fails. 

254 """ 

255 if not isinstance(attr_name, str): 

256 raise TypeError("Attribute name must be a string") 

257 if not isinstance(input_index, int): 

258 raise TypeError("Input index must be an integer") 

259 

260 attr_value = _AttrValue() 

261 attr_name_bytes = attr_name.encode("utf-8") 

262 

263 ret = graph_lib.GeApiWrapper_GNode_GetInputAttr( 

264 self._handle, 

265 attr_name_bytes, 

266 ctypes.c_uint32(input_index), 

267 attr_value._av_ptr, 

268 ) 

269 if ret != 0: # GRAPH_SUCCESS 

270 raise RuntimeError(f"Failed to get Node {self.name} input attribute '{attr_name}' for index {input_index}") 

271 

272 return attr_value.get_value() 

273 

274 def set_input_attr(self, attr_name: str, input_index: int, value: Any) -> None: 

275 """Set input attribute. 

276 

277 Args: 

278 attr_name: Attribute name. 

279 input_index: Input index. 

280 value: Attribute value. 

281 

282 Raises: 

283 TypeError: If arguments have wrong types. 

284 RuntimeError: If attribute setting fails. 

285 """ 

286 if not isinstance(attr_name, str): 

287 raise TypeError("Attribute name must be a string") 

288 if not isinstance(input_index, int): 

289 raise TypeError("Input index must be an integer") 

290 

291 attr_name_bytes = attr_name.encode("utf-8") 

292 attr_value = _AttrValue() 

293 attr_value.set_value(value) 

294 ret = graph_lib.GeApiWrapper_GNode_SetInputAttr( 

295 self._handle, 

296 attr_name_bytes, 

297 ctypes.c_uint32(input_index), 

298 attr_value._av_ptr, 

299 ) 

300 if ret != 0: # GRAPH_SUCCESS 

301 raise RuntimeError(f"Failed to set Node {self.name} input attribute '{attr_name}' for index {input_index}") 

302 

303 def get_output_attr(self, attr_name: str, output_index: int) -> Any: 

304 """Get output attribute. 

305 

306 Args: 

307 attr_name: Attribute name. 

308 output_index: Output index. 

309 

310 Returns: 

311 Attribute value. 

312 

313 Raises: 

314 TypeError: If arguments have wrong types. 

315 RuntimeError: If attribute retrieval fails. 

316 """ 

317 if not isinstance(attr_name, str): 

318 raise TypeError("Attribute name must be a string") 

319 if not isinstance(output_index, int): 

320 raise TypeError("Output index must be an integer") 

321 

322 attr_value = _AttrValue() 

323 attr_name_bytes = attr_name.encode("utf-8") 

324 

325 ret = graph_lib.GeApiWrapper_GNode_GetOutputAttr( 

326 self._handle, 

327 attr_name_bytes, 

328 ctypes.c_uint32(output_index), 

329 attr_value._av_ptr, 

330 ) 

331 if ret != 0: # GRAPH_SUCCESS 

332 raise RuntimeError( 

333 f"Failed to get Node {self.name} output attribute '{attr_name}' for index {output_index}" 

334 ) 

335 

336 return attr_value.get_value() 

337 

338 def set_output_attr(self, attr_name: str, output_index: int, value: Any) -> None: 

339 """Set output attribute. 

340 

341 Args: 

342 attr_name: Attribute name. 

343 output_index: Output index. 

344 value: Attribute value. 

345 

346 Raises: 

347 TypeError: If arguments have wrong types. 

348 RuntimeError: If attribute setting fails. 

349 """ 

350 if not isinstance(attr_name, str): 

351 raise TypeError("Attribute name must be a string") 

352 if not isinstance(output_index, int): 

353 raise TypeError("Output index must be an integer") 

354 

355 attr_name_bytes = attr_name.encode("utf-8") 

356 attr_value = _AttrValue() 

357 attr_value.set_value(value) 

358 ret = graph_lib.GeApiWrapper_GNode_SetOutputAttr( 

359 self._handle, 

360 attr_name_bytes, 

361 ctypes.c_uint32(output_index), 

362 attr_value._av_ptr, 

363 ) 

364 if ret != 0: # GRAPH_SUCCESS 

365 raise RuntimeError( 

366 f"Failed to set Node {self.name} output attribute '{attr_name}' for index {output_index}" 

367 ) 

368 

369 def get_out_control_nodes(self) -> List["Node"]: 

370 """Get output control nodes. 

371 

372 Returns: 

373 List of output control Node objects. 

374 """ 

375 node_num = ctypes.c_size_t() 

376 nodes = graph_lib.GeApiWrapper_GNode_GetOutControlNodes(self._handle, ctypes.byref(node_num)) 

377 if not nodes: 

378 return [] 

379 

380 try: 

381 return [Node._create_from(nodes[i]) for i in range(node_num.value)] 

382 finally: 

383 graph_lib.GeApiWrapper_GNode_FreeGNodeArray(nodes) 

384 

385 def get_inputs_size(self) -> int: 

386 """Get input size 

387 

388 Returns: 

389 Number of input size 

390 """ 

391 return graph_lib.GeApiWrapper_GNode_GetInputsSize(self._handle) 

392 

393 def get_outputs_size(self) -> int: 

394 """Get output size 

395 

396 Returns: 

397 Number of output size 

398 """ 

399 return graph_lib.GeApiWrapper_GNode_GetOutputsSize(self._handle) 

400 

401 def has_attr(self, attr_name: str) -> bool: 

402 """Has attribute 

403 

404 Args: 

405 attr_name: Attribute name. 

406 

407 Returns: 

408 If node has attribute of this name. 

409 """ 

410 if not isinstance(attr_name, str): 

411 raise TypeError("attr_name must be a string") 

412 

413 attr_name_bytes = attr_name.encode("utf-8") 

414 return graph_lib.GeApiWrapper_GNode_HasAttr(self._handle, attr_name_bytes) 

415 

416 def get_input_desc(self, index: int) -> TensorDesc: 

417 """Get input tensor descriptor. 

418 

419 Args: 

420 index: Input index. 

421 

422 Returns: 

423 TensorDesc object. 

424 """ 

425 if not isinstance(index, int): 

426 raise TypeError("Input index must be an integer") 

427 

428 desc_handle = graph_lib.GeApiWrapper_GNode_GetInputDesc(self._handle, ctypes.c_int32(index)) 

429 if not desc_handle: 

430 raise RuntimeError(f"Failed to get input desc for index {index}") 

431 return TensorDesc._create_from(desc_handle) 

432 

433 def update_input_desc(self, index: int, tensor_desc: TensorDesc) -> None: 

434 """Update input tensor descriptor. 

435 

436 Args: 

437 index: Input index. 

438 tensor_desc: New tensor descriptor. 

439 """ 

440 if not isinstance(index, int): 

441 raise TypeError("Input index must be an integer") 

442 if not isinstance(tensor_desc, TensorDesc): 

443 raise TypeError("tensor_desc must be a TensorDesc") 

444 

445 ret = graph_lib.GeApiWrapper_GNode_UpdateInputDesc(self._handle, ctypes.c_int32(index), tensor_desc._handle) 

446 if ret != 0: 

447 raise RuntimeError(f"Failed to update input desc for index {index}") 

448 

449 def get_output_desc(self, index: int) -> TensorDesc: 

450 """Get output tensor descriptor. 

451 

452 Args: 

453 index: Output index. 

454 

455 Returns: 

456 TensorDesc object. 

457 """ 

458 if not isinstance(index, int): 

459 raise TypeError("Output index must be an integer") 

460 

461 desc_handle = graph_lib.GeApiWrapper_GNode_GetOutputDesc(self._handle, ctypes.c_int32(index)) 

462 if not desc_handle: 

463 raise RuntimeError(f"Failed to get output desc for index {index}") 

464 return TensorDesc._create_from(desc_handle) 

465 

466 def update_output_desc(self, index: int, tensor_desc: TensorDesc) -> None: 

467 """Update output tensor descriptor. 

468 

469 Args: 

470 index: Output index. 

471 tensor_desc: New tensor descriptor. 

472 """ 

473 if not isinstance(index, int): 

474 raise TypeError("Output index must be an integer") 

475 if not isinstance(tensor_desc, TensorDesc): 

476 raise TypeError("tensor_desc must be a TensorDesc") 

477 

478 ret = graph_lib.GeApiWrapper_GNode_UpdateOutputDesc(self._handle, ctypes.c_int32(index), tensor_desc._handle) 

479 if ret != 0: 

480 raise RuntimeError(f"Failed to update output desc for index {index}")