Coverage for tsfpga/build_project_list.py: 95%
191 statements
« prev ^ index » next coverage.py v7.10.2, created at 2025-08-07 20:55 +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 (projects_path / project.name / "project").exists():
151 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
152 build_wrappers.append(build_wrapper)
154 if not build_wrappers:
155 # Return straight away if no projects need to be created. To avoid extra
156 # "No tests were run!" printout from creation step that is very misleading.
157 return True
159 return self._run_build_wrappers(
160 projects_path=projects_path,
161 build_wrappers=build_wrappers,
162 num_parallel_builds=num_parallel_builds,
163 )
165 def build(
166 self,
167 projects_path: Path,
168 num_parallel_builds: int,
169 num_threads_per_build: int,
170 output_path: Path | None = None,
171 collect_artifacts: Callable[[VivadoProject, Path], bool] | None = None,
172 **kwargs: Any, # noqa: ANN401
173 ) -> bool:
174 """
175 Build all the projects in the list.
177 Arguments:
178 projects_path: The projects are placed here.
179 num_parallel_builds: The number of projects that will be built in parallel.
180 num_threads_per_build: The number threads that will be used for each
181 parallel build process.
182 output_path: Where the artifacts should be placed.
183 Will default to within the ``projects_path`` if not set.
184 collect_artifacts: Callback to collect artifacts.
185 Takes two named arguments:
187 | **project** (:class:`.VivadoProject`): The project that is being built.
189 | **output_path** (pathlib.Path): Where the build artifacts should be placed.
191 | Must return True.
192 kwargs: Other arguments as accepted by :meth:`.VivadoProject.build`.
194 .. Note::
195 Argument ``project_path`` can not be set, it is set by this class
196 based on the ``project_paths`` argument to this function.
198 Argument ``num_threads`` is set by the ``num_threads_per_build``
199 argument to this function. This naming difference is done to avoid
200 confusion with regards to ``num_parallel_builds``.
202 Return:
203 True if everything went well.
204 """
205 if collect_artifacts:
206 thread_safe_collect_artifacts = ThreadSafeCollectArtifacts(
207 collect_artifacts
208 ).collect_artifacts
209 else:
210 thread_safe_collect_artifacts = None
212 build_wrappers = []
213 for project in self.projects:
214 project_output_path = self.get_build_project_output_path(
215 project=project, projects_path=projects_path, output_path=output_path
216 )
218 build_wrapper = BuildProjectBuildWrapper(
219 project=project,
220 collect_artifacts=thread_safe_collect_artifacts,
221 output_path=project_output_path,
222 num_threads=num_threads_per_build,
223 **kwargs,
224 )
225 build_wrappers.append(build_wrapper)
227 return self._run_build_wrappers(
228 projects_path=projects_path,
229 build_wrappers=build_wrappers,
230 num_parallel_builds=num_parallel_builds,
231 )
233 @staticmethod
234 def get_build_project_output_path(
235 project: VivadoProject, projects_path: Path, output_path: Path | None = None
236 ) -> Path:
237 """
238 Find where build artifacts will be placed for a project.
239 Arguments are the same as for :meth:`.build`.
240 """
241 if output_path:
242 return output_path.resolve() / project.name
244 return projects_path / project.name
246 def open(self, projects_path: Path) -> bool:
247 """
248 Open the projects in EDA GUI.
250 Arguments:
251 projects_path: The projects are placed here.
253 Return:
254 True if everything went well.
255 """
256 build_wrappers = [BuildProjectOpenWrapper(project=project) for project in self.projects]
258 return self._run_build_wrappers(
259 projects_path=projects_path,
260 build_wrappers=build_wrappers,
261 # For open there is no performance limitation. Set a high value.
262 num_parallel_builds=20,
263 )
265 def _run_build_wrappers(
266 self,
267 projects_path: Path,
268 build_wrappers: list[BuildProjectCreateWrapper]
269 | list[BuildProjectBuildWrapper]
270 | list[BuildProjectOpenWrapper],
271 num_parallel_builds: int,
272 ) -> bool:
273 if not build_wrappers:
274 # Return straight away if no builds are supplied
275 return True
277 start_time = time.time()
279 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER
280 report = BuildReport(printer=color_printer)
282 test_list = TestList()
283 for build_wrapper in build_wrappers:
284 test_list.add_test(build_wrapper)
286 verbosity = BuildRunner.VERBOSITY_QUIET
287 test_runner = BuildRunner(
288 report=report,
289 output_path=projects_path,
290 verbosity=verbosity,
291 num_threads=num_parallel_builds,
292 )
293 test_runner.run(test_list)
295 all_builds_ok: bool = report.all_ok()
296 report.set_real_total_time(time.time() - start_time)
298 # True if the builds are for the "build" step (not "create" or "open")
299 builds_are_build_step = isinstance(build_wrappers[0], BuildProjectBuildWrapper)
300 # If we are building, we should print the summary that is at the end of the console output.
301 # (however if we are creating or opening a project we should not print anything extra).
302 # However if anything has failed, we should also print.
303 if builds_are_build_step:
304 # The length of the build summary depends on if we are working with netlist builds or
305 # regular ones, so set the length given by one of the project objects.
306 report_length_lines = build_wrappers[0].build_result_report_length
307 report.set_report_length(report_length_lines=report_length_lines)
309 # If all are OK then we should print the resource utilization numbers.
310 # If not, then we print a few last lines of the log output.
311 if builds_are_build_step or not all_builds_ok:
312 report.print_str()
314 return all_builds_ok
316 def _iterate_projects(
317 self, project_filters: list[str], include_netlist_not_top_builds: bool
318 ) -> Iterable[VivadoProject]:
319 available_projects = []
320 for module in self._modules:
321 available_projects += module.get_build_projects()
323 for project in available_projects:
324 if project.is_netlist_build == include_netlist_not_top_builds:
325 if not project_filters:
326 yield project
328 else:
329 for project_filter in project_filters:
330 if fnmatch.filter([project.name], project_filter):
331 yield project
333 # Do not continue with further filters if we have already matched this
334 # project.
335 # Multiple filters might match the same project, and multiple objects
336 # of the same project will break build
337 break
340class BuildProjectWrapper(ABC):
341 """
342 Mimics a VUnit test case object.
343 """
345 def get_seed(self) -> str:
346 """
347 Required since VUnit version 5.0.0.dev6, where a 'get_seed' method was added
348 to the 'TestSuiteWrapper' class, which calls a 'get_seed' method expected to be implemented
349 in the test case object.
350 This mechanism is not used by tsfpga, but is required in order to avoid errors.
351 Adding a dummy implementation like this makes sure it works with older as well as newer
352 versions of VUnit.
353 """
354 return ""
356 @abstractmethod
357 def run(
358 self,
359 output_path: Path,
360 read_output: Any, # noqa: ANN401
361 ) -> bool:
362 pass
365class BuildProjectCreateWrapper(BuildProjectWrapper):
366 """
367 Wrapper to create a build project, for usage in the build runner.
368 """
370 def __init__(
371 self,
372 project: VivadoProject,
373 **kwargs: Any, # noqa: ANN401
374 ) -> None:
375 self.name = project.name
376 self._project = project
377 self._create_arguments = kwargs
379 def run(
380 self,
381 output_path: Path,
382 read_output: Any, # noqa: ANN401, ARG002
383 ) -> bool:
384 """
385 Argument 'read_output' sent by VUnit test runner is unused by us.
386 """
387 this_projects_path = Path(output_path) / "project"
388 return self._project.create(project_path=this_projects_path, **self._create_arguments)
391class BuildProjectBuildWrapper(BuildProjectWrapper):
392 """
393 Wrapper to build a project, for usage in the build runner.
394 """
396 def __init__(
397 self,
398 project: VivadoProject,
399 collect_artifacts: Callable[..., bool] | None,
400 **kwargs: Any, # noqa: ANN401
401 ) -> None:
402 self.name = project.name
403 self._project = project
404 self._collect_artifacts = collect_artifacts
405 self._build_arguments = kwargs
407 def run(
408 self,
409 output_path: Path,
410 read_output: Any, # noqa: ANN401, ARG002
411 ) -> bool:
412 """
413 Argument 'read_output' sent by VUnit test runner is unused by us.
414 """
415 this_projects_path = Path(output_path) / "project"
416 build_result = self._project.build(project_path=this_projects_path, **self._build_arguments)
418 if not build_result.success:
419 self._print_build_result(build_result)
420 return build_result.success
422 # Proceed to artifact collection only if build succeeded.
423 if self._collect_artifacts and not self._collect_artifacts(
424 project=self._project, output_path=self._build_arguments["output_path"]
425 ):
426 build_result.success = False
428 # Print size at the absolute end
429 self._print_build_result(build_result=build_result)
430 return build_result.success
432 @staticmethod
433 def _print_build_result(build_result: build_result.BuildResult) -> None:
434 build_report = build_result.report()
435 if build_report:
436 # Add an empty line before the build result report, to have margin in how many lines are
437 # printed. See the comments in BuildResult for an explanation.
438 print()
439 print(build_report)
441 @property
442 def build_result_report_length(self) -> int:
443 """
444 The number of lines in the build_result report from this project.
445 """
446 # The size summary, as returned by tsfpga.vivado.project.BuildResult is a JSON formatted
447 # string with one line for each utilization category.
448 # For Xilinx 7 series, there are 8 categories (Total LUTs, Logic LUTs, LUTRAMs,
449 # SRLs, FFs, RAMB36, RAMB18, DSP Blocks). For UltraScale series there is one
450 # extra (URAM).
451 # Additionally, the size summary contains three extra lines for JSON braces and a title.
452 #
453 # This value is enough lines so the whole summary gets printed to console.
454 # For 7 series, this will mean an extra blank line before the summary.
455 #
456 # This is a hack. Works for now, but is far from reliable.
457 length_of_size_report = 3 + 8 + 1
459 if self._project.is_netlist_build:
460 # The logic level distribution report is five lines, plus a title line.
461 # This report is only printed for netlist builds, where there is no configured clock
462 # present. If there were many clocks present in the build, the report would be longer.
463 length_of_logic_level_report = 5 + 1
464 return length_of_size_report + length_of_logic_level_report
466 return length_of_size_report
469class BuildProjectOpenWrapper(BuildProjectWrapper):
470 """
471 Wrapper to open a build project, for usage in the build runner.
472 """
474 def __init__(self, project: VivadoProject) -> None:
475 self.name = project.name
476 self._project = project
478 def run(
479 self,
480 output_path: Path,
481 read_output: Any, # noqa: ANN401, ARG002
482 ) -> bool:
483 """
484 Argument 'read_output' sent by VUnit test runner is unused by us.
485 """
486 this_projects_path = Path(output_path) / "project"
487 return self._project.open(project_path=this_projects_path)
490class BuildRunner(TestRunner):
491 """
492 Build runner that mimics a VUnit TestRunner. Most things are used as they are in the
493 base class, but some behavior is overridden.
494 """
496 def _create_test_mapping_file(
497 self,
498 test_suites: Any, # noqa: ANN401
499 ) -> None:
500 """
501 Overloaded from super class.
503 Do not create this file.
505 We do not need it since folder name is the same as project name.
506 """
508 def _get_output_path(self, test_suite_name: str) -> str:
509 """
510 Overloaded from super class.
512 Output folder name is the same as the project name.
514 Original function adds a hash at the end of the folder name.
515 We do not want that necessarily.
516 """
517 return str(Path(self._output_path) / test_suite_name)
519 @staticmethod
520 def _prepare_test_suite_output_path(output_path: str) -> None:
521 """
522 Overloaded from super class.
524 Create the directory unless it already exists.
526 Original function wipes the path before running a test. We do not want to do that
527 since e.g. a Vivado project takes a long time to create and might contain a state
528 that the user wants to keep.
529 """
530 create_directory(Path(output_path), empty=False)
533class ThreadSafeCollectArtifacts:
534 """
535 A thread-safe wrapper around a user-supplied function that makes sure the function
536 is not launched more than once at the same time. When two builds finish at the
537 same time, race conditions can arise depending on what the function does.
539 Note that this is a VERY fringe case, since builds usually take >20 minutes, and the
540 collection probably only takes a few seconds. But it happens sometimes with the tsfpga
541 example projects which are identical and quite fast (roughly three minutes).
542 """
544 def __init__(self, collect_artifacts: Callable[..., bool]) -> None:
545 self._collect_artifacts = collect_artifacts
546 self._lock = Lock()
548 def collect_artifacts(self, project: VivadoProject, output_path: Path) -> bool:
549 with self._lock:
550 return self._collect_artifacts(project=project, output_path=output_path)
553class BuildReport(TestReport):
554 def add_result(
555 self,
556 *args: Any, # noqa: ANN401
557 **kwargs: Any, # noqa: ANN401
558 ) -> None:
559 """
560 Overloaded from super class.
562 Add a a test result.
564 Uses a different Result class than the super method.
565 """
566 result = BuildResult(*args, **kwargs)
567 self._test_results[result.name] = result
568 self._test_names_in_order.append(result.name)
570 def set_report_length(self, report_length_lines: int) -> None:
571 """
572 Set the report length for all test results that have been added to the report.
573 """
574 for test_result in self._test_results.values():
575 test_result.set_report_length(report_length_lines)
577 def print_latest_status(self, total_tests: int) -> None:
578 """
579 Overloaded from super class.
581 This method is called for each build when it should print its result just as it finished,
582 but other builds may not be finished yet.
584 Inherited and adapted from the VUnit function:
585 * Removed support for the "skipped" result.
586 * Do not use abbreviations in the printout.
587 * Use f-strings.
588 """
589 result = self._last_test_result()
590 passed, failed, _ = self._split()
592 if result.passed:
593 self._printer.write("pass", fg="gi")
594 elif result.failed:
595 self._printer.write("fail", fg="ri")
596 else:
597 raise AssertionError
599 count_summary = f"pass={len(passed)} fail={len(failed)} total={total_tests}"
600 self._printer.write(f" ({count_summary}) {result.name} ({result.time:.1f} seconds)\n")
603class BuildResult(TestResult):
604 report_length_lines = None
606 def _print_output(
607 self,
608 printer: ColorPrinter,
609 num_lines: int,
610 ) -> None:
611 """
612 Print the last lines from the output file.
613 """
614 output_tail = read_last_lines_of_file(Path(self._output_file_name), num_lines=num_lines)
615 printer.write(output_tail)
617 def set_report_length(self, report_length_lines: int) -> None:
618 """
619 Set how many lines shall be printed when this result is printed.
620 """
621 self.report_length_lines = report_length_lines
623 def print_status(
624 self,
625 printer: ColorPrinter,
626 padding: int = 0,
627 **kwargs: dict[str, Any],
628 ) -> None:
629 """
630 Overloaded from super class.
632 This method is called for each build when it should print its result in the "Summary" at
633 the end when all builds have finished.
635 Inherited and adapted from the VUnit function.
637 Note that a ``max_time`` integer argument is added in VUnit >4.7.0, but at the time of
638 writing this is un-released on the VUnit ``master`` branch.
639 In order to be compatible with both older and newer versions, we use ``**kwargs`` for this.
640 """
641 if self.passed and self.report_length_lines is not None:
642 # Build passed, print build summary of the specified length. The length is only
643 # set if this is a "build" result (not "create" or "open").
644 self._print_output(printer=printer, num_lines=self.report_length_lines)
645 else:
646 # The build failed, which can either be caused by
647 # 1. IDE build failure
648 # 2. IDE build succeeded, but post build hook, or size checkers failed.
649 # 3. Other python error (directory already exists, ...)
650 # In the case of IDE build failed, we want a significant portion of the output, to be
651 # able to see an indication of what failed. In the case of size checkers, we want to see
652 # all the printouts from all checkers, to see which one failed. Since there are at most
653 # eight resource categories, it is reasonable to assume that there will never be more
654 # than eight size checkers.
655 self._print_output(printer=printer, num_lines=25)
657 # Print the regular output from the VUnit class.
658 # A little extra margin between build name and execution time makes the output more readable
659 super().print_status(printer=printer, padding=padding + 2, **kwargs)
660 # Add an empty line between each build, for readability.
661 printer.write("\n")