Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / config / configurable.py: 85%

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"""Declarative configuration base types for HyperParallel components.""" 

16 

17from __future__ import annotations 

18 

19import dataclasses 

20import logging 

21from collections.abc import Iterator 

22from dataclasses import dataclass, fields, replace 

23from typing import Any, ClassVar 

24 

25logger = logging.getLogger(__name__) 

26 

27 

28def _build_from_config( 

29 owner: type["Configurable"], 

30 config: "Configurable.Config", 

31 **kwargs: Any, 

32) -> "Configurable": 

33 """Instantiate *owner* from a config snapshot. 

34 

35 Thin wrapper around ``owner(config=config, **kwargs)`` so static analyzers 

36 can resolve the constructor call site. 

37 

38 Args: 

39 owner: Configurable subclass to construct (from ``Config._owner``). 

40 config: Config instance passed to ``owner.__init__``. 

41 **kwargs: Extra keyword arguments forwarded to ``owner.__init__``. 

42 

43 Returns: 

44 A new instance of *owner*. 

45 """ 

46 return owner(config=config, **kwargs) 

47 

48 

49class Configurable: 

50 """Base class for components built from nested dataclass configs. 

51 

52 Subclasses declare a nested ``Config`` and implement ``__init__(self, config)``. 

53 ``Config.build()`` constructs the outer class; ``__init_subclass__`` wires 

54 ``Config._owner`` when the nested class is defined on the subclass. 

55 

56 ``build()`` supports two forms: 

57 

58 - No extra kwargs: ``owner(config=replace(self))``. 

59 - With kwargs: runtime-only values (not stored on the config) are forwarded 

60 to ``owner.__init__``; names must not overlap config fields. 

61 

62 Module configs additionally require ``@dataclass(kw_only=True, slots=True)`` 

63 via :func:`enforce_module_config_slots` in 

64 :mod:`hyper_parallel.dmodule.module`. 

65 """ 

66 

67 Config: type["Configurable.Config"] 

68 

69 @dataclass(kw_only=True, slots=True) 

70 class Config: 

71 """Base dataclass for component configuration trees.""" 

72 

73 _owner: ClassVar[type["Configurable"] | None] = None 

74 

75 def replace(self, **kwargs) -> "Configurable.Config": 

76 """Return a shallow copy with selected fields replaced. 

77 

78 Args: 

79 **kwargs: Field names and new values (same semantics as 

80 :func:`dataclasses.replace`). 

81 

82 Returns: 

83 A new config instance; the original is unchanged. 

84 """ 

85 return replace(self, **kwargs) 

86 

87 def to_dict(self) -> dict: 

88 """Serialize this config tree to a JSON-friendly dict. 

89 

90 Recurses into nested configs and containers. Callable fields (for 

91 example ``param_init`` hooks) are converted with ``repr()``; other 

92 non-JSON types log a warning and fall back to ``repr()``. 

93 

94 Fields whose names start with ``_`` (such as ``_owner``) are omitted. 

95 

96 Returns: 

97 Plain ``dict`` suitable for ``json.dumps``. 

98 """ 

99 

100 def _convert(val): 

101 if hasattr(val, "to_dict"): 

102 return val.to_dict() 

103 if dataclasses.is_dataclass(val): 

104 return dataclasses.asdict(val) 

105 if isinstance(val, (list, tuple)): 

106 return type(val)(_convert(v) for v in val) 

107 if isinstance(val, dict): 

108 return {k: _convert(v) for k, v in val.items()} 

109 if isinstance(val, (str, int, float, bool, type(None))): 

110 return val 

111 if callable(val): 

112 return repr(val) 

113 logger.warning( 

114 "Config field value of type %s may not be JSON serializable", 

115 type(val).__name__, 

116 ) 

117 return repr(val) 

118 

119 return { 

120 f.name: _convert(getattr(self, f.name)) 

121 for f in fields(self) 

122 if not f.name.startswith("_") 

123 } 

124 

125 def traverse( 

126 self, 

127 config_cls: type, 

128 *, 

129 _prefix: str = "", 

130 ) -> Iterator[tuple[str, "Configurable.Config", object, str | int]]: 

131 """Walk nested configs and yield nodes matching *config_cls*. 

132 

133 Traverses dataclass fields and list elements recursively. Each 

134 yielded fully-qualified name (*fqn*) mirrors the attribute path 

135 under the root config (for example ``"layers.0.attention"``). 

136 

137 Args: 

138 config_cls: Config type to match (for example a specific 

139 :class:`~hyper_parallel.dmodule.module.Module` nested 

140 ``Config`` subclass). 

141 _prefix: Internal path prefix used during recursion. 

142 

143 Yields: 

144 Tuples ``(fqn, config, parent, field_name)`` where *parent* is 

145 the object holding *config* (another config or a ``list``), and 

146 *field_name* is the attribute name or list index. Callers can 

147 replace nodes in place, for example via ``setattr(parent, attr, 

148 new_cfg)`` or ``parent[attr] = new_cfg`` when *parent* is a list. 

149 """ 

150 for field in fields(self): 

151 val = getattr(self, field.name) 

152 fqn = f"{_prefix}.{field.name}" if _prefix else field.name 

153 if isinstance(val, config_cls): 

154 yield fqn, val, self, field.name 

155 elif isinstance(val, list): 

156 for index, item in enumerate(val): 

157 item_fqn = f"{fqn}.{index}" 

158 if isinstance(item, config_cls): 

159 yield item_fqn, item, val, index 

160 elif hasattr(item, "traverse"): 

161 yield from item.traverse(config_cls, _prefix=item_fqn) 

162 elif hasattr(val, "traverse"): 

163 yield from val.traverse(config_cls, _prefix=fqn) 

164 

165 def build(self, **kwargs): 

166 """Construct the :class:`Configurable` subclass bound to this config. 

167 

168 Uses ``Config._owner`` set by :meth:`Configurable.__init_subclass__`. 

169 The config is copied with :func:`dataclasses.replace` before 

170 construction so the template instance is not mutated. 

171 

172 Args: 

173 **kwargs: Runtime-only arguments for ``owner.__init__``. Must not 

174 name fields already present on this config. 

175 

176 Returns: 

177 New instance of ``self._owner``. 

178 

179 Raises: 

180 NotImplementedError: If ``_owner`` was never wired (config not 

181 defined inside a :class:`Configurable` subclass). 

182 ValueError: If any ``kwargs`` key overlaps a config field name. 

183 """ 

184 if self._owner is None: 

185 raise NotImplementedError( 

186 f"{type(self).__name__} has no owner class. " 

187 "Define Config inside a Configurable subclass." 

188 ) 

189 owner_cls: type[Configurable] = self._owner 

190 built_config = replace(self) 

191 if not kwargs: 

192 return _build_from_config(owner_cls, built_config) 

193 

194 config_field_names = {f.name for f in fields(self)} 

195 overlap = config_field_names & kwargs.keys() 

196 if overlap: 

197 raise ValueError( 

198 f"build() kwargs {overlap} overlap with config fields. " 

199 "Put these values in the Config, not in build() kwargs." 

200 ) 

201 return _build_from_config(owner_cls, built_config, **kwargs) 

202 

203 def __init_subclass__(cls, **kwargs): 

204 """Bind nested ``Config._owner`` when a subclass defines its own ``Config``.""" 

205 super().__init_subclass__(**kwargs) 

206 if "Config" in cls.__dict__: 

207 config_cls = cls.__dict__["Config"] 

208 if issubclass(config_cls, Configurable.Config): 

209 config_cls._owner = cls 

210 

211 

212def enforce_module_config_slots(config_cls: type, owner_name: str) -> None: 

213 """Require ``@dataclass(kw_only=True, slots=True)`` on a Module nested config. 

214 

215 Called from :class:`~hyper_parallel.dmodule.module.Module` ``__init_subclass__`` 

216 for configs declared directly on Module subclasses. 

217 

218 Args: 

219 config_cls: Nested config class to validate. 

220 owner_name: Outer class name (used in error messages). 

221 

222 Raises: 

223 TypeError: If *config_cls* lacks ``slots=True`` or has positional-init 

224 fields. 

225 """ 

226 if "__slots__" not in config_cls.__dict__: 

227 raise TypeError( 

228 f"{owner_name}.Config must use @dataclass(kw_only=True, slots=True)" 

229 ) 

230 for field in fields(config_cls): 

231 if field.init and not field.kw_only: 

232 raise TypeError( 

233 f"{owner_name}.Config field '{field.name}' must be keyword-only" 

234 ) 

235 

236 

237__all__ = ["Configurable", "enforce_module_config_slots"]