Coverage for tsfpga/build_project_list.py: 94%

190 statements  

« prev     ^ index     » next       coverage.py v7.6.9, created at 2024-12-07 20:51 +0000

1# -------------------------------------------------------------------------------------------------- 

2# Copyright (c) Lukas Vik. All rights reserved. 

3# 

4# This file is part of the tsfpga project, a project platform for modern FPGA development. 

5# https://tsfpga.com 

6# https://github.com/tsfpga/tsfpga 

7# -------------------------------------------------------------------------------------------------- 

8 

9# Standard libraries 

10import fnmatch 

11import time 

12from pathlib import Path 

13from threading import Lock 

14from typing import TYPE_CHECKING, Any, Callable, Iterable, Optional, Union 

15 

16# Third party libraries 

17from vunit.color_printer import COLOR_PRINTER, NO_COLOR_PRINTER 

18from vunit.test.list import TestList 

19from vunit.test.report import TestReport, TestResult 

20from vunit.test.runner import TestRunner 

21 

22# First party libraries 

23from tsfpga.system_utils import create_directory, read_last_lines_of_file 

24 

25if TYPE_CHECKING: 

26 # Local folder libraries 

27 from .module_list import ModuleList 

28 from .vivado import build_result 

29 from .vivado.project import VivadoProject 

30 

31 

32class BuildProjectList: 

33 """ 

34 Interface to handle a list of FPGA build projects. 

35 Enables building many projects in parallel. 

36 """ 

37 

38 def __init__( 

39 self, 

40 modules: "ModuleList", 

41 project_filters: list[str], 

42 include_netlist_not_top_builds: bool = False, 

43 no_color: bool = False, 

44 ): 

45 """ 

46 Arguments: 

47 modules: Module objects that can define build projects. 

48 project_filters: Project name filters. Can use wildcards (*). Leave empty for all. 

49 include_netlist_not_top_builds: Set True to get only netlist builds, 

50 instead of only top level builds. 

51 no_color: Disable color in printouts. 

52 """ 

53 self._modules = modules 

54 self._no_color = no_color 

55 

56 self.projects = list( 

57 self._iterate_projects( 

58 project_filters=project_filters, 

59 include_netlist_not_top_builds=include_netlist_not_top_builds, 

60 ) 

61 ) 

62 

63 if not self.projects: 

64 print(f"No projects matched this filter: {' '.join(project_filters)}") 

65 

66 def __str__(self) -> str: 

67 """ 

68 Returns a string with a description list of the projects. 

69 

70 Will print some information about each project (name, generics, part, ...) so can become 

71 long if there are many projects present. 

72 An alternative in that case would be :meth:`.get_short_str`. 

73 """ 

74 result = "\n".join([str(project) for project in self.projects]) 

75 result += "\n" 

76 result += "\n" 

77 result += f"Listed {len(self.projects)} builds" 

78 

79 return result 

80 

81 def get_short_str(self) -> str: 

82 """ 

83 Returns a short string with a description list of the projects. 

84 

85 This is an alternative function that is more compact than ``__str__``. 

86 """ 

87 result = "\n".join([project.name for project in self.projects]) 

88 result += "\n" 

89 result += f"Listed {len(self.projects)} builds" 

90 

91 return result 

92 

93 def create(self, projects_path: Path, num_parallel_builds: int, **kwargs: Any) -> bool: 

94 """ 

95 Create build project on disk for all the projects in the list. 

96 

97 Arguments: 

98 projects_path: The projects will be placed here. 

99 num_parallel_builds: The number of projects that will be created in parallel. 

100 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`. 

101 

102 .. Note:: 

103 Argument ``project_path`` can not be set, it is set by this class 

104 based on the ``project_paths`` argument to this function. 

105 

106 Return: 

107 True if everything went well. 

108 """ 

109 build_wrappers = [] 

110 for project in self.projects: 

111 build_wrapper = BuildProjectCreateWrapper(project, **kwargs) 

112 build_wrappers.append(build_wrapper) 

113 

114 return self._run_build_wrappers( 

115 projects_path=projects_path, 

116 build_wrappers=build_wrappers, 

117 num_parallel_builds=num_parallel_builds, 

118 ) 

119 

120 def create_unless_exists( 

121 self, projects_path: Path, num_parallel_builds: int, **kwargs: Any 

122 ) -> bool: 

123 """ 

124 Create build project for all the projects in the list, unless the project already 

125 exists. 

126 

127 Arguments: 

128 projects_path: The projects will be placed here. 

129 num_parallel_builds: The number of projects that will be created in parallel. 

130 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`. 

131 

132 .. Note:: 

133 Argument ``project_path`` can not be set, it is set by this class 

134 based on the ``project_paths`` argument to this function. 

135 

136 Return: 

137 True if everything went well. 

138 """ 

139 build_wrappers = [] 

140 for project in self.projects: 

141 if not (projects_path / project.name / "project").exists(): 

142 build_wrapper = BuildProjectCreateWrapper(project, **kwargs) 

143 build_wrappers.append(build_wrapper) 

144 

145 if not build_wrappers: 

146 # Return straight away if no projects need to be created. To avoid extra 

147 # "No tests were run!" printout from creation step that is very misleading. 

148 return True 

149 

150 return self._run_build_wrappers( 

151 projects_path=projects_path, 

152 build_wrappers=build_wrappers, 

153 num_parallel_builds=num_parallel_builds, 

154 ) 

155 

156 def build( 

157 self, 

158 projects_path: Path, 

159 num_parallel_builds: int, 

160 num_threads_per_build: int, 

161 output_path: Optional[Path] = None, 

162 collect_artifacts: Optional[Callable[["VivadoProject", Path], bool]] = None, 

163 **kwargs: Any, 

164 ) -> bool: 

165 """ 

166 Build all the projects in the list. 

167 

168 Arguments: 

169 projects_path: The projects are placed here. 

170 num_parallel_builds: The number of projects that will be built in parallel. 

171 num_threads_per_build: The number threads that will be used for each 

172 parallel build process. 

173 output_path: Where the artifacts should be placed. 

174 Will default to within the ``projects_path`` if not set. 

175 collect_artifacts: Callback to collect artifacts. 

176 Takes two named arguments: 

177 

178 | **project** (:class:`.VivadoProject`): The project that is being built. 

179 

180 | **output_path** (pathlib.Path): Where the build artifacts should be placed. 

181 

182 | Must return True. 

183 kwargs: Other arguments as accepted by :meth:`.VivadoProject.build`. 

184 

185 .. Note:: 

186 Argument ``project_path`` can not be set, it is set by this class 

187 based on the ``project_paths`` argument to this function. 

188 

189 Argument ``num_threads`` is set by the ``num_threads_per_build`` 

190 argument to this function. This naming difference is done to avoid 

191 confusion with regards to ``num_parallel_builds``. 

192 

193 Return: 

194 True if everything went well. 

195 """ 

196 if collect_artifacts: 

197 thread_safe_collect_artifacts = ThreadSafeCollectArtifacts( 

198 collect_artifacts 

199 ).collect_artifacts 

200 else: 

201 thread_safe_collect_artifacts = None 

202 

203 build_wrappers = [] 

204 for project in self.projects: 

205 project_output_path = self.get_build_project_output_path( 

206 project=project, projects_path=projects_path, output_path=output_path 

207 ) 

208 

209 build_wrapper = BuildProjectBuildWrapper( 

210 project=project, 

211 collect_artifacts=thread_safe_collect_artifacts, 

212 output_path=project_output_path, 

213 num_threads=num_threads_per_build, 

214 **kwargs, 

215 ) 

216 build_wrappers.append(build_wrapper) 

217 

218 return self._run_build_wrappers( 

219 projects_path=projects_path, 

220 build_wrappers=build_wrappers, 

221 num_parallel_builds=num_parallel_builds, 

222 ) 

223 

224 @staticmethod 

225 def get_build_project_output_path( 

226 project: "VivadoProject", projects_path: Path, output_path: Optional[Path] = None 

227 ) -> Path: 

228 """ 

229 Find where build artifacts will be placed for a project. 

230 Arguments are the same as for :meth:`.build`. 

231 """ 

232 if output_path: 

233 return output_path.resolve() / project.name 

234 

235 return projects_path / project.name 

236 

237 def open(self, projects_path: Path) -> bool: 

238 """ 

239 Open the projects in EDA GUI. 

240 

241 Arguments: 

242 projects_path: The projects are placed here. 

243 

244 Return: 

245 True if everything went well. 

246 """ 

247 build_wrappers = [] 

248 for project in self.projects: 

249 build_wrappers.append(BuildProjectOpenWrapper(project)) 

250 

251 return self._run_build_wrappers( 

252 projects_path=projects_path, 

253 build_wrappers=build_wrappers, 

254 # For open there is no performance limitation. Set a high value. 

255 num_parallel_builds=20, 

256 ) 

257 

258 def _run_build_wrappers( 

259 self, 

260 projects_path: Path, 

261 build_wrappers: Union[ 

262 list["BuildProjectCreateWrapper"], 

263 list["BuildProjectBuildWrapper"], 

264 list["BuildProjectOpenWrapper"], 

265 ], 

266 num_parallel_builds: int, 

267 ) -> bool: 

268 if not build_wrappers: 

269 # Return straight away if no builds are supplied 

270 return True 

271 

272 start_time = time.time() 

273 

274 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER 

275 report = BuildReport(printer=color_printer) 

276 

277 test_list = TestList() 

278 for build_wrapper in build_wrappers: 

279 test_list.add_test(build_wrapper) 

280 

281 verbosity = BuildRunner.VERBOSITY_QUIET 

282 test_runner = BuildRunner( 

283 report=report, 

284 output_path=projects_path, 

285 verbosity=verbosity, 

286 num_threads=num_parallel_builds, 

287 ) 

288 test_runner.run(test_list) 

289 

290 all_builds_ok: bool = report.all_ok() 

291 report.set_real_total_time(time.time() - start_time) 

292 

293 # True if the builds are for the "build" step (not "create" or "open") 

294 builds_are_build_step = isinstance(build_wrappers[0], BuildProjectBuildWrapper) 

295 # If we are building, we should print the summary that is at the end of the console output. 

296 # (however if we are creating or opening a project we should not print anything extra). 

297 # However if anything has failed, we should also print. 

298 if builds_are_build_step: 

299 # The length of the build summary depends on if we are working with netlist builds or 

300 # regular ones, so set the length given by one of the project objects. 

301 # Ignore typing error that 'Create' or 'Open' build wrappers do not have 

302 # 'build_result_report_length'. We only end up here if it is a Build. 

303 report_length_lines = build_wrappers[ 

304 0 

305 ].build_result_report_length # type: ignore[union-attr] 

306 report.set_report_length(report_length_lines=report_length_lines) 

307 

308 # If all are OK then we should print the resource utilization numbers. 

309 # If not, then we print a few last lines of the log output. 

310 if builds_are_build_step or not all_builds_ok: 

311 report.print_str() 

312 

313 return all_builds_ok 

314 

315 def _iterate_projects( 

316 self, project_filters: list[str], include_netlist_not_top_builds: bool 

317 ) -> Iterable["VivadoProject"]: 

318 available_projects = [] 

319 for module in self._modules: 

320 available_projects += module.get_build_projects() 

321 

322 for project in available_projects: 

323 if project.is_netlist_build == include_netlist_not_top_builds: 

324 if not project_filters: 

325 yield project 

326 

327 else: 

328 for project_filter in project_filters: 

329 if fnmatch.filter([project.name], project_filter): 

330 yield project 

331 

332 # Do not continue with further filters if we have already matched this 

333 # project. 

334 # Multiple filters might match the same project, and multiple objects 

335 # of the same project will break build 

336 break 

337 

338 

339class BuildProjectCreateWrapper: 

340 """ 

341 Wrapper to create a build project, for usage in the build runner. 

342 Mimics a VUnit test object. 

343 """ 

344 

345 def __init__(self, project: "VivadoProject", **kwargs: Any) -> None: 

346 self.name = project.name 

347 self._project = project 

348 self._create_arguments = kwargs 

349 

350 def run(self, output_path: Path, read_output: Any) -> bool: # pylint: disable=unused-argument 

351 """ 

352 VUnit test runner sends another argument "read_output" which we don't use. 

353 """ 

354 this_projects_path = Path(output_path) / "project" 

355 return self._project.create(project_path=this_projects_path, **self._create_arguments) 

356 

357 

358class BuildProjectBuildWrapper: 

359 """ 

360 Wrapper to build a project, for usage in the build runner. 

361 Mimics a VUnit test object. 

362 """ 

363 

364 def __init__( 

365 self, 

366 project: "VivadoProject", 

367 collect_artifacts: Optional[Callable[..., bool]], 

368 **kwargs: Any, 

369 ) -> None: 

370 self.name = project.name 

371 self._project = project 

372 self._collect_artifacts = collect_artifacts 

373 self._build_arguments = kwargs 

374 

375 def run(self, output_path: Path, read_output: Any) -> bool: # pylint: disable=unused-argument 

376 """ 

377 VUnit test runner sends another argument "read_output" which we don't use. 

378 """ 

379 this_projects_path = Path(output_path) / "project" 

380 build_result = self._project.build(project_path=this_projects_path, **self._build_arguments) 

381 

382 if not build_result.success: 

383 self._print_build_result(build_result) 

384 return build_result.success 

385 

386 # Proceed to artifact collection only if build succeeded. 

387 if self._collect_artifacts: 

388 if not self._collect_artifacts( 

389 project=self._project, output_path=self._build_arguments["output_path"] 

390 ): 

391 build_result.success = False 

392 

393 # Print size at the absolute end 

394 self._print_build_result(build_result=build_result) 

395 return build_result.success 

396 

397 @staticmethod 

398 def _print_build_result(build_result: "build_result.BuildResult") -> None: 

399 build_report = build_result.report() 

400 if build_report: 

401 # Add an empty line before the build result report, to have margin in how many lines are 

402 # printed. See the comments in BuildResult for an explanation. 

403 print() 

404 print(build_report) 

405 

406 @property 

407 def build_result_report_length(self) -> int: 

408 """ 

409 The number of lines in the build_result report from this project. 

410 """ 

411 # The size summary, as returned by tsfpga.vivado.project.BuildResult is a JSON formatted 

412 # string with one line for each utilization category. 

413 # For Xilinx 7 series, there are 8 categories (Total LUTs, Logic LUTs, LUTRAMs, 

414 # SRLs, FFs, RAMB36, RAMB18, DSP Blocks). For UltraScale series there is one 

415 # extra (URAM). 

416 # Additionally, the size summary contains three extra lines for JSON braces and a title. 

417 # 

418 # This value is enough lines so the whole summary gets printed to console. 

419 # For 7 series, this will mean an extra blank line before the summary. 

420 # 

421 # This is a hack. Works for now, but is far from reliable. 

422 length_of_size_report = 3 + 8 + 1 

423 

424 if self._project.is_netlist_build: 

425 # The logic level distribution report is five lines, plus a title line. 

426 # This report is only printed for netlist builds, where there is no configured clock 

427 # present. If there were many clocks present in the build, the report would be longer. 

428 length_of_logic_level_report = 5 + 1 

429 return length_of_size_report + length_of_logic_level_report 

430 

431 return length_of_size_report 

432 

433 

434class BuildProjectOpenWrapper: 

435 """ 

436 Wrapper to open a build project, for usage in the build runner. 

437 Mimics a VUnit test object. 

438 """ 

439 

440 def __init__(self, project: "VivadoProject") -> None: 

441 self.name = project.name 

442 self._project = project 

443 

444 def run(self, output_path: Path, read_output: Any) -> bool: # pylint: disable=unused-argument 

445 """ 

446 VUnit test runner sends another argument "read_output" which we don't use. 

447 """ 

448 this_projects_path = Path(output_path) / "project" 

449 return self._project.open(project_path=this_projects_path) 

450 

451 

452class BuildRunner(TestRunner): 

453 """ 

454 Build runner that mimics a VUnit TestRunner. Most things are used as they are in the 

455 base class, but some behavior is overridden. 

456 """ 

457 

458 def _create_test_mapping_file(self, test_suites: Any) -> None: 

459 """ 

460 Overloaded from super class. 

461 

462 Do not create this file. 

463 

464 We do not need it since folder name is the same as project name. 

465 """ 

466 

467 def _get_output_path(self, test_suite_name: str) -> str: 

468 """ 

469 Overloaded from super class. 

470 

471 Output folder name is the same as the project name. 

472 

473 Original function adds a hash at the end of the folder name. 

474 We do not want that necessarily. 

475 """ 

476 return str(Path(self._output_path) / test_suite_name) 

477 

478 @staticmethod 

479 def _prepare_test_suite_output_path(output_path: str) -> None: 

480 """ 

481 Overloaded from super class. 

482 

483 Create the directory unless it already exists. 

484 

485 Original function wipes the path before running a test. We do not want to do that 

486 since e.g. a Vivado project takes a long time to create and might contain a state 

487 that the user wants to keep. 

488 """ 

489 create_directory(Path(output_path), empty=False) 

490 

491 

492class ThreadSafeCollectArtifacts: 

493 """ 

494 A thread-safe wrapper around a user-supplied function that makes sure the function 

495 is not launched more than once at the same time. When two builds finish at the 

496 same time, race conditions can arise depending on what the function does. 

497 

498 Note that this is a VERY fringe case, since builds usually take >20 minutes, and the 

499 collection probably only takes a few seconds. But it happens sometimes with the tsfpga 

500 example projects which are identical and quite fast (roughly three minutes). 

501 """ 

502 

503 def __init__(self, collect_artifacts: Callable[..., bool]) -> None: 

504 self._collect_artifacts = collect_artifacts 

505 self._lock = Lock() 

506 

507 def collect_artifacts(self, project: "VivadoProject", output_path: Path) -> bool: 

508 with self._lock: 

509 return self._collect_artifacts(project=project, output_path=output_path) 

510 

511 

512class BuildReport(TestReport): 

513 def add_result(self, *args: Any, **kwargs: Any) -> None: 

514 """ 

515 Overloaded from super class. 

516 

517 Add a a test result. 

518 

519 Uses a different Result class than the super method. 

520 """ 

521 result = BuildResult(*args, **kwargs) 

522 self._test_results[result.name] = result 

523 self._test_names_in_order.append(result.name) 

524 

525 def set_report_length(self, report_length_lines: int) -> None: 

526 """ 

527 Set the report length for all test results that have been added to the report. 

528 """ 

529 for test_result in self._test_results.values(): 

530 test_result.set_report_length(report_length_lines) 

531 

532 def print_latest_status(self, total_tests: int) -> None: 

533 """ 

534 Overloaded from super class. 

535 

536 This method is called for each build when it should print its result just as it finished, 

537 but other builds may not be finished yet. 

538 

539 Inherited and adapted from the VUnit function: 

540 * Removed support for the "skipped" result. 

541 * Do not use abbreviations in the printout. 

542 * Use f-strings. 

543 """ 

544 result = self._last_test_result() 

545 passed, failed, _ = self._split() 

546 

547 if result.passed: 

548 self._printer.write("pass", fg="gi") 

549 elif result.failed: 

550 self._printer.write("fail", fg="ri") 

551 else: 

552 assert False 

553 

554 count_summary = f"pass={len(passed)} fail={len(failed)} total={total_tests}" 

555 self._printer.write(f" ({count_summary}) {result.name} ({result.time:.1f} seconds)\n") 

556 

557 

558class BuildResult(TestResult): 

559 report_length_lines = None 

560 

561 def _print_output(self, printer: Any, num_lines: int) -> None: 

562 """ 

563 Print the last lines from the output file. 

564 

565 The ``printer`` argument should of type ``ColorPrinter`` from VUnit. 

566 """ 

567 output_tail = read_last_lines_of_file(Path(self._output_file_name), num_lines=num_lines) 

568 printer.write(output_tail) 

569 

570 def set_report_length(self, report_length_lines: int) -> None: 

571 """ 

572 Set how many lines shall be printed when this result is printed. 

573 """ 

574 self.report_length_lines = report_length_lines 

575 

576 # pylint: disable=arguments-differ 

577 def print_status( # type: ignore[override] 

578 self, printer: Any, padding: int = 0, **kwargs: dict[str, Any] 

579 ) -> None: 

580 """ 

581 Overloaded from super class. 

582 

583 The ``printer`` argument should of type ``ColorPrinter`` from VUnit. 

584 

585 This method is called for each build when it should print its result in the "Summary" at 

586 the end when all builds have finished. 

587 

588 Inherited and adapted from the VUnit function. 

589 

590 Note that a ``max_time`` integer argument is added in VUnit >4.7.0, but at the time of 

591 writing this is un-released on the VUnit ``master`` branch. 

592 In order to be compatible with both older and newer versions, we use ``**kwargs`` for this. 

593 """ 

594 if self.passed and self.report_length_lines is not None: 

595 # Build passed, print build summary of the specified length. The length is only 

596 # set if this is a "build" result (not "create" or "open"). 

597 self._print_output(printer=printer, num_lines=self.report_length_lines) 

598 else: 

599 # The build failed, which can either be caused by 

600 # 1. IDE build failure 

601 # 2. IDE build succeeded, but post build hook, or size checkers failed. 

602 # 3. Other python error (directory already exists, ...) 

603 # In the case of IDE build failed, we want a significant portion of the output, to be 

604 # able to see an indication of what failed. In the case of size checkers, we want to see 

605 # all the printouts from all checkers, to see which one failed. Since there are at most 

606 # eight resource categories, it is reasonable to assume that there will never be more 

607 # than eight size checkers. 

608 self._print_output(printer=printer, num_lines=25) 

609 

610 # Print the regular output from the VUnit class. 

611 # A little extra margin between build name and execution time makes the output more readable 

612 super().print_status(printer=printer, padding=padding + 2, **kwargs) 

613 # Add an empty line between each build, for readability. 

614 printer.write("\n")