Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / codegen / cli.py: 0%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# Licensed under the Apache License, Version 2.0 

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

4"""Codegen CLI: thin command-line wrapper over manager / check / artifact. 

5 

6``python -m hyper_parallel.codegen`` dispatches to :func:`main`. No business 

7logic — argument parsing, YAML discovery, and exit-code mapping only. 

8 

9Exit codes are 0 for success or a skipped pin check, 1 for a check/clean 

10failure, and 2 for a usage error. 

11""" 

12from __future__ import annotations 

13 

14import argparse 

15import logging 

16import os 

17import sys 

18from typing import Iterator, Optional 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23def main(argv: Optional[list[str]] = None) -> int: 

24 """argparse entry; returns the process exit code.""" 

25 parser = _build_parser() 

26 args, config_overrides = parser.parse_known_args(argv) 

27 setattr(args, "_config_overrides", config_overrides) 

28 if not getattr(args, "command", None): 

29 parser.print_usage(sys.stderr) 

30 return 2 

31 _configure_logging(getattr(args, "verbose", False)) 

32 try: 

33 return args.func(args) 

34 except Exception as exc: # noqa: BLE001 - a CLI reports, does not traceback 

35 logger.error("codegen: %s", exc) 

36 return 1 

37 

38 

39# --------------------------------------------------------------------------- 

40# commands 

41# --------------------------------------------------------------------------- 

42 

43def cmd_generate(args: argparse.Namespace) -> int: 

44 """``generate --config <yaml>`` — build/reuse the bundle for one YAML.""" 

45 config = _load_config(args.config, _cli_overrides(args)) 

46 from hyper_parallel.codegen.manager import ensure_codegen_artifact 

47 

48 if not getattr(config, "codegen", False): 

49 logger.info("codegen: generate skipped — codegen is false") 

50 return 0 

51 layout = ensure_codegen_artifact(config, args.config) 

52 if layout is None: 

53 return 0 

54 from hyper_parallel.codegen.meta import load_codegen_meta 

55 

56 meta = load_codegen_meta(layout.meta_path) 

57 signature = meta.signature if meta is not None else "?" 

58 logger.info("codegen: artifact %s (signature %s)", layout.artifact_dir, signature) 

59 return 0 

60 

61 

62def cmd_check(args: argparse.Namespace) -> int: 

63 """``check --config <yaml>`` — full preflight for one YAML.""" 

64 config = _load_config(args.config, _cli_overrides(args)) 

65 if not getattr(config, "codegen", False): 

66 logger.info("codegen: SKIP (codegen disabled)") 

67 return 0 

68 from hyper_parallel.codegen.manager import preflight_integrity_check 

69 

70 preflight_integrity_check(config, args.config) 

71 logger.info("codegen: PASS %s", args.config) 

72 return 0 

73 

74 

75def cmd_check_all(args: argparse.Namespace) -> int: 

76 """Run preflight for every discovered YAML and check cached artifact drift.""" 

77 if _cli_overrides(args): 

78 raise ValueError("check --all does not accept per-config CLI overrides") 

79 failed = 0 

80 for yaml_path in discover_yaml_files(args.root): 

81 try: 

82 config = _load_config(yaml_path) 

83 if not getattr(config, "codegen", False): 

84 logger.info("codegen: SKIP (codegen disabled) %s", yaml_path) 

85 continue 

86 from hyper_parallel.codegen.manager import preflight_integrity_check 

87 

88 preflight_integrity_check(config, yaml_path) 

89 logger.info("codegen: PASS %s", yaml_path) 

90 # Drift can only be checked when an artifact already exists. 

91 from hyper_parallel.codegen.check.drift import verify_drift_intact 

92 from hyper_parallel.codegen.meta import load_codegen_meta 

93 from hyper_parallel.codegen.artifact import resolve_artifact_layout 

94 

95 layout = resolve_artifact_layout(yaml_path) 

96 meta = load_codegen_meta(layout.meta_path) 

97 if meta is not None: 

98 drifts = verify_drift_intact(meta, yaml_path) 

99 if drifts: 

100 failed = 1 

101 logger.error("codegen: drift in %s: %s", yaml_path, "; ".join(drifts)) 

102 else: 

103 logger.info("codegen: drift intact %s", yaml_path) 

104 except Exception as exc: # noqa: BLE001 - aggregate, keep scanning 

105 failed = 1 

106 logger.error("codegen: FAIL %s: %s", yaml_path, exc) 

107 return failed 

108 

109 

110def cmd_check_pin(args: argparse.Namespace) -> int: 

111 """Validate a Transformers version constraint without regenerating artifacts.""" 

112 from hyper_parallel.codegen.check.pin import verify_transformers_pin 

113 

114 status, installed, spec = verify_transformers_pin(args.transformers_pin) 

115 logger.info("codegen: transformers pin %s — installed %s (%s)", spec, installed, status) 

116 return 0 # OK and SKIP both exit 0 

117 

118 

119# --------------------------------------------------------------------------- 

120# discovery & helpers 

121# --------------------------------------------------------------------------- 

122 

123_SKIP_DIRS = {"generated", ".git", ".venv", "venv", "__pycache__", ".tmp"} 

124 

125 

126def discover_yaml_files(root: str) -> Iterator[str]: 

127 """Walk ``root`` for ``*.yaml`` / ``*.yml``, skipping generated/venv/git dirs.""" 

128 if not os.path.isdir(root): 

129 return 

130 for dirpath, dirnames, filenames in os.walk(root): 

131 dirnames[:] = [d for d in dirnames if d not in _SKIP_DIRS] 

132 for name in sorted(filenames): 

133 if name.endswith((".yaml", ".yml")): 

134 yield os.path.join(dirpath, name) 

135 

136 

137def _load_config(yaml_path: str, overrides: Optional[list[str]] = None): 

138 """Load a ``TrainerConfig`` from a YAML path.""" 

139 from hyper_parallel.trainer.config.parser import load_training_config 

140 

141 return load_training_config(yaml_path, overrides or []) 

142 

143 

144def _model_name(config: object) -> Optional[str]: 

145 from hyper_parallel.codegen.manager import ( # pylint: disable=import-outside-toplevel 

146 _model_name as _manager_model_name, 

147 _resolve_hf_config_for_layout, 

148 ) 

149 

150 return _manager_model_name(config, _resolve_hf_config_for_layout(config)) 

151 

152 

153def _model_name_for(yaml_path: str, overrides: Optional[list[str]] = None) -> Optional[str]: 

154 """Model name from a YAML (used by ``clean`` before config load).""" 

155 try: 

156 return _model_name(_load_config(yaml_path, overrides)) 

157 except Exception: # noqa: BLE001 - clean falls back to default naming 

158 return None 

159 

160 

161def _cli_overrides(args: argparse.Namespace) -> list[str]: 

162 """Typed config overrides captured after argparse consumes codegen flags.""" 

163 return list(getattr(args, "_config_overrides", None) or []) 

164 

165 

166def _configure_logging(verbose: bool) -> None: 

167 logging.basicConfig( 

168 level=logging.INFO if verbose else logging.WARNING, 

169 format="%(levelname)s:%(name)s:%(message)s", 

170 ) 

171 

172 

173def _build_parser() -> argparse.ArgumentParser: 

174 """Build the argparse parser for the codegen CLI.""" 

175 parser = argparse.ArgumentParser( 

176 prog="hyper_parallel.codegen", 

177 description="Codegen: generate/check/clean codegen artifacts.", 

178 ) 

179 sub = parser.add_subparsers(dest="command") 

180 parser.add_argument("--verbose", action="store_true", help="verbose logging") 

181 

182 gen = sub.add_parser("generate", help="generate or reuse the bundle for one YAML") 

183 gen.add_argument("--config", required=True, help="path to the training YAML") 

184 gen.set_defaults(func=cmd_generate) 

185 

186 check = sub.add_parser("check", help="preflight integrity check") 

187 check.add_argument("--config", help="path to the training YAML") 

188 check.add_argument("--all", action="store_true", help="check every discovered YAML") 

189 check.add_argument("--root", default=".", help="root to search when --all") 

190 check.add_argument("--transformers-pin", help="Transformers version constraint (e.g. ==5.15.0)") 

191 check.set_defaults(func=_cmd_check_dispatch) 

192 

193 return parser 

194 

195 

196def _cmd_check_dispatch(args: argparse.Namespace) -> int: 

197 """Route the ``check`` subcommand's mutually-exclusive modes.""" 

198 modes = [bool(args.all), bool(args.transformers_pin)] 

199 if sum(modes) > 1: 

200 raise ValueError("check accepts only one of --all / --transformers-pin") 

201 if args.all: 

202 return cmd_check_all(args) 

203 if args.transformers_pin: 

204 return cmd_check_pin(args) 

205 if not args.config: 

206 raise ValueError("check needs --config, --all, or --transformers-pin") 

207 return cmd_check(args) 

208 

209 

210__all__ = [ 

211 "cmd_check", 

212 "cmd_check_all", 

213 "cmd_check_pin", 

214 "cmd_generate", 

215 "discover_yaml_files", 

216 "main", 

217]