Coverage for tsfpga/build_project_list.py: 96%
186 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-21 23:39 +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 . import build_result
29 from .module_list import ModuleList
30 from .vivado.project import VivadoProject
31 from .yosys.project import YosysNetlistBuild
33 #: A build project managed by :class:`.BuildProjectList`. Either a full, place-and-route
34 #: capable :class:`.VivadoProject`, or a synthesis-only :class:`.YosysNetlistBuild`.
35 #: The two share a duck-typed interface (``create``/``build``/``open``, ``name``,
36 #: ``is_netlist_build``) so they can be mixed freely in the same list.
37 BuildProject = VivadoProject | YosysNetlistBuild
40class BuildProjectList:
41 """
42 Interface to handle a list of FPGA build projects.
43 Enables building many projects in parallel.
44 """
46 def __init__(self, projects: Sequence[BuildProject], no_color: bool = False) -> None:
47 """
48 Arguments:
49 projects: The FPGA build projects that will be executed.
50 no_color: Disable color in printouts.
51 """
52 self.projects = projects
53 self._no_color = no_color
55 def __str__(self) -> str:
56 """
57 Returns a string with a description list of the projects.
59 Will print some information about each project (name, generics, part, ...) so can become
60 long if there are many projects present.
61 An alternative in that case would be :meth:`.get_short_str`.
62 """
63 result = "\n".join([str(project) for project in self.projects])
64 result += "\n"
65 result += "\n"
66 result += f"Listed {len(self.projects)} builds"
68 return result
70 def get_short_str(self) -> str:
71 """
72 Returns a short string with a description list of the projects.
74 This is an alternative function that is more compact than ``__str__``.
75 """
76 result = "\n".join([project.name for project in self.projects])
77 result += "\n"
78 result += f"Listed {len(self.projects)} builds"
80 return result
82 def create(
83 self,
84 projects_path: Path,
85 num_parallel_builds: int,
86 **kwargs: Any, # noqa: ANN401
87 ) -> bool:
88 """
89 Create build project on disk for all the projects in the list.
91 Arguments:
92 projects_path: The projects will be placed here.
93 num_parallel_builds: The number of projects that will be created in parallel.
94 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
96 .. Note::
97 Argument ``project_path`` can not be set, it is set by this class
98 based on the ``project_paths`` argument to this function.
100 Return:
101 True if everything went well.
102 """
103 build_wrappers = []
104 for project in self.projects:
105 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
106 build_wrappers.append(build_wrapper)
108 return self._run_build_wrappers(
109 projects_path=projects_path,
110 build_wrappers=build_wrappers,
111 num_parallel_builds=num_parallel_builds,
112 )
114 def create_unless_exists(
115 self,
116 projects_path: Path,
117 num_parallel_builds: int,
118 **kwargs: Any, # noqa: ANN401
119 ) -> bool:
120 """
121 Create build project for all the projects in the list, unless the project already
122 exists.
124 Arguments:
125 projects_path: The projects will be placed here.
126 num_parallel_builds: The number of projects that will be created in parallel.
127 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
129 .. Note::
130 Argument ``project_path`` can not be set, it is set by this class
131 based on the ``project_paths`` argument to this function.
133 Return:
134 True if everything went well.
135 """
136 build_wrappers = []
137 for project in self.projects:
138 if not self.get_build_project_path(
139 project=project, projects_path=projects_path
140 ).exists():
141 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
142 build_wrappers.append(build_wrapper)
144 if not build_wrappers:
145 # Return straight away if no projects need to be created. To avoid extra
146 # "No tests were run!" printout from creation step that is very misleading.
147 return True
149 return self._run_build_wrappers(
150 projects_path=projects_path,
151 build_wrappers=build_wrappers,
152 num_parallel_builds=num_parallel_builds,
153 )
155 def build(
156 self,
157 projects_path: Path,
158 num_parallel_builds: int,
159 num_threads_per_build: int,
160 output_path: Path | None = None,
161 collect_artifacts: Callable[[BuildProject, Path], bool] | None = None,
162 **kwargs: Any, # noqa: ANN401
163 ) -> bool:
164 """
165 Build all the projects in the list.
167 Arguments:
168 projects_path: The projects are placed here.
169 num_parallel_builds: The number of projects that will be built in parallel.
170 num_threads_per_build: The number threads that will be used for each
171 parallel build process.
172 output_path: Where the artifacts should be placed.
173 Will default to within the ``projects_path`` if not set.
174 collect_artifacts: Callback to collect artifacts.
175 Takes two named arguments:
177 | **project** (:class:`.VivadoProject`): The project that is being built.
179 | **output_path** (pathlib.Path): Where the build artifacts should be placed.
181 | Must return True.
182 kwargs: Other arguments as accepted by :meth:`.VivadoProject.build`.
184 .. Note::
185 Argument ``project_path`` can not be set, it is set by this class
186 based on the ``project_paths`` argument to this function.
188 Argument ``num_threads`` is set by the ``num_threads_per_build``
189 argument to this function. This naming difference is done to avoid
190 confusion with regards to ``num_parallel_builds``.
192 Return:
193 True if everything went well.
194 """
195 if collect_artifacts:
196 thread_safe_collect_artifacts = ThreadSafeCollectArtifacts(
197 collect_artifacts=collect_artifacts
198 ).collect_artifacts
199 else:
200 thread_safe_collect_artifacts = None
202 build_wrappers = []
203 for project in self.projects:
204 project_output_path = self.get_build_project_output_path(
205 project=project, projects_path=projects_path, output_path=output_path
206 )
208 build_wrapper = BuildProjectBuildWrapper(
209 project=project,
210 collect_artifacts=thread_safe_collect_artifacts,
211 output_path=project_output_path,
212 num_threads=num_threads_per_build,
213 **kwargs,
214 )
215 build_wrappers.append(build_wrapper)
217 return self._run_build_wrappers(
218 projects_path=projects_path,
219 build_wrappers=build_wrappers,
220 num_parallel_builds=num_parallel_builds,
221 )
223 @staticmethod
224 def get_build_project_path(project: BuildProject, projects_path: Path) -> Path:
225 """
226 Find where the project files for a specific project will be placed.
227 Arguments are the same as for :meth:`.create`.
228 """
229 return projects_path / project.name / "project"
231 @staticmethod
232 def get_build_project_output_path(
233 project: BuildProject, projects_path: Path, output_path: Path | None = None
234 ) -> Path:
235 """
236 Find where build artifacts will be placed for a project.
237 Arguments are the same as for :meth:`.build`.
238 """
239 if output_path:
240 return output_path.resolve() / project.name
242 return projects_path / project.name
244 def open(self, projects_path: Path) -> bool:
245 """
246 Open the projects in EDA GUI.
248 Arguments:
249 projects_path: The projects are placed here.
251 Return:
252 True if everything went well.
253 """
254 build_wrappers = [BuildProjectOpenWrapper(project=project) for project in self.projects]
256 return self._run_build_wrappers(
257 projects_path=projects_path,
258 build_wrappers=build_wrappers,
259 # For open there is no performance limitation. Set a high value.
260 num_parallel_builds=20,
261 )
263 def _run_build_wrappers(
264 self,
265 projects_path: Path,
266 build_wrappers: list[BuildProjectCreateWrapper]
267 | list[BuildProjectBuildWrapper]
268 | list[BuildProjectOpenWrapper],
269 num_parallel_builds: int,
270 ) -> bool:
271 if not build_wrappers:
272 # Return straight away if no builds are supplied
273 return True
275 start_time = time.time()
277 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER
278 report = BuildReport(printer=color_printer)
280 test_list = TestList()
281 for build_wrapper in build_wrappers:
282 test_list.add_test(build_wrapper)
284 verbosity = BuildRunner.VERBOSITY_QUIET
285 test_runner = BuildRunner(
286 report=report,
287 output_path=projects_path,
288 verbosity=verbosity,
289 num_threads=num_parallel_builds,
290 run_script_path=None,
291 )
292 test_runner.run(test_list)
294 all_builds_ok: bool = report.all_ok()
295 report.set_real_total_time(time.time() - start_time)
297 # True if the builds are for the "build" step (not "create" or "open")
298 builds_are_build_step = isinstance(build_wrappers[0], BuildProjectBuildWrapper)
300 if builds_are_build_step:
301 for build_wrapper in build_wrappers:
302 # Update the 'report' object with info about how many lines to print for each build.
303 # This information is only available after the build has finished.
304 report.set_report_length(
305 name=build_wrapper.name,
306 report_length_lines=build_wrapper.report_length_lines,
307 )
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
317class BuildProjectWrapper(ABC):
318 """
319 Mimics a VUnit test case object.
320 """
322 def get_seed(self) -> str:
323 """
324 Required since VUnit version 5.0.0.dev6, where a 'get_seed' method was added
325 to the 'TestSuiteWrapper' class, which calls a 'get_seed' method expected to be implemented
326 in the test case object.
327 This mechanism is not used by tsfpga, but is required in order to avoid errors.
328 Adding a dummy implementation like this makes sure it works with older as well as newer
329 versions of VUnit.
330 """
331 return ""
333 @abstractmethod
334 def run(
335 self,
336 output_path: Path,
337 read_output: Any, # noqa: ANN401
338 ) -> bool:
339 pass
342class BuildProjectCreateWrapper(BuildProjectWrapper):
343 """
344 Wrapper to create a build project, for usage in the build runner.
345 """
347 def __init__(
348 self,
349 project: BuildProject,
350 **kwargs: Any, # noqa: ANN401
351 ) -> None:
352 self.name = project.name
353 self._project = project
354 self._create_arguments = kwargs
356 def run(
357 self,
358 output_path: Path,
359 read_output: Any, # noqa: ANN401, ARG002
360 run_script_path: Any, # noqa: ANN401, ARG002
361 ) -> bool:
362 """
363 Arguments 'read_output' and 'run_script_path' sent by VUnit test runner are unused by us.
364 """
365 this_project_path = Path(output_path) / "project"
366 return self._project.create(project_path=this_project_path, **self._create_arguments)
369class BuildProjectBuildWrapper(BuildProjectWrapper):
370 """
371 Wrapper to build a project, for usage in the build runner.
372 """
374 def __init__(
375 self,
376 project: BuildProject,
377 collect_artifacts: Callable[..., bool] | None,
378 **kwargs: Any, # noqa: ANN401
379 ) -> None:
380 self.name = project.name
381 self._project = project
382 self._collect_artifacts = collect_artifacts
383 self._build_arguments = kwargs
385 self._report_length_lines: int | None = None
387 def run(
388 self,
389 output_path: Path,
390 read_output: Any, # noqa: ANN401, ARG002
391 run_script_path: Any, # noqa: ANN401, ARG002
392 ) -> bool:
393 """
394 Arguments 'read_output' and 'run_script_path' sent by VUnit test runner are unused by us.
395 """
396 this_project_path = Path(output_path) / "project"
397 build_result = self._project.build(project_path=this_project_path, **self._build_arguments)
399 if not build_result.success:
400 self._print_build_result(build_result=build_result)
401 return build_result.success
403 # Proceed to artifact collection only if build succeeded.
404 if self._collect_artifacts is not None:
405 build_result.success &= self._collect_artifacts(
406 project=self._project, output_path=self._build_arguments["output_path"]
407 )
409 # Print size at the absolute end.
410 self._print_build_result(build_result=build_result)
411 return build_result.success
413 def _print_build_result(self, build_result: build_result.BuildResult) -> None:
414 build_report = build_result.report()
416 if build_report:
417 print(build_report)
418 self._report_length_lines = build_report.count("\n") + 1
420 @property
421 def report_length_lines(self) -> int | None:
422 """
423 The number of lines in the ``build_result`` report from this project.
424 A value of ``None`` would indicate a build failure, either in the IDE or in the
425 post-build steps.
426 """
427 return self._report_length_lines
430class BuildProjectOpenWrapper(BuildProjectWrapper):
431 """
432 Wrapper to open a build project, for usage in the build runner.
433 """
435 def __init__(self, project: BuildProject) -> None:
436 self.name = project.name
437 self._project = project
439 def run(
440 self,
441 output_path: Path,
442 read_output: Any, # noqa: ANN401, ARG002
443 run_script_path: Any, # noqa: ANN401, ARG002
444 ) -> bool:
445 """
446 Arguments 'read_output' and 'run_script_path' sent by VUnit test runner are unused by us.
447 """
448 this_project_path = Path(output_path) / "project"
449 return self._project.open(project_path=this_project_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(
459 self,
460 test_suites: Any, # noqa: ANN401
461 ) -> None:
462 """
463 Overloaded from super class.
465 Do not create this file.
467 We do not need it since folder name is the same as project name.
468 """
470 def _get_output_path(self, test_suite_name: str) -> str:
471 """
472 Overloaded from super class.
474 Output folder name is the same as the project name.
476 Original function adds a hash at the end of the folder name.
477 We do not want that necessarily.
478 """
479 return str(Path(self._output_path) / test_suite_name)
481 @staticmethod
482 def _prepare_test_suite_output_path(output_path: str) -> None:
483 """
484 Overloaded from super class.
486 Create the directory unless it already exists.
488 Original function wipes the path before running a test. We do not want to do that
489 since e.g. a Vivado project takes a long time to create and might contain a state
490 that the user wants to keep.
491 """
492 create_directory(Path(output_path), empty=False)
495class ThreadSafeCollectArtifacts:
496 """
497 A thread-safe wrapper around a user-supplied function that makes sure the function
498 is not launched more than once at the same time. When two builds finish at the
499 same time, race conditions can arise depending on what the function does.
501 Note that this is a VERY fringe case, since builds usually take >20 minutes, and the
502 collection probably only takes a few seconds. But it happens sometimes with the tsfpga
503 example projects which are identical and quite fast (roughly three minutes).
504 """
506 def __init__(self, collect_artifacts: Callable[[BuildProject, Path], bool]) -> None:
507 self._collect_artifacts = collect_artifacts
508 self._lock = Lock()
510 def collect_artifacts(self, project: BuildProject, output_path: Path) -> bool:
511 with self._lock:
512 return self._collect_artifacts(project=project, output_path=output_path)
515class BuildReport(TestReport):
516 def add_result(
517 self,
518 *args: Any, # noqa: ANN401
519 **kwargs: Any, # noqa: ANN401
520 ) -> None:
521 """
522 Overloaded from super class.
524 Add a a test result.
526 Uses a different Result class than the super method.
527 """
528 result = BuildResult(*args, **kwargs)
529 self._test_results[result.name] = result
530 self._test_names_in_order.append(result.name)
532 def set_report_length(self, name: str, report_length_lines: int | None) -> None:
533 """
534 Set how many lines shall be printed for this build.
535 Can be ``None`` to indicate that the build failed, and we don't how much to print.
536 """
537 self._test_results[name].set_report_length(report_length_lines=report_length_lines)
539 def print_latest_status(self, total_tests: int) -> None:
540 """
541 Overloaded from super class.
543 This method is called for each build when it should print its result just as it finished,
544 but other builds may not be finished yet.
546 Inherited and adapted from the VUnit function:
547 * Removed support for the "skipped" result.
548 * Do not use abbreviations in the printout.
549 * Use f-strings.
550 """
551 result = self._last_test_result()
552 passed, failed, _ = self._split()
554 if result.passed:
555 self._printer.write("pass", fg="gi")
556 elif result.failed:
557 self._printer.write("fail", fg="ri")
558 else:
559 raise AssertionError
561 count_summary = f"pass={len(passed)} fail={len(failed)} total={total_tests}"
562 self._printer.write(f" ({count_summary}) {result.name} ({result.time:.1f} seconds)\n")
565class BuildResult(TestResult):
566 _report_length_lines: int | None = None
568 def _print_output(
569 self,
570 printer: ColorPrinter,
571 num_lines: int,
572 ) -> None:
573 """
574 Print the last lines from the output file.
575 """
576 output_tail = read_last_lines_of_file(Path(self._output_file_name), num_lines=num_lines)
577 printer.write(output_tail)
579 def set_report_length(self, report_length_lines: int) -> None:
580 """
581 Set how many lines shall be printed when this result is printed.
582 Can be ``None`` to indicate that the build failed, and we don't how much to print.
583 """
584 self._report_length_lines = report_length_lines
586 def print_status(
587 self,
588 printer: ColorPrinter,
589 padding: int = 0,
590 **kwargs: dict[str, Any],
591 ) -> None:
592 """
593 Overloaded from super class.
595 This method is called for each build when it should print its result in the "Summary" at
596 the end when all builds have finished.
598 Inherited and adapted from the VUnit function.
600 Note that a ``max_time`` integer argument is added in VUnit >4.7.0, but at the time of
601 writing this is un-released on the VUnit ``master`` branch.
602 In order to be compatible with both older and newer versions, we use ``**kwargs`` for this.
603 """
604 if self.passed and self._report_length_lines is not None:
605 # Build passed, print build summary of the specified length. The length is only
606 # set if this is a "build" result (not "create" or "open").
607 self._print_output(printer=printer, num_lines=self._report_length_lines)
609 else:
610 # The build failed, which can either be caused by
611 # 1. IDE build failure
612 # 2. IDE build succeeded, but post build hook, or size checkers failed.
613 # 3. Other python error (directory already exists, ...)
614 # In the case of IDE build failed, we want a significant portion of the output, to be
615 # able to see an indication of what failed.
616 # In the case of size checkers, we want to see all the printouts from all checkers,
617 # to see which one failed.
618 self._print_output(printer=printer, num_lines=25)
620 # Print the regular output from the VUnit class.
621 # A little extra margin between build name and execution time makes the output more readable
622 super().print_status(printer=printer, padding=padding + 2, **kwargs)
623 # Add an empty line between each build, for readability.
624 printer.write("\n")
627def get_build_projects(
628 modules: ModuleList, project_filters: list[str], include_netlist_not_full_builds: bool = False
629) -> list[BuildProject]:
630 """
631 Get build projects from the given modules that match the given filters.
632 Note that the result of this function is a list of "raw" build project objects
633 (:class:`.VivadoProject` or :class:`.YosysNetlistBuild`).
634 These are meant to be passed to :class:`.BuildProjectList` for execution.
636 Arguments:
637 modules: Module objects that can define build projects.
638 project_filters: Project name filters.
639 Can use wildcards (*).
640 Leave empty for all.
641 include_netlist_not_full_builds:
642 Set True to get only netlist builds, instead of only full top level builds.
643 """
644 result = []
645 for module in modules:
646 for project in module.get_build_projects():
647 if project.is_netlist_build == include_netlist_not_full_builds:
648 if not project_filters:
649 result.append(project)
651 else:
652 for project_filter in project_filters:
653 if fnmatch.filter([project.name], project_filter):
654 result.append(project)
656 # Do not continue with further filters if we have already matched
657 # this project.
658 # Multiple filters might match the same project, and we dont't
659 # want duplicates.
660 break
662 return result