Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

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

3# 

4# This file is part of the tsfpga project. 

5# https://tsfpga.com 

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

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

8 

9import fnmatch 

10import time 

11from pathlib import Path 

12from threading import Lock 

13 

14from vunit.test.list import TestList 

15from vunit.test.runner import TestRunner 

16from vunit.test.report import TestReport, TestResult 

17from vunit.color_printer import COLOR_PRINTER, NO_COLOR_PRINTER 

18 

19from tsfpga.system_utils import create_directory, read_last_lines_of_file 

20 

21 

22class BuildProjectList: 

23 

24 """ 

25 Interface to handle a list of FPGA build projects. 

26 Enables building many projects in parallel. 

27 """ 

28 

29 def __init__( 

30 self, modules, project_filters=None, include_netlist_not_top_builds=False, no_color=False 

31 ): 

32 """ 

33 Arguments: 

34 modules (list(:class:`.BaseModule`)): Module objects that can define build 

35 projects. 

36 project_filters (list(str)): Project name filters. Can use wildcards (*). 

37 Leave empty for all. 

38 include_netlist_not_top_builds (bool): Set True to get only netlist builds, 

39 instead of only top level builds. 

40 no_color (bool): Disable color in printouts. 

41 """ 

42 self._modules = modules 

43 self._no_color = no_color 

44 self.projects = list( 

45 self._iterate_projects(project_filters, include_netlist_not_top_builds) 

46 ) 

47 

48 if not self.projects: 

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

50 

51 def __str__(self): 

52 """ 

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

54 """ 

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

56 result += "\n" 

57 result += "\n" 

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

59 return result 

60 

61 def create(self, projects_path, num_parallel_builds, **kwargs): 

62 """ 

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

64 

65 Arguments: 

66 projects_path (`pathlib.Path`): The projects will be placed here. 

67 num_parallel_builds (int): The number of projects that will be created in 

68 parallel. 

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

70 

71 .. Note:: 

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

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

74 

75 Returns: 

76 bool: True if everything went well. 

77 """ 

78 build_wrappers = [] 

79 for project in self.projects: 

80 build_wrapper = BuildProjectCreateWrapper(project, **kwargs) 

81 build_wrappers.append(build_wrapper) 

82 

83 return self._run_build_wrappers( 

84 projects_path=projects_path, 

85 build_wrappers=build_wrappers, 

86 num_parallel_builds=num_parallel_builds, 

87 ) 

88 

89 def create_unless_exists(self, projects_path, num_parallel_builds, **kwargs): 

90 """ 

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

92 exists. 

93 

94 Arguments: 

95 projects_path (`pathlib.Path`): The projects will be placed here. 

96 num_parallel_builds (int): The number of projects that will be created in 

97 parallel. 

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

99 

100 .. Note:: 

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

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

103 

104 Returns: 

105 bool: True if everything went well. 

106 """ 

107 build_wrappers = [] 

108 for project in self.projects: 

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

110 build_wrapper = BuildProjectCreateWrapper(project, **kwargs) 

111 build_wrappers.append(build_wrapper) 

112 

113 if not build_wrappers: 

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

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

116 return True 

117 

118 return self._run_build_wrappers( 

119 projects_path=projects_path, 

120 build_wrappers=build_wrappers, 

121 num_parallel_builds=num_parallel_builds, 

122 ) 

123 

124 def build( 

125 self, 

126 projects_path, 

127 num_parallel_builds, 

128 num_threads_per_build, 

129 output_path=None, 

130 collect_artifacts=None, 

131 **kwargs, 

132 ): 

133 """ 

134 Build all the projects in the list. 

135 

136 Arguments: 

137 projects_path (`pathlib.Path`): The projects will be placed here. 

138 num_parallel_builds (int): The number of projects that will be built in 

139 parallel. 

140 num_threads_per_build (int): The number threads that will be used for each 

141 parallel build process. 

142 output_path (`pathlib.Path`): Where the artifacts should be placed. 

143 collect_artifacts (`function`): Callback to collect artifacts. Takes two named 

144 arguments: 

145 

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

147 

148 | **output_path** (`pathlib.Path`): Where the artifacts should be placed. 

149 

150 

151 | Must return True. 

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

153 

154 .. Note:: 

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

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

157 

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

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

160 confusion with regards to ``num_parallel_builds``. 

161 

162 

163 Returns: 

164 bool: True if everything went well. 

165 """ 

166 if collect_artifacts: 

167 thread_safe_collect_artifacts = ThreadSafeCollectArtifacts( 

168 collect_artifacts 

169 ).collect_artifacts 

170 else: 

171 thread_safe_collect_artifacts = None 

172 

173 build_wrappers = [] 

174 for project in self.projects: 

175 if output_path: 

176 this_projects_output_path = output_path.resolve() / project.name 

177 else: 

178 this_projects_output_path = projects_path / project.name 

179 

180 build_wrapper = BuildProjectBuildWrapper( 

181 project, 

182 output_path=this_projects_output_path, 

183 collect_artifacts=thread_safe_collect_artifacts, 

184 num_threads=num_threads_per_build, 

185 **kwargs, 

186 ) 

187 build_wrappers.append(build_wrapper) 

188 

189 return self._run_build_wrappers( 

190 projects_path=projects_path, 

191 build_wrappers=build_wrappers, 

192 num_parallel_builds=num_parallel_builds, 

193 ) 

194 

195 def open(self, projects_path): 

196 """ 

197 Open the projects in EDA GUI. 

198 

199 Arguments: 

200 projects_path (`pathlib.Path`): The projects are placed here. 

201 

202 Returns: 

203 bool: True if everything went well. 

204 """ 

205 build_wrappers = [] 

206 for project in self.projects: 

207 build_wrapper = BuildProjectOpenWrapper(project) 

208 build_wrappers.append(build_wrapper) 

209 

210 return self._run_build_wrappers( 

211 projects_path=projects_path, 

212 build_wrappers=build_wrappers, 

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

214 num_parallel_builds=20, 

215 ) 

216 

217 def _run_build_wrappers(self, projects_path, build_wrappers, num_parallel_builds): 

218 if not build_wrappers: 

219 # Return straight away if no builds are supplied 

220 return True 

221 

222 start_time = time.time() 

223 

224 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER 

225 report = BuildReport(printer=color_printer) 

226 

227 test_list = TestList() 

228 for build_wrapper in build_wrappers: 

229 test_list.add_test(build_wrapper) 

230 

231 verbosity = BuildRunner.VERBOSITY_QUIET 

232 test_runner = BuildRunner( 

233 report=report, 

234 output_path=projects_path, 

235 verbosity=verbosity, 

236 num_threads=num_parallel_builds, 

237 ) 

238 test_runner.run(test_list) 

239 

240 all_builds_ok = report.all_ok() 

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

242 

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

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

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

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

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

248 if builds_are_build_step: 

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

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

251 report.set_report_length(build_wrappers[0].build_result_report_length) 

252 if builds_are_build_step or not all_builds_ok: 

253 report.print_str() 

254 

255 return all_builds_ok 

256 

257 def _iterate_projects(self, project_filters, include_netlist_not_top_builds): 

258 available_projects = [] 

259 for module in self._modules: 

260 available_projects += module.get_build_projects() 

261 

262 for project in available_projects: 

263 if project.is_netlist_build == include_netlist_not_top_builds: 

264 if not project_filters: 

265 yield project 

266 else: 

267 for project_filter in project_filters: 

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

269 yield project 

270 

271 

272class BuildProjectCreateWrapper: 

273 

274 """ 

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

276 Mimics a VUnit test object. 

277 """ 

278 

279 def __init__(self, project, **kwargs): 

280 self.name = project.name 

281 self._project = project 

282 self._create_arguments = kwargs 

283 

284 # pylint: disable=unused-argument 

285 def run(self, output_path, read_output): 

286 """ 

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

288 """ 

289 this_projects_path = Path(output_path) / "project" 

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

291 

292 

293class BuildProjectBuildWrapper: 

294 

295 """ 

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

297 Mimics a VUnit test object. 

298 """ 

299 

300 def __init__(self, project, collect_artifacts, **kwargs): 

301 self.name = project.name 

302 self._project = project 

303 self._collect_artifacts = collect_artifacts 

304 self._build_arguments = kwargs 

305 

306 # pylint: disable=unused-argument 

307 def run(self, output_path, read_output): 

308 """ 

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

310 """ 

311 this_projects_path = Path(output_path) / "project" 

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

313 

314 if not build_result.success: 

315 self._print_build_result(build_result) 

316 return build_result.success 

317 

318 # Proceed to artifact collection only if build succeeded. 

319 if self._collect_artifacts is not None: 

320 if not self._collect_artifacts( 

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

322 ): 

323 build_result.success = False 

324 

325 # Print size at the absolute end 

326 self._print_build_result(build_result) 

327 return build_result.success 

328 

329 @staticmethod 

330 def _print_build_result(build_result): 

331 build_report = build_result.report() 

332 if build_report: 

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

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

335 print() 

336 print(build_report) 

337 

338 @property 

339 def build_result_report_length(self): 

340 """ 

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

342 """ 

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

344 # string with one line for each utilization category. 

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

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

347 # extra (URAM). 

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

349 # 

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

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

352 # 

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

354 length_of_size_report = 3 + 8 + 1 

355 

356 if self._project.is_netlist_build: 

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

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

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

360 length_of_logic_level_report = 5 + 1 

361 return length_of_size_report + length_of_logic_level_report 

362 

363 return length_of_size_report 

364 

365 

366class BuildProjectOpenWrapper: 

367 

368 """ 

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

370 Mimics a VUnit test object. 

371 """ 

372 

373 def __init__(self, project): 

374 self.name = project.name 

375 self._project = project 

376 

377 # pylint: disable=unused-argument 

378 def run(self, output_path, read_output): 

379 """ 

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

381 """ 

382 this_projects_path = Path(output_path) / "project" 

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

384 

385 

386class BuildRunner(TestRunner): 

387 

388 """ 

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

390 base class, but some behavior is overridden. 

391 """ 

392 

393 def _create_test_mapping_file(self, test_suites): 

394 """ 

395 Do not create this file. 

396 

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

398 """ 

399 

400 def _get_output_path(self, test_suite_name): 

401 """ 

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

403 

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

405 We do not want that necessarily. 

406 """ 

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

408 

409 @staticmethod 

410 def _prepare_test_suite_output_path(output_path): 

411 """ 

412 Create the directory unless it already exists. 

413 

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

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

416 that the user wants to keep. 

417 """ 

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

419 

420 

421class ThreadSafeCollectArtifacts: 

422 

423 """ 

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

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

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

427 

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

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

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

431 """ 

432 

433 def __init__(self, collect_artifacts): 

434 self._collect_artifacts = collect_artifacts 

435 self._lock = Lock() 

436 

437 def collect_artifacts(self, project, output_path): 

438 with self._lock: 

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

440 

441 

442class BuildReport(TestReport): 

443 def add_result(self, *args, **kwargs): 

444 """ 

445 Add a a test result. 

446 

447 Inherited and adapted from the VUnit function. Uses a different Result class. 

448 """ 

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

450 self._test_results[result.name] = result 

451 self._test_names_in_order.append(result.name) 

452 

453 def set_report_length(self, report_length_lines): 

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

455 test_result.set_report_length(report_length_lines) 

456 

457 def print_latest_status(self, total_tests): 

458 """ 

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

460 but other builds may not be finished yet. 

461 

462 Inherited and adapted from the VUnit function: 

463 * Removed support for the "skipped" result. 

464 * Do not use abbreviations in the printout. 

465 * Use f-strings. 

466 """ 

467 result = self._last_test_result() 

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

469 

470 if result.passed: 

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

472 elif result.failed: 

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

474 else: 

475 assert False 

476 

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

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

479 

480 

481class BuildResult(TestResult): 

482 

483 report_length_lines = None 

484 

485 def _print_output(self, printer, num_lines): 

486 """ 

487 Print the last lines from the output file. 

488 """ 

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

490 printer.write(output_tail) 

491 

492 def set_report_length(self, report_length_lines): 

493 self.report_length_lines = report_length_lines 

494 

495 def print_status(self, printer, padding=0): 

496 """ 

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

498 the end when all builds have finished. 

499 

500 Inherited and adapted from the VUnit function. 

501 """ 

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

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

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

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

506 else: 

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

508 # 1. IDE build failure 

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

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

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

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

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

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

515 # than eight size checkers. 

516 self._print_output(printer, num_lines=25) 

517 

518 # Print the regular output from the VUnit class. 

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

520 super().print_status(printer, padding + 2) 

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

522 printer.write("\n")