Coverage for tsfpga/build_project_list.py: 93%
189 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-02-21 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 pathlib import Path
14from threading import Lock
15from typing import TYPE_CHECKING, Any, Callable
17from vunit.color_printer import COLOR_PRINTER, NO_COLOR_PRINTER, ColorPrinter
18from vunit.test.list import TestList
19from vunit.test.report import TestReport, TestResult
20from vunit.test.runner import TestRunner
22from tsfpga.system_utils import create_directory, read_last_lines_of_file
24if TYPE_CHECKING:
25 from collections.abc import Iterable
27 from .module_list import ModuleList
28 from .vivado import build_result
29 from .vivado.project import VivadoProject
32class BuildProjectList:
33 """
34 Interface to handle a list of FPGA build projects.
35 Enables building many projects in parallel.
36 """
38 def __init__(
39 self,
40 modules: ModuleList,
41 project_filters: list[str],
42 include_netlist_not_top_builds: bool = False,
43 no_color: bool = False,
44 ) -> None:
45 """
46 Arguments:
47 modules: Module objects that can define build projects.
48 project_filters: Project name filters. Can use wildcards (*). Leave empty for all.
49 include_netlist_not_top_builds: Set True to get only netlist builds,
50 instead of only top level builds.
51 no_color: Disable color in printouts.
52 """
53 self._modules = modules
54 self._no_color = no_color
56 self.projects = list(
57 self._iterate_projects(
58 project_filters=project_filters,
59 include_netlist_not_top_builds=include_netlist_not_top_builds,
60 )
61 )
63 if not self.projects:
64 print(f"No projects matched this filter: {' '.join(project_filters)}")
66 def __str__(self) -> str:
67 """
68 Returns a string with a description list of the projects.
70 Will print some information about each project (name, generics, part, ...) so can become
71 long if there are many projects present.
72 An alternative in that case would be :meth:`.get_short_str`.
73 """
74 result = "\n".join([str(project) for project in self.projects])
75 result += "\n"
76 result += "\n"
77 result += f"Listed {len(self.projects)} builds"
79 return result
81 def get_short_str(self) -> str:
82 """
83 Returns a short string with a description list of the projects.
85 This is an alternative function that is more compact than ``__str__``.
86 """
87 result = "\n".join([project.name for project in self.projects])
88 result += "\n"
89 result += f"Listed {len(self.projects)} builds"
91 return result
93 def create(
94 self,
95 projects_path: Path,
96 num_parallel_builds: int,
97 **kwargs: Any, # noqa: ANN401
98 ) -> bool:
99 """
100 Create build project on disk for all the projects in the list.
102 Arguments:
103 projects_path: The projects will be placed here.
104 num_parallel_builds: The number of projects that will be created in parallel.
105 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
107 .. Note::
108 Argument ``project_path`` can not be set, it is set by this class
109 based on the ``project_paths`` argument to this function.
111 Return:
112 True if everything went well.
113 """
114 build_wrappers = []
115 for project in self.projects:
116 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
117 build_wrappers.append(build_wrapper)
119 return self._run_build_wrappers(
120 projects_path=projects_path,
121 build_wrappers=build_wrappers,
122 num_parallel_builds=num_parallel_builds,
123 )
125 def create_unless_exists(
126 self,
127 projects_path: Path,
128 num_parallel_builds: int,
129 **kwargs: Any, # noqa: ANN401
130 ) -> bool:
131 """
132 Create build project for all the projects in the list, unless the project already
133 exists.
135 Arguments:
136 projects_path: The projects will be placed here.
137 num_parallel_builds: The number of projects that will be created in parallel.
138 kwargs: Other arguments as accepted by :meth:`.VivadoProject.create`.
140 .. Note::
141 Argument ``project_path`` can not be set, it is set by this class
142 based on the ``project_paths`` argument to this function.
144 Return:
145 True if everything went well.
146 """
147 build_wrappers = []
148 for project in self.projects:
149 if not (projects_path / project.name / "project").exists():
150 build_wrapper = BuildProjectCreateWrapper(project, **kwargs)
151 build_wrappers.append(build_wrapper)
153 if not build_wrappers:
154 # Return straight away if no projects need to be created. To avoid extra
155 # "No tests were run!" printout from creation step that is very misleading.
156 return True
158 return self._run_build_wrappers(
159 projects_path=projects_path,
160 build_wrappers=build_wrappers,
161 num_parallel_builds=num_parallel_builds,
162 )
164 def build(
165 self,
166 projects_path: Path,
167 num_parallel_builds: int,
168 num_threads_per_build: int,
169 output_path: Path | None = None,
170 collect_artifacts: Callable[[VivadoProject, Path], bool] | None = None,
171 **kwargs: Any, # noqa: ANN401
172 ) -> bool:
173 """
174 Build all the projects in the list.
176 Arguments:
177 projects_path: The projects are placed here.
178 num_parallel_builds: The number of projects that will be built in parallel.
179 num_threads_per_build: The number threads that will be used for each
180 parallel build process.
181 output_path: Where the artifacts should be placed.
182 Will default to within the ``projects_path`` if not set.
183 collect_artifacts: Callback to collect artifacts.
184 Takes two named arguments:
186 | **project** (:class:`.VivadoProject`): The project that is being built.
188 | **output_path** (pathlib.Path): Where the build artifacts should be placed.
190 | Must return True.
191 kwargs: Other arguments as accepted by :meth:`.VivadoProject.build`.
193 .. Note::
194 Argument ``project_path`` can not be set, it is set by this class
195 based on the ``project_paths`` argument to this function.
197 Argument ``num_threads`` is set by the ``num_threads_per_build``
198 argument to this function. This naming difference is done to avoid
199 confusion with regards to ``num_parallel_builds``.
201 Return:
202 True if everything went well.
203 """
204 if collect_artifacts:
205 thread_safe_collect_artifacts = ThreadSafeCollectArtifacts(
206 collect_artifacts
207 ).collect_artifacts
208 else:
209 thread_safe_collect_artifacts = None
211 build_wrappers = []
212 for project in self.projects:
213 project_output_path = self.get_build_project_output_path(
214 project=project, projects_path=projects_path, output_path=output_path
215 )
217 build_wrapper = BuildProjectBuildWrapper(
218 project=project,
219 collect_artifacts=thread_safe_collect_artifacts,
220 output_path=project_output_path,
221 num_threads=num_threads_per_build,
222 **kwargs,
223 )
224 build_wrappers.append(build_wrapper)
226 return self._run_build_wrappers(
227 projects_path=projects_path,
228 build_wrappers=build_wrappers,
229 num_parallel_builds=num_parallel_builds,
230 )
232 @staticmethod
233 def get_build_project_output_path(
234 project: VivadoProject, projects_path: Path, output_path: Path | None = None
235 ) -> Path:
236 """
237 Find where build artifacts will be placed for a project.
238 Arguments are the same as for :meth:`.build`.
239 """
240 if output_path:
241 return output_path.resolve() / project.name
243 return projects_path / project.name
245 def open(self, projects_path: Path) -> bool:
246 """
247 Open the projects in EDA GUI.
249 Arguments:
250 projects_path: The projects are placed here.
252 Return:
253 True if everything went well.
254 """
255 build_wrappers = [BuildProjectOpenWrapper(project=project) for project in self.projects]
257 return self._run_build_wrappers(
258 projects_path=projects_path,
259 build_wrappers=build_wrappers,
260 # For open there is no performance limitation. Set a high value.
261 num_parallel_builds=20,
262 )
264 def _run_build_wrappers(
265 self,
266 projects_path: Path,
267 build_wrappers: list[BuildProjectCreateWrapper]
268 | list[BuildProjectBuildWrapper]
269 | list[BuildProjectOpenWrapper],
270 num_parallel_builds: int,
271 ) -> bool:
272 if not build_wrappers:
273 # Return straight away if no builds are supplied
274 return True
276 start_time = time.time()
278 color_printer = NO_COLOR_PRINTER if self._no_color else COLOR_PRINTER
279 report = BuildReport(printer=color_printer)
281 test_list = TestList()
282 for build_wrapper in build_wrappers:
283 test_list.add_test(build_wrapper)
285 verbosity = BuildRunner.VERBOSITY_QUIET
286 test_runner = BuildRunner(
287 report=report,
288 output_path=projects_path,
289 verbosity=verbosity,
290 num_threads=num_parallel_builds,
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)
299 # If we are building, we should print the summary that is at the end of the console output.
300 # (however if we are creating or opening a project we should not print anything extra).
301 # However if anything has failed, we should also print.
302 if builds_are_build_step:
303 # The length of the build summary depends on if we are working with netlist builds or
304 # regular ones, so set the length given by one of the project objects.
305 report_length_lines = build_wrappers[0].build_result_report_length
306 report.set_report_length(report_length_lines=report_length_lines)
308 # If all are OK then we should print the resource utilization numbers.
309 # If not, then we print a few last lines of the log output.
310 if builds_are_build_step or not all_builds_ok:
311 report.print_str()
313 return all_builds_ok
315 def _iterate_projects(
316 self, project_filters: list[str], include_netlist_not_top_builds: bool
317 ) -> Iterable[VivadoProject]:
318 available_projects = []
319 for module in self._modules:
320 available_projects += module.get_build_projects()
322 for project in available_projects:
323 if project.is_netlist_build == include_netlist_not_top_builds:
324 if not project_filters:
325 yield project
327 else:
328 for project_filter in project_filters:
329 if fnmatch.filter([project.name], project_filter):
330 yield project
332 # Do not continue with further filters if we have already matched this
333 # project.
334 # Multiple filters might match the same project, and multiple objects
335 # of the same project will break build
336 break
339class BuildProjectCreateWrapper:
340 """
341 Wrapper to create a build project, for usage in the build runner.
342 Mimics a VUnit test object.
343 """
345 def __init__(
346 self,
347 project: VivadoProject,
348 **kwargs: Any, # noqa: ANN401
349 ) -> None:
350 self.name = project.name
351 self._project = project
352 self._create_arguments = kwargs
354 def run(
355 self,
356 output_path: Path,
357 read_output: Any, # noqa: ANN401, ARG002
358 ) -> bool:
359 """
360 VUnit test runner sends another argument "read_output" which we don't use.
361 """
362 this_projects_path = Path(output_path) / "project"
363 return self._project.create(project_path=this_projects_path, **self._create_arguments)
366class BuildProjectBuildWrapper:
367 """
368 Wrapper to build a project, for usage in the build runner.
369 Mimics a VUnit test object.
370 """
372 def __init__(
373 self,
374 project: VivadoProject,
375 collect_artifacts: Callable[..., bool] | None,
376 **kwargs: Any, # noqa: ANN401
377 ) -> None:
378 self.name = project.name
379 self._project = project
380 self._collect_artifacts = collect_artifacts
381 self._build_arguments = kwargs
383 def run(
384 self,
385 output_path: Path,
386 read_output: Any, # noqa: ANN401, ARG002
387 ) -> bool:
388 """
389 VUnit test runner sends another argument "read_output" which we don't use.
390 """
391 this_projects_path = Path(output_path) / "project"
392 build_result = self._project.build(project_path=this_projects_path, **self._build_arguments)
394 if not build_result.success:
395 self._print_build_result(build_result)
396 return build_result.success
398 # Proceed to artifact collection only if build succeeded.
399 if self._collect_artifacts and not self._collect_artifacts(
400 project=self._project, output_path=self._build_arguments["output_path"]
401 ):
402 build_result.success = False
404 # Print size at the absolute end
405 self._print_build_result(build_result=build_result)
406 return build_result.success
408 @staticmethod
409 def _print_build_result(build_result: build_result.BuildResult) -> None:
410 build_report = build_result.report()
411 if build_report:
412 # Add an empty line before the build result report, to have margin in how many lines are
413 # printed. See the comments in BuildResult for an explanation.
414 print()
415 print(build_report)
417 @property
418 def build_result_report_length(self) -> int:
419 """
420 The number of lines in the build_result report from this project.
421 """
422 # The size summary, as returned by tsfpga.vivado.project.BuildResult is a JSON formatted
423 # string with one line for each utilization category.
424 # For Xilinx 7 series, there are 8 categories (Total LUTs, Logic LUTs, LUTRAMs,
425 # SRLs, FFs, RAMB36, RAMB18, DSP Blocks). For UltraScale series there is one
426 # extra (URAM).
427 # Additionally, the size summary contains three extra lines for JSON braces and a title.
428 #
429 # This value is enough lines so the whole summary gets printed to console.
430 # For 7 series, this will mean an extra blank line before the summary.
431 #
432 # This is a hack. Works for now, but is far from reliable.
433 length_of_size_report = 3 + 8 + 1
435 if self._project.is_netlist_build:
436 # The logic level distribution report is five lines, plus a title line.
437 # This report is only printed for netlist builds, where there is no configured clock
438 # present. If there were many clocks present in the build, the report would be longer.
439 length_of_logic_level_report = 5 + 1
440 return length_of_size_report + length_of_logic_level_report
442 return length_of_size_report
445class BuildProjectOpenWrapper:
446 """
447 Wrapper to open a build project, for usage in the build runner.
448 Mimics a VUnit test object.
449 """
451 def __init__(self, project: VivadoProject) -> None:
452 self.name = project.name
453 self._project = project
455 def run(
456 self,
457 output_path: Path,
458 read_output: Any, # noqa: ANN401, ARG002
459 ) -> bool:
460 """
461 VUnit test runner sends another argument "read_output" which we don't use.
462 """
463 this_projects_path = Path(output_path) / "project"
464 return self._project.open(project_path=this_projects_path)
467class BuildRunner(TestRunner):
468 """
469 Build runner that mimics a VUnit TestRunner. Most things are used as they are in the
470 base class, but some behavior is overridden.
471 """
473 def _create_test_mapping_file(
474 self,
475 test_suites: Any, # noqa: ANN401
476 ) -> None:
477 """
478 Overloaded from super class.
480 Do not create this file.
482 We do not need it since folder name is the same as project name.
483 """
485 def _get_output_path(self, test_suite_name: str) -> str:
486 """
487 Overloaded from super class.
489 Output folder name is the same as the project name.
491 Original function adds a hash at the end of the folder name.
492 We do not want that necessarily.
493 """
494 return str(Path(self._output_path) / test_suite_name)
496 @staticmethod
497 def _prepare_test_suite_output_path(output_path: str) -> None:
498 """
499 Overloaded from super class.
501 Create the directory unless it already exists.
503 Original function wipes the path before running a test. We do not want to do that
504 since e.g. a Vivado project takes a long time to create and might contain a state
505 that the user wants to keep.
506 """
507 create_directory(Path(output_path), empty=False)
510class ThreadSafeCollectArtifacts:
511 """
512 A thread-safe wrapper around a user-supplied function that makes sure the function
513 is not launched more than once at the same time. When two builds finish at the
514 same time, race conditions can arise depending on what the function does.
516 Note that this is a VERY fringe case, since builds usually take >20 minutes, and the
517 collection probably only takes a few seconds. But it happens sometimes with the tsfpga
518 example projects which are identical and quite fast (roughly three minutes).
519 """
521 def __init__(self, collect_artifacts: Callable[..., bool]) -> None:
522 self._collect_artifacts = collect_artifacts
523 self._lock = Lock()
525 def collect_artifacts(self, project: VivadoProject, output_path: Path) -> bool:
526 with self._lock:
527 return self._collect_artifacts(project=project, output_path=output_path)
530class BuildReport(TestReport):
531 def add_result(
532 self,
533 *args: Any, # noqa: ANN401
534 **kwargs: Any, # noqa: ANN401
535 ) -> None:
536 """
537 Overloaded from super class.
539 Add a a test result.
541 Uses a different Result class than the super method.
542 """
543 result = BuildResult(*args, **kwargs)
544 self._test_results[result.name] = result
545 self._test_names_in_order.append(result.name)
547 def set_report_length(self, report_length_lines: int) -> None:
548 """
549 Set the report length for all test results that have been added to the report.
550 """
551 for test_result in self._test_results.values():
552 test_result.set_report_length(report_length_lines)
554 def print_latest_status(self, total_tests: int) -> None:
555 """
556 Overloaded from super class.
558 This method is called for each build when it should print its result just as it finished,
559 but other builds may not be finished yet.
561 Inherited and adapted from the VUnit function:
562 * Removed support for the "skipped" result.
563 * Do not use abbreviations in the printout.
564 * Use f-strings.
565 """
566 result = self._last_test_result()
567 passed, failed, _ = self._split()
569 if result.passed:
570 self._printer.write("pass", fg="gi")
571 elif result.failed:
572 self._printer.write("fail", fg="ri")
573 else:
574 raise AssertionError
576 count_summary = f"pass={len(passed)} fail={len(failed)} total={total_tests}"
577 self._printer.write(f" ({count_summary}) {result.name} ({result.time:.1f} seconds)\n")
580class BuildResult(TestResult):
581 report_length_lines = None
583 def _print_output(
584 self,
585 printer: ColorPrinter,
586 num_lines: int,
587 ) -> None:
588 """
589 Print the last lines from the output file.
590 """
591 output_tail = read_last_lines_of_file(Path(self._output_file_name), num_lines=num_lines)
592 printer.write(output_tail)
594 def set_report_length(self, report_length_lines: int) -> None:
595 """
596 Set how many lines shall be printed when this result is printed.
597 """
598 self.report_length_lines = report_length_lines
600 def print_status(
601 self,
602 printer: ColorPrinter,
603 padding: int = 0,
604 **kwargs: dict[str, Any],
605 ) -> None:
606 """
607 Overloaded from super class.
609 This method is called for each build when it should print its result in the "Summary" at
610 the end when all builds have finished.
612 Inherited and adapted from the VUnit function.
614 Note that a ``max_time`` integer argument is added in VUnit >4.7.0, but at the time of
615 writing this is un-released on the VUnit ``master`` branch.
616 In order to be compatible with both older and newer versions, we use ``**kwargs`` for this.
617 """
618 if self.passed and self.report_length_lines is not None:
619 # Build passed, print build summary of the specified length. The length is only
620 # set if this is a "build" result (not "create" or "open").
621 self._print_output(printer=printer, num_lines=self.report_length_lines)
622 else:
623 # The build failed, which can either be caused by
624 # 1. IDE build failure
625 # 2. IDE build succeeded, but post build hook, or size checkers failed.
626 # 3. Other python error (directory already exists, ...)
627 # In the case of IDE build failed, we want a significant portion of the output, to be
628 # able to see an indication of what failed. In the case of size checkers, we want to see
629 # all the printouts from all checkers, to see which one failed. Since there are at most
630 # eight resource categories, it is reasonable to assume that there will never be more
631 # than eight size checkers.
632 self._print_output(printer=printer, num_lines=25)
634 # Print the regular output from the VUnit class.
635 # A little extra margin between build name and execution time makes the output more readable
636 super().print_status(printer=printer, padding=padding + 2, **kwargs)
637 # Add an empty line between each build, for readability.
638 printer.write("\n")