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

143 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-09-23 03:53 +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 os 

10import sys 

11import time 

12import shutil 

13 

14from datetime import datetime 

15from google.protobuf import text_format 

16from apio.common.debug_util import is_debug 

17from apio.common import apio_console, proto_util 

18from apio.common.apio_console import ( 

19 cout, 

20 cstyle, 

21 cunstyle, 

22 fatal_error, 

23) 

24from apio.common.apio_styles import SUCCESS, ERROR, EMPH3 

25from apio.utils import util 

26from apio.apio_context import ApioContext 

27from apio.managers.scons_filter import SconsFilter 

28from apio.managers import xilinx_chipdb 

29from apio.common.proto.apio_common_pb2 import ApioArch 

30from apio.common.proto.apio_scons_pb2 import ( 

31 FORCE_PIPE, 

32 FORCE_TERMINAL, 

33 Verbosity, 

34 Environment, 

35 SconsParams, 

36 TargetParams, 

37 FpgaInfo, 

38 ApioEnvParams, 

39 Ice40Params, 

40 Ecp5FpgaParams, 

41 GowinParams, 

42 XilinxParams, 

43 GraphParams, 

44 LintParams, 

45 SimParams, 

46 ApioTestParams, 

47 UploadParams, 

48) 

49 

50 

51class SConsManager: 

52 """Class for managing the scons tools""" 

53 

54 def __init__(self, apio_ctx: ApioContext): 

55 """Initialization.""" 

56 # -- Cache the apio context. 

57 self.apio_ctx = apio_ctx 

58 

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

60 os.chdir(apio_ctx.project_dir) 

61 

62 def graph( 

63 self, graph_params: GraphParams, verbosity: Verbosity 

64 ) -> int | None: 

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

66 exit code, 0 if ok.""" 

67 

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

69 scons_params = self.construct_scons_params( 

70 target_params=TargetParams(graph=graph_params), 

71 verbosity=verbosity, 

72 ) 

73 

74 # -- Run the scons process. 

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

76 

77 def lint(self, lint_params: LintParams) -> int | None: 

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

79 exit code, 0 if ok.""" 

80 

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

82 scons_params = self.construct_scons_params( 

83 target_params=TargetParams(lint=lint_params) 

84 ) 

85 

86 # -- Run the scons process. 

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

88 

89 def sim(self, sim_params: SimParams) -> int | None: 

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

91 exit code, 0 if ok.""" 

92 

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

94 scons_params = self.construct_scons_params( 

95 target_params=TargetParams(sim=sim_params) 

96 ) 

97 

98 # -- Run the scons process. 

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

100 

101 def test(self, test_params: ApioTestParams) -> int | None: 

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

103 exit code, 0 if ok.""" 

104 

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

106 scons_params = self.construct_scons_params( 

107 target_params=TargetParams(test=test_params) 

108 ) 

109 

110 # -- Run the scons process. 

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

112 

113 def build(self, verbosity: Verbosity) -> int | None: 

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

115 exit code, 0 if ok.""" 

116 

117 # -- Construct the scons params object. 

118 scons_params = self.construct_scons_params( 

119 verbosity=verbosity, 

120 ) 

121 

122 # -- Run the scons process. 

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

124 

125 def report(self, verbosity: Verbosity) -> int | None: 

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

127 exit code, 0 if ok.""" 

128 

129 # -- Construct the scons params object. 

130 scons_params = self.construct_scons_params( 

131 verbosity=verbosity, 

132 ) 

133 

134 # -- Run the scons process. 

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

136 

137 def upload(self, upload_params: UploadParams) -> int | None: 

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

139 exit code, 0 if ok. 

140 """ 

141 

142 # -- Construct the scons params. 

143 scons_params = self.construct_scons_params( 

144 target_params=TargetParams(upload=upload_params) 

145 ) 

146 

147 # -- Execute Scons for uploading! 

148 exit_code = self._run_scons_subprocess( 

149 "upload", scons_params=scons_params 

150 ) 

151 

152 return exit_code 

153 

154 def construct_scons_params( 

155 self, 

156 *, 

157 target_params: TargetParams | None = None, 

158 verbosity: Verbosity | None = None, 

159 ) -> SconsParams: 

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

161 process.""" 

162 

163 # pylint: disable=too-many-statements 

164 # pylint: disable=too-many-locals 

165 

166 # -- Create a shortcut. 

167 apio_ctx = self.apio_ctx 

168 

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

170 result = SconsParams() 

171 

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

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

174 ts = datetime.now() 

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

176 

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

178 # -- to be in a project context. 

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

180 project = apio_ctx.project 

181 

182 # -- Get the project resources. 

183 pr = apio_ctx.project_resources 

184 fpga_definition = pr.fpga_definition 

185 

186 # -- Populate the common values of FpgaInfo. 

187 proto_util.check_is_required(fpga_definition, "part_num", "size") 

188 result.fpga_info.MergeFrom( 

189 FpgaInfo( 

190 fpga_id=pr.fpga_id, 

191 part_num=fpga_definition.part_num, 

192 size=fpga_definition.size, 

193 ) 

194 ) 

195 

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

197 proto_util.check_is_required(fpga_definition, "arch") 

198 fpga_arch = fpga_definition.arch 

199 match fpga_arch: 

200 case ApioArch.ice40: 

201 assert fpga_definition.HasField("ice40_params") 

202 ice40_params = fpga_definition.ice40_params 

203 result.arch = ApioArch.ice40 

204 proto_util.check_is_required(ice40_params, "type", "package") 

205 result.fpga_info.ice40_params.MergeFrom( 

206 Ice40Params( 

207 type=ice40_params.type, 

208 package=ice40_params.package, 

209 ) 

210 ) 

211 case ApioArch.ecp5: 

212 assert fpga_definition.HasField("ecp5_params") 

213 epp5_params = fpga_definition.ecp5_params 

214 result.arch = ApioArch.ecp5 

215 proto_util.check_is_required( 

216 epp5_params, "type", "package", "speed" 

217 ) 

218 result.fpga_info.ecp5_params.MergeFrom( 

219 Ecp5FpgaParams( 

220 type=epp5_params.type, 

221 package=epp5_params.package, 

222 speed=epp5_params.speed, 

223 ) 

224 ) 

225 case ApioArch.gowin: 

226 assert fpga_definition.HasField("gowin_params") 

227 gowin_params = fpga_definition.gowin_params 

228 result.arch = ApioArch.gowin 

229 proto_util.check_is_required( 

230 gowin_params, 

231 "yosys_family", 

232 "nextpnr_family", 

233 "packer_device", 

234 ) 

235 result.fpga_info.gowin_params.MergeFrom( 

236 GowinParams( 

237 yosys_family=gowin_params.yosys_family, 

238 nextpnr_family=gowin_params.nextpnr_family, 

239 packer_device=gowin_params.packer_device, 

240 ) 

241 ) 

242 case ApioArch.xilinx: 242 ↛ 267line 242 didn't jump to line 267 because the pattern on line 242 always matched

243 assert fpga_definition.HasField("xilinx_params") 

244 xilinx_params = fpga_definition.xilinx_params 

245 result.arch = ApioArch.xilinx 

246 proto_util.check_is_required( 

247 xilinx_params, 

248 "yosys_family", 

249 "yosys_arch", 

250 "yosys_part", 

251 "speed", 

252 ) 

253 # -- Get a path to the chipdb file for this yosys part. 

254 # -- If it doesn't exist, it is fetched on the fly from 

255 # -- the release of the installed openxc7 package. 

256 chipdb_file_path = xilinx_chipdb.chipdb_file_on_demand( 

257 apio_ctx, xilinx_params.yosys_part 

258 ) 

259 result.fpga_info.xilinx_params.MergeFrom( 

260 XilinxParams( 

261 yosys_family=xilinx_params.yosys_family, 

262 yosys_arch=xilinx_params.yosys_arch, 

263 yosys_part=xilinx_params.yosys_part, 

264 chipdb_file_path=str(chipdb_file_path), 

265 ) 

266 ) 

267 case _: 

268 fatal_error(f"Unexpected fpga_arch value {fpga_arch}") 

269 

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

271 assert result.fpga_info.IsInitialized(), result 

272 

273 # -- Populate the optional Verbosity params. 

274 if verbosity: 

275 result.verbosity.MergeFrom(verbosity) 

276 assert result.verbosity.IsInitialized(), result 

277 

278 # -- Populate the Environment params. 

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

280 oss_define_consts = apio_ctx.all_packages["oss-cad-suite"]["env"][ 

281 "define-consts" 

282 ] 

283 assert "YOSYS_LIB" in oss_define_consts, oss_define_consts 

284 assert "TRELLIS" in oss_define_consts, oss_define_consts 

285 

286 openxc7_define_consts = apio_ctx.all_packages["openxc7"]["env"][ 

287 "define-consts" 

288 ] 

289 assert "PRJXRAY_DB_DIR" in openxc7_define_consts, openxc7_define_consts 

290 

291 result.environment.MergeFrom( 

292 Environment( 

293 platform_id=apio_ctx.platform_id, 

294 is_windows=apio_ctx.is_windows, 

295 terminal_mode=( 

296 FORCE_TERMINAL 

297 if apio_console.is_terminal() 

298 else FORCE_PIPE 

299 ), 

300 theme_name=apio_console.current_theme_name(), 

301 yosys_path=oss_define_consts["YOSYS_LIB"], 

302 trellis_path=oss_define_consts["TRELLIS"], 

303 scons_shell_id=apio_ctx.scons_shell_id, 

304 xilinx_prjxray_db_path=openxc7_define_consts["PRJXRAY_DB_DIR"], 

305 ) 

306 ) 

307 assert result.environment.IsInitialized(), result 

308 

309 # -- Populate the Project params. 

310 result.apio_env_params.MergeFrom( 

311 ApioEnvParams( 

312 env_name=apio_ctx.project.env_name, 

313 board_id=pr.board_id, 

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

315 defines=apio_ctx.project.get_list_option( 

316 "defines", default=[] 

317 ), 

318 yosys_extra_options=apio_ctx.project.get_list_option( 

319 "yosys-extra-options", None 

320 ), 

321 nextpnr_extra_options=apio_ctx.project.get_list_option( 

322 "nextpnr-extra-options", None 

323 ), 

324 gtkwave_extra_options=apio_ctx.project.get_list_option( 

325 "gtkwave-extra-options", None 

326 ), 

327 verilator_extra_options=apio_ctx.project.get_list_option( 

328 "verilator-extra-options", None 

329 ), 

330 constraint_file=apio_ctx.project.get_str_option( 

331 "constraint-file", None 

332 ), 

333 ) 

334 ) 

335 assert result.apio_env_params.IsInitialized(), result 

336 

337 # -- Populate the optional command specific params. 

338 if target_params: 

339 result.target.MergeFrom(target_params) 

340 assert result.target.IsInitialized(), result 

341 

342 # -- All done. 

343 assert result.IsInitialized(), result 

344 return result 

345 

346 def _run_scons_subprocess( 

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

348 ) -> int | None: 

349 """Invoke an scons subprocess.""" 

350 

351 # pylint: disable=too-many-locals 

352 

353 # -- Create a shortcut. 

354 apio_ctx = self.apio_ctx 

355 

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

357 # -- should use. 

358 scons_dir = util.get_path_in_apio_package("scons") 

359 scons_file_path = scons_dir / "SConstruct" 

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

361 

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

363 # -- to the project root. 

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

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

366 

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

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

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

370 

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

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

373 # -- variables of the scons arg parser. 

374 apio_ctx.set_env_for_packages() 

375 

376 if is_debug(1): 376 ↛ 377line 376 didn't jump to line 377 because the condition on line 376 was never true

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

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

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

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

381 cout() 

382 

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

384 terminal_width, _ = shutil.get_terminal_size() 

385 

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

387 # -- to execute the apio command) 

388 start_time = time.time() 

389 

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

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

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

393 terminal_width -= 1 

394 

395 # -- Print a horizontal line 

396 cout("-" * terminal_width) 

397 

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

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

400 debug_options = ( 

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

402 if is_debug(1) 

403 else [] 

404 ) 

405 

406 # -- Construct the scons command line. 

407 # -- 

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

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

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

411 # -- information. 

412 # -- 

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

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

415 cmd = ( 

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

417 + ["-Q", scons_target] 

418 + debug_options 

419 + variables 

420 ) 

421 

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

423 # -- needed and write them to stdout. 

424 scons_filter = SconsFilter( 

425 colors_enabled=apio_console.is_colors_enabled() 

426 ) 

427 

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

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

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

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

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

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

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

435 

436 if is_debug(1): 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true

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

438 

439 # -- Execute the scons builder! 

440 result = util.exec_command( 

441 cmd, 

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

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

444 ) 

445 

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

447 is_error = result.exit_code != 0 

448 

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

450 duration = time.time() - start_time 

451 

452 # -- Determine status message 

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

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

455 else: 

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

457 

458 # -- Determine the summary text 

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

460 

461 # -- Construct the entire message. 

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

463 msg_len = len(cunstyle(styled_msg)) 

464 

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

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

467 # -- widths. 

468 pad1_len = (terminal_width - msg_len) // 2 

469 pad2_len = terminal_width - pad1_len - msg_len 

470 

471 # -- Print the entire line. 

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

473 

474 # -- Return the exit code 

475 return result.exit_code