Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / trainer / config / resolver.py: 71%
371 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-25 05:45 +0800
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"""Resolve target-selected YAML groups into a typed trainer config tree."""
17from __future__ import annotations
19__all__ = [
20 "ConfigResolutionError",
21 "replace_override_path",
22 "resolve_config",
23]
25import dataclasses
26import difflib
27import importlib
28import inspect
29import types
30from collections.abc import Iterable, Mapping
31from dataclasses import MISSING, fields, is_dataclass, replace
32from typing import Any, Literal, Union, get_args, get_origin, get_type_hints
34from hyper_parallel.trainer.config.data import (
35 DataLoaderConfig,
36 DatasetConfig,
37 ModelAssetsConfig,
38)
39from hyper_parallel.trainer.config.optimization import (
40 OptimizerConfig,
41 OptimizerSwapConfig,
42)
43from hyper_parallel.trainer.config.target import Target
44from hyper_parallel.trainer.config.trainer import TrainerConfig
47_NONE_TYPE = types.NoneType # pylint: disable=no-member
48_UNION_TYPE = types.UnionType # pylint: disable=no-member
51class ConfigResolutionError(ValueError):
52 """A target or typed configuration value is invalid."""
54 def __init__(self, location: str, message: str) -> None:
55 """Prefix the error message with its configuration location."""
56 super().__init__(f"{location}: {message}")
59def import_target(target_path: str, *, location: str) -> object:
60 """Import a callable from a dotted Python path.
62 Args:
63 target_path: Module path followed by callable attributes.
64 location: Configuration location used in error messages.
66 Returns:
67 The imported callable, without invoking it.
69 Raises:
70 ConfigResolutionError: The path is invalid, cannot be imported, or
71 does not identify a callable.
72 """
73 if not isinstance(target_path, str) or not target_path.strip():
74 raise ConfigResolutionError(location, "_target_ must be a non-empty dotted path")
76 parts = target_path.split(".")
77 if any(not part for part in parts):
78 raise ConfigResolutionError(location, f"invalid target path {target_path!r}")
80 for split_at in range(len(parts), 0, -1):
81 module_name = ".".join(parts[:split_at])
82 try:
83 target = importlib.import_module(module_name)
84 except ModuleNotFoundError as exc:
85 if exc.name == module_name or module_name.startswith(f"{exc.name}."):
86 continue
87 raise ConfigResolutionError(
88 location,
89 f"target {target_path!r} failed while importing dependency {exc.name!r}",
90 ) from exc
91 except ImportError as exc:
92 raise ConfigResolutionError(location, f"target {target_path!r} could not be imported: {exc}") from exc
94 for attribute in parts[split_at:]:
95 if not hasattr(target, attribute):
96 raise ConfigResolutionError(
97 location,
98 f"target {target_path!r} has no attribute {attribute!r}",
99 )
100 target = getattr(target, attribute)
102 if not callable(target):
103 raise ConfigResolutionError(location, f"target {target_path!r} is not callable")
104 return target
106 raise ConfigResolutionError(location, f"target {target_path!r} could not be imported")
109def _is_union(annotation: object) -> bool:
110 """Return whether the annotation is a ``Union`` or a PEP 604 union."""
111 return get_origin(annotation) in (Union, _UNION_TYPE)
114def _annotation_name(annotation: object) -> str:
115 """Format an annotation for configuration error messages."""
116 if annotation is Any:
117 return "Any"
118 if isinstance(annotation, type):
119 return annotation.__qualname__
120 return str(annotation).replace("typing.", "")
123def _target_hints(target: object, *, path: str) -> dict[str, object]:
124 """Resolve annotations for a target function or class constructor."""
125 hint_source = target.__init__ if inspect.isclass(target) else target
126 try:
127 return get_type_hints(hint_source)
128 except (NameError, TypeError) as exc:
129 raise ConfigResolutionError(path, f"could not resolve target type annotations: {exc}") from exc
132# Annotations whose payload is a container of other values.
133_COLLECTION_TYPES: tuple[type, ...] = (list, tuple, dict, Mapping)
136def _normalize_sequence(
137 value: object,
138 item_types: tuple,
139 *,
140 uniform: bool,
141 kind: str,
142 path: str,
143) -> list | tuple:
144 """Normalize sequence items against their declared types.
146 With ``uniform=True``, one item type applies to every element. Otherwise,
147 ``item_types`` must contain one type per element. Empty ``item_types``
148 produces a tuple without normalizing its items.
149 """
150 if not isinstance(value, (list, tuple)):
151 raise ConfigResolutionError(path, f"expected {kind}, got {type(value).__name__}")
152 if not item_types:
153 return tuple(value)
154 if uniform:
155 item_types = item_types * len(value)
156 if len(item_types) != len(value):
157 raise ConfigResolutionError(
158 path,
159 f"expected {kind} of length {len(item_types)}, got {len(value)}",
160 )
161 normalized = [
162 normalize_value(item, item_type, path=f"{path}[{index}]")
163 for index, (item, item_type) in enumerate(zip(value, item_types))
164 ]
165 return tuple(normalized) if kind == "tuple" else normalized
168def _require_none_allowed(annotation: object, *, path: str) -> None:
169 """Accept ``None`` only when the annotation permits it."""
170 if annotation is _NONE_TYPE or (
171 _is_union(annotation) and _NONE_TYPE in get_args(annotation)
172 ):
173 return None
174 raise ConfigResolutionError(path, f"expected {_annotation_name(annotation)}, got None")
177def _normalize_union(value: object, annotation: object, *, path: str) -> object:
178 """Normalize a value against the first compatible union member."""
179 members = get_args(annotation)
180 non_none_members = tuple(member for member in members if member is not _NONE_TYPE)
181 if len(non_none_members) == 1 and len(non_none_members) != len(members):
182 return normalize_value(value, non_none_members[0], path=path)
184 for member in members:
185 try:
186 return normalize_value(value, member, path=path)
187 except ConfigResolutionError:
188 continue
189 raise ConfigResolutionError(
190 path,
191 f"expected {_annotation_name(annotation)}, got {type(value).__name__}",
192 )
195def _normalize_scalar(value: object, annotation: object, *, path: str) -> object:
196 """Validate a scalar's type, allowing integers to normalize to floats."""
197 if annotation is bool and isinstance(value, bool):
198 return value
199 if annotation is int and isinstance(value, int) and not isinstance(value, bool):
200 return value
201 if annotation is float and isinstance(value, (int, float)) and not isinstance(value, bool):
202 return float(value)
203 if annotation is str and isinstance(value, str):
204 return value
205 raise ConfigResolutionError(
206 path,
207 f"expected {_annotation_name(annotation)}, got {type(value).__name__}",
208 )
211def _normalize_literal(value: object, choices: tuple, *, path: str) -> object:
212 """Validate a literal value, restoring YAML boolean aliases when allowed."""
213 # PyYAML 1.1 parses unquoted on/off/yes/no/true/false scalars as bool.
214 # Map the bool back to the matching word when it is one of the choices.
215 if isinstance(value, bool):
216 words = ("on", "yes", "true") if value else ("off", "no", "false")
217 for word in words:
218 if word in choices:
219 return word
220 if any(type(value) is type(choice) and value == choice for choice in choices):
221 return value
222 expected = ", ".join(repr(choice) for choice in choices)
223 raise ConfigResolutionError(path, f"expected one of ({expected}), got {value!r}")
226def _normalize_collection(
227 value: object,
228 annotation: object,
229 origin: object,
230 args: tuple,
231 *,
232 path: str,
233) -> object:
234 """Normalize a value against a list, tuple, or mapping annotation."""
235 if origin is list or annotation is list:
236 return _normalize_sequence(
237 value,
238 (args[0] if args else Any,),
239 uniform=True,
240 kind="list",
241 path=path,
242 )
243 if origin is tuple or annotation is tuple:
244 if len(args) == 2 and args[1] is Ellipsis:
245 return _normalize_sequence(value, args[:1], uniform=True, kind="tuple", path=path)
246 return _normalize_sequence(value, args, uniform=False, kind="tuple", path=path)
247 if not isinstance(value, Mapping):
248 raise ConfigResolutionError(path, f"expected mapping, got {type(value).__name__}")
249 return dict(value)
252def _normalize_class(value: object, annotation: object, *, path: str) -> object:
253 """Validate that a value is an instance of the annotated class."""
254 if not isinstance(annotation, type):
255 raise ConfigResolutionError(path, f"unsupported type annotation {_annotation_name(annotation)}")
256 if isinstance(value, annotation):
257 return value
258 raise ConfigResolutionError(
259 path,
260 f"expected {_annotation_name(annotation)}, got {type(value).__name__}",
261 )
264def _normalize_override_scalar(value: object, annotation: object) -> object:
265 """Convert a CLI string to the numeric type declared by its annotation.
267 Supports ``int``, ``float``, and their optional forms. YAML may leave
268 numeric CLI input such as ``1e-4`` as a string. Failed conversions leave
269 the value unchanged for subsequent validation by ``normalize_value``.
270 """
271 if not isinstance(value, str):
272 return value
273 target = annotation
274 if get_origin(target) in (Union, types.UnionType):
275 members = [member for member in get_args(target) if member is not types.NoneType]
276 if len(members) != 1:
277 return value
278 target = members[0]
279 if target is int:
280 try:
281 return int(value)
282 except ValueError:
283 return value
284 if target is float:
285 try:
286 return float(value)
287 except ValueError:
288 return value
289 return value
292def normalize_value(value: object, annotation: object, *, path: str) -> object:
293 """Validate and normalize a parsed value against its annotation.
295 Args:
296 value: Parsed value to validate or convert.
297 annotation: Type annotation defining the accepted value.
298 path: Configuration path used in error messages.
300 Returns:
301 The validated or converted value.
303 Raises:
304 ConfigResolutionError: The value does not match the annotation or the
305 annotation is unsupported.
306 """
307 # A nested target is a deferred instance of the annotated runtime type.
308 # Its constructor arguments are validated when its own node is resolved;
309 # the parent Target materializes it immediately before invocation.
310 if isinstance(value, Target):
311 return value
312 if annotation in (Any, object):
313 return value
314 if isinstance(annotation, dataclasses.InitVar):
315 return normalize_value(value, annotation.type, path=path)
316 if value is None:
317 return _require_none_allowed(annotation, path=path)
318 if _is_union(annotation):
319 return _normalize_union(value, annotation, path=path)
320 if annotation in (bool, int, float, str):
321 return _normalize_scalar(value, annotation, path=path)
323 origin = get_origin(annotation)
324 args = get_args(annotation)
325 if origin is Literal:
326 return _normalize_literal(value, args, path=path)
327 if origin in _COLLECTION_TYPES or annotation in _COLLECTION_TYPES:
328 return _normalize_collection(value, annotation, origin, args, path=path)
329 if isinstance(annotation, type) and dataclasses.is_dataclass(annotation):
330 # Nested dataclass items resolve from mappings in the same way as
331 # top-level dataclass components.
332 return _resolve_dataclass(value, annotation, path=path)
333 return _normalize_class(value, annotation, path=path)
336def _suggestion(name: str, candidates: Iterable[str]) -> str:
337 """Return a spelling suggestion, or an empty string if no name is close."""
338 matches = difflib.get_close_matches(name, candidates, n=1)
339 return f"; did you mean {matches[0]!r}?" if matches else ""
342def _replace_target_path(
343 config: Target[Any],
344 parts: list[str],
345 value: object,
346 *,
347 path: str,
348) -> Target[Any]:
349 """Return a ``Target`` copy with an argument or nested value replaced."""
350 name = parts[0]
351 full_path = f"{path}.{name}" if path else name
352 if name == "_target_":
353 raise ConfigResolutionError(
354 f"CLI.{full_path}", "changing _target_ through an override is not supported"
355 )
357 signature = inspect.signature(config.callable)
358 parameter = signature.parameters.get(name)
359 has_var_kwargs = any(
360 item.kind is inspect.Parameter.VAR_KEYWORD
361 for item in signature.parameters.values()
362 )
363 if parameter is None and not has_var_kwargs:
364 candidates = [
365 item.name
366 for item in signature.parameters.values()
367 if item.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
368 ]
369 raise ConfigResolutionError(
370 f"CLI.{full_path}",
371 f"unknown target argument {name!r}{_suggestion(name, candidates)}",
372 )
374 if len(parts) > 1:
375 try:
376 child = getattr(config, name)
377 except AttributeError as exc:
378 raise ConfigResolutionError(
379 f"CLI.{full_path}", "target argument is not configured"
380 ) from exc
381 return config.replace(**{name: replace_override_path(child, parts[1:], value, path=full_path)})
383 normalized = value
384 if parameter is not None:
385 annotation = _target_hints(config.callable, path=path).get(
386 name,
387 parameter.annotation,
388 )
389 if annotation not in (Any, object, inspect.Signature.empty):
390 normalized = normalize_value(
391 _normalize_override_scalar(value, annotation),
392 annotation,
393 path=f"CLI.{full_path}",
394 )
395 return config.replace(**{name: normalized})
398def replace_override_path(config: object, parts: list[str], value: object, *, path: str) -> object:
399 """Return a configuration copy with a dotted-path override applied.
401 Args:
402 config: Resolved configuration object or nested value to update.
403 parts: Nonempty sequence of remaining dotted-path components.
404 value: Parsed override value to validate and assign.
405 path: Path already traversed, or an empty string at the root.
407 Returns:
408 A copy with the override applied; the input is not modified.
410 Raises:
411 ConfigResolutionError: The path cannot be updated or the value does
412 not match the selected field's annotation.
413 """
414 if isinstance(config, Target):
415 return _replace_target_path(config, parts, value, path=path)
417 if isinstance(config, Mapping):
418 name = parts[0]
419 full_path = f"{path}.{name}" if path else name
420 if name not in config:
421 raise ConfigResolutionError(f"CLI.{path}", f"unknown mapping key {name!r}")
422 if len(parts) == 1:
423 return {**config, name: value}
424 return {**config, name: replace_override_path(config[name], parts[1:], value, path=full_path)}
426 location = f"CLI.{path}" if path else "CLI"
427 if not is_dataclass(config):
428 raise ConfigResolutionError(
429 location, f"value of type {type(config).__name__} has no configurable fields"
430 )
432 config_fields = {field.name: field for field in fields(config)}
433 name = parts[0]
434 if name not in config_fields:
435 target = getattr(config, "target", None)
436 if isinstance(target, Target):
437 return replace(config, target=_replace_target_path(target, parts, value, path=path))
438 raise ConfigResolutionError(
439 location,
440 f"unknown field {name!r} on {type(config).__name__}"
441 f"{_suggestion(name, config_fields)}",
442 )
444 full_path = f"{path}.{name}" if path else name
445 if len(parts) == 1:
446 annotation = get_type_hints(type(config))[name]
447 normalized = normalize_value(
448 _normalize_override_scalar(value, annotation), annotation, path=f"CLI.{full_path}"
449 )
450 return replace(config, **{name: normalized})
452 child = getattr(config, name)
453 if child is None:
454 raise ConfigResolutionError(
455 f"CLI.{full_path}", "component was not selected by the YAML"
456 )
457 return replace(config, **{name: replace_override_path(child, parts[1:], value, path=full_path)})
460def _resolve_union(node: object, annotation: object, *, path: str) -> object:
461 """Resolve a YAML value against a compatible non-``None`` union member."""
462 non_none_members = [
463 member for member in get_args(annotation) if member is not _NONE_TYPE
464 ]
465 if len(non_none_members) == 1:
466 # Single-member union (e.g. Optional[Target]): resolve directly so the
467 # member's specific error (e.g. an unexpected target argument) reaches
468 # the user instead of being swallowed by the generic union failure.
469 return resolve_component(node, annotation=non_none_members[0], path=path)
471 last_error: ConfigResolutionError | None = None
472 for member in non_none_members:
473 try:
474 return resolve_component(node, annotation=member, path=path)
475 except ConfigResolutionError as exc:
476 last_error = exc
477 raise ConfigResolutionError(
478 path,
479 f"expected {_annotation_name(annotation)}, got {type(node).__name__}",
480 ) from last_error
483def _resolve_dataclass(node: object, config_type: type, *, path: str) -> object:
484 """Resolve YAML fields and construct a configuration dataclass."""
485 if not isinstance(node, Mapping):
486 raise ConfigResolutionError(path, "configuration section must be a YAML mapping")
488 config_fields = {field.name: field for field in fields(config_type)}
489 unknown = sorted(set(node) - set(config_fields))
490 if unknown:
491 raise ConfigResolutionError(path, f"unknown configuration fields: {unknown}")
493 required = [
494 field.name
495 for field in config_fields.values()
496 if field.default is MISSING and field.default_factory is MISSING
497 ]
498 missing = [name for name in required if name not in node]
499 if missing:
500 raise ConfigResolutionError(path, f"missing required configuration fields: {missing}")
502 try:
503 hints = get_type_hints(config_type)
504 except (NameError, TypeError) as exc:
505 raise ConfigResolutionError(path, f"could not resolve configuration type annotations: {exc}") from exc
507 resolved = {
508 name: resolve_component(
509 value,
510 annotation=hints[name],
511 path=f"{path}.{name}",
512 )
513 for name, value in node.items()
514 }
515 try:
516 return config_type(**resolved)
517 except TypeError as exc:
518 raise ConfigResolutionError(path, f"could not construct {config_type.__name__}: {exc}") from exc
521def _resolve_nested_target_nodes(value: object, *, path: str) -> object:
522 """Resolve reserved ``_target_`` nodes inside a target argument tree."""
523 if isinstance(value, Mapping):
524 if "_target_" in value:
525 return _resolve_target(value, path=path)
526 return {
527 key: _resolve_nested_target_nodes(item, path=f"{path}.{key}")
528 for key, item in value.items()
529 }
530 if isinstance(value, list):
531 return [
532 _resolve_nested_target_nodes(item, path=f"{path}[{index}]")
533 for index, item in enumerate(value)
534 ]
535 if isinstance(value, tuple):
536 return tuple(
537 _resolve_nested_target_nodes(item, path=f"{path}[{index}]")
538 for index, item in enumerate(value)
539 )
540 return value
543def _resolve_target_args(
544 raw_args: Mapping[str, object],
545 signature: inspect.Signature,
546 hints: Mapping[str, object],
547 *,
548 path: str,
549) -> dict[str, object]:
550 """Validate target arguments and fill omitted keyword-callable defaults.
552 Positional-only defaults are not added because ``Target`` invokes its
553 callable with keyword arguments.
554 """
555 try:
556 signature.bind_partial(**raw_args)
557 except TypeError as exc:
558 raise ConfigResolutionError(path, f"target arguments are invalid: {exc}") from exc
560 normalized = {}
561 for name, value in raw_args.items():
562 parameter = signature.parameters.get(name)
563 value = _resolve_nested_target_nodes(value, path=f"{path}.{name}")
564 if parameter is None:
565 normalized[name] = value
566 continue
568 annotation = hints.get(name, parameter.annotation)
569 if annotation in (Any, object, inspect.Signature.empty):
570 normalized[name] = value
571 else:
572 normalized[name] = normalize_value(
573 value,
574 annotation,
575 path=f"{path}.{name}",
576 )
578 for name, parameter in signature.parameters.items():
579 if (
580 name in normalized
581 or parameter.default is inspect.Signature.empty
582 or parameter.kind is inspect.Parameter.POSITIONAL_ONLY
583 ):
584 continue
585 normalized[name] = parameter.default
587 return normalized
590def _resolve_target(node: object, *, path: str) -> Target[Any]:
591 """Resolve a YAML mapping into a ``Target`` without invoking its callable."""
592 if not isinstance(node, Mapping):
593 raise ConfigResolutionError(path, "target section must be a YAML mapping")
594 if "_target_" not in node:
595 raise ConfigResolutionError(path, "target section is missing required _target_")
597 target_path = node["_target_"]
598 target = import_target(target_path, location=f"{path}._target_")
599 try:
600 signature = inspect.signature(target)
601 except (TypeError, ValueError) as exc:
602 raise ConfigResolutionError(path, f"target signature is unavailable: {exc}") from exc
604 for parameter in signature.parameters.values():
605 if (
606 parameter.kind is inspect.Parameter.POSITIONAL_ONLY
607 and parameter.default is inspect.Signature.empty
608 ):
609 raise ConfigResolutionError(
610 path,
611 f"target parameter {parameter.name!r} must be callable by keyword",
612 )
614 raw_args = {key: value for key, value in node.items() if key != "_target_"}
615 normalized_args = _resolve_target_args(
616 raw_args,
617 signature,
618 _target_hints(target, path=path),
619 path=path,
620 )
621 return Target(
622 target,
623 target_path=target_path,
624 **normalized_args,
625 )
628def _resolve_dataloader_config(node: object, *, path: str) -> DataLoaderConfig:
629 """Resolve a DataLoader target with nested collator and batch runtime."""
630 if not isinstance(node, Mapping):
631 raise ConfigResolutionError(path, "DataLoader configuration must be a YAML mapping")
633 target_node = dict(node)
634 if "dataloader_type" in target_node:
635 raise ConfigResolutionError(f"{path}.dataloader_type", "renamed to sampler_type")
636 if "batch_adapter" in target_node:
637 raise ConfigResolutionError(
638 f"{path}.batch_adapter",
639 "removed by the Omni data lifecycle; use dataset.data_transform for "
640 "sample/batch encoding and dataloader.get_batch.runtime_input_adapter "
641 "for model-owned forward metadata",
642 )
643 collate_node = target_node.pop("collate_fn", None)
644 get_batch_node = target_node.pop("get_batch", None)
645 sampler_type = normalize_value(
646 target_node.pop("sampler_type", "single"),
647 Literal["single", "cyclic"],
648 path=f"{path}.sampler_type",
649 )
650 data_rearrange_map = target_node.pop("data_rearrange_map", None)
651 data_sharding = normalize_value(
652 target_node.pop("data_sharding", False),
653 bool,
654 path=f"{path}.data_sharding",
655 )
656 use_background_prefetcher = normalize_value(
657 target_node.pop("use_background_prefetcher", False),
658 bool,
659 path=f"{path}.use_background_prefetcher",
660 )
661 target = _resolve_target(target_node, path=path)
662 collate_fn = (
663 None
664 if collate_node is None
665 else _resolve_target(collate_node, path=f"{path}.collate_fn")
666 )
667 get_batch = (
668 None
669 if get_batch_node is None
670 else _resolve_target(get_batch_node, path=f"{path}.get_batch")
671 )
672 return DataLoaderConfig(
673 target=target,
674 collate_fn=collate_fn,
675 get_batch=get_batch,
676 sampler_type=sampler_type,
677 data_rearrange_map=data_rearrange_map,
678 data_sharding=data_sharding,
679 use_background_prefetcher=use_background_prefetcher,
680 )
683def _resolve_dataset_config(node: object, *, path: str) -> DatasetConfig:
684 """Resolve a ``DatasetConfig`` with its model assets and sample transform."""
685 if not isinstance(node, Mapping):
686 raise ConfigResolutionError(path, "Dataset configuration must be a YAML mapping")
688 target_node = dict(node)
689 model_assets_node = target_node.pop("model_assets", {})
690 data_transform_node = target_node.pop("data_transform", None)
691 target = _resolve_target(target_node, path=path)
692 if isinstance(model_assets_node, Mapping) and "_target_" in model_assets_node:
693 model_assets = _resolve_target(
694 model_assets_node,
695 path=f"{path}.model_assets",
696 )
697 else:
698 model_assets = resolve_component(
699 model_assets_node,
700 annotation=ModelAssetsConfig,
701 path=f"{path}.model_assets",
702 )
703 data_transform = (
704 None
705 if data_transform_node is None
706 else _resolve_target(
707 data_transform_node,
708 path=f"{path}.data_transform",
709 )
710 )
711 dataset_config = DatasetConfig(
712 target=target,
713 model_assets=model_assets,
714 data_transform=data_transform,
715 )
716 return dataset_config
719def _resolve_optimizer_config(node: object, *, path: str) -> OptimizerConfig:
720 """Resolve an optimizer target with its precision and swap policies."""
721 if not isinstance(node, Mapping):
722 raise ConfigResolutionError(path, "Optimizer configuration must be a YAML mapping")
724 target_node = dict(node)
725 fp32_main_params = normalize_value(
726 target_node.pop("fp32_main_params", False),
727 bool,
728 path=f"{path}.fp32_main_params",
729 )
730 swap = normalize_value(
731 target_node.pop("swap", {}),
732 OptimizerSwapConfig,
733 path=f"{path}.swap",
734 )
735 return OptimizerConfig(
736 target=_resolve_target(target_node, path=path),
737 fp32_main_params=fp32_main_params,
738 swap=swap,
739 )
742def resolve_component(node: object, *, annotation: object, path: str) -> object:
743 """Resolve a YAML value according to its configuration annotation.
745 Args:
746 node: YAML value to resolve.
747 annotation: Type annotation defining the configuration component.
748 path: Configuration path used in error messages.
750 Returns:
751 The resolved value, configuration object, or unbuilt ``Target``.
753 Raises:
754 ConfigResolutionError: The value or target declaration is invalid for
755 the annotation.
756 """
757 if node is None:
758 return _require_none_allowed(annotation, path=path)
759 if _is_union(annotation):
760 return _resolve_union(node, annotation, path=path)
762 origin = get_origin(annotation)
763 if origin is Target or annotation is Target:
764 return _resolve_target(node, path=path)
765 if annotation is DatasetConfig:
766 return _resolve_dataset_config(node, path=path)
767 if annotation is DataLoaderConfig:
768 return _resolve_dataloader_config(node, path=path)
769 if annotation is OptimizerConfig:
770 return _resolve_optimizer_config(node, path=path)
771 if isinstance(node, Mapping) and "_target_" in node:
772 return _resolve_target(node, path=path)
773 if isinstance(annotation, type) and dataclasses.is_dataclass(annotation):
774 return _resolve_dataclass(node, annotation, path=path)
775 return normalize_value(node, annotation, path=path)
778def resolve_config(raw: object) -> TrainerConfig:
779 """Resolve a YAML root mapping into a ``TrainerConfig``.
781 Args:
782 raw: Parsed YAML root containing training configuration fields.
784 Returns:
785 The resolved configuration with unbuilt ``Target`` objects.
787 Raises:
788 ConfigResolutionError: The root is not a mapping, required fields are
789 missing, or a field or target declaration is invalid.
790 """
791 if not isinstance(raw, Mapping):
792 raise ConfigResolutionError("$", "YAML root must be a mapping")
794 root_fields = {field.name: field for field in fields(TrainerConfig)}
795 unknown = sorted(set(raw) - set(root_fields))
796 if unknown:
797 raise ConfigResolutionError("$", f"unknown configuration fields: {unknown}")
799 required = [
800 field.name
801 for field in root_fields.values()
802 if field.default is MISSING and field.default_factory is MISSING
803 ]
804 missing = [name for name in required if name not in raw]
805 if missing:
806 raise ConfigResolutionError("$", f"missing required configuration fields: {missing}")
808 root_hints = get_type_hints(TrainerConfig)
809 resolved = {
810 name: resolve_component(
811 node,
812 annotation=root_hints[name],
813 path=f"$.{name}",
814 )
815 for name, node in raw.items()
816 }
817 try:
818 return TrainerConfig(**resolved)
819 except TypeError as exc:
820 raise ConfigResolutionError("$", f"could not construct TrainerConfig: {exc}") from exc