Coverage for apio/managers/scons_filter.py: 79%
96 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-06 10:20 +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.
21LINE_COLORING_TABLE = [
22 # -- Info patterns
23 (r"^info:", INFO),
24 # -- Warning patterns
25 (r"^warning:", WARNING),
26 # -- Error patterns.
27 (r"^error:", ERROR),
28 (r"fail: ", ERROR),
29 (r"fatal: ", ERROR),
30 (r"^fatal error:", ERROR),
31 (r"assertion failed", ERROR),
32 # -- Success patterns
33 (r"is up to date", SUCCESS),
34 (r"[$]finish called", SUCCESS),
35 (r"^verify ok$", SUCCESS),
36 (r"^done$", SUCCESS),
37]
40class PipeId(Enum):
41 """Represent the two output streams from the scons subprocess."""
43 STDOUT = 1
44 STDERR = 2
47class RangeEvents(Enum):
48 """An stdout/err line can trigger one of these events, when detecting a
49 range of lines."""
51 START_BEFORE = 1 # Range starts before the current line.
52 START_AFTER = 2 # Range starts after the current line.
53 END_BEFORE = 3 # Range ends before the current line.
54 END_AFTER = 4 # Range ends, after the current line.
57class RangeDetector:
58 """Base detector of a range of lines within the sequence of stdout/err
59 lines recieves from the scons subprocess."""
61 def __init__(self):
62 self._in_range = False
64 def update(self, pipe_id: PipeId, line: str) -> bool:
65 """Updates the range detector with the next stdout/err line.
66 return True iff detector classified this line to be within a range."""
68 prev_state = self._in_range
69 event = self.classify_line(pipe_id, line)
71 if event == RangeEvents.START_BEFORE: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 self._in_range = True
73 return self._in_range
75 if event == RangeEvents.START_AFTER:
76 self._in_range = True
77 return prev_state
79 if event == RangeEvents.END_BEFORE: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 self._in_range = False
81 return self._in_range
83 if event == RangeEvents.END_AFTER:
84 self._in_range = False
85 return prev_state
87 assert event is None, event
88 return self._in_range
90 def classify_line(
91 self, pipe_id: PipeId, line: str
92 ) -> Optional[RangeEvents]: # pragma: no cover
93 """Tests if the next stdout/err line affects the range begin/end.
94 Subclasses should implement this with the necessary logic for the
95 range that is being detected.
96 Returns the event of None if no event."""
97 raise NotImplementedError("Should be implemented by a subclass")
100class PnrRangeDetector(RangeDetector):
101 """Implements a RangeDetector for the nextpnr command verbose
102 log lines."""
104 def classify_line(self, pipe_id: PipeId, line: str) -> RangeEvents:
105 # -- Break line into words.
106 tokens = line.split()
108 # -- Range start: A nextpnr command on stdout without
109 # -- the -q (quiet) flag.
110 # --
111 # -- IMPORTANT: Each of the supported architecture has a different
112 # -- nextpnr command, including 'nextpnr', 'nextpnr-ecp5', and
113 # -- 'nextpnr-himbaechel'.
114 if (
115 pipe_id == PipeId.STDOUT
116 and line.startswith("nextpnr")
117 and "-q" not in tokens
118 ):
119 return RangeEvents.START_AFTER
121 # Range end: The end message of nextnpr.
122 if pipe_id == PipeId.STDERR and "Program finished normally." in line:
123 return RangeEvents.END_AFTER
125 return None
128class SconsFilter:
129 """Implements the filtering and printing of the stdout/err streams of the
130 scons subprocess. Accepts a line one at a time, detects lines ranges of
131 interest, mutates and colors the lines where applicable, and print to
132 stdout."""
134 def __init__(self, colors_enabled: bool):
135 self.colors_enabled = colors_enabled
136 self._pnr_detector = PnrRangeDetector()
138 # self._iverilog_detector = IVerilogRangeDetector()
139 # self._iceprog_detector = IceProgRangeDetector()
141 # -- We cache the values to avoid reevaluating sys env.
142 self._is_debug = util.is_debug(1)
143 self._is_verbose_debug = util.is_debug(5)
145 # -- Accumulates string pieces until we write and flush them. This
146 # -- mechanism is used to display progress bar correctly, Writing the
147 # -- erasure string only when a new value is available.
148 self._output_bfr: str = ""
150 # -- The stdout and stderr are called from independent threads, so we
151 # -- protect the handling method with this lock.
152 # --
153 # -- We don't protect the third thread which is main(). We hope that
154 # -- it doesn't any print console output while these two threads are
155 # -- active, otherwise it can mingle the output.
156 self._thread_lock = threading.Lock()
158 def on_stdout_line(self, line: str, terminator: str) -> None:
159 """Stdout pipe calls this on each line. Called from the stdout thread
160 in AsyncPipe."""
161 with self._thread_lock:
162 self.on_line(PipeId.STDOUT, line, terminator)
164 def on_stderr_line(self, line: str, terminator: str) -> None:
165 """Stderr pipe calls this on each line. Called from the stderr thread
166 in AsyncPipe."""
167 with self._thread_lock:
168 self.on_line(PipeId.STDERR, line, terminator)
170 @staticmethod
171 def _assign_line_color(
172 line: str, patterns: List[Tuple[str, str]], default_color: str = None
173 ) -> Optional[str]:
174 """Assigns a color for a given line using a list of (regex, color)
175 pairs. Returns the color of the first matching regex (case
176 insensitive), or default_color if none match.
177 """
178 for regex, color in patterns:
179 if re.search(regex, line, re.IGNORECASE):
180 return color
181 return default_color
183 def _output_line(
184 self, line: str, style: Optional[str], terminator: str
185 ) -> None:
186 """Output a line. If a style is given, force that style, otherwise,
187 pass on any color information it may have. The implementation takes
188 into consideration progress bars such as when uploading with the
189 iceprog programmer. These progress bar require certain timing between
190 the chars to have sufficient time to display the text before erasing
191 it."""
193 # -- Apply style if needed.
194 if style:
195 line_part = cstyle(cunstyle(line), style=style)
196 else:
197 line_part = line
199 # -- Get line conditions.
200 is_white = len(line.strip()) == 0
201 is_cr = terminator == "\r"
203 if not is_cr: 203 ↛ 208line 203 didn't jump to line 208 because the condition on line 203 was always true
204 # -- Terminator is EOF or \n. We flush everything.
205 self._output_bfr += line_part + terminator
206 leftover = ""
207 flush = True
208 elif is_white:
209 # -- Terminator is \r and line is white space (progress bar
210 # -- eraser). We queue and and wait for the updated text.
211 self._output_bfr += line_part + terminator
212 leftover = ""
213 flush = False
214 else:
215 # -- Terminator is \r and line has actual text, we flush it out
216 # -- but save queue the \r because on windows 10 cmd it clears the
217 # -- line(?)
218 self._output_bfr += line_part
219 leftover = terminator
220 flush = True
222 if flush: 222 ↛ 228line 222 didn't jump to line 228 because the condition on line 222 was always true
223 # -- Flush the buffer and queue the optional leftover terminator.
224 cwrite(self._output_bfr)
225 self._output_bfr = leftover
226 else:
227 # -- We just queued. Should have no leftover here.
228 assert not leftover
230 def _ignore_line(self, line: str) -> None:
231 """Handle an ignored line. It's dumped if in debug mode."""
232 if self._is_debug:
233 cout(f"IGNORED: {line}")
235 def on_line(self, pipe_id: PipeId, line: str, terminator) -> None:
236 """A shared handler for stdout/err lines from the scons sub process.
237 The handler writes both stdout and stderr lines to stdout, possibly
238 with modifications such as text deletion, coloring, and cursor
239 directives.
241 For the possible values of terminator, see AsyncPipe.__init__().
243 NOTE: Ideally, the program specific patterns such as for Fumo and
244 Iceprog should should be condition by a range detector for lines that
245 came from that program. That is to minimize the risk of matching lines
246 from other programs. See the PNR detector for an example.
247 """
249 if self._is_verbose_debug: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 cout(
251 f"*** LINE: [{pipe_id}], [{repr(line)}], [{repr(terminator)}]",
252 style=INFO,
253 )
255 # -- Update the range detectors.
256 in_pnr_verbose_range = self._pnr_detector.update(pipe_id, line)
258 # -- Remove the 'Info: ' prefix. Nextpnr write a long log where
259 # -- each line starts with "Info: "
260 if ( 260 ↛ 265line 260 didn't jump to line 265 because the condition on line 260 was never true
261 pipe_id == PipeId.STDERR
262 and in_pnr_verbose_range
263 and line.startswith("Info: ")
264 ):
265 line = line[6:]
267 # -- Output the line in the appropriate style.
268 line_color = self._assign_line_color(line, LINE_COLORING_TABLE)
269 self._output_line(line, line_color, terminator)