Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/dttest/api/python/ge/ge/passes/base.py: 87%

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"""Base definitions for Python GE passes.""" 

14 

15from __future__ import annotations 

16 

17from enum import Enum 

18from typing import TYPE_CHECKING, Iterable, List, Optional, Union 

19 

20from ._native import MatchResult, PassContext, PatternMatcherConfig, PatternMatcherConfigBuilder, SubgraphInput, SubgraphOutput, SubgraphBoundary, SubgraphRewriter 

21 

22if TYPE_CHECKING: 

23 from ge.graph.graph import Graph 

24 from ge.graph.node import Node 

25 

26 from .pattern import Pattern 

27 

28 

29PatternOrGraph = Union["Pattern", "Graph"] 

30StatusLike = Optional[Union[bool, int]] 

31 

32 

33class PassStage(str, Enum): 

34 """Python-facing pass stage names.""" 

35 

36 BEFORE_INFER_SHAPE = "BeforeInferShape" 

37 AFTER_INFER_SHAPE = "AfterInferShape" 

38 AFTER_BUILTIN_FUSION_PASS = "AfterBuiltinFusionPass" 

39 AFTER_ORIGIN_GRAPH_OPTIMIZE = "AfterOriginGraphOptimize" 

40 

41 

42class FusionBasePass: 

43 """Python FusionBasePass contract.""" 

44 

45 def run(self, graph: Graph, context: PassContext) -> StatusLike: 

46 raise NotImplementedError("FusionBasePass.run must be implemented") 

47 

48 

49class PatternFusionPass(FusionBasePass): 

50 """Python PatternFusionPass contract. 

51 

52 The execution engine calls ``patterns()``, ``meet_requirements()``, and 

53 ``replacement()`` — **not** ``run()``. Overriding ``run()`` in a 

54 ``PatternFusionPass`` subclass has no effect; implement the three hook 

55 methods above instead. 

56 

57 Subclasses can use either the legacy graph-building hooks: 

58 - ``patterns(self) -> Iterable[Pattern | Graph]`` 

59 - ``replacement(self, match_result) -> Graph`` 

60 

61 or expression-style pattern methods: 

62 - ``@pattern def name(self, inputs) -> TensorHolder | list[TensorHolder] | tuple[TensorHolder, ...]`` 

63 - ``replacement(self, inputs) -> TensorHolder | list[TensorHolder] | tuple[TensorHolder, ...] | Graph`` 

64 - ``replacement(self, inputs, match_result)`` when match details are needed 

65 

66 In a ``@pattern`` method, a list/tuple means multiple outputs of one 

67 pattern, not multiple patterns. Declare several ``@pattern`` methods for 

68 a multiple-pattern pass. Expression-style patterns automatically capture 

69 accessed inputs in input-index order, then returned outputs in return order. 

70 """ 

71 

72 def __init__(self, matcher_config: Optional[PatternMatcherConfig] = None) -> None: 

73 self._matcher_config = matcher_config 

74 

75 def __init_subclass__(cls, **kwargs) -> None: 

76 super().__init_subclass__(**kwargs) 

77 if "run" in cls.__dict__: 

78 raise TypeError( 

79 f"{cls.__name__} overrides run(), which is never invoked " 

80 f"by the PatternFusionPass execution path. " 

81 f"Implement patterns()/replacement() instead." 

82 ) 

83 from .pattern import ( 

84 _adapt_decorated_pattern_methods, 

85 _adapt_expression_replacement, 

86 _has_decorated_pattern_methods, 

87 ) 

88 

89 if _has_decorated_pattern_methods(cls): 

90 if "patterns" in cls.__dict__: 

91 raise TypeError( 

92 f"{cls.__name__} cannot combine @pattern methods with patterns(). " 

93 f"Use one style for declaring patterns." 

94 ) 

95 cls.patterns = _adapt_decorated_pattern_methods(cls) 

96 if "replacement" in cls.__dict__: 

97 cls.replacement = _adapt_expression_replacement(cls.__dict__["replacement"]) 

98 

99 @property 

100 def matcher_config(self) -> Optional[PatternMatcherConfig]: 

101 return self._matcher_config 

102 

103 def patterns(self) -> Iterable[PatternOrGraph]: 

104 raise NotImplementedError("PatternFusionPass.patterns must be implemented") 

105 

106 def meet_requirements(self, match_result: MatchResult) -> bool: 

107 return True 

108 

109 def replacement(self, match_result: MatchResult) -> "Graph": 

110 raise NotImplementedError("PatternFusionPass.replacement must be implemented") 

111 

112 

113class DecomposePass(FusionBasePass): 

114 """Python DecomposePass contract. 

115 

116 The execution engine calls ``meet_requirements()`` and ``replacement()`` 

117 for matched nodes — **not** ``run()``. Overriding ``run()`` in a 

118 ``DecomposePass`` subclass has no effect; implement the two hook methods 

119 above instead. 

120 """ 

121 

122 op_types: Optional[List[str]] = None 

123 

124 def __init_subclass__(cls, **kwargs) -> None: 

125 super().__init_subclass__(**kwargs) 

126 if "run" in cls.__dict__: 

127 raise TypeError( 

128 f"{cls.__name__} overrides run(), which is never invoked " 

129 f"by the DecomposePass execution path. " 

130 f"Implement meet_requirements()/replacement() instead." 

131 ) 

132 

133 def meet_requirements(self, node: "Node") -> bool: 

134 return True 

135 

136 def replacement(self, node: "Node") -> "Graph": 

137 raise NotImplementedError("DecomposePass.replacement must be implemented")