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

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 

15_OPERATOR_FACTORY_TOKEN = object() 

16 

17 

18class Operator: 

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

20 

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

22 

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

24 if token is not _OPERATOR_FACTORY_TOKEN: 

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

26 if handle is None: 

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

28 self._handle = handle 

29 self._valid = True 

30 

31 def __copy__(self) -> None: 

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

33 

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

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

36 

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

38 return self 

39 

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

41 if not self._valid: 

42 return 

43 self._valid = False 

44 self._handle.invalidate() 

45 

46 @staticmethod 

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

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

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

50 

51 @property 

52 def name(self) -> str: 

53 self._ensure_valid() 

54 return self._handle.get_name() 

55 

56 @property 

57 def type(self) -> str: 

58 self._ensure_valid() 

59 return self._handle.get_type() 

60 

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

62 self._ensure_valid() 

63 self._validate_name(name, "attribute") 

64 if type(value) is int: 

65 if value < -(1 << 63) or value >= 1 << 63: 

66 raise ValueError("Operator int attribute must be in int64 range") 

67 elif type(value) is not float: 

68 raise TypeError("Operator set_attr only supports int and float values") 

69 self._handle.set_attr(name, value) 

70 

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

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

73 

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

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

76 

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

78 self._ensure_valid() 

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

80 if type(count) is not int: 

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

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

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

84 if is_input: 

85 self._handle.register_dynamic_input(name, count) 

86 else: 

87 self._handle.register_dynamic_output(name, count) 

88 

89 def _ensure_valid(self) -> None: 

90 if not self._valid: 

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

92 

93 

94def create_operator(handle) -> Operator: 

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

96 

97 return Operator(handle, _OPERATOR_FACTORY_TOKEN)