Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2004-2005 Elemental Security, Inc. All Rights Reserved.
2# Licensed to PSF under a Contributor Agreement.
4# Modifications:
5# Copyright 2006 Google, Inc. All Rights Reserved.
6# Licensed to PSF under a Contributor Agreement.
8"""Parser driver.
10This provides a high-level interface to parse a file into a syntax tree.
12"""
14__author__ = "Guido van Rossum <guido@python.org>"
16__all__ = ["Driver", "load_grammar"]
18# Python imports
19import logging
20import os
21import pickle
22import sys
23from collections.abc import Iterable, Iterator
24from contextlib import contextmanager
25from dataclasses import dataclass, field
26from logging import Logger
27from typing import Any, Union, cast
29from blib2to3.pgen2.grammar import Grammar
30from blib2to3.pgen2.tokenize import TokenInfo
31from blib2to3.pytree import NL
33# Pgen imports
34from . import grammar, parse, pgen, token, tokenize
36Path = Union[str, "os.PathLike[str]"]
39@dataclass
40class ReleaseRange:
41 start: int
42 end: int | None = None
43 tokens: list[Any] = field(default_factory=list)
45 def lock(self) -> None:
46 total_eaten = len(self.tokens)
47 self.end = self.start + total_eaten
50class TokenProxy:
51 def __init__(self, generator: Any) -> None:
52 self._tokens = generator
53 self._counter = 0
54 self._release_ranges: list[ReleaseRange] = []
56 @contextmanager
57 def release(self) -> Iterator["TokenProxy"]:
58 release_range = ReleaseRange(self._counter)
59 self._release_ranges.append(release_range)
60 try:
61 yield self
62 finally:
63 # Lock the last release range to the final position that
64 # has been eaten.
65 release_range.lock()
67 def eat(self, point: int) -> Any:
68 eaten_tokens = self._release_ranges[-1].tokens
69 if point < len(eaten_tokens):
70 return eaten_tokens[point]
71 else:
72 while point >= len(eaten_tokens):
73 token = next(self._tokens)
74 eaten_tokens.append(token)
75 return token
77 def __iter__(self) -> "TokenProxy":
78 return self
80 def __next__(self) -> Any:
81 # If the current position is already compromised (looked up)
82 # return the eaten token, if not just go further on the given
83 # token producer.
84 # Ranges are appended in counter order and the counter only moves
85 # forward, so any range we've already advanced past can never match
86 # again. Drop that exhausted prefix; otherwise the scan below grows
87 # with the number of soft-keyword lookaheads and parsing a file full
88 # of them (e.g. many `match`/`case` blocks) becomes quadratic.
89 ranges = self._release_ranges
90 while ranges and ranges[0].end is not None and ranges[0].end <= self._counter:
91 ranges.pop(0)
93 for release_range in ranges:
94 assert release_range.end is not None
96 start, end = release_range.start, release_range.end
97 if start <= self._counter < end:
98 token = release_range.tokens[self._counter - start]
99 break
100 else:
101 token = next(self._tokens)
102 self._counter += 1
103 return token
105 def can_advance(self, to: int) -> bool:
106 # Try to eat, fail if it can't. The eat operation is cached
107 # so there won't be any additional cost of eating here
108 try:
109 self.eat(to)
110 except StopIteration:
111 return False
112 else:
113 return True
116class Driver:
117 def __init__(self, grammar: Grammar, logger: Logger | None = None) -> None:
118 self.grammar = grammar
119 if logger is None:
120 logger = logging.getLogger(__name__)
121 self.logger = logger
123 def parse_tokens(self, tokens: Iterable[TokenInfo], debug: bool = False) -> NL:
124 """Parse a series of tokens and return the syntax tree."""
125 # XXX Move the prefix computation into a wrapper around tokenize.
126 proxy = TokenProxy(tokens)
128 p = parse.Parser(self.grammar)
129 p.setup(proxy=proxy)
131 lineno = 1
132 column = 0
133 indent_columns: list[int] = []
134 type = value = start = end = line_text = None
135 prefix = ""
137 for quintuple in proxy:
138 type, value, start, end, line_text = quintuple
139 if start != (lineno, column):
140 assert (lineno, column) <= start, ((lineno, column), start)
141 s_lineno, s_column = start
142 if lineno < s_lineno:
143 prefix += "\n" * (s_lineno - lineno)
144 lineno = s_lineno
145 column = 0
146 if column < s_column:
147 prefix += line_text[column:s_column]
148 column = s_column
149 if type in (tokenize.COMMENT, tokenize.NL):
150 prefix += value
151 lineno, column = end
152 if value.endswith("\n"):
153 lineno += 1
154 column = 0
155 continue
156 if type == token.OP:
157 type = grammar.opmap[value]
158 if debug:
159 assert type is not None
160 self.logger.debug(
161 "%s %r (prefix=%r)", token.tok_name[type], value, prefix
162 )
163 if type == token.INDENT:
164 indent_columns.append(len(value))
165 _prefix = prefix + value
166 prefix = ""
167 value = ""
168 elif type == token.DEDENT:
169 _indent_col = indent_columns.pop()
170 prefix, _prefix = self._partially_consume_prefix(prefix, _indent_col)
171 if p.addtoken(cast(int, type), value, (prefix, start)):
172 if debug:
173 self.logger.debug("Stop.")
174 break
175 prefix = ""
176 if type in {token.INDENT, token.DEDENT}:
177 prefix = _prefix
178 lineno, column = end
179 # FSTRING_MIDDLE and TSTRING_MIDDLE are the only token that can end with a
180 # newline, and `end` will point to the next line. For that case, don't
181 # increment lineno.
182 if value.endswith("\n") and type not in (
183 token.FSTRING_MIDDLE,
184 token.TSTRING_MIDDLE,
185 ):
186 lineno += 1
187 column = 0
188 else:
189 # We never broke out -- EOF is too soon (how can this happen???)
190 assert start is not None
191 raise parse.ParseError("incomplete input", type, value, (prefix, start))
192 assert p.rootnode is not None
193 return p.rootnode
195 def parse_string(self, text: str, debug: bool = False) -> NL:
196 """Parse a string and return the syntax tree."""
197 tokens = tokenize.tokenize(text, grammar=self.grammar)
198 return self.parse_tokens(tokens, debug)
200 def _partially_consume_prefix(self, prefix: str, column: int) -> tuple[str, str]:
201 lines: list[str] = []
202 current_line = ""
203 current_column = 0
204 wait_for_nl = False
205 for char in prefix:
206 current_line += char
207 if wait_for_nl:
208 if char == "\n":
209 if current_line.strip() and current_column < column:
210 res = "".join(lines)
211 return res, prefix[len(res) :]
213 lines.append(current_line)
214 current_line = ""
215 current_column = 0
216 wait_for_nl = False
217 elif char in " \t":
218 current_column += 1
219 elif char == "\n":
220 # unexpected empty line
221 current_column = 0
222 elif char == "\f":
223 current_column = 0
224 else:
225 # indent is finished
226 wait_for_nl = True
227 return "".join(lines), current_line
230def _generate_pickle_name(gt: Path, cache_dir: Path | None = None) -> str:
231 head, tail = os.path.splitext(gt)
232 if tail == ".txt":
233 tail = ""
234 name = head + tail + ".".join(map(str, sys.version_info)) + ".pickle"
235 if cache_dir:
236 return os.path.join(cache_dir, os.path.basename(name))
237 else:
238 return name
241def load_grammar(
242 gt: str = "Grammar.txt",
243 gp: str | None = None,
244 save: bool = True,
245 force: bool = False,
246 logger: Logger | None = None,
247) -> Grammar:
248 """Load the grammar (maybe from a pickle)."""
249 if logger is None:
250 logger = logging.getLogger(__name__)
252 gp = _generate_pickle_name(gt) if gp is None else gp
254 if not force and _newer(gp, gt):
255 try:
256 g_unpickled = grammar.Grammar()
257 g_unpickled.load(gp)
258 return g_unpickled
259 except (
260 pickle.UnpicklingError,
261 EOFError,
262 ValueError,
263 IndexError,
264 PermissionError,
265 ):
266 # Ignore error and continue on to generate a new grammar
267 pass
269 g_generated: grammar.Grammar = pgen.generate_grammar(gt)
270 if save:
271 try:
272 g_generated.dump(gp)
273 except OSError:
274 # Ignore error, caching is not vital.
275 pass
276 return g_generated
279def _newer(a: str, b: str) -> bool:
280 """Inquire whether file a was written since file b."""
281 if not os.path.exists(a):
282 return False
283 if not os.path.exists(b):
284 return True
285 return os.path.getmtime(a) >= os.path.getmtime(b)
288def load_packaged_grammar(
289 grammar_source: str, cache_dir: Path | None = None, *, save: bool = True
290) -> grammar.Grammar:
291 """Loads a grammar by doing `load_grammar(grammar_source)`.
292 This facilitates using a packaged grammar file when needed but preserves
293 load_grammar's automatic regeneration behavior when possible.
294 """
295 gp = _generate_pickle_name(grammar_source, cache_dir) if cache_dir else None
296 return load_grammar(grammar_source, gp=gp, save=save)
299def main(*args: str) -> bool:
300 """Main program, when run as a script: produce grammar pickle files.
302 Calls load_grammar for each argument, a path to a grammar text file.
303 """
304 if not args:
305 args = tuple(sys.argv[1:])
306 logging.basicConfig(level=logging.INFO, stream=sys.stdout, format="%(message)s")
307 for gt in args:
308 load_grammar(gt, save=True, force=True)
309 return True
312if __name__ == "__main__":
313 sys.exit(int(not main()))