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

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"""Project native replacement and injection observations into source rules.""" 

16 

17from __future__ import annotations 

18 

19from typing import Any, Iterable 

20 

21from hyper_parallel.codegen.inline.ir import SourceObservation 

22from hyper_parallel.codegen.inline.targets import projected_target_path 

23 

24 

25def collect_source_observations(meta: Any) -> tuple[SourceObservation, ...]: 

26 """Return normalized source rules from the already observed native plan.""" 

27 rules: list[SourceObservation] = [] 

28 for record in getattr(meta, "module_overrides", None) or []: 

29 target = _replacement_target(record) 

30 if target: 

31 rules.append( 

32 SourceObservation( 

33 match=_matches(record), 

34 module_type=record.get("module_type"), 

35 replace_target=target, 

36 ) 

37 ) 

38 for record in getattr(meta, "injections", None) or []: 

39 local = projected_target_path(record.get("local_compute_fn")) 

40 inner = projected_target_path(record.get("inner_wrapper")) 

41 if local is not None or inner is not None: 

42 rules.append( 

43 SourceObservation( 

44 match=_matches(record), 

45 inner_wrapper_target=inner, 

46 local_compute_target=local, 

47 ) 

48 ) 

49 return tuple(rules) 

50 

51 

52def _replacement_target(record: dict[str, Any]) -> Any: 

53 """Prefer model-owned declarations and otherwise use the observed factory.""" 

54 declared = record.get("replacement_target") 

55 parts = str(declared or "").split(".") 

56 if len(parts) >= 4 and parts[:2] == ["hyper_parallel", "models"] and parts[3] == "adapter": 

57 return declared 

58 return record.get("factory") 

59 

60 

61def _matches(record: dict[str, Any]) -> tuple[str, ...]: 

62 """Normalize one observation's selected FQNs.""" 

63 value = record.get("fqns") or record.get("match") or record.get("fqn") 

64 if value is None: 

65 return () 

66 if isinstance(value, str): 

67 return (value,) 

68 if isinstance(value, Iterable): 

69 return tuple(str(item) for item in value) 

70 return (str(value),) 

71 

72 

73__all__ = ["collect_source_observations"]