Coverage for tsfpga/build_project_list.py: 94%
190 statements
« prev ^ index » next coverage.py v7.6.9, created at 2024-12-20 20:52 +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:
33 """
34 Interface to handle a list of FPGA build projects.
35 Enables building many projects in parallel.
36 """
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
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 )
63 if not self.projects:
64 print(f"No projects matched this filter: {' '.join(project_filters)}")
66 def __str__(self) -> str:
67 """
68 Returns a string with a description list of the projects.
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"
79 return result
81 def get_short_str(self) -> str:
82 """
83 Returns a short string with a description list of the projects.
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"
91 return result
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.
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`.
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.
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)
114 return self._run_build_wrappers(
115 projects_path=projects_path,
116 build_wrappers=build_wrappers,
117 num_parallel_builds=num_parallel_builds,
118 )
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.
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`.
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.
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)
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
150 return self._run_build_wrappers(
151 projects_path=projects_path,
152 build_wrappers=build_wrappers,
153 num_parallel_builds=num_parallel_builds,
154 )
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.
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:
178 | **project** (:class:`.VivadoProject`): The project that is being built.
180 | **output_path** (pathlib.Path): Where the build artifacts should be placed.
182 | Must return True.
183 kwargs: Other arguments as accepted by :meth:`.VivadoProject.build`.
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.
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``.
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
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 )
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)
218 return self._run_build_wrappers(
219 projects_path=projects_path,
220 build_wrappers=build_wrappers,
221 num_parallel_builds=num_parallel_builds,
222 )
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
235 return projects_path / project.name
237 def open(self, projects_path: Path) -> bool:
238 """
239 Open the projects in EDA GUI.
241 Arguments:
242 projects_path: The projects are placed here.
244 Return:
245 True if everything went well.
246 """
247 build_wrappers = []
248 for project in self.projects:
249 build_wrappers.append(BuildProjectOpenWrapper(project))
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 )
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
272 start_time = time.time()
274 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER
275 report = BuildReport(printer=color_printer)
277 test_list = TestList()
278 for build_wrapper in build_wrappers:
279 test_list.add_test(build_wrapper)
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)
290 all_builds_ok: bool = report.all_ok()
291 report.set_real_total_time(time.time() - start_time)
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)
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()
313 return all_builds_ok
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()
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
327 else:
328 for project_filter in project_filters:
329 if fnmatch.filter([project.name], project_filter):
330 yield project
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
339class BuildProjectCreateWrapper:
340 """
341 Wrapper to create a build project, for usage in the build runner.
342 Mimics a VUnit test object.
343 """
345 def __init__(self, project: "VivadoProject", **kwargs: Any) -> None:
346 self.name = project.name
347 self._project = project
348 self._create_arguments = kwargs
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)
358class BuildProjectBuildWrapper:
359 """
360 Wrapper to build a project, for usage in the build runner.
361 Mimics a VUnit test object.
362 """
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
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)
382 if not build_result.success:
383 self._print_build_result(build_result)
384 return build_result.success
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
393 # Print size at the absolute end
394 self._print_build_result(build_result=build_result)
395 return build_result.success
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)
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
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
431 return length_of_size_report
434class BuildProjectOpenWrapper:
435 """
436 Wrapper to open a build project, for usage in the build runner.
437 Mimics a VUnit test object.
438 """
440 def __init__(self, project: "VivadoProject") -> None:
441 self.name = project.name
442 self._project = project
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)
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 """
458 def _create_test_mapping_file(self, test_suites: Any) -> None:
459 """
460 Overloaded from super class.
462 Do not create this file.
464 We do not need it since folder name is the same as project name.
465 """
467 def _get_output_path(self, test_suite_name: str) -> str:
468 """
469 Overloaded from super class.
471 Output folder name is the same as the project name.
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)
478 @staticmethod
479 def _prepare_test_suite_output_path(output_path: str) -> None:
480 """
481 Overloaded from super class.
483 Create the directory unless it already exists.
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)
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.
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 """
503 def __init__(self, collect_artifacts: Callable[..., bool]) -> None:
504 self._collect_artifacts = collect_artifacts
505 self._lock = Lock()
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)
512class BuildReport(TestReport):
513 def add_result(self, *args: Any, **kwargs: Any) -> None:
514 """
515 Overloaded from super class.
517 Add a a test result.
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)
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)
532 def print_latest_status(self, total_tests: int) -> None:
533 """
534 Overloaded from super class.
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.
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()
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
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")
558class BuildResult(TestResult):
559 report_length_lines = None
561 def _print_output(self, printer: Any, num_lines: int) -> None:
562 """
563 Print the last lines from the output file.
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)
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
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.
583 The ``printer`` argument should of type ``ColorPrinter`` from VUnit.
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.
588 Inherited and adapted from the VUnit function.
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)
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")