Coverage for tsfpga/build_project_list.py: 96%
186 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-25 23:35 +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
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 Callable, Sequence
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__(self, projects: Sequence[VivadoProject], no_color: bool = False) -> None:
40 """
41 Arguments:
42 projects: The FPGA build projects that will be executed.
43 no_color: Disable color in printouts.
44 """
45 self.projects = projects
46 self._no_color = no_color
48 def __str__(self) -> str:
49 """
50 Returns a string with a description list of the projects.
52 Will print some information about each project (name, generics, part, ...) so can become
53 long if there are many projects present.
54 An alternative in that case would be :meth:`.get_short_str`.
55 """
56 result = "\n".join([str(project) for project in self.projects])
57 result += "\n"
58 result += "\n"
59 result += f"Listed {len(self.projects)} builds"
61 return result
63 def get_short_str(self) -> str:
64 """
65 Returns a short string with a description list of the projects.
67 This is an alternative function that is more compact than ``__str__``.
68 """
69 result = "\n".join([project.name for project in self.projects])
70 result += "\n"
71 result += f"Listed {len(self.projects)} builds"
73 return result
75 def create(
76 self,
77 projects_path: Path,
78 num_parallel_builds: int,
79 **kwargs: Any, # noqa: ANN401
80 ) -> bool:
81 """
82 Create build project on disk for all the projects in the list.
84 Arguments:
85 projects_path: The projects will be placed here.
86 num_parallel_builds: The number of projects that will be created in parallel.
87 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
89 .. Note::
90 Argument ``project_path`` can not be set, it is set by this class
91 based on the ``project_paths`` argument to this function.
93 Return:
94 True if everything went well.
95 """
96 build_wrappers = []
97 for project in self.projects:
98 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
99 build_wrappers.append(build_wrapper)
101 return self._run_build_wrappers(
102 projects_path=projects_path,
103 build_wrappers=build_wrappers,
104 num_parallel_builds=num_parallel_builds,
105 )
107 def create_unless_exists(
108 self,
109 projects_path: Path,
110 num_parallel_builds: int,
111 **kwargs: Any, # noqa: ANN401
112 ) -> bool:
113 """
114 Create build project for all the projects in the list, unless the project already
115 exists.
117 Arguments:
118 projects_path: The projects will be placed here.
119 num_parallel_builds: The number of projects that will be created in parallel.
120 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
122 .. Note::
123 Argument ``project_path`` can not be set, it is set by this class
124 based on the ``project_paths`` argument to this function.
126 Return:
127 True if everything went well.
128 """
129 build_wrappers = []
130 for project in self.projects:
131 if not self.get_build_project_path(
132 project=project, projects_path=projects_path
133 ).exists():
134 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
135 build_wrappers.append(build_wrapper)
137 if not build_wrappers:
138 # Return straight away if no projects need to be created. To avoid extra
139 # "No tests were run!" printout from creation step that is very misleading.
140 return True
142 return self._run_build_wrappers(
143 projects_path=projects_path,
144 build_wrappers=build_wrappers,
145 num_parallel_builds=num_parallel_builds,
146 )
148 def build(
149 self,
150 projects_path: Path,
151 num_parallel_builds: int,
152 num_threads_per_build: int,
153 output_path: Path | None = None,
154 collect_artifacts: Callable[[VivadoProject, Path], bool] | None = None,
155 **kwargs: Any, # noqa: ANN401
156 ) -> bool:
157 """
158 Build all the projects in the list.
160 Arguments:
161 projects_path: The projects are placed here.
162 num_parallel_builds: The number of projects that will be built in parallel.
163 num_threads_per_build: The number threads that will be used for each
164 parallel build process.
165 output_path: Where the artifacts should be placed.
166 Will default to within the ``projects_path`` if not set.
167 collect_artifacts: Callback to collect artifacts.
168 Takes two named arguments:
170 | **project** (:class:`.VivadoProject`): The project that is being built.
172 | **output_path** (pathlib.Path): Where the build artifacts should be placed.
174 | Must return True.
175 kwargs: Other arguments as accepted by :meth:`.VivadoProject.build`.
177 .. Note::
178 Argument ``project_path`` can not be set, it is set by this class
179 based on the ``project_paths`` argument to this function.
181 Argument ``num_threads`` is set by the ``num_threads_per_build``
182 argument to this function. This naming difference is done to avoid
183 confusion with regards to ``num_parallel_builds``.
185 Return:
186 True if everything went well.
187 """
188 if collect_artifacts:
189 thread_safe_collect_artifacts = ThreadSafeCollectArtifacts(
190 collect_artifacts=collect_artifacts
191 ).collect_artifacts
192 else:
193 thread_safe_collect_artifacts = None
195 build_wrappers = []
196 for project in self.projects:
197 project_output_path = self.get_build_project_output_path(
198 project=project, projects_path=projects_path, output_path=output_path
199 )
201 build_wrapper = BuildProjectBuildWrapper(
202 project=project,
203 collect_artifacts=thread_safe_collect_artifacts,
204 output_path=project_output_path,
205 num_threads=num_threads_per_build,
206 **kwargs,
207 )
208 build_wrappers.append(build_wrapper)
210 return self._run_build_wrappers(
211 projects_path=projects_path,
212 build_wrappers=build_wrappers,
213 num_parallel_builds=num_parallel_builds,
214 )
216 @staticmethod
217 def get_build_project_path(project: VivadoProject, projects_path: Path) -> Path:
218 """
219 Find where the project files for a specific project will be placed.
220 Arguments are the same as for :meth:`.create`.
221 """
222 return projects_path / project.name / "project"
224 @staticmethod
225 def get_build_project_output_path(
226 project: VivadoProject, projects_path: Path, output_path: Path | None = 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 = [BuildProjectOpenWrapper(project=project) for project in self.projects]
249 return self._run_build_wrappers(
250 projects_path=projects_path,
251 build_wrappers=build_wrappers,
252 # For open there is no performance limitation. Set a high value.
253 num_parallel_builds=20,
254 )
256 def _run_build_wrappers(
257 self,
258 projects_path: Path,
259 build_wrappers: list[BuildProjectCreateWrapper]
260 | list[BuildProjectBuildWrapper]
261 | list[BuildProjectOpenWrapper],
262 num_parallel_builds: int,
263 ) -> bool:
264 if not build_wrappers:
265 # Return straight away if no builds are supplied
266 return True
268 start_time = time.time()
270 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER
271 report = BuildReport(printer=color_printer)
273 test_list = TestList()
274 for build_wrapper in build_wrappers:
275 test_list.add_test(build_wrapper)
277 verbosity = BuildRunner.VERBOSITY_QUIET
278 test_runner = BuildRunner(
279 report=report,
280 output_path=projects_path,
281 verbosity=verbosity,
282 num_threads=num_parallel_builds,
283 run_script_path=None,
284 )
285 test_runner.run(test_list)
287 all_builds_ok: bool = report.all_ok()
288 report.set_real_total_time(time.time() - start_time)
290 # True if the builds are for the "build" step (not "create" or "open")
291 builds_are_build_step = isinstance(build_wrappers[0], BuildProjectBuildWrapper)
293 if builds_are_build_step:
294 for build_wrapper in build_wrappers:
295 # Update the 'report' object with info about how many lines to print for each build.
296 # This information is only available after the build has finished.
297 report.set_report_length(
298 name=build_wrapper.name,
299 report_length_lines=build_wrapper.report_length_lines,
300 )
302 # If all are OK then we should print the resource utilization numbers.
303 # If not, then we print a few last lines of the log output.
304 if builds_are_build_step or not all_builds_ok:
305 report.print_str()
307 return all_builds_ok
310class BuildProjectWrapper(ABC):
311 """
312 Mimics a VUnit test case object.
313 """
315 def get_seed(self) -> str:
316 """
317 Required since VUnit version 5.0.0.dev6, where a 'get_seed' method was added
318 to the 'TestSuiteWrapper' class, which calls a 'get_seed' method expected to be implemented
319 in the test case object.
320 This mechanism is not used by tsfpga, but is required in order to avoid errors.
321 Adding a dummy implementation like this makes sure it works with older as well as newer
322 versions of VUnit.
323 """
324 return ""
326 @abstractmethod
327 def run(
328 self,
329 output_path: Path,
330 read_output: Any, # noqa: ANN401
331 ) -> bool:
332 pass
335class BuildProjectCreateWrapper(BuildProjectWrapper):
336 """
337 Wrapper to create a build project, for usage in the build runner.
338 """
340 def __init__(
341 self,
342 project: VivadoProject,
343 **kwargs: Any, # noqa: ANN401
344 ) -> None:
345 self.name = project.name
346 self._project = project
347 self._create_arguments = kwargs
349 def run(
350 self,
351 output_path: Path,
352 read_output: Any, # noqa: ANN401, ARG002
353 run_script_path: Any, # noqa: ANN401, ARG002
354 ) -> bool:
355 """
356 Arguments 'read_output' and 'run_script_path' sent by VUnit test runner are unused by us.
357 """
358 this_project_path = Path(output_path) / "project"
359 return self._project.create(project_path=this_project_path, **self._create_arguments)
362class BuildProjectBuildWrapper(BuildProjectWrapper):
363 """
364 Wrapper to build a project, for usage in the build runner.
365 """
367 def __init__(
368 self,
369 project: VivadoProject,
370 collect_artifacts: Callable[..., bool] | None,
371 **kwargs: Any, # noqa: ANN401
372 ) -> None:
373 self.name = project.name
374 self._project = project
375 self._collect_artifacts = collect_artifacts
376 self._build_arguments = kwargs
378 self._report_length_lines: int | None = None
380 def run(
381 self,
382 output_path: Path,
383 read_output: Any, # noqa: ANN401, ARG002
384 run_script_path: Any, # noqa: ANN401, ARG002
385 ) -> bool:
386 """
387 Arguments 'read_output' and 'run_script_path' sent by VUnit test runner are unused by us.
388 """
389 this_project_path = Path(output_path) / "project"
390 build_result = self._project.build(project_path=this_project_path, **self._build_arguments)
392 if not build_result.success:
393 self._print_build_result(build_result=build_result)
394 return build_result.success
396 # Proceed to artifact collection only if build succeeded.
397 if self._collect_artifacts is not None:
398 build_result.success &= self._collect_artifacts(
399 project=self._project, output_path=self._build_arguments["output_path"]
400 )
402 # Print size at the absolute end.
403 self._print_build_result(build_result=build_result)
404 return build_result.success
406 def _print_build_result(self, build_result: build_result.BuildResult) -> None:
407 build_report = build_result.report()
409 if build_report:
410 print(build_report)
411 self._report_length_lines = build_report.count("\n") + 1
413 @property
414 def report_length_lines(self) -> int | None:
415 """
416 The number of lines in the ``build_result`` report from this project.
417 A value of ``None`` would indicate a build failure, either in the IDE or in the
418 post-build steps.
419 """
420 return self._report_length_lines
423class BuildProjectOpenWrapper(BuildProjectWrapper):
424 """
425 Wrapper to open a build project, for usage in the build runner.
426 """
428 def __init__(self, project: VivadoProject) -> None:
429 self.name = project.name
430 self._project = project
432 def run(
433 self,
434 output_path: Path,
435 read_output: Any, # noqa: ANN401, ARG002
436 run_script_path: Any, # noqa: ANN401, ARG002
437 ) -> bool:
438 """
439 Arguments 'read_output' and 'run_script_path' sent by VUnit test runner are unused by us.
440 """
441 this_project_path = Path(output_path) / "project"
442 return self._project.open(project_path=this_project_path)
445class BuildRunner(TestRunner):
446 """
447 Build runner that mimics a VUnit TestRunner. Most things are used as they are in the
448 base class, but some behavior is overridden.
449 """
451 def _create_test_mapping_file(
452 self,
453 test_suites: Any, # noqa: ANN401
454 ) -> None:
455 """
456 Overloaded from super class.
458 Do not create this file.
460 We do not need it since folder name is the same as project name.
461 """
463 def _get_output_path(self, test_suite_name: str) -> str:
464 """
465 Overloaded from super class.
467 Output folder name is the same as the project name.
469 Original function adds a hash at the end of the folder name.
470 We do not want that necessarily.
471 """
472 return str(Path(self._output_path) / test_suite_name)
474 @staticmethod
475 def _prepare_test_suite_output_path(output_path: str) -> None:
476 """
477 Overloaded from super class.
479 Create the directory unless it already exists.
481 Original function wipes the path before running a test. We do not want to do that
482 since e.g. a Vivado project takes a long time to create and might contain a state
483 that the user wants to keep.
484 """
485 create_directory(Path(output_path), empty=False)
488class ThreadSafeCollectArtifacts:
489 """
490 A thread-safe wrapper around a user-supplied function that makes sure the function
491 is not launched more than once at the same time. When two builds finish at the
492 same time, race conditions can arise depending on what the function does.
494 Note that this is a VERY fringe case, since builds usually take >20 minutes, and the
495 collection probably only takes a few seconds. But it happens sometimes with the tsfpga
496 example projects which are identical and quite fast (roughly three minutes).
497 """
499 def __init__(self, collect_artifacts: Callable[[VivadoProject, Path], bool]) -> None:
500 self._collect_artifacts = collect_artifacts
501 self._lock = Lock()
503 def collect_artifacts(self, project: VivadoProject, output_path: Path) -> bool:
504 with self._lock:
505 return self._collect_artifacts(project=project, output_path=output_path)
508class BuildReport(TestReport):
509 def add_result(
510 self,
511 *args: Any, # noqa: ANN401
512 **kwargs: Any, # noqa: ANN401
513 ) -> 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, name: str, report_length_lines: int | None) -> None:
526 """
527 Set how many lines shall be printed for this build.
528 Can be ``None`` to indicate that the build failed, and we don't how much to print.
529 """
530 self._test_results[name].set_report_length(report_length_lines=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 raise AssertionError
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: int | None = None
561 def _print_output(
562 self,
563 printer: ColorPrinter,
564 num_lines: int,
565 ) -> None:
566 """
567 Print the last lines from the output file.
568 """
569 output_tail = read_last_lines_of_file(Path(self._output_file_name), num_lines=num_lines)
570 printer.write(output_tail)
572 def set_report_length(self, report_length_lines: int) -> None:
573 """
574 Set how many lines shall be printed when this result is printed.
575 Can be ``None`` to indicate that the build failed, and we don't how much to print.
576 """
577 self._report_length_lines = report_length_lines
579 def print_status(
580 self,
581 printer: ColorPrinter,
582 padding: int = 0,
583 **kwargs: dict[str, Any],
584 ) -> None:
585 """
586 Overloaded from super class.
588 This method is called for each build when it should print its result in the "Summary" at
589 the end when all builds have finished.
591 Inherited and adapted from the VUnit function.
593 Note that a ``max_time`` integer argument is added in VUnit >4.7.0, but at the time of
594 writing this is un-released on the VUnit ``master`` branch.
595 In order to be compatible with both older and newer versions, we use ``**kwargs`` for this.
596 """
597 if self.passed and self._report_length_lines is not None:
598 # Build passed, print build summary of the specified length. The length is only
599 # set if this is a "build" result (not "create" or "open").
600 self._print_output(printer=printer, num_lines=self._report_length_lines)
602 else:
603 # The build failed, which can either be caused by
604 # 1. IDE build failure
605 # 2. IDE build succeeded, but post build hook, or size checkers failed.
606 # 3. Other python error (directory already exists, ...)
607 # In the case of IDE build failed, we want a significant portion of the output, to be
608 # able to see an indication of what failed.
609 # In the case of size checkers, we want to see all the printouts from all checkers,
610 # to see which one failed.
611 self._print_output(printer=printer, num_lines=25)
613 # Print the regular output from the VUnit class.
614 # A little extra margin between build name and execution time makes the output more readable
615 super().print_status(printer=printer, padding=padding + 2, **kwargs)
616 # Add an empty line between each build, for readability.
617 printer.write("\n")
620def get_build_projects(
621 modules: ModuleList, project_filters: list[str], include_netlist_not_full_builds: bool = False
622) -> list[VivadoProject]:
623 """
624 Get build projects from the given modules that match the given filters.
625 Note that the result of this function is a list of "raw" :class:`.VivadoProject` objects.
626 These are meant to be passed to :class:`.BuildProjectList` for execution.
628 Arguments:
629 modules: Module objects that can define build projects.
630 project_filters: Project name filters.
631 Can use wildcards (*).
632 Leave empty for all.
633 include_netlist_not_full_builds:
634 Set True to get only netlist builds, instead of only full top level builds.
635 """
636 result = []
637 for module in modules:
638 for project in module.get_build_projects():
639 if project.is_netlist_build == include_netlist_not_full_builds:
640 if not project_filters:
641 result.append(project)
643 else:
644 for project_filter in project_filters:
645 if fnmatch.filter([project.name], project_filter):
646 result.append(project)
648 # Do not continue with further filters if we have already matched
649 # this project.
650 # Multiple filters might match the same project, and we dont't
651 # want duplicates.
652 break
654 return result