Coverage for /pythoncovmergedfiles/medio/medio/src/black/src/blib2to3/pgen2/driver.py: 87%

1# Copyright 2004-2005 Elemental Security, Inc. All Rights Reserved. 

2# Licensed to PSF under a Contributor Agreement. 

3 

4# Modifications: 

5# Copyright 2006 Google, Inc. All Rights Reserved. 

6# Licensed to PSF under a Contributor Agreement. 

7 

8"""Parser driver. 

9 

10This provides a high-level interface to parse a file into a syntax tree. 

11 

12""" 

13 

14__author__ = "Guido van Rossum <guido@python.org>" 

15 

16__all__ = ["Driver", "load_grammar"] 

17 

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 

28 

29from blib2to3.pgen2.grammar import Grammar 

30from blib2to3.pgen2.tokenize import TokenInfo 

31from blib2to3.pytree import NL 

32 

33# Pgen imports 

34from . import grammar, parse, pgen, token, tokenize 

35 

36Path = Union[str, "os.PathLike[str]"] 

37 

38 

39@dataclass 

40class ReleaseRange: 

41 start: int 

42 end: int | None = None 

43 tokens: list[Any] = field(default_factory=list) 

44 

45 def lock(self) -> None: 

46 total_eaten = len(self.tokens) 

47 self.end = self.start + total_eaten 

48 

49 

50class TokenProxy: 

51 def __init__(self, generator: Any) -> None: 

52 self._tokens = generator 

53 self._counter = 0 

54 self._release_ranges: list[ReleaseRange] = [] 

55 

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() 

66 

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 

76 

77 def __iter__(self) -> "TokenProxy": 

78 return self 

79 

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) 

92 

93 for release_range in ranges: 

94 assert release_range.end is not None 

95 

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 

104 

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 

114 

115 

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 

122 

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) 

127 

128 p = parse.Parser(self.grammar) 

129 p.setup(proxy=proxy) 

130 

131 lineno = 1 

132 column = 0 

133 indent_columns: list[int] = [] 

134 type = value = start = end = line_text = None 

135 prefix = "" 

136 

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 

194 

195 def parse_file( 

196 self, filename: Path, encoding: str | None = None, debug: bool = False 

197 ) -> NL: 

198 """Parse a file and return the syntax tree.""" 

199 with open(filename, encoding=encoding) as stream: 

200 text = stream.read() 

201 return self.parse_string(text, debug) 

202 

203 def parse_string(self, text: str, debug: bool = False) -> NL: 

204 """Parse a string and return the syntax tree.""" 

205 tokens = tokenize.tokenize(text, grammar=self.grammar) 

206 return self.parse_tokens(tokens, debug) 

207 

208 def _partially_consume_prefix(self, prefix: str, column: int) -> tuple[str, str]: 

209 lines: list[str] = [] 

210 current_line = "" 

211 current_column = 0 

212 wait_for_nl = False 

213 for char in prefix: 

214 current_line += char 

215 if wait_for_nl: 

216 if char == "\n": 

217 if current_line.strip() and current_column < column: 

218 res = "".join(lines) 

219 return res, prefix[len(res) :] 

220 

221 lines.append(current_line) 

222 current_line = "" 

223 current_column = 0 

224 wait_for_nl = False 

225 elif char in " \t": 

226 current_column += 1 

227 elif char == "\n": 

228 # unexpected empty line 

229 current_column = 0 

230 elif char == "\f": 

231 current_column = 0 

232 else: 

233 # indent is finished 

234 wait_for_nl = True 

235 return "".join(lines), current_line 

236 

237 

238def _generate_pickle_name(gt: Path, cache_dir: Path | None = None) -> str: 

239 head, tail = os.path.splitext(gt) 

240 if tail == ".txt": 

241 tail = "" 

242 name = head + tail + ".".join(map(str, sys.version_info)) + ".pickle" 

243 if cache_dir: 

244 return os.path.join(cache_dir, os.path.basename(name)) 

245 else: 

246 return name 

247 

248 

249def load_grammar( 

250 gt: str = "Grammar.txt", 

251 gp: str | None = None, 

252 save: bool = True, 

253 force: bool = False, 

254 logger: Logger | None = None, 

255) -> Grammar: 

256 """Load the grammar (maybe from a pickle).""" 

257 if logger is None: 

258 logger = logging.getLogger(__name__) 

259 

260 gp = _generate_pickle_name(gt) if gp is None else gp 

261 

262 if not force and _newer(gp, gt): 

263 try: 

264 g_unpickled = grammar.Grammar() 

265 g_unpickled.load(gp) 

266 return g_unpickled 

267 except ( 

268 pickle.UnpicklingError, 

269 EOFError, 

270 ValueError, 

271 IndexError, 

272 PermissionError, 

273 ): 

274 # Ignore error and continue on to generate a new grammar 

275 pass 

276 

277 g_generated: grammar.Grammar = pgen.generate_grammar(gt) 

278 if save: 

279 try: 

280 g_generated.dump(gp) 

281 except OSError: 

282 # Ignore error, caching is not vital. 

283 pass 

284 return g_generated 

285 

286 

287def _newer(a: str, b: str) -> bool: 

288 """Inquire whether file a was written since file b.""" 

289 if not os.path.exists(a): 

290 return False 

291 if not os.path.exists(b): 

292 return True 

293 return os.path.getmtime(a) >= os.path.getmtime(b) 

294 

295 

296def load_packaged_grammar( 

297 grammar_source: str, cache_dir: Path | None = None 

298) -> grammar.Grammar: 

299 """Loads a grammar by doing `load_grammar(grammar_source)`. 

300 This facilitates using a packaged grammar file when needed but preserves 

301 load_grammar's automatic regeneration behavior when possible. 

302 """ 

303 gp = _generate_pickle_name(grammar_source, cache_dir) if cache_dir else None 

304 return load_grammar(grammar_source, gp=gp) 

305 

306 

307def main(*args: str) -> bool: 

308 """Main program, when run as a script: produce grammar pickle files. 

309 

310 Calls load_grammar for each argument, a path to a grammar text file. 

311 """ 

312 if not args: 

313 args = tuple(sys.argv[1:]) 

314 logging.basicConfig(level=logging.INFO, stream=sys.stdout, format="%(message)s") 

315 for gt in args: 

316 load_grammar(gt, save=True, force=True) 

317 return True 

318 

319 

320if __name__ == "__main__": 

321 sys.exit(int(not main()))