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

1#!/usr/bin/env python3 

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

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

4# Copyright (c) 2026 Huawei Technologies Co., Ltd. 

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

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

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

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

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

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

11# ----------------------------------------------------------------------------------------------------------- 

12 

13"""GE operator object for reading and updating definition information.""" 

14 

15from __future__ import annotations 

16 

17import ctypes 

18 

19from ge._capi.pygraph_wrapper import graph_lib 

20 

21from ._attr import _AttrValue 

22 

23_OPERATOR_FACTORY_TOKEN = object() 

24 

25 

26class Operator: 

27 """GE operator borrowed for the duration of a callback. 

28 

29 The ctypes handle is borrowed from the C++ callback owner and is never 

30 created or destroyed by this wrapper. 

31 """ 

32 

33 __slots__ = ("_handle", "_valid") 

34 

35 def __init__(self, handle=None, token=None) -> None: 

36 if token is not _OPERATOR_FACTORY_TOKEN: 

37 raise RuntimeError("Operator objects should not be created directly.") 

38 if handle is None: 

39 raise ValueError("Operator handle cannot be None") 

40 

41 if isinstance(handle, int): 

42 handle = ctypes.c_void_p(handle) 

43 if not isinstance(handle, ctypes.c_void_p) or not handle: 

44 raise ValueError("Operator handle cannot be null") 

45 self._handle = handle 

46 self._valid = True 

47 

48 def __copy__(self) -> None: 

49 raise RuntimeError("Operator does not support copy") 

50 

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

52 raise RuntimeError("Operator does not support deepcopy") 

53 

54 def __enter__(self) -> "Operator": 

55 return self 

56 

57 def __exit__(self, exc_type, exc_value, traceback) -> None: 

58 if not self._valid: 

59 return 

60 self._valid = False 

61 self._handle = ctypes.c_void_p() 

62 

63 @staticmethod 

64 def _validate_name(name: str, kind: str) -> None: 

65 if not isinstance(name, str) or not name: 

66 raise TypeError(f"Operator {kind} name must be a non-empty string") 

67 

68 @property 

69 def name(self) -> str: 

70 self._ensure_valid() 

71 return self._get_string(graph_lib.GeApiWrapper_Operator_GetName) 

72 

73 @property 

74 def type(self) -> str: 

75 self._ensure_valid() 

76 return self._get_string(graph_lib.GeApiWrapper_Operator_GetType) 

77 

78 def set_attr(self, name: str, value: object) -> None: 

79 self._ensure_valid() 

80 self._validate_name(name, "attribute") 

81 

82 attr_value = _AttrValue() 

83 attr_value.set_value(value) 

84 ret = graph_lib.GeApiWrapper_Operator_SetAttr( 

85 self._handle, name.encode("utf-8"), attr_value._av_ptr 

86 ) 

87 if ret != 0: 

88 raise RuntimeError( 

89 f"Failed to set attribute '{name}' on Operator {self.name}" 

90 ) 

91 

92 def register_input(self, name: str) -> None: 

93 self._register_port( 

94 name, 

95 "input", 

96 "register_input", 

97 graph_lib.GeApiWrapper_Operator_InputRegister, 

98 ) 

99 

100 def register_optional_input(self, name: str) -> None: 

101 self._register_port( 

102 name, 

103 "optional input", 

104 "register_optional_input", 

105 graph_lib.GeApiWrapper_Operator_OptionalInputRegister, 

106 ) 

107 

108 def register_output(self, name: str) -> None: 

109 self._register_port( 

110 name, 

111 "output", 

112 "register_output", 

113 graph_lib.GeApiWrapper_Operator_OutputRegister, 

114 ) 

115 

116 def register_dynamic_input(self, name: str, count: int) -> None: 

117 self._register_dynamic_port(name, count, is_input=True) 

118 

119 def register_dynamic_output(self, name: str, count: int) -> None: 

120 self._register_dynamic_port(name, count, is_input=False) 

121 

122 def _register_port(self, name: str, kind: str, method_name: str, c_func) -> None: 

123 self._ensure_valid() 

124 self._validate_name(name, kind) 

125 ret = c_func(self._handle, name.encode("utf-8")) 

126 if ret != 0: 

127 raise RuntimeError( 

128 f"Failed to {method_name} '{name}' on Operator {self.name}" 

129 ) 

130 

131 def _register_dynamic_port(self, name: str, count: int, *, is_input: bool) -> None: 

132 self._ensure_valid() 

133 self._validate_name(name, "dynamic port") 

134 if type(count) is not int: 

135 raise TypeError("Operator dynamic port count must be an integer") 

136 if count < 0 or count >= 1 << 32: 

137 raise ValueError("Operator dynamic port count must be in uint32 range") 

138 c_func = ( 

139 graph_lib.GeApiWrapper_Operator_DynamicInputRegister 

140 if is_input 

141 else graph_lib.GeApiWrapper_Operator_DynamicOutputRegister 

142 ) 

143 ret = c_func(self._handle, name.encode("utf-8"), ctypes.c_uint32(count)) 

144 if ret != 0: 

145 direction = "input" if is_input else "output" 

146 raise RuntimeError( 

147 f"Failed to register dynamic {direction} '{name}' on Operator {self.name}" 

148 ) 

149 

150 def _get_string(self, c_func) -> str: 

151 c_str = c_func(self._handle) 

152 if not c_str: 

153 raise RuntimeError("Failed to get Operator name or type") 

154 try: 

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

156 finally: 

157 graph_lib.GeApiWrapper_FreeString(c_str) 

158 

159 def _ensure_valid(self) -> None: 

160 if not self._valid: 

161 raise RuntimeError("Operator is only valid inside parse_node") 

162 

163 

164def create_operator(handle) -> Operator: 

165 """Create a callback-bound Operator for internal bridge use.""" 

166 

167 return Operator(handle, _OPERATOR_FACTORY_TOKEN)