Coverage for apio/managers/scons_manager.py: 80%

160 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-06-24 03:51 +0000

1"""A manager class for to dispatch the Apio SCONS targets.""" 

2 

3# -*- coding: utf-8 -*- 

4# -- This file is part of the Apio project 

5# -- (C) 2016-2019 FPGAwars 

6# -- Author Jesús Arroyo 

7# -- License GPLv2 

8 

9import traceback 

10import os 

11import sys 

12import time 

13import shutil 

14from functools import wraps 

15from datetime import datetime 

16from typing import Optional 

17from google.protobuf import text_format 

18from apio.common import apio_console 

19from apio.common.apio_console import cout, cerror, cstyle, cunstyle 

20from apio.common.apio_styles import SUCCESS, ERROR, EMPH3, INFO 

21from apio.utils import util 

22from apio.apio_context import ApioContext 

23from apio.managers.scons_filter import SconsFilter 

24from apio.common.proto.apio_pb2 import ( 

25 FORCE_PIPE, 

26 FORCE_TERMINAL, 

27 Verbosity, 

28 Environment, 

29 SconsParams, 

30 TargetParams, 

31 FpgaInfo, 

32 ApioEnvParams, 

33 Ice40FpgaParams, 

34 Ecp5FpgaParams, 

35 GowinFpgaParams, 

36 XilinxFpgaParams, 

37 ApioArch, 

38 GraphParams, 

39 LintParams, 

40 SimParams, 

41 ApioTestParams, 

42 UploadParams, 

43) 

44 

45# from apio.common import rich_lib_windows 

46 

47 

48# W0703: Catching too general exception Exception (broad-except) 

49# pylint: disable=W0703 

50# 

51# -- Based on 

52# -- https://stackoverflow.com/questions/5929107/decorators-with-parameters 

53def on_exception(*, exit_code: int): 

54 """Decorator for functions that return int exit code. If the function 

55 throws an exception, the error message is printed, and the caller see the 

56 returned value exit_code instead of the exception. 

57 """ 

58 

59 def decorator(function): 

60 @wraps(function) 

61 def wrapper(*args, **kwargs): 

62 try: 

63 return function(*args, **kwargs) 

64 except Exception as exc: 

65 if util.is_debug(1): 

66 traceback.print_tb(exc.__traceback__) 

67 

68 if str(exc): 

69 cerror(str(exc)) 

70 

71 if not util.is_debug(1): 

72 cout( 

73 "Setting env var APIO_DEBUG=1 may provide " 

74 "additional diagnostic information.", 

75 style=INFO, 

76 ) 

77 return exit_code 

78 

79 return wrapper 

80 

81 return decorator 

82 

83 

84class SConsManager: 

85 """Class for managing the scons tools""" 

86 

87 def __init__(self, apio_ctx: ApioContext): 

88 """Initialization.""" 

89 # -- Cache the apio context. 

90 self.apio_ctx = apio_ctx 

91 

92 # -- Change to the project's folder. 

93 os.chdir(apio_ctx.project_dir) 

94 

95 @on_exception(exit_code=1) 

96 def graph( 

97 self, graph_params: GraphParams, verbosity: Verbosity 

98 ) -> Optional[int]: 

99 """Runs a scons subprocess with the 'graph' target. Returns process 

100 exit code, 0 if ok.""" 

101 

102 # -- Construct scons params with graph command info. 

103 scons_params = self.construct_scons_params( 

104 target_params=TargetParams(graph=graph_params), 

105 verbosity=verbosity, 

106 ) 

107 

108 # -- Run the scons process. 

109 return self._run_scons_subprocess("graph", scons_params=scons_params) 

110 

111 @on_exception(exit_code=1) 

112 def lint(self, lint_params: LintParams) -> Optional[int]: 

113 """Runs a scons subprocess with the 'lint' target. Returns process 

114 exit code, 0 if ok.""" 

115 

116 # -- Construct scons params with graph command info. 

117 scons_params = self.construct_scons_params( 

118 target_params=TargetParams(lint=lint_params) 

119 ) 

120 

121 # -- Run the scons process. 

122 return self._run_scons_subprocess("lint", scons_params=scons_params) 

123 

124 @on_exception(exit_code=1) 

125 def sim(self, sim_params: SimParams) -> Optional[int]: 

126 """Runs a scons subprocess with the 'sim' target. Returns process 

127 exit code, 0 if ok.""" 

128 

129 # -- Construct scons params with graph command info. 

130 scons_params = self.construct_scons_params( 

131 target_params=TargetParams(sim=sim_params) 

132 ) 

133 

134 # -- Run the scons process. 

135 return self._run_scons_subprocess("sim", scons_params=scons_params) 

136 

137 @on_exception(exit_code=1) 

138 def test(self, test_params: ApioTestParams) -> Optional[int]: 

139 """Runs a scons subprocess with the 'test' target. Returns process 

140 exit code, 0 if ok.""" 

141 

142 # -- Construct scons params with graph command info. 

143 scons_params = self.construct_scons_params( 

144 target_params=TargetParams(test=test_params) 

145 ) 

146 

147 # -- Run the scons process. 

148 return self._run_scons_subprocess("test", scons_params=scons_params) 

149 

150 @on_exception(exit_code=1) 

151 def build(self, nextpnr_gui: bool, verbosity: Verbosity) -> Optional[int]: 

152 """Runs a scons subprocess with the 'build' target. Returns process 

153 exit code, 0 if ok.""" 

154 

155 # -- Construct the scons params object. 

156 scons_params = self.construct_scons_params( 

157 nextpnr_gui=nextpnr_gui, 

158 verbosity=verbosity, 

159 ) 

160 

161 # -- Run the scons process. 

162 return self._run_scons_subprocess("build", scons_params=scons_params) 

163 

164 @on_exception(exit_code=1) 

165 def report(self, verbosity: Verbosity) -> Optional[int]: 

166 """Runs a scons subprocess with the 'report' target. Returns process 

167 exit code, 0 if ok.""" 

168 

169 # -- Construct the scons params object. 

170 scons_params = self.construct_scons_params(verbosity=verbosity) 

171 

172 # -- Run the scons process. 

173 return self._run_scons_subprocess("report", scons_params=scons_params) 

174 

175 @on_exception(exit_code=1) 

176 def upload(self, upload_params: UploadParams) -> Optional[int]: 

177 """Runs a scons subprocess with the 'time' target. Returns process 

178 exit code, 0 if ok. 

179 """ 

180 

181 # -- Construct the scons params. 

182 scons_params = self.construct_scons_params( 

183 target_params=TargetParams(upload=upload_params) 

184 ) 

185 

186 # -- Execute Scons for uploading! 

187 exit_code = self._run_scons_subprocess( 

188 "upload", scons_params=scons_params 

189 ) 

190 

191 return exit_code 

192 

193 def construct_scons_params( 

194 self, 

195 *, 

196 target_params: TargetParams | None = None, 

197 nextpnr_gui: bool = False, 

198 verbosity: Verbosity | None = None, 

199 ) -> SconsParams: 

200 """Populate and return the SconsParam proto to pass to the scons 

201 process.""" 

202 

203 # -- Create a shortcut. 

204 apio_ctx = self.apio_ctx 

205 

206 # -- Create an empty proto object that will be populated. 

207 result = SconsParams() 

208 

209 # -- Set the nextpnr_gui if True. 

210 if nextpnr_gui: 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true

211 result.nextpnr_gui = True 

212 

213 # -- Populate the timestamp. We use to to make sure scons reads the 

214 # -- correct version of the scons.params file. 

215 ts = datetime.now() 

216 result.timestamp = ts.strftime("%d%H%M%S%f")[:-3] 

217 

218 # -- Get the project data. All commands that invoke scons are expected 

219 # -- to be in a project context. 

220 assert apio_ctx.has_project, "Scons encountered a missing project." 

221 project = apio_ctx.project 

222 

223 # -- Get the project resources. 

224 pr = apio_ctx.project_resources 

225 fpga_info = pr.fpga_info 

226 

227 # -- Populate the common values of FpgaInfo. 

228 result.fpga_info.MergeFrom( 

229 FpgaInfo( 

230 fpga_id=pr.fpga_id, 

231 part_num=fpga_info["part-num"], 

232 size=fpga_info["size"], 

233 ) 

234 ) 

235 

236 # - Populate the architecture specific values of result.fpga_info. 

237 fpga_arch = fpga_info["arch"] 

238 match fpga_arch: 

239 case "ice40": 

240 params = fpga_info["ice40-params"] 

241 result.arch = ApioArch.ICE40 

242 result.fpga_info.ice40_params.MergeFrom( 

243 Ice40FpgaParams( 

244 type=params["type"], 

245 package=params["package"], 

246 ) 

247 ) 

248 case "ecp5": 

249 params = fpga_info["ecp5-params"] 

250 result.arch = ApioArch.ECP5 

251 result.fpga_info.ecp5_params.MergeFrom( 

252 Ecp5FpgaParams( 

253 type=params["type"], 

254 package=params["package"], 

255 speed=params["speed"], 

256 ) 

257 ) 

258 case "gowin": 

259 params = fpga_info["gowin-params"] 

260 result.arch = ApioArch.GOWIN 

261 result.fpga_info.gowin_params.MergeFrom( 

262 GowinFpgaParams( 

263 yosys_family=params["yosys-family"], 

264 nextpnr_family=params["nextpnr-family"], 

265 packer_device=params["packer-device"], 

266 ) 

267 ) 

268 case "xilinx": 268 ↛ 279line 268 didn't jump to line 279 because the pattern on line 268 always matched

269 params = fpga_info["xilinx-params"] 

270 result.arch = ApioArch.XILINX 

271 result.fpga_info.xilinx_params.MergeFrom( 

272 XilinxFpgaParams( 

273 family=params["family"], 

274 yosys_arch=params["yosys-arch"], 

275 package=params["package"], 

276 speed=params["speed"], 

277 ) 

278 ) 

279 case _: 

280 cerror(f"Unexpected fpga_arch value {fpga_arch}") 

281 sys.exit(1) 

282 

283 # -- We are done populating The FpgaInfo params.. 

284 assert result.fpga_info.IsInitialized(), result 

285 

286 # -- Populate the optional Verbosity params. 

287 if verbosity: 

288 result.verbosity.MergeFrom(verbosity) 

289 assert result.verbosity.IsInitialized(), result 

290 

291 # -- Populate the Environment params. 

292 assert apio_ctx.platform_id, "Missing platform_id in apio context" 

293 oss_set_vars = apio_ctx.all_packages["oss-cad-suite"]["env"][ 

294 "set-vars" 

295 ] 

296 assert "YOSYS_LIB" in oss_set_vars, oss_set_vars 

297 assert "TRELLIS" in oss_set_vars, oss_set_vars 

298 

299 # -- openxc7 is available only on the platforms listed in its 

300 # -- "restricted-to-platforms" field (see packages.jsonc). On a 

301 # -- platform where it isn't available, all_packages["openxc7"] is 

302 # -- absent, so fall back to empty paths. 

303 try: 

304 openxc7_set_vars = apio_ctx.all_packages["openxc7"]["env"][ 

305 "set-vars" 

306 ] 

307 

308 # -- Platform not supported. Ignore it! 

309 except KeyError: 

310 openxc7_set_vars = {"PRJXRAY_DB_DIR": "", "CHIPDB_DIR": ""} 

311 

312 result.environment.MergeFrom( 

313 Environment( 

314 platform_id=apio_ctx.platform_id, 

315 is_windows=apio_ctx.is_windows, 

316 terminal_mode=( 

317 FORCE_TERMINAL 

318 if apio_console.is_terminal() 

319 else FORCE_PIPE 

320 ), 

321 theme_name=apio_console.current_theme_name(), 

322 debug_level=util.debug_level(), 

323 yosys_path=oss_set_vars["YOSYS_LIB"], 

324 trellis_path=oss_set_vars["TRELLIS"], 

325 scons_shell_id=apio_ctx.scons_shell_id, 

326 prjxray_db_path=openxc7_set_vars["PRJXRAY_DB_DIR"], 

327 chipdb_path=openxc7_set_vars["CHIPDB_DIR"], 

328 apio_home_path=str(apio_ctx.apio_home_dir), 

329 env_build_path=str(apio_ctx.env_build_path), 

330 ) 

331 ) 

332 assert result.environment.IsInitialized(), result 

333 

334 # -- Populate the Project params. 

335 result.apio_env_params.MergeFrom( 

336 ApioEnvParams( 

337 env_name=apio_ctx.project.env_name, 

338 board_id=pr.board_id, 

339 top_module=project.get_str_option("top-module"), 

340 defines=apio_ctx.project.get_list_option( 

341 "defines", default=[] 

342 ), 

343 yosys_extra_options=apio_ctx.project.get_list_option( 

344 "yosys-extra-options", None 

345 ), 

346 nextpnr_extra_options=apio_ctx.project.get_list_option( 

347 "nextpnr-extra-options", None 

348 ), 

349 gtkwave_extra_options=apio_ctx.project.get_list_option( 

350 "gtkwave-extra-options", None 

351 ), 

352 verilator_extra_options=apio_ctx.project.get_list_option( 

353 "verilator-extra-options", None 

354 ), 

355 constraint_file=apio_ctx.project.get_str_option( 

356 "constraint-file", None 

357 ), 

358 ) 

359 ) 

360 assert result.apio_env_params.IsInitialized(), result 

361 

362 # -- Populate the optional command specific params. 

363 if target_params: 

364 result.target.MergeFrom(target_params) 

365 assert result.target.IsInitialized(), result 

366 

367 # -- All done. 

368 assert result.IsInitialized(), result 

369 return result 

370 

371 def _run_scons_subprocess( 

372 self, scons_command: str, *, scons_params: SconsParams 

373 ) -> Optional[int]: 

374 """Invoke an scons subprocess.""" 

375 

376 # pylint: disable=too-many-locals 

377 

378 # -- Create a shortcut. 

379 apio_ctx = self.apio_ctx 

380 

381 # -- Pass to the scons process the name of the sconstruct file it 

382 # -- should use. 

383 scons_dir = util.get_path_in_apio_package("scons") 

384 scons_file_path = scons_dir / "SConstruct" 

385 variables = ["-f", f"{scons_file_path}"] 

386 

387 # -- Pass the path to the proto params file. The path is relative 

388 # -- to the project root. 

389 params_file_path = apio_ctx.env_build_path / "scons.params" 

390 variables += [f"params={str(params_file_path)}"] 

391 

392 # -- Pass to the scons process the timestamp of the scons params we 

393 # -- pass via a file. This is for verification purposes only. 

394 variables += [f"timestamp={scons_params.timestamp}"] 

395 

396 # -- We set the env variables also for a command such as 'clean' 

397 # -- which doesn't use the packages, to satisfy the required env 

398 # -- variables of the scons arg parser. 

399 apio_ctx.set_env_for_packages() 

400 

401 if util.is_debug(1): 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true

402 cout("\nSCONS CALL:", style=EMPH3) 

403 cout(f"* command: {scons_command}") 

404 cout(f"* variables: {variables}") 

405 cout(f"* scons params: \n{scons_params}") 

406 cout() 

407 

408 # -- Get the terminal width (typically 80) 

409 terminal_width, _ = shutil.get_terminal_size() 

410 

411 # -- Read the time (for measuring how long does it take 

412 # -- to execute the apio command) 

413 start_time = time.time() 

414 

415 # -- Subtracting 1 to avoid line overflow on windows, Observed with 

416 # -- Windows 10 and cmd.exe shell. 

417 if apio_ctx.is_windows: 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true

418 terminal_width -= 1 

419 

420 # -- Print a horizontal line 

421 cout("-" * terminal_width) 

422 

423 # -- Create the scons debug options. See details at 

424 # -- https://scons.org/doc/2.4.1/HTML/scons-man.html 

425 debug_options = ( 

426 ["--debug=explain,prepare,stacktrace", "--tree=all"] 

427 if util.is_debug(1) 

428 else [] 

429 ) 

430 

431 # -- Construct the scons command line. 

432 # -- 

433 # -- sys.executable is resolved to the full path of the python 

434 # -- interpreter or to apio if running from a pyinstall setup. 

435 # -- See https://github.com/orgs/pyinstaller/discussions/9023 for more 

436 # -- information. 

437 # -- 

438 # -- We use exec -m SCons instead of scones also for non pyinstaller 

439 # -- deployment in case the scons binary is not on the PATH. 

440 cmd = ( 

441 [sys.executable, "-m", "apio", "--scons"] 

442 + ["-Q", scons_command] 

443 + debug_options 

444 + variables 

445 ) 

446 

447 # -- An output filter that manipulates the scons stdout/err lines as 

448 # -- needed and write them to stdout. 

449 scons_filter = SconsFilter( 

450 colors_enabled=apio_console.is_colors_enabled() 

451 ) 

452 

453 # -- Write the scons parameters to a temp file in the build 

454 # -- directory. It will be cleaned up as part of 'apio cleanup'. 

455 # -- At this point, the project is the current directory, even if 

456 # -- the command used the --project-dir option. 

457 os.makedirs(apio_ctx.env_build_path, exist_ok=True) 

458 with open(params_file_path, "w", encoding="utf8") as f: 

459 f.write(text_format.MessageToString(scons_params)) 

460 

461 if util.is_debug(1): 461 ↛ 462line 461 didn't jump to line 462 because the condition on line 461 was never true

462 cout(f"\nFull scons command: {cmd}\n\n") 

463 

464 # -- Execute the scons builder! 

465 result = util.exec_command( 

466 cmd, 

467 stdout=util.AsyncPipe(scons_filter.on_stdout_line), 

468 stderr=util.AsyncPipe(scons_filter.on_stderr_line), 

469 ) 

470 

471 # -- Is there an error? True/False 

472 is_error = result.exit_code != 0 

473 

474 # -- Calculate the time it took to execute the command 

475 duration = time.time() - start_time 

476 

477 # -- Determine status message 

478 if is_error: 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true

479 styled_status = cstyle("ERROR", style=ERROR) 

480 else: 

481 styled_status = cstyle("SUCCESS", style=SUCCESS) 

482 

483 # -- Determine the summary text 

484 summary = f"Took {duration:.2f} seconds" 

485 

486 # -- Construct the entire message. 

487 styled_msg = f" [{styled_status}] {summary} " 

488 msg_len = len(cunstyle(styled_msg)) 

489 

490 # -- Determine the lengths of the paddings before and after 

491 # -- the message. Should be correct for odd and even terminal 

492 # -- widths. 

493 pad1_len = (terminal_width - msg_len) // 2 

494 pad2_len = terminal_width - pad1_len - msg_len 

495 

496 # -- Print the entire line. 

497 cout(f"{'=' * pad1_len}{styled_msg}{'=' * pad2_len}") 

498 

499 # -- Return the exit code 

500 return result.exit_code