Coverage for tsfpga/build_project_list.py: 96%
186 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-06 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 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 )
284 test_runner.run(test_list)
286 all_builds_ok: bool = report.all_ok()
287 report.set_real_total_time(time.time() - start_time)
289 # True if the builds are for the "build" step (not "create" or "open")
290 builds_are_build_step = isinstance(build_wrappers[0], BuildProjectBuildWrapper)
292 if builds_are_build_step:
293 for build_wrapper in build_wrappers:
294 # Update the 'report' object with info about how many lines to print for each build.
295 # This information is only available after the build has finished.
296 report.set_report_length(
297 name=build_wrapper.name,
298 report_length_lines=build_wrapper.report_length_lines,
299 )
301 # If all are OK then we should print the resource utilization numbers.
302 # If not, then we print a few last lines of the log output.
303 if builds_are_build_step or not all_builds_ok:
304 report.print_str()
306 return all_builds_ok
309class BuildProjectWrapper(ABC):
310 """
311 Mimics a VUnit test case object.
312 """
314 def get_seed(self) -> str:
315 """
316 Required since VUnit version 5.0.0.dev6, where a 'get_seed' method was added
317 to the 'TestSuiteWrapper' class, which calls a 'get_seed' method expected to be implemented
318 in the test case object.
319 This mechanism is not used by tsfpga, but is required in order to avoid errors.
320 Adding a dummy implementation like this makes sure it works with older as well as newer
321 versions of VUnit.
322 """
323 return ""
325 @abstractmethod
326 def run(
327 self,
328 output_path: Path,
329 read_output: Any, # noqa: ANN401
330 ) -> bool:
331 pass
334class BuildProjectCreateWrapper(BuildProjectWrapper):
335 """
336 Wrapper to create a build project, for usage in the build runner.
337 """
339 def __init__(
340 self,
341 project: VivadoProject,
342 **kwargs: Any, # noqa: ANN401
343 ) -> None:
344 self.name = project.name
345 self._project = project
346 self._create_arguments = kwargs
348 def run(
349 self,
350 output_path: Path,
351 read_output: Any, # noqa: ANN401, ARG002
352 ) -> bool:
353 """
354 Argument 'read_output' sent by VUnit test runner is unused by us.
355 """
356 this_project_path = Path(output_path) / "project"
357 return self._project.create(project_path=this_project_path, **self._create_arguments)
360class BuildProjectBuildWrapper(BuildProjectWrapper):
361 """
362 Wrapper to build a project, for usage in the build runner.
363 """
365 def __init__(
366 self,
367 project: VivadoProject,
368 collect_artifacts: Callable[..., bool] | None,
369 **kwargs: Any, # noqa: ANN401
370 ) -> None:
371 self.name = project.name
372 self._project = project
373 self._collect_artifacts = collect_artifacts
374 self._build_arguments = kwargs
376 self._report_length_lines: int | None = None
378 def run(
379 self,
380 output_path: Path,
381 read_output: Any, # noqa: ANN401, ARG002
382 ) -> bool:
383 """
384 Argument 'read_output' sent by VUnit test runner is unused by us.
385 """
386 this_project_path = Path(output_path) / "project"
387 build_result = self._project.build(project_path=this_project_path, **self._build_arguments)
389 if not build_result.success:
390 self._print_build_result(build_result=build_result)
391 return build_result.success
393 # Proceed to artifact collection only if build succeeded.
394 if self._collect_artifacts is not None:
395 build_result.success &= self._collect_artifacts(
396 project=self._project, output_path=self._build_arguments["output_path"]
397 )
399 # Print size at the absolute end.
400 self._print_build_result(build_result=build_result)
401 return build_result.success
403 def _print_build_result(self, build_result: build_result.BuildResult) -> None:
404 build_report = build_result.report()
406 if build_report:
407 print(build_report)
408 self._report_length_lines = build_report.count("\n") + 1
410 @property
411 def report_length_lines(self) -> int | None:
412 """
413 The number of lines in the ``build_result`` report from this project.
414 A value of ``None`` would indicate a build failure, either in the IDE or in the
415 post-build steps.
416 """
417 return self._report_length_lines
420class BuildProjectOpenWrapper(BuildProjectWrapper):
421 """
422 Wrapper to open a build project, for usage in the build runner.
423 """
425 def __init__(self, project: VivadoProject) -> None:
426 self.name = project.name
427 self._project = project
429 def run(
430 self,
431 output_path: Path,
432 read_output: Any, # noqa: ANN401, ARG002
433 ) -> bool:
434 """
435 Argument 'read_output' sent by VUnit test runner is unused by us.
436 """
437 this_project_path = Path(output_path) / "project"
438 return self._project.open(project_path=this_project_path)
441class BuildRunner(TestRunner):
442 """
443 Build runner that mimics a VUnit TestRunner. Most things are used as they are in the
444 base class, but some behavior is overridden.
445 """
447 def _create_test_mapping_file(
448 self,
449 test_suites: Any, # noqa: ANN401
450 ) -> None:
451 """
452 Overloaded from super class.
454 Do not create this file.
456 We do not need it since folder name is the same as project name.
457 """
459 def _get_output_path(self, test_suite_name: str) -> str:
460 """
461 Overloaded from super class.
463 Output folder name is the same as the project name.
465 Original function adds a hash at the end of the folder name.
466 We do not want that necessarily.
467 """
468 return str(Path(self._output_path) / test_suite_name)
470 @staticmethod
471 def _prepare_test_suite_output_path(output_path: str) -> None:
472 """
473 Overloaded from super class.
475 Create the directory unless it already exists.
477 Original function wipes the path before running a test. We do not want to do that
478 since e.g. a Vivado project takes a long time to create and might contain a state
479 that the user wants to keep.
480 """
481 create_directory(Path(output_path), empty=False)
484class ThreadSafeCollectArtifacts:
485 """
486 A thread-safe wrapper around a user-supplied function that makes sure the function
487 is not launched more than once at the same time. When two builds finish at the
488 same time, race conditions can arise depending on what the function does.
490 Note that this is a VERY fringe case, since builds usually take >20 minutes, and the
491 collection probably only takes a few seconds. But it happens sometimes with the tsfpga
492 example projects which are identical and quite fast (roughly three minutes).
493 """
495 def __init__(self, collect_artifacts: Callable[[VivadoProject, Path], bool]) -> None:
496 self._collect_artifacts = collect_artifacts
497 self._lock = Lock()
499 def collect_artifacts(self, project: VivadoProject, output_path: Path) -> bool:
500 with self._lock:
501 return self._collect_artifacts(project=project, output_path=output_path)
504class BuildReport(TestReport):
505 def add_result(
506 self,
507 *args: Any, # noqa: ANN401
508 **kwargs: Any, # noqa: ANN401
509 ) -> None:
510 """
511 Overloaded from super class.
513 Add a a test result.
515 Uses a different Result class than the super method.
516 """
517 result = BuildResult(*args, **kwargs)
518 self._test_results[result.name] = result
519 self._test_names_in_order.append(result.name)
521 def set_report_length(self, name: str, report_length_lines: int | None) -> None:
522 """
523 Set how many lines shall be printed for this build.
524 Can be ``None`` to indicate that the build failed, and we don't how much to print.
525 """
526 self._test_results[name].set_report_length(report_length_lines=report_length_lines)
528 def print_latest_status(self, total_tests: int) -> None:
529 """
530 Overloaded from super class.
532 This method is called for each build when it should print its result just as it finished,
533 but other builds may not be finished yet.
535 Inherited and adapted from the VUnit function:
536 * Removed support for the "skipped" result.
537 * Do not use abbreviations in the printout.
538 * Use f-strings.
539 """
540 result = self._last_test_result()
541 passed, failed, _ = self._split()
543 if result.passed:
544 self._printer.write("pass", fg="gi")
545 elif result.failed:
546 self._printer.write("fail", fg="ri")
547 else:
548 raise AssertionError
550 count_summary = f"pass={len(passed)} fail={len(failed)} total={total_tests}"
551 self._printer.write(f" ({count_summary}) {result.name} ({result.time:.1f} seconds)\n")
554class BuildResult(TestResult):
555 _report_length_lines: int | None = None
557 def _print_output(
558 self,
559 printer: ColorPrinter,
560 num_lines: int,
561 ) -> None:
562 """
563 Print the last lines from the output file.
564 """
565 output_tail = read_last_lines_of_file(Path(self._output_file_name), num_lines=num_lines)
566 printer.write(output_tail)
568 def set_report_length(self, report_length_lines: int) -> None:
569 """
570 Set how many lines shall be printed when this result is printed.
571 Can be ``None`` to indicate that the build failed, and we don't how much to print.
572 """
573 self._report_length_lines = report_length_lines
575 def print_status(
576 self,
577 printer: ColorPrinter,
578 padding: int = 0,
579 **kwargs: dict[str, Any],
580 ) -> None:
581 """
582 Overloaded from super class.
584 This method is called for each build when it should print its result in the "Summary" at
585 the end when all builds have finished.
587 Inherited and adapted from the VUnit function.
589 Note that a ``max_time`` integer argument is added in VUnit >4.7.0, but at the time of
590 writing this is un-released on the VUnit ``master`` branch.
591 In order to be compatible with both older and newer versions, we use ``**kwargs`` for this.
592 """
593 if self.passed and self._report_length_lines is not None:
594 # Build passed, print build summary of the specified length. The length is only
595 # set if this is a "build" result (not "create" or "open").
596 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.
605 # In the case of size checkers, we want to see all the printouts from all checkers,
606 # to see which one failed.
607 self._print_output(printer=printer, num_lines=25)
609 # Print the regular output from the VUnit class.
610 # A little extra margin between build name and execution time makes the output more readable
611 super().print_status(printer=printer, padding=padding + 2, **kwargs)
612 # Add an empty line between each build, for readability.
613 printer.write("\n")
616def get_build_projects(
617 modules: ModuleList, project_filters: list[str], include_netlist_not_full_builds: bool
618) -> list[VivadoProject]:
619 """
620 Get build projects from the given modules that match the given filters.
621 Note that the result of this function is a list of "raw" :class:`.VivadoProject` objects.
622 These are meant to be passed to :class:`.BuildProjectList` for execution.
624 Arguments:
625 modules: Module objects that can define build projects.
626 project_filters: Project name filters.
627 Can use wildcards (*).
628 Leave empty for all.
629 include_netlist_not_full_builds:
630 Set True to get only netlist builds, instead of only full top level builds.
631 """
632 result = []
633 for module in modules:
634 for project in module.get_build_projects():
635 if project.is_netlist_build == include_netlist_not_full_builds:
636 if not project_filters:
637 result.append(project)
639 else:
640 for project_filter in project_filters:
641 if fnmatch.filter([project.name], project_filter):
642 result.append(project)
644 # Do not continue with further filters if we have already matched
645 # this project.
646 # Multiple filters might match the same project, and we dont't
647 # want duplicates.
648 break
650 return result