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

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# Licensed under the Apache License, Version 2.0 

3# ============================================================================ 

4"""Recover inline rules from existing YAML-derived codegen metadata. 

5 

6The user YAML is not extended. Codegen reads the already frozen 

7``module_overrides`` and ``injections`` metadata, whose fields originate from 

8``plan_overrides.replace_module`` / ``inner_wrapper`` / ``local_compute_fn``. 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import Any, Iterable, Optional 

14 

15from hyper_parallel.codegen.inline.ir import InlineRule 

16 

17 

18def collect_inline_rules(meta: Any) -> tuple[InlineRule, ...]: 

19 """Return normalized inline rules recovered from ``meta``.""" 

20 

21 rules: list[InlineRule] = [] 

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

23 target = record.get("factory") 

24 if not target: 

25 continue 

26 rules.append( 

27 InlineRule( 

28 match=_matches(record), 

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

30 replace_target=target, 

31 ) 

32 ) 

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

34 local = _target_path(record.get("local_compute_fn")) 

35 inner = _target_path(record.get("inner_wrapper")) 

36 if local is None and inner is None: 

37 continue 

38 rules.append( 

39 InlineRule( 

40 match=_matches(record), 

41 inner_wrapper_target=inner, 

42 local_compute_target=local, 

43 ) 

44 ) 

45 return tuple(rules) 

46 

47 

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

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

50 if value is None: 

51 return () 

52 if isinstance(value, str): 

53 return (value,) 

54 if isinstance(value, Iterable): 

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

56 return (str(value),) 

57 

58 

59def _target_path(value: Any) -> Optional[str]: 

60 if value is None: 

61 return None 

62 if isinstance(value, str): 

63 return value 

64 if isinstance(value, dict): 

65 return value.get("_target_") or value.get("target") 

66 return getattr(value, "__qualname__", None)