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

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 XILINX, 

28 Verbosity, 

29 Environment, 

30 SconsParams, 

31 TargetParams, 

32 FpgaInfo, 

33 ApioEnvParams, 

34 Ice40FpgaParams, 

35 Ecp5FpgaParams, 

36 GowinFpgaParams, 

37 XilinxFpgaParams, 

38 ApioArch, 

39 GraphParams, 

40 ReportParams, 

41 LintParams, 

42 SimParams, 

43 ApioTestParams, 

44 UploadParams, 

45) 

46 

47# from apio.common import rich_lib_windows 

48 

49 

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

51# pylint: disable=W0703 

52# 

53# -- Based on 

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

55def on_exception(*, exit_code: int): 

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

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

58 returned value exit_code instead of the exception. 

59 """ 

60 

61 def decorator(function): 

62 @wraps(function) 

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

64 try: 

65 return function(*args, **kwargs) 

66 except Exception as exc: 

67 if util.is_debug(1): 

68 traceback.print_tb(exc.__traceback__) 

69 

70 if str(exc): 

71 cerror(str(exc)) 

72 

73 if not util.is_debug(1): 

74 cout( 

75 "Setting env var APIO_DEBUG=1 may provide " 

76 "additional diagnostic information.", 

77 style=INFO, 

78 ) 

79 return exit_code 

80 

81 return wrapper 

82 

83 return decorator 

84 

85 

86class SConsManager: 

87 """Class for managing the scons tools""" 

88 

89 def __init__(self, apio_ctx: ApioContext): 

90 """Initialization.""" 

91 # -- Cache the apio context. 

92 self.apio_ctx = apio_ctx 

93 

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

95 os.chdir(apio_ctx.project_dir) 

96 

97 def _fetch_support_files_on_demand( 

98 self, scons_target, scons_params: SconsParams 

99 ): 

100 """Called before invoking scons to optionally fetch missing support 

101 files on the file.""" 

102 

103 # -- Only these targets may require support files. For all the rest 

104 # -- we don't need to do anything. 

105 if scons_target not in ["build", "report", "upload"]: 

106 return 

107 

108 # -- Only xilinx architecture may requires support files. For all the 

109 # -- rest we don't need to do anything. 

110 if scons_params.arch != XILINX: 

111 return 

112 

113 # -- TODO: Implement the xilinx on-demand chipdb file download. 

114 

115 @on_exception(exit_code=1) 

116 def graph( 

117 self, graph_params: GraphParams, verbosity: Verbosity 

118 ) -> Optional[int]: 

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

120 exit code, 0 if ok.""" 

121 

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

123 scons_params = self.construct_scons_params( 

124 target_params=TargetParams(graph=graph_params), 

125 verbosity=verbosity, 

126 ) 

127 

128 # -- Run the scons process. 

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

130 

131 @on_exception(exit_code=1) 

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

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

134 exit code, 0 if ok.""" 

135 

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

137 scons_params = self.construct_scons_params( 

138 target_params=TargetParams(lint=lint_params) 

139 ) 

140 

141 # -- Run the scons process. 

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

143 

144 @on_exception(exit_code=1) 

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

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

147 exit code, 0 if ok.""" 

148 

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

150 scons_params = self.construct_scons_params( 

151 target_params=TargetParams(sim=sim_params) 

152 ) 

153 

154 # -- Run the scons process. 

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

156 

157 @on_exception(exit_code=1) 

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

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

160 exit code, 0 if ok.""" 

161 

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

163 scons_params = self.construct_scons_params( 

164 target_params=TargetParams(test=test_params) 

165 ) 

166 

167 # -- Run the scons process. 

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

169 

170 @on_exception(exit_code=1) 

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

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

173 exit code, 0 if ok.""" 

174 

175 # -- Construct the scons params object. 

176 scons_params = self.construct_scons_params( 

177 nextpnr_gui=nextpnr_gui, 

178 verbosity=verbosity, 

179 ) 

180 

181 # -- Run the scons process. 

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

183 

184 @on_exception(exit_code=1) 

185 def report( 

186 self, report_params: ReportParams, verbosity: Verbosity 

187 ) -> Optional[int]: 

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

189 exit code, 0 if ok.""" 

190 

191 # -- Construct the scons params object. 

192 scons_params = self.construct_scons_params( 

193 target_params=TargetParams(report=report_params), 

194 verbosity=verbosity, 

195 ) 

196 

197 # -- Run the scons process. 

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

199 

200 @on_exception(exit_code=1) 

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

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

203 exit code, 0 if ok. 

204 """ 

205 

206 # -- Construct the scons params. 

207 scons_params = self.construct_scons_params( 

208 target_params=TargetParams(upload=upload_params) 

209 ) 

210 

211 # -- Execute Scons for uploading! 

212 exit_code = self._run_scons_subprocess( 

213 "upload", scons_params=scons_params 

214 ) 

215 

216 return exit_code 

217 

218 def construct_scons_params( 

219 self, 

220 *, 

221 target_params: TargetParams | None = None, 

222 nextpnr_gui: bool = False, 

223 verbosity: Verbosity | None = None, 

224 ) -> SconsParams: 

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

226 process.""" 

227 

228 # -- Create a shortcut. 

229 apio_ctx = self.apio_ctx 

230 

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

232 result = SconsParams() 

233 

234 # -- Set the nextpnr_gui if True. 

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

236 result.nextpnr_gui = True 

237 

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

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

240 ts = datetime.now() 

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

242 

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

244 # -- to be in a project context. 

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

246 project = apio_ctx.project 

247 

248 # -- Get the project resources. 

249 pr = apio_ctx.project_resources 

250 fpga_info = pr.fpga_info 

251 

252 # -- Populate the common values of FpgaInfo. 

253 result.fpga_info.MergeFrom( 

254 FpgaInfo( 

255 fpga_id=pr.fpga_id, 

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

257 size=fpga_info["size"], 

258 ) 

259 ) 

260 

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

262 fpga_arch = fpga_info["arch"] 

263 match fpga_arch: 

264 case "ice40": 

265 params = fpga_info["ice40-params"] 

266 result.arch = ApioArch.ICE40 

267 result.fpga_info.ice40_params.MergeFrom( 

268 Ice40FpgaParams( 

269 type=params["type"], 

270 package=params["package"], 

271 ) 

272 ) 

273 case "ecp5": 

274 params = fpga_info["ecp5-params"] 

275 result.arch = ApioArch.ECP5 

276 result.fpga_info.ecp5_params.MergeFrom( 

277 Ecp5FpgaParams( 

278 type=params["type"], 

279 package=params["package"], 

280 speed=params["speed"], 

281 ) 

282 ) 

283 case "gowin": 

284 params = fpga_info["gowin-params"] 

285 result.arch = ApioArch.GOWIN 

286 result.fpga_info.gowin_params.MergeFrom( 

287 GowinFpgaParams( 

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

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

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

291 ) 

292 ) 

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

294 params = fpga_info["xilinx-params"] 

295 result.arch = ApioArch.XILINX 

296 result.fpga_info.xilinx_params.MergeFrom( 

297 XilinxFpgaParams( 

298 family=params["family"], 

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

300 package=params["package"], 

301 speed=params["speed"], 

302 ) 

303 ) 

304 case _: 

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

306 sys.exit(1) 

307 

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

309 assert result.fpga_info.IsInitialized(), result 

310 

311 # -- Populate the optional Verbosity params. 

312 if verbosity: 

313 result.verbosity.MergeFrom(verbosity) 

314 assert result.verbosity.IsInitialized(), result 

315 

316 # -- Populate the Environment params. 

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

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

319 "set-vars" 

320 ] 

321 assert "YOSYS_LIB" in oss_set_vars, oss_set_vars 

322 assert "TRELLIS" in oss_set_vars, oss_set_vars 

323 

324 openxc7_set_vars = apio_ctx.all_packages["openxc7"]["env"]["set-vars"] 

325 assert "PRJXRAY_DB_DIR" in openxc7_set_vars, open 

326 assert "CHIPDB_DIR" in openxc7_set_vars, open 

327 

328 result.environment.MergeFrom( 

329 Environment( 

330 platform_id=apio_ctx.platform_id, 

331 is_windows=apio_ctx.is_windows, 

332 terminal_mode=( 

333 FORCE_TERMINAL 

334 if apio_console.is_terminal() 

335 else FORCE_PIPE 

336 ), 

337 theme_name=apio_console.current_theme_name(), 

338 debug_level=util.debug_level(), 

339 yosys_path=oss_set_vars["YOSYS_LIB"], 

340 trellis_path=oss_set_vars["TRELLIS"], 

341 scons_shell_id=apio_ctx.scons_shell_id, 

342 xilinx_prjxray_db_path=openxc7_set_vars["PRJXRAY_DB_DIR"], 

343 xilinx_chipdb_path=openxc7_set_vars["CHIPDB_DIR"], 

344 ) 

345 ) 

346 assert result.environment.IsInitialized(), result 

347 

348 # -- Populate the Project params. 

349 result.apio_env_params.MergeFrom( 

350 ApioEnvParams( 

351 env_name=apio_ctx.project.env_name, 

352 board_id=pr.board_id, 

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

354 defines=apio_ctx.project.get_list_option( 

355 "defines", default=[] 

356 ), 

357 yosys_extra_options=apio_ctx.project.get_list_option( 

358 "yosys-extra-options", None 

359 ), 

360 nextpnr_extra_options=apio_ctx.project.get_list_option( 

361 "nextpnr-extra-options", None 

362 ), 

363 gtkwave_extra_options=apio_ctx.project.get_list_option( 

364 "gtkwave-extra-options", None 

365 ), 

366 verilator_extra_options=apio_ctx.project.get_list_option( 

367 "verilator-extra-options", None 

368 ), 

369 constraint_file=apio_ctx.project.get_str_option( 

370 "constraint-file", None 

371 ), 

372 ) 

373 ) 

374 assert result.apio_env_params.IsInitialized(), result 

375 

376 # -- Populate the optional command specific params. 

377 if target_params: 

378 result.target.MergeFrom(target_params) 

379 assert result.target.IsInitialized(), result 

380 

381 # -- All done. 

382 assert result.IsInitialized(), result 

383 return result 

384 

385 def _run_scons_subprocess( 

386 self, scons_target: str, *, scons_params: SconsParams 

387 ) -> Optional[int]: 

388 """Invoke an scons subprocess.""" 

389 

390 # pylint: disable=too-many-locals 

391 

392 # -- Create a shortcut. 

393 apio_ctx = self.apio_ctx 

394 

395 # -- Fetch missing files on demand. 

396 self._fetch_support_files_on_demand(scons_target, scons_params) 

397 

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

399 # -- should use. 

400 scons_dir = util.get_path_in_apio_package("scons") 

401 scons_file_path = scons_dir / "SConstruct" 

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

403 

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

405 # -- to the project root. 

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

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

408 

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

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

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

412 

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

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

415 # -- variables of the scons arg parser. 

416 apio_ctx.set_env_for_packages() 

417 

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

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

420 cout(f"* target: {scons_target}") 

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

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

423 cout() 

424 

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

426 terminal_width, _ = shutil.get_terminal_size() 

427 

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

429 # -- to execute the apio command) 

430 start_time = time.time() 

431 

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

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

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

435 terminal_width -= 1 

436 

437 # -- Print a horizontal line 

438 cout("-" * terminal_width) 

439 

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

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

442 debug_options = ( 

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

444 if util.is_debug(1) 

445 else [] 

446 ) 

447 

448 # -- Construct the scons command line. 

449 # -- 

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

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

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

453 # -- information. 

454 # -- 

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

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

457 cmd = ( 

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

459 + ["-Q", scons_target] 

460 + debug_options 

461 + variables 

462 ) 

463 

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

465 # -- needed and write them to stdout. 

466 scons_filter = SconsFilter( 

467 colors_enabled=apio_console.is_colors_enabled() 

468 ) 

469 

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

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

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

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

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

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

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

477 

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

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

480 

481 # -- Execute the scons builder! 

482 result = util.exec_command( 

483 cmd, 

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

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

486 ) 

487 

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

489 is_error = result.exit_code != 0 

490 

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

492 duration = time.time() - start_time 

493 

494 # -- Determine status message 

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

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

497 else: 

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

499 

500 # -- Determine the summary text 

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

502 

503 # -- Construct the entire message. 

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

505 msg_len = len(cunstyle(styled_msg)) 

506 

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

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

509 # -- widths. 

510 pad1_len = (terminal_width - msg_len) // 2 

511 pad2_len = terminal_width - pad1_len - msg_len 

512 

513 # -- Print the entire line. 

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

515 

516 # -- Return the exit code 

517 return result.exit_code