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", "_read_only") 

34 

35 def __init__(self, handle=None, token=None, read_only=False) -> 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 self._read_only = read_only 

48 

49 def __copy__(self) -> None: 

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

51 

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

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

54 

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

56 return self 

57 

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

59 if not self._valid: 

60 return 

61 self._valid = False 

62 self._handle = ctypes.c_void_p() 

63 

64 @staticmethod 

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

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

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

68 

69 @property 

70 def name(self) -> str: 

71 self._ensure_valid() 

72 return self._get_string(graph_lib.GeApiWrapper_Operator_GetName) 

73 

74 @property 

75 def type(self) -> str: 

76 self._ensure_valid() 

77 return self._get_string(graph_lib.GeApiWrapper_Operator_GetType) 

78 

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

80 self._ensure_mutable() 

81 self._validate_name(name, "attribute") 

82 

83 attr_value = _AttrValue() 

84 attr_value.set_value(value) 

85 ret = graph_lib.GeApiWrapper_Operator_SetAttr( 

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

87 ) 

88 if ret != 0: 

89 raise RuntimeError( 

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

91 ) 

92 

93 def get_attr(self, name: str): 

94 self._ensure_valid() 

95 self._validate_name(name, "attribute") 

96 attr_value = _AttrValue() 

97 ret = graph_lib.GeApiWrapper_Operator_GetAttr( 

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

99 ) 

100 if ret != 0: 

101 raise RuntimeError(f"Failed to get attribute '{name}' from Operator") 

102 return attr_value.get_value() 

103 

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

105 self._register_port( 

106 name, 

107 "input", 

108 "register_input", 

109 graph_lib.GeApiWrapper_Operator_InputRegister, 

110 ) 

111 

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

113 self._register_port( 

114 name, 

115 "optional input", 

116 "register_optional_input", 

117 graph_lib.GeApiWrapper_Operator_OptionalInputRegister, 

118 ) 

119 

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

121 self._register_port( 

122 name, 

123 "output", 

124 "register_output", 

125 graph_lib.GeApiWrapper_Operator_OutputRegister, 

126 ) 

127 

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

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

130 

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

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

133 

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

135 self._ensure_mutable() 

136 self._validate_name(name, kind) 

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

138 if ret != 0: 

139 raise RuntimeError( 

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

141 ) 

142 

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

144 self._ensure_mutable() 

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

146 if type(count) is not int: 

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

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

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

150 c_func = ( 

151 graph_lib.GeApiWrapper_Operator_DynamicInputRegister 

152 if is_input 

153 else graph_lib.GeApiWrapper_Operator_DynamicOutputRegister 

154 ) 

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

156 if ret != 0: 

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

158 raise RuntimeError( 

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

160 ) 

161 

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

163 c_str = c_func(self._handle) 

164 if not c_str: 

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

166 try: 

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

168 finally: 

169 graph_lib.GeApiWrapper_FreeString(c_str) 

170 

171 def _ensure_valid(self) -> None: 

172 if not self._valid: 

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

174 

175 def _ensure_mutable(self) -> None: 

176 self._ensure_valid() 

177 if self._read_only: 

178 raise RuntimeError("Source Operator is read-only") 

179 

180 

181def create_operator(handle, *, read_only=False) -> Operator: 

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

183 

184 return Operator(handle, _OPERATOR_FACTORY_TOKEN, read_only)