Coverage for tsfpga/build_project_list.py: 94%
187 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-10 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# --------------------------------------------------------------------------------------------------
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
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
22# First party libraries
23from tsfpga.system_utils import create_directory, read_last_lines_of_file
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
32class BuildProjectList:
34 """
35 Interface to handle a list of FPGA build projects.
36 Enables building many projects in parallel.
37 """
39 def __init__(
40 self,
41 modules: "ModuleList",
42 project_filters: list[str],
43 include_netlist_not_top_builds: bool = False,
44 no_color: bool = False,
45 ):
46 """
47 Arguments:
48 modules: Module objects that can define build projects.
49 project_filters: Project name filters. Can use wildcards (*). Leave empty for all.
50 include_netlist_not_top_builds: Set True to get only netlist builds,
51 instead of only top level builds.
52 no_color: Disable color in printouts.
53 """
54 self._modules = modules
55 self._no_color = no_color
57 self.projects = list(
58 self._iterate_projects(
59 project_filters=project_filters,
60 include_netlist_not_top_builds=include_netlist_not_top_builds,
61 )
62 )
64 if not self.projects:
65 print(f"No projects matched this filter: {' '.join(project_filters)}")
67 def __str__(self) -> str:
68 """
69 Returns a string with a description list of the projects.
71 Will print some information about each project (name, generics, part, ...) so can become
72 long if there are many projects present.
73 An alternative in that case would be :meth:`.get_short_str`.
74 """
75 result = "\n".join([str(project) for project in self.projects])
76 result += "\n"
77 result += "\n"
78 result += f"Listed {len(self.projects)} builds"
80 return result
82 def get_short_str(self) -> str:
83 """
84 Returns a short string with a description list of the projects.
86 This is an alternative function that is more compact than ``__str__``.
87 """
88 result = "\n".join([project.name for project in self.projects])
89 result += "\n"
90 result += f"Listed {len(self.projects)} builds"
92 return result
94 def create(self, projects_path: Path, num_parallel_builds: int, **kwargs: Any) -> bool:
95 """
96 Create build project on disk for all the projects in the list.
98 Arguments:
99 projects_path: The projects will be placed here.
100 num_parallel_builds: The number of projects that will be created in parallel.
101 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
103 .. Note::
104 Argument ``project_path`` can not be set, it is set by this class
105 based on the ``project_paths`` argument to this function.
107 Return:
108 True if everything went well.
109 """
110 build_wrappers = []
111 for project in self.projects:
112 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
113 build_wrappers.append(build_wrapper)
115 return self._run_build_wrappers(
116 projects_path=projects_path,
117 build_wrappers=build_wrappers,
118 num_parallel_builds=num_parallel_builds,
119 )
121 def create_unless_exists(
122 self, projects_path: Path, num_parallel_builds: int, **kwargs: Any
123 ) -> bool:
124 """
125 Create build project for all the projects in the list, unless the project already
126 exists.
128 Arguments:
129 projects_path: The projects will be placed here.
130 num_parallel_builds: The number of projects that will be created in parallel.
131 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
133 .. Note::
134 Argument ``project_path`` can not be set, it is set by this class
135 based on the ``project_paths`` argument to this function.
137 Return:
138 True if everything went well.
139 """
140 build_wrappers = []
141 for project in self.projects:
142 if not (projects_path / project.name / "project").exists():
143 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
144 build_wrappers.append(build_wrapper)
146 if not build_wrappers:
147 # Return straight away if no projects need to be created. To avoid extra
148 # "No tests were run!" printout from creation step that is very misleading.
149 return True
151 return self._run_build_wrappers(
152 projects_path=projects_path,
153 build_wrappers=build_wrappers,
154 num_parallel_builds=num_parallel_builds,
155 )
157 def build(
158 self,
159 projects_path: Path,
160 num_parallel_builds: int,
161 num_threads_per_build: int,
162 output_path: Optional[Path] = None,
163 collect_artifacts: Optional[Callable[["VivadoProject", Path], bool]] = None,
164 **kwargs: Any,
165 ) -> bool:
166 """
167 Build all the projects in the list.
169 Arguments:
170 projects_path: The projects will be placed here.
171 num_parallel_builds: The number of projects that will be built in parallel.
172 num_threads_per_build: The number threads that will be used for each
173 parallel build process.
174 output_path: Where the artifacts should be placed.
175 collect_artifacts: Callback to collect artifacts.
176 Takes two named arguments:
178 | **project** (:class:`.VivadoProject`): The project that is being built.
180 | **output_path** (pathlib.Path): Where the artifacts should be placed.
183 | Must return True.
184 kwargs: Other arguments as accepted by :meth:`.VivadoProject.build`.
186 .. Note::
187 Argument ``project_path`` can not be set, it is set by this class
188 based on the ``project_paths`` argument to this function.
190 Argument ``num_threads`` is set by the ``num_threads_per_build``
191 argument to this function. This naming difference is done to avoid
192 confusion with regards to ``num_parallel_builds``.
195 Return:
196 True if everything went well.
197 """
198 if collect_artifacts:
199 thread_safe_collect_artifacts = ThreadSafeCollectArtifacts(
200 collect_artifacts
201 ).collect_artifacts
202 else:
203 thread_safe_collect_artifacts = None
205 build_wrappers = []
206 for project in self.projects:
207 if output_path:
208 this_projects_output_path = output_path.resolve() / project.name
209 else:
210 this_projects_output_path = projects_path / project.name
212 build_wrapper = BuildProjectBuildWrapper(
213 project,
214 output_path=this_projects_output_path,
215 collect_artifacts=thread_safe_collect_artifacts,
216 num_threads=num_threads_per_build,
217 **kwargs,
218 )
219 build_wrappers.append(build_wrapper)
221 return self._run_build_wrappers(
222 projects_path=projects_path,
223 build_wrappers=build_wrappers,
224 num_parallel_builds=num_parallel_builds,
225 )
227 def open(self, projects_path: Path) -> bool:
228 """
229 Open the projects in EDA GUI.
231 Arguments:
232 projects_path: The projects are placed here.
234 Return:
235 True if everything went well.
236 """
237 build_wrappers = []
238 for project in self.projects:
239 build_wrappers.append(BuildProjectOpenWrapper(project))
241 return self._run_build_wrappers(
242 projects_path=projects_path,
243 build_wrappers=build_wrappers,
244 # For open there is no performance limitation. Set a high value.
245 num_parallel_builds=20,
246 )
248 def _run_build_wrappers(
249 self,
250 projects_path: Path,
251 build_wrappers: Union[
252 list["BuildProjectCreateWrapper"],
253 list["BuildProjectBuildWrapper"],
254 list["BuildProjectOpenWrapper"],
255 ],
256 num_parallel_builds: int,
257 ) -> bool:
258 if not build_wrappers:
259 # Return straight away if no builds are supplied
260 return True
262 start_time = time.time()
264 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER
265 report = BuildReport(printer=color_printer)
267 test_list = TestList()
268 for build_wrapper in build_wrappers:
269 test_list.add_test(build_wrapper)
271 verbosity = BuildRunner.VERBOSITY_QUIET
272 test_runner = BuildRunner(
273 report=report,
274 output_path=projects_path,
275 verbosity=verbosity,
276 num_threads=num_parallel_builds,
277 )
278 test_runner.run(test_list)
280 all_builds_ok: bool = report.all_ok()
281 report.set_real_total_time(time.time() - start_time)
283 # True if the builds are for the "build" step (not "create" or "open")
284 builds_are_build_step = isinstance(build_wrappers[0], BuildProjectBuildWrapper)
285 # If we are building, we should print the summary that is at the end of the console output.
286 # (however if we are creating or opening a project we should not print anything extra).
287 # However if anything has failed, we should also print.
288 if builds_are_build_step:
289 # The length of the build summary depends on if we are working with netlist builds or
290 # regular ones, so set the length given by one of the project objects.
291 # Ignore typing error that 'Create' or 'Open' build wrappers do not have
292 # 'build_result_report_length'. We only end up here if it is a Build.
293 report_length_lines = build_wrappers[
294 0
295 ].build_result_report_length # type: ignore[union-attr]
296 report.set_report_length(report_length_lines=report_length_lines)
298 # If all are OK then we should print the resource utilization numbers.
299 # If not, then we print a few last lines of the log output.
300 if builds_are_build_step or not all_builds_ok:
301 report.print_str()
303 return all_builds_ok
305 def _iterate_projects(
306 self, project_filters: list[str], include_netlist_not_top_builds: bool
307 ) -> Iterable["VivadoProject"]:
308 available_projects = []
309 for module in self._modules:
310 available_projects += module.get_build_projects()
312 for project in available_projects:
313 if project.is_netlist_build == include_netlist_not_top_builds:
314 if not project_filters:
315 yield project
317 else:
318 for project_filter in project_filters:
319 if fnmatch.filter([project.name], project_filter):
320 yield project
322 # Do not continue with further filters if we have already matched this
323 # project.
324 # Multiple filters might match the same project, and multiple objects
325 # of the same project will break build
326 break
329class BuildProjectCreateWrapper:
331 """
332 Wrapper to create a build project, for usage in the build runner.
333 Mimics a VUnit test object.
334 """
336 def __init__(self, project: "VivadoProject", **kwargs: Any) -> None:
337 self.name = project.name
338 self._project = project
339 self._create_arguments = kwargs
341 def run(self, output_path: Path, read_output: Any) -> bool: # pylint: disable=unused-argument
342 """
343 VUnit test runner sends another argument "read_output" which we don't use.
344 """
345 this_projects_path = Path(output_path) / "project"
346 return self._project.create(project_path=this_projects_path, **self._create_arguments)
349class BuildProjectBuildWrapper:
351 """
352 Wrapper to build a project, for usage in the build runner.
353 Mimics a VUnit test object.
354 """
356 def __init__(
357 self,
358 project: "VivadoProject",
359 collect_artifacts: Optional[Callable[..., bool]],
360 **kwargs: Any,
361 ) -> None:
362 self.name = project.name
363 self._project = project
364 self._collect_artifacts = collect_artifacts
365 self._build_arguments = kwargs
367 def run(self, output_path: Path, read_output: Any) -> bool: # pylint: disable=unused-argument
368 """
369 VUnit test runner sends another argument "read_output" which we don't use.
370 """
371 this_projects_path = Path(output_path) / "project"
372 build_result = self._project.build(project_path=this_projects_path, **self._build_arguments)
374 if not build_result.success:
375 self._print_build_result(build_result)
376 return build_result.success
378 # Proceed to artifact collection only if build succeeded.
379 if self._collect_artifacts:
380 if not self._collect_artifacts(
381 project=self._project, output_path=self._build_arguments["output_path"]
382 ):
383 build_result.success = False
385 # Print size at the absolute end
386 self._print_build_result(build_result=build_result)
387 return build_result.success
389 @staticmethod
390 def _print_build_result(build_result: "build_result.BuildResult") -> None:
391 build_report = build_result.report()
392 if build_report:
393 # Add an empty line before the build result report, to have margin in how many lines are
394 # printed. See the comments in BuildResult for an explanation.
395 print()
396 print(build_report)
398 @property
399 def build_result_report_length(self) -> int:
400 """
401 The number of lines in the build_result report from this project.
402 """
403 # The size summary, as returned by tsfpga.vivado.project.BuildResult is a JSON formatted
404 # string with one line for each utilization category.
405 # For Xilinx 7 series, there are 8 categories (Total LUTs, Logic LUTs, LUTRAMs,
406 # SRLs, FFs, RAMB36, RAMB18, DSP Blocks). For UltraScale series there is one
407 # extra (URAM).
408 # Additionally, the size summary contains three extra lines for JSON braces and a title.
409 #
410 # This value is enough lines so the whole summary gets printed to console.
411 # For 7 series, this will mean an extra blank line before the summary.
412 #
413 # This is a hack. Works for now, but is far from reliable.
414 length_of_size_report = 3 + 8 + 1
416 if self._project.is_netlist_build:
417 # The logic level distribution report is five lines, plus a title line.
418 # This report is only printed for netlist builds, where there is no configured clock
419 # present. If there were many clocks present in the build, the report would be longer.
420 length_of_logic_level_report = 5 + 1
421 return length_of_size_report + length_of_logic_level_report
423 return length_of_size_report
426class BuildProjectOpenWrapper:
428 """
429 Wrapper to open a build project, for usage in the build runner.
430 Mimics a VUnit test object.
431 """
433 def __init__(self, project: "VivadoProject") -> None:
434 self.name = project.name
435 self._project = project
437 def run(self, output_path: Path, read_output: Any) -> bool: # pylint: disable=unused-argument
438 """
439 VUnit test runner sends another argument "read_output" which we don't use.
440 """
441 this_projects_path = Path(output_path) / "project"
442 return self._project.open(project_path=this_projects_path)
445class BuildRunner(TestRunner):
447 """
448 Build runner that mimics a VUnit TestRunner. Most things are used as they are in the
449 base class, but some behavior is overridden.
450 """
452 def _create_test_mapping_file(self, test_suites: Any) -> None:
453 """
454 Overloaded from super class.
456 Do not create this file.
458 We do not need it since folder name is the same as project name.
459 """
461 def _get_output_path(self, test_suite_name: str) -> str:
462 """
463 Overloaded from super class.
465 Output folder name is the same as the project name.
467 Original function adds a hash at the end of the folder name.
468 We do not want that necessarily.
469 """
470 return str(Path(self._output_path) / test_suite_name)
472 @staticmethod
473 def _prepare_test_suite_output_path(output_path: str) -> None:
474 """
475 Overloaded from super class.
477 Create the directory unless it already exists.
479 Original function wipes the path before running a test. We do not want to do that
480 since e.g. a Vivado project takes a long time to create and might contain a state
481 that the user wants to keep.
482 """
483 create_directory(Path(output_path), empty=False)
486class ThreadSafeCollectArtifacts:
488 """
489 A thread-safe wrapper around a user-supplied function that makes sure the function
490 is not launched more than once at the same time. When two builds finish at the
491 same time, race conditions can arise depending on what the function does.
493 Note that this is a VERY fringe case, since builds usually take >20 minutes, and the
494 collection probably only takes a few seconds. But it happens sometimes with the tsfpga
495 example projects which are identical and quite fast (roughly three minutes).
496 """
498 def __init__(self, collect_artifacts: Callable[..., bool]) -> None:
499 self._collect_artifacts = collect_artifacts
500 self._lock = Lock()
502 def collect_artifacts(self, project: "VivadoProject", output_path: Path) -> bool:
503 with self._lock:
504 return self._collect_artifacts(project=project, output_path=output_path)
507class BuildReport(TestReport):
508 def add_result(self, *args: Any, **kwargs: Any) -> None:
509 """
510 Overloaded from super class.
512 Add a a test result.
514 Uses a different Result class than the super method.
515 """
516 result = BuildResult(*args, **kwargs)
517 self._test_results[result.name] = result
518 self._test_names_in_order.append(result.name)
520 def set_report_length(self, report_length_lines: int) -> None:
521 """
522 Set the report length for all test results that have been added to the report.
523 """
524 for test_result in self._test_results.values():
525 test_result.set_report_length(report_length_lines)
527 def print_latest_status(self, total_tests: int) -> None:
528 """
529 Overloaded from super class.
531 This method is called for each build when it should print its result just as it finished,
532 but other builds may not be finished yet.
534 Inherited and adapted from the VUnit function:
535 * Removed support for the "skipped" result.
536 * Do not use abbreviations in the printout.
537 * Use f-strings.
538 """
539 result = self._last_test_result()
540 passed, failed, _ = self._split()
542 if result.passed:
543 self._printer.write("pass", fg="gi")
544 elif result.failed:
545 self._printer.write("fail", fg="ri")
546 else:
547 assert False
549 count_summary = f"pass={len(passed)} fail={len(failed)} total={total_tests}"
550 self._printer.write(f" ({count_summary}) {result.name} ({result.time:.1f} seconds)\n")
553class BuildResult(TestResult):
554 report_length_lines = None
556 def _print_output(self, printer: Any, num_lines: int) -> None:
557 """
558 Print the last lines from the output file.
560 The ``printer`` argument should of type ``ColorPrinter`` from VUnit.
561 """
562 output_tail = read_last_lines_of_file(Path(self._output_file_name), num_lines=num_lines)
563 printer.write(output_tail)
565 def set_report_length(self, report_length_lines: int) -> None:
566 """
567 Set how many lines shall be printed when this result is printed.
568 """
569 self.report_length_lines = report_length_lines
571 def print_status(self, printer: Any, padding: int = 0) -> None:
572 """
573 Overloaded from super class.
575 The ``printer`` argument should of type ``ColorPrinter`` from VUnit.
577 This method is called for each build when it should print its result in the "Summary" at
578 the end when all builds have finished.
580 Inherited and adapted from the VUnit function.
581 """
582 if self.passed and self.report_length_lines is not None:
583 # Build passed, print build summary of the specified length. The length is only
584 # set if this is a "build" result (not "create" or "open").
585 self._print_output(printer=printer, num_lines=self.report_length_lines)
586 else:
587 # The build failed, which can either be caused by
588 # 1. IDE build failure
589 # 2. IDE build succeeded, but post build hook, or size checkers failed.
590 # 3. Other python error (directory already exists, ...)
591 # In the case of IDE build failed, we want a significant portion of the output, to be
592 # able to see an indication of what failed. In the case of size checkers, we want to see
593 # all the printouts from all checkers, to see which one failed. Since there are at most
594 # eight resource categories, it is reasonable to assume that there will never be more
595 # than eight size checkers.
596 self._print_output(printer=printer, num_lines=25)
598 # Print the regular output from the VUnit class.
599 # A little extra margin between build name and execution time makes the output more readable
600 super().print_status(printer=printer, padding=padding + 2)
601 # Add an empty line between each build, for readability.
602 printer.write("\n")