Coverage for tsfpga/build_project_list.py: 95%
194 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-08-29 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# --------------------------------------------------------------------------------------------------
9from __future__ import annotations
11import fnmatch
12import time
13from abc import ABC, abstractmethod
14from pathlib import Path
15from threading import Lock
16from typing import TYPE_CHECKING, Any, Callable
18from vunit.color_printer import COLOR_PRINTER, NO_COLOR_PRINTER, ColorPrinter
19from vunit.test.list import TestList
20from vunit.test.report import TestReport, TestResult
21from vunit.test.runner import TestRunner
23from tsfpga.system_utils import create_directory, read_last_lines_of_file
25if TYPE_CHECKING:
26 from collections.abc import Iterable
28 from .module_list import ModuleList
29 from .vivado import build_result
30 from .vivado.project import VivadoProject
33class 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 ) -> None:
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(
95 self,
96 projects_path: Path,
97 num_parallel_builds: int,
98 **kwargs: Any, # noqa: ANN401
99 ) -> bool:
100 """
101 Create build project on disk for all the projects in the list.
103 Arguments:
104 projects_path: The projects will be placed here.
105 num_parallel_builds: The number of projects that will be created in parallel.
106 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
108 .. Note::
109 Argument ``project_path`` can not be set, it is set by this class
110 based on the ``project_paths`` argument to this function.
112 Return:
113 True if everything went well.
114 """
115 build_wrappers = []
116 for project in self.projects:
117 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
118 build_wrappers.append(build_wrapper)
120 return self._run_build_wrappers(
121 projects_path=projects_path,
122 build_wrappers=build_wrappers,
123 num_parallel_builds=num_parallel_builds,
124 )
126 def create_unless_exists(
127 self,
128 projects_path: Path,
129 num_parallel_builds: int,
130 **kwargs: Any, # noqa: ANN401
131 ) -> bool:
132 """
133 Create build project for all the projects in the list, unless the project already
134 exists.
136 Arguments:
137 projects_path: The projects will be placed here.
138 num_parallel_builds: The number of projects that will be created in parallel.
139 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
141 .. Note::
142 Argument ``project_path`` can not be set, it is set by this class
143 based on the ``project_paths`` argument to this function.
145 Return:
146 True if everything went well.
147 """
148 build_wrappers = []
149 for project in self.projects:
150 if not self.get_build_project_path(
151 project=project, projects_path=projects_path
152 ).exists():
153 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
154 build_wrappers.append(build_wrapper)
156 if not build_wrappers:
157 # Return straight away if no projects need to be created. To avoid extra
158 # "No tests were run!" printout from creation step that is very misleading.
159 return True
161 return self._run_build_wrappers(
162 projects_path=projects_path,
163 build_wrappers=build_wrappers,
164 num_parallel_builds=num_parallel_builds,
165 )
167 def build(
168 self,
169 projects_path: Path,
170 num_parallel_builds: int,
171 num_threads_per_build: int,
172 output_path: Path | None = None,
173 collect_artifacts: Callable[[VivadoProject, Path], bool] | None = None,
174 **kwargs: Any, # noqa: ANN401
175 ) -> bool:
176 """
177 Build all the projects in the list.
179 Arguments:
180 projects_path: The projects are placed here.
181 num_parallel_builds: The number of projects that will be built in parallel.
182 num_threads_per_build: The number threads that will be used for each
183 parallel build process.
184 output_path: Where the artifacts should be placed.
185 Will default to within the ``projects_path`` if not set.
186 collect_artifacts: Callback to collect artifacts.
187 Takes two named arguments:
189 | **project** (:class:`.VivadoProject`): The project that is being built.
191 | **output_path** (pathlib.Path): Where the build artifacts should be placed.
193 | Must return True.
194 kwargs: Other arguments as accepted by :meth:`.VivadoProject.build`.
196 .. Note::
197 Argument ``project_path`` can not be set, it is set by this class
198 based on the ``project_paths`` argument to this function.
200 Argument ``num_threads`` is set by the ``num_threads_per_build``
201 argument to this function. This naming difference is done to avoid
202 confusion with regards to ``num_parallel_builds``.
204 Return:
205 True if everything went well.
206 """
207 if collect_artifacts:
208 thread_safe_collect_artifacts = ThreadSafeCollectArtifacts(
209 collect_artifacts=collect_artifacts
210 ).collect_artifacts
211 else:
212 thread_safe_collect_artifacts = None
214 build_wrappers = []
215 for project in self.projects:
216 project_output_path = self.get_build_project_output_path(
217 project=project, projects_path=projects_path, output_path=output_path
218 )
220 build_wrapper = BuildProjectBuildWrapper(
221 project=project,
222 collect_artifacts=thread_safe_collect_artifacts,
223 output_path=project_output_path,
224 num_threads=num_threads_per_build,
225 **kwargs,
226 )
227 build_wrappers.append(build_wrapper)
229 return self._run_build_wrappers(
230 projects_path=projects_path,
231 build_wrappers=build_wrappers,
232 num_parallel_builds=num_parallel_builds,
233 )
235 @staticmethod
236 def get_build_project_path(project: VivadoProject, projects_path: Path) -> Path:
237 """
238 Find where the project files for a specific project will be placed.
239 Arguments are the same as for :meth:`.create`.
240 """
241 return projects_path / project.name / "project"
243 @staticmethod
244 def get_build_project_output_path(
245 project: VivadoProject, projects_path: Path, output_path: Path | None = None
246 ) -> Path:
247 """
248 Find where build artifacts will be placed for a project.
249 Arguments are the same as for :meth:`.build`.
250 """
251 if output_path:
252 return output_path.resolve() / project.name
254 return projects_path / project.name
256 def open(self, projects_path: Path) -> bool:
257 """
258 Open the projects in EDA GUI.
260 Arguments:
261 projects_path: The projects are placed here.
263 Return:
264 True if everything went well.
265 """
266 build_wrappers = [BuildProjectOpenWrapper(project=project) for project in self.projects]
268 return self._run_build_wrappers(
269 projects_path=projects_path,
270 build_wrappers=build_wrappers,
271 # For open there is no performance limitation. Set a high value.
272 num_parallel_builds=20,
273 )
275 def _run_build_wrappers(
276 self,
277 projects_path: Path,
278 build_wrappers: list[BuildProjectCreateWrapper]
279 | list[BuildProjectBuildWrapper]
280 | list[BuildProjectOpenWrapper],
281 num_parallel_builds: int,
282 ) -> bool:
283 if not build_wrappers:
284 # Return straight away if no builds are supplied
285 return True
287 start_time = time.time()
289 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER
290 report = BuildReport(printer=color_printer)
292 test_list = TestList()
293 for build_wrapper in build_wrappers:
294 test_list.add_test(build_wrapper)
296 verbosity = BuildRunner.VERBOSITY_QUIET
297 test_runner = BuildRunner(
298 report=report,
299 output_path=projects_path,
300 verbosity=verbosity,
301 num_threads=num_parallel_builds,
302 )
303 test_runner.run(test_list)
305 all_builds_ok: bool = report.all_ok()
306 report.set_real_total_time(time.time() - start_time)
308 # True if the builds are for the "build" step (not "create" or "open")
309 builds_are_build_step = isinstance(build_wrappers[0], BuildProjectBuildWrapper)
310 # If we are building, we should print the summary that is at the end of the console output.
311 # (however if we are creating or opening a project we should not print anything extra).
312 # However if anything has failed, we should also print.
313 if builds_are_build_step:
314 # The length of the build summary depends on if we are working with netlist builds or
315 # regular ones, so set the length given by one of the project objects.
316 report_length_lines = build_wrappers[0].build_result_report_length
317 report.set_report_length(report_length_lines=report_length_lines)
319 # If all are OK then we should print the resource utilization numbers.
320 # If not, then we print a few last lines of the log output.
321 if builds_are_build_step or not all_builds_ok:
322 report.print_str()
324 return all_builds_ok
326 def _iterate_projects(
327 self, project_filters: list[str], include_netlist_not_top_builds: bool
328 ) -> Iterable[VivadoProject]:
329 available_projects = []
330 for module in self._modules:
331 available_projects += module.get_build_projects()
333 for project in available_projects:
334 if project.is_netlist_build == include_netlist_not_top_builds:
335 if not project_filters:
336 yield project
338 else:
339 for project_filter in project_filters:
340 if fnmatch.filter([project.name], project_filter):
341 yield project
343 # Do not continue with further filters if we have already matched this
344 # project.
345 # Multiple filters might match the same project, and multiple objects
346 # of the same project will break build
347 break
350class BuildProjectWrapper(ABC):
351 """
352 Mimics a VUnit test case object.
353 """
355 def get_seed(self) -> str:
356 """
357 Required since VUnit version 5.0.0.dev6, where a 'get_seed' method was added
358 to the 'TestSuiteWrapper' class, which calls a 'get_seed' method expected to be implemented
359 in the test case object.
360 This mechanism is not used by tsfpga, but is required in order to avoid errors.
361 Adding a dummy implementation like this makes sure it works with older as well as newer
362 versions of VUnit.
363 """
364 return ""
366 @abstractmethod
367 def run(
368 self,
369 output_path: Path,
370 read_output: Any, # noqa: ANN401
371 ) -> bool:
372 pass
375class BuildProjectCreateWrapper(BuildProjectWrapper):
376 """
377 Wrapper to create a build project, for usage in the build runner.
378 """
380 def __init__(
381 self,
382 project: VivadoProject,
383 **kwargs: Any, # noqa: ANN401
384 ) -> None:
385 self.name = project.name
386 self._project = project
387 self._create_arguments = kwargs
389 def run(
390 self,
391 output_path: Path,
392 read_output: Any, # noqa: ANN401, ARG002
393 ) -> bool:
394 """
395 Argument 'read_output' sent by VUnit test runner is unused by us.
396 """
397 this_project_path = Path(output_path) / "project"
398 return self._project.create(project_path=this_project_path, **self._create_arguments)
401class BuildProjectBuildWrapper(BuildProjectWrapper):
402 """
403 Wrapper to build a project, for usage in the build runner.
404 """
406 def __init__(
407 self,
408 project: VivadoProject,
409 collect_artifacts: Callable[..., bool] | None,
410 **kwargs: Any, # noqa: ANN401
411 ) -> None:
412 self.name = project.name
413 self._project = project
414 self._collect_artifacts = collect_artifacts
415 self._build_arguments = kwargs
417 def run(
418 self,
419 output_path: Path,
420 read_output: Any, # noqa: ANN401, ARG002
421 ) -> bool:
422 """
423 Argument 'read_output' sent by VUnit test runner is unused by us.
424 """
425 this_project_path = Path(output_path) / "project"
426 build_result = self._project.build(project_path=this_project_path, **self._build_arguments)
428 if not build_result.success:
429 self._print_build_result(build_result=build_result)
430 return build_result.success
432 # Proceed to artifact collection only if build succeeded.
433 if self._collect_artifacts is not None:
434 build_result.success &= self._collect_artifacts(
435 project=self._project, output_path=self._build_arguments["output_path"]
436 )
438 # Print size at the absolute end.
439 self._print_build_result(build_result=build_result)
440 return build_result.success
442 @staticmethod
443 def _print_build_result(build_result: build_result.BuildResult) -> None:
444 build_report = build_result.report()
445 if build_report:
446 # Add an empty line before the build result report, to have margin in how many lines are
447 # printed. See the comments in BuildResult for an explanation.
448 print()
449 print(build_report)
451 @property
452 def build_result_report_length(self) -> int:
453 """
454 The number of lines in the build_result report from this project.
455 """
456 # The size summary, as returned by tsfpga.vivado.project.BuildResult is a JSON formatted
457 # string with one line for each utilization category.
458 # For Xilinx 7 series, there are 8 categories (Total LUTs, Logic LUTs, LUTRAMs,
459 # SRLs, FFs, RAMB36, RAMB18, DSP Blocks). For UltraScale series there is one
460 # extra (URAM).
461 # Additionally, the size summary contains three extra lines for JSON braces and a title.
462 #
463 # This value is enough lines so the whole summary gets printed to console.
464 # For 7 series, this will mean an extra blank line before the summary.
465 #
466 # This is a hack. Works for now, but is far from reliable.
467 length_of_size_report = 3 + 8 + 1
469 if self._project.is_netlist_build:
470 # The logic level distribution report is five lines, plus a title line.
471 # This report is only printed for netlist builds, where there is no configured clock
472 # present. If there were many clocks present in the build, the report would be longer.
473 length_of_logic_level_report = 5 + 1
474 return length_of_size_report + length_of_logic_level_report
476 return length_of_size_report
479class BuildProjectOpenWrapper(BuildProjectWrapper):
480 """
481 Wrapper to open a build project, for usage in the build runner.
482 """
484 def __init__(self, project: VivadoProject) -> None:
485 self.name = project.name
486 self._project = project
488 def run(
489 self,
490 output_path: Path,
491 read_output: Any, # noqa: ANN401, ARG002
492 ) -> bool:
493 """
494 Argument 'read_output' sent by VUnit test runner is unused by us.
495 """
496 this_project_path = Path(output_path) / "project"
497 return self._project.open(project_path=this_project_path)
500class BuildRunner(TestRunner):
501 """
502 Build runner that mimics a VUnit TestRunner. Most things are used as they are in the
503 base class, but some behavior is overridden.
504 """
506 def _create_test_mapping_file(
507 self,
508 test_suites: Any, # noqa: ANN401
509 ) -> None:
510 """
511 Overloaded from super class.
513 Do not create this file.
515 We do not need it since folder name is the same as project name.
516 """
518 def _get_output_path(self, test_suite_name: str) -> str:
519 """
520 Overloaded from super class.
522 Output folder name is the same as the project name.
524 Original function adds a hash at the end of the folder name.
525 We do not want that necessarily.
526 """
527 return str(Path(self._output_path) / test_suite_name)
529 @staticmethod
530 def _prepare_test_suite_output_path(output_path: str) -> None:
531 """
532 Overloaded from super class.
534 Create the directory unless it already exists.
536 Original function wipes the path before running a test. We do not want to do that
537 since e.g. a Vivado project takes a long time to create and might contain a state
538 that the user wants to keep.
539 """
540 create_directory(Path(output_path), empty=False)
543class ThreadSafeCollectArtifacts:
544 """
545 A thread-safe wrapper around a user-supplied function that makes sure the function
546 is not launched more than once at the same time. When two builds finish at the
547 same time, race conditions can arise depending on what the function does.
549 Note that this is a VERY fringe case, since builds usually take >20 minutes, and the
550 collection probably only takes a few seconds. But it happens sometimes with the tsfpga
551 example projects which are identical and quite fast (roughly three minutes).
552 """
554 def __init__(self, collect_artifacts: Callable[[VivadoProject, Path], bool]) -> None:
555 self._collect_artifacts = collect_artifacts
556 self._lock = Lock()
558 def collect_artifacts(self, project: VivadoProject, output_path: Path) -> bool:
559 with self._lock:
560 return self._collect_artifacts(project=project, output_path=output_path)
563class BuildReport(TestReport):
564 def add_result(
565 self,
566 *args: Any, # noqa: ANN401
567 **kwargs: Any, # noqa: ANN401
568 ) -> None:
569 """
570 Overloaded from super class.
572 Add a a test result.
574 Uses a different Result class than the super method.
575 """
576 result = BuildResult(*args, **kwargs)
577 self._test_results[result.name] = result
578 self._test_names_in_order.append(result.name)
580 def set_report_length(self, report_length_lines: int) -> None:
581 """
582 Set the report length for all test results that have been added to the report.
583 """
584 for test_result in self._test_results.values():
585 test_result.set_report_length(report_length_lines)
587 def print_latest_status(self, total_tests: int) -> None:
588 """
589 Overloaded from super class.
591 This method is called for each build when it should print its result just as it finished,
592 but other builds may not be finished yet.
594 Inherited and adapted from the VUnit function:
595 * Removed support for the "skipped" result.
596 * Do not use abbreviations in the printout.
597 * Use f-strings.
598 """
599 result = self._last_test_result()
600 passed, failed, _ = self._split()
602 if result.passed:
603 self._printer.write("pass", fg="gi")
604 elif result.failed:
605 self._printer.write("fail", fg="ri")
606 else:
607 raise AssertionError
609 count_summary = f"pass={len(passed)} fail={len(failed)} total={total_tests}"
610 self._printer.write(f" ({count_summary}) {result.name} ({result.time:.1f} seconds)\n")
613class BuildResult(TestResult):
614 report_length_lines = None
616 def _print_output(
617 self,
618 printer: ColorPrinter,
619 num_lines: int,
620 ) -> None:
621 """
622 Print the last lines from the output file.
623 """
624 output_tail = read_last_lines_of_file(Path(self._output_file_name), num_lines=num_lines)
625 printer.write(output_tail)
627 def set_report_length(self, report_length_lines: int) -> None:
628 """
629 Set how many lines shall be printed when this result is printed.
630 """
631 self.report_length_lines = report_length_lines
633 def print_status(
634 self,
635 printer: ColorPrinter,
636 padding: int = 0,
637 **kwargs: dict[str, Any],
638 ) -> None:
639 """
640 Overloaded from super class.
642 This method is called for each build when it should print its result in the "Summary" at
643 the end when all builds have finished.
645 Inherited and adapted from the VUnit function.
647 Note that a ``max_time`` integer argument is added in VUnit >4.7.0, but at the time of
648 writing this is un-released on the VUnit ``master`` branch.
649 In order to be compatible with both older and newer versions, we use ``**kwargs`` for this.
650 """
651 if self.passed and self.report_length_lines is not None:
652 # Build passed, print build summary of the specified length. The length is only
653 # set if this is a "build" result (not "create" or "open").
654 self._print_output(printer=printer, num_lines=self.report_length_lines)
655 else:
656 # The build failed, which can either be caused by
657 # 1. IDE build failure
658 # 2. IDE build succeeded, but post build hook, or size checkers failed.
659 # 3. Other python error (directory already exists, ...)
660 # In the case of IDE build failed, we want a significant portion of the output, to be
661 # able to see an indication of what failed. In the case of size checkers, we want to see
662 # all the printouts from all checkers, to see which one failed. Since there are at most
663 # eight resource categories, it is reasonable to assume that there will never be more
664 # than eight size checkers.
665 self._print_output(printer=printer, num_lines=25)
667 # Print the regular output from the VUnit class.
668 # A little extra margin between build name and execution time makes the output more readable
669 super().print_status(printer=printer, padding=padding + 2, **kwargs)
670 # Add an empty line between each build, for readability.
671 printer.write("\n")