Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / codegen / inline / source_plan.py: 42%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Build one source patch plan from native replacement and strategy observations.""" 

16 

17from __future__ import annotations 

18 

19from collections.abc import Collection, Mapping 

20 

21from hyper_parallel.codegen.inline.ir import InlinePatchSet, SourceObservation 

22from hyper_parallel.codegen.inline.replacement_pass import build_replacement_patches 

23from hyper_parallel.codegen.inline.strategy_pass import build_strategy_patches 

24 

25 

26def build_source_patch_plan( 

27 observations: tuple[SourceObservation, ...], 

28 strategy_observations: tuple[SourceObservation, ...], 

29 model_type: str | None, 

30 boundary_classes: Mapping[str, str] | None, 

31 external_state_classes: Collection[str] | None, 

32) -> InlinePatchSet: 

33 """Build the complete ordered patch set for one modeling source.""" 

34 replacement = build_replacement_patches(observations, model_type) 

35 strategy = build_strategy_patches( 

36 strategy_observations, 

37 model_type, 

38 boundary_classes, 

39 external_state_classes, 

40 ) 

41 return merge_patch_sets(replacement, strategy) 

42 

43 

44def merge_patch_sets(*patch_sets: InlinePatchSet) -> InlinePatchSet: 

45 """Merge pass outputs without changing their established application order.""" 

46 merged = InlinePatchSet() 

47 for patch_set in patch_sets: 

48 merged.imports.extend(patch_set.imports) 

49 merged.module_snippets.extend(patch_set.module_snippets) 

50 merged.constructor_replaces.extend(patch_set.constructor_replaces) 

51 merged.forward_extracts.extend(patch_set.forward_extracts) 

52 merged.class_removals.extend(patch_set.class_removals) 

53 return merged 

54 

55 

56__all__ = ["build_source_patch_plan", "merge_patch_sets"]