Coverage for apio / managers / scons_filter.py: 82%
101 statements
« prev ^ index » next coverage.py v7.13.3, created at 2026-02-08 02:47 +0000
1"""DOC: TODO"""
3# -*- coding: utf-8 -*-
4# -- This file is part of the Apio project
5# -- (C) 2016-2019 FPGAwars
6# -- Author Jesús Arroyo
7# -- License GPLv2
10import re
11import threading
12from enum import Enum
13from typing import List, Optional, Tuple
14from apio.common.apio_console import cout, cunstyle, cwrite, cstyle
15from apio.common.apio_styles import INFO, WARNING, SUCCESS, ERROR
16from apio.utils import util
19# -- A table with line coloring rules. If a line matches any regex, it gets
20# -- The style of the first regex it matches. Patterns are case insensitive.
21LINE_COLORING_TABLE = [
22 # -- Info patterns
23 (r"^info:", INFO),
24 # -- Warning patterns
25 (r"^warning:", WARNING),
26 (r"%warning-", WARNING), # Lint/Verilator
27 # -- Error patterns.
28 (r"^error:", ERROR),
29 (r"^%Error:", ERROR),
30 (r" error: ", ERROR),
31 (r"fail: ", ERROR),
32 (r"fatal: ", ERROR),
33 (r"^fatal error:", ERROR),
34 (r"assertion failed", ERROR),
35 # -- Success patterns
36 (r"is up to date", SUCCESS),
37 (r"[$]finish called", SUCCESS),
38 (r"^verify ok$", SUCCESS),
39 (r"^done$", SUCCESS),
40]
42# -- Lines that contain a substring that match any of these regex's are
43# -- ignored. Regexs are case insensitive.
44LINE_IGNORE_LIST = [
45 # -- Per https://github.com/fpgawars/apio/issues/824
46 # -- TODO: Remove when fixed.
47 r"Warning: define gw1n not used in the library",
48]
51class PipeId(Enum):
52 """Represent the two output streams from the scons subprocess."""
54 STDOUT = 1
55 STDERR = 2
58class RangeEvents(Enum):
59 """An stdout/err line can trigger one of these events, when detecting a
60 range of lines."""
62 START_BEFORE = 1 # Range starts before the current line.
63 START_AFTER = 2 # Range starts after the current line.
64 END_BEFORE = 3 # Range ends before the current line.
65 END_AFTER = 4 # Range ends, after the current line.
68class RangeDetector:
69 """Base detector of a range of lines within the sequence of stdout/err
70 lines recieves from the scons subprocess."""
72 def __init__(self):
73 self._in_range = False
75 def update(self, pipe_id: PipeId, line: str) -> bool:
76 """Updates the range detector with the next stdout/err line.
77 return True iff detector classified this line to be within a range."""
79 prev_state = self._in_range
80 event = self.classify_line(pipe_id, line)
82 if event == RangeEvents.START_BEFORE: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 self._in_range = True
84 return self._in_range
86 if event == RangeEvents.START_AFTER:
87 self._in_range = True
88 return prev_state
90 if event == RangeEvents.END_BEFORE: 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true
91 self._in_range = False
92 return self._in_range
94 if event == RangeEvents.END_AFTER:
95 self._in_range = False
96 return prev_state
98 assert event is None, event
99 return self._in_range
101 def classify_line(
102 self, pipe_id: PipeId, line: str
103 ) -> Optional[RangeEvents]: # pragma: no cover
104 """Tests if the next stdout/err line affects the range begin/end.
105 Subclasses should implement this with the necessary logic for the
106 range that is being detected.
107 Returns the event of None if no event."""
108 raise NotImplementedError("Should be implemented by a subclass")
111class PnrRangeDetector(RangeDetector):
112 """Implements a RangeDetector for the nextpnr command verbose
113 log lines."""
115 def classify_line(self, pipe_id: PipeId, line: str) -> RangeEvents:
116 # -- Break line into words.
117 tokens = line.split()
119 # -- Range start: A nextpnr command on stdout without
120 # -- the -q (quiet) flag.
121 # --
122 # -- IMPORTANT: Each of the supported architecture has a different
123 # -- nextpnr command, including 'nextpnr', 'nextpnr-ecp5', and
124 # -- 'nextpnr-himbaechel'.
125 if (
126 pipe_id == PipeId.STDOUT
127 and line.startswith("nextpnr")
128 and "-q" not in tokens
129 ):
130 return RangeEvents.START_AFTER
132 # Range end: The end message of nextnpr.
133 if pipe_id == PipeId.STDERR and "Program finished normally." in line:
134 return RangeEvents.END_AFTER
136 return None
139class SconsFilter:
140 """Implements the filtering and printing of the stdout/err streams of the
141 scons subprocess. Accepts a line one at a time, detects lines ranges of
142 interest, mutates and colors the lines where applicable, and print to
143 stdout."""
145 def __init__(self, colors_enabled: bool):
146 self.colors_enabled = colors_enabled
147 self._pnr_detector = PnrRangeDetector()
149 # self._iverilog_detector = IVerilogRangeDetector()
150 # self._iceprog_detector = IceProgRangeDetector()
152 # -- We cache the values to avoid reevaluating sys env.
153 self._is_debug = util.is_debug(1)
154 self._is_verbose_debug = util.is_debug(5)
156 # -- Accumulates string pieces until we write and flush them. This
157 # -- mechanism is used to display progress bar correctly, Writing the
158 # -- erasure string only when a new value is available.
159 self._output_bfr: str = ""
161 # -- The stdout and stderr are called from independent threads, so we
162 # -- protect the handling method with this lock.
163 # --
164 # -- We don't protect the third thread which is main(). We hope that
165 # -- it doesn't any print console output while these two threads are
166 # -- active, otherwise it can mingle the output.
167 self._thread_lock = threading.Lock()
169 def on_stdout_line(self, line: str, terminator: str) -> None:
170 """Stdout pipe calls this on each line. Called from the stdout thread
171 in AsyncPipe."""
172 with self._thread_lock:
173 self.on_line(PipeId.STDOUT, line, terminator)
175 def on_stderr_line(self, line: str, terminator: str) -> None:
176 """Stderr pipe calls this on each line. Called from the stderr thread
177 in AsyncPipe."""
178 with self._thread_lock:
179 self.on_line(PipeId.STDERR, line, terminator)
181 @staticmethod
182 def _assign_line_color(
183 line: str, patterns: List[Tuple[str, str]], default_color: str = None
184 ) -> Optional[str]:
185 """Assigns a color for a given line using a list of (regex, color)
186 pairs. Returns the color of the first matching regex (case
187 insensitive), or default_color if none match.
188 """
189 for regex, color in patterns:
190 if re.search(regex, line, re.IGNORECASE):
191 return color
192 return default_color
194 def _output_line(
195 self, line: str, style: Optional[str], terminator: str
196 ) -> None:
197 """Output a line. If a style is given, force that style, otherwise,
198 pass on any color information it may have. The implementation takes
199 into consideration progress bars such as when uploading with the
200 iceprog programmer. These progress bar require certain timing between
201 the chars to have sufficient time to display the text before erasing
202 it."""
204 # -- Apply style if needed.
205 if style:
206 line_part = cstyle(cunstyle(line), style=style)
207 else:
208 line_part = line
210 # -- Get line conditions.
211 is_white = len(line.strip()) == 0
212 is_cr = terminator == "\r"
214 if not is_cr: 214 ↛ 219line 214 didn't jump to line 219 because the condition on line 214 was always true
215 # -- Terminator is EOF or \n. We flush everything.
216 self._output_bfr += line_part + terminator
217 leftover = ""
218 flush = True
219 elif is_white:
220 # -- Terminator is \r and line is white space (progress bar
221 # -- eraser). We queue and and wait for the updated text.
222 self._output_bfr += line_part + terminator
223 leftover = ""
224 flush = False
225 else:
226 # -- Terminator is \r and line has actual text, we flush it out
227 # -- but save queue the \r because on windows 10 cmd it clears the
228 # -- line(?)
229 self._output_bfr += line_part
230 leftover = terminator
231 flush = True
233 if flush: 233 ↛ 239line 233 didn't jump to line 239 because the condition on line 233 was always true
234 # -- Flush the buffer and queue the optional leftover terminator.
235 cwrite(self._output_bfr)
236 self._output_bfr = leftover
237 else:
238 # -- We just queued. Should have no leftover here.
239 assert not leftover
241 def _ignore_line(self, line: str) -> None:
242 """Handle an ignored line. It's dumped if in debug mode."""
243 if self._is_debug: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 cout(f"IGNORED: {line}")
246 def on_line(self, pipe_id: PipeId, line: str, terminator) -> None:
247 """A shared handler for stdout/err lines from the scons sub process.
248 The handler writes both stdout and stderr lines to stdout, possibly
249 with modifications such as text deletion, coloring, and cursor
250 directives.
252 For the possible values of terminator, see AsyncPipe.__init__().
254 NOTE: Ideally, the program specific patterns such as for Fumo and
255 Iceprog should should be condition by a range detector for lines that
256 came from that program. That is to minimize the risk of matching lines
257 from other programs. See the PNR detector for an example.
258 """
260 if self._is_verbose_debug: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 cout(
262 f"*** LINE: [{pipe_id}], [{repr(line)}], [{repr(terminator)}]",
263 style=INFO,
264 )
266 # -- Update the range detectors.
267 in_pnr_verbose_range = self._pnr_detector.update(pipe_id, line)
269 # -- If the line match any of the ignore patterns ignore it.
270 for regex in LINE_IGNORE_LIST:
271 if re.search(regex, line, re.IGNORECASE):
272 self._ignore_line(line)
273 return
275 # -- Remove the 'Info: ' prefix. Nextpnr write a long log where
276 # -- each line starts with "Info: "
277 if ( 277 ↛ 282line 277 didn't jump to line 282 because the condition on line 277 was never true
278 pipe_id == PipeId.STDERR
279 and in_pnr_verbose_range
280 and line.startswith("Info: ")
281 ):
282 line = line[6:]
284 # -- Output the line in the appropriate style.
285 line_color = self._assign_line_color(line, LINE_COLORING_TABLE)
286 self._output_line(line, line_color, terminator)