Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / trainer / config / target.py: 95%

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"""Delayed-invocation target wrapper and config serialization helper. 

16 

17Split from ``auto_models/trainer/config.py`` in stage 7 (05 §15.2.5): 

18``Target`` is the YAML transport wrapper; ``_serialize_config_value`` is 

19shared by every nested config's ``to_dict`` and therefore lives beside it 

20(the plan lists it under ``trainer.py``; placing it here avoids a circular 

21import, the public surface is unchanged). 

22""" 

23 

24import inspect 

25from dataclasses import fields, is_dataclass 

26from typing import Any, Callable, Generic, TypeVar 

27 

28_T = TypeVar("_T") 

29 

30 

31def _serialize_config_value(value: Any) -> Any: 

32 """Convert one target argument to a plain serializable value.""" 

33 if inspect.isroutine(value): 

34 return f"{value.__module__}.{value.__qualname__}" 

35 if hasattr(value, "to_dict"): 

36 return value.to_dict() 

37 if is_dataclass(value): 

38 return { 

39 config_field.name: _serialize_config_value( 

40 getattr(value, config_field.name) 

41 ) 

42 for config_field in fields(value) 

43 } 

44 if isinstance(value, dict): 

45 return { 

46 key: _serialize_config_value(item) 

47 for key, item in value.items() 

48 } 

49 if isinstance(value, (list, tuple)): 

50 return [_serialize_config_value(item) for item in value] 

51 return value 

52 

53 

54class Target(Generic[_T]): 

55 """Configuration for one callable whose invocation is delayed until runtime.""" 

56 

57 def __init__( 

58 self, 

59 _target_: Callable[..., _T], # pylint: disable=invalid-name 

60 *, 

61 target_path: str, 

62 **kwargs: Any, 

63 ) -> None: 

64 """Store the resolved callable, its source path, and configured arguments.""" 

65 if not callable(_target_): 

66 raise TypeError("_target_ must be callable") 

67 if not isinstance(target_path, str) or not target_path.strip(): 

68 raise ValueError("target_path must be a non-empty string") 

69 

70 self._target_ = _target_ 

71 self._target_path = target_path 

72 self._kwargs = dict(kwargs) 

73 

74 def __getattr__(self, name: str) -> Any: 

75 kwargs = object.__getattribute__(self, "_kwargs") 

76 try: 

77 return kwargs[name] 

78 except KeyError as exc: 

79 raise AttributeError(name) from exc 

80 

81 @property 

82 def callable(self) -> Callable[..., _T]: 

83 """Return the wrapped callable, read-only, for callers outside this class.""" 

84 return self._target_ 

85 

86 def build(self, **runtime_kwargs: Any) -> _T: 

87 """Invoke the target with configured and applicable runtime arguments. 

88 

89 Nested :class:`Target` values are built immediately before their 

90 parent. This lets configuration inject a model-owned adapter into a 

91 generic framework target without introducing a model-specific wrapper 

92 factory. 

93 """ 

94 signature = inspect.signature(self._target_) 

95 if not any( 

96 parameter.kind is inspect.Parameter.VAR_KEYWORD 

97 for parameter in signature.parameters.values() 

98 ): 

99 runtime_kwargs = { 

100 name: value 

101 for name, value in runtime_kwargs.items() 

102 if name in signature.parameters 

103 } 

104 

105 kwargs = {**self._kwargs, **runtime_kwargs} 

106 kwargs = { 

107 name: _build_nested_targets(value) 

108 for name, value in kwargs.items() 

109 } 

110 return self._target_(**kwargs) 

111 

112 def replace(self, **changes: Any) -> "Target[_T]": 

113 """Return a new target with selected configured arguments replaced.""" 

114 kwargs = dict(self._kwargs) 

115 kwargs.update(changes) 

116 return type(self)( 

117 self._target_, 

118 target_path=self._target_path, 

119 **kwargs, 

120 ) 

121 

122 def to_dict(self) -> dict[str, Any]: 

123 """Serialize this target back to its YAML-compatible form.""" 

124 return { 

125 "_target_": self._target_path, 

126 **{ 

127 name: _serialize_config_value(value) 

128 for name, value in self._kwargs.items() 

129 }, 

130 } 

131 

132 

133def _build_nested_targets(value: Any) -> Any: 

134 """Materialize deferred targets inside one configured argument tree.""" 

135 if isinstance(value, Target): 

136 return value.build() 

137 if isinstance(value, dict): 

138 return { 

139 key: _build_nested_targets(item) 

140 for key, item in value.items() 

141 } 

142 if isinstance(value, list): 

143 return [_build_nested_targets(item) for item in value] 

144 if isinstance(value, tuple): 

145 return tuple(_build_nested_targets(item) for item in value) 

146 return value