Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/lark/parser_frontends.py: 63%

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

198 statements  

1from typing import Any, Callable, Dict, Optional, Collection, Union, TYPE_CHECKING 

2 

3from .exceptions import ConfigurationError, GrammarError, assert_config 

4from .utils import get_regexp_width, Serialize, TextOrSlice, TextSlice, LarkInput 

5from .lexer import LexerThread, BasicLexer, ContextualLexer, Lexer 

6from .parsers import earley, xearley, cyk 

7from .parsers.lalr_parser import LALR_Parser 

8from .tree import Tree 

9from .common import LexerConf, ParserConf, _ParserArgType, _LexerArgType 

10 

11if TYPE_CHECKING: 

12 from .parsers.lalr_analysis import ParseTableBase 

13 

14 

15###{standalone 

16 

17def _wrap_lexer(lexer_class): 

18 future_interface = getattr(lexer_class, '__future_interface__', 0) 

19 if future_interface == 2: 

20 return lexer_class 

21 elif future_interface == 1: 

22 class CustomLexerWrapper1(Lexer): 

23 def __init__(self, lexer_conf): 

24 self.lexer = lexer_class(lexer_conf) 

25 def lex(self, lexer_state, parser_state): 

26 if isinstance(lexer_state.text, TextSlice) and not lexer_state.text.is_complete_text(): 

27 raise TypeError("Interface=1 Custom Lexer don't support TextSlice") 

28 lexer_state.text = lexer_state.text 

29 return self.lexer.lex(lexer_state, parser_state) 

30 return CustomLexerWrapper1 

31 elif future_interface == 0: 

32 class CustomLexerWrapper0(Lexer): 

33 def __init__(self, lexer_conf): 

34 self.lexer = lexer_class(lexer_conf) 

35 

36 def lex(self, lexer_state, parser_state): 

37 if isinstance(lexer_state.text, TextSlice): 

38 if not lexer_state.text.is_complete_text(): 

39 raise TypeError("Interface=0 Custom Lexer don't support TextSlice") 

40 return self.lexer.lex(lexer_state.text.text) 

41 return self.lexer.lex(lexer_state.text) 

42 return CustomLexerWrapper0 

43 else: 

44 raise ValueError(f"Unknown __future_interface__ value {future_interface}, integer 0-2 expected") 

45 

46 

47def _deserialize_parsing_frontend(data, memo, lexer_conf, callbacks, options): 

48 parser_conf = ParserConf.deserialize(data['parser_conf'], memo) 

49 cls = (options and options._plugins.get('LALR_Parser')) or LALR_Parser 

50 parser = cls.deserialize(data['parser'], memo, callbacks, options.debug) 

51 parser_conf.callbacks = callbacks 

52 return ParsingFrontend(lexer_conf, parser_conf, options, parser=parser) 

53 

54 

55_parser_creators: 'Dict[str, Callable[[LexerConf, Any, Any], Any]]' = {} 

56 

57 

58class ParsingFrontend(Serialize): 

59 __serialize_fields__ = 'lexer_conf', 'parser_conf', 'parser' 

60 

61 lexer_conf: LexerConf 

62 parser_conf: ParserConf 

63 options: Any 

64 

65 def __init__(self, lexer_conf: LexerConf, parser_conf: ParserConf, options, parser=None): 

66 self.parser_conf = parser_conf 

67 self.lexer_conf = lexer_conf 

68 self.options = options 

69 

70 # Set-up parser 

71 if parser: # From cache 

72 self.parser = parser 

73 else: 

74 create_parser = _parser_creators.get(parser_conf.parser_type) 

75 assert create_parser is not None, "{} is not supported in standalone mode".format( 

76 parser_conf.parser_type 

77 ) 

78 self.parser = create_parser(lexer_conf, parser_conf, options) 

79 

80 # Set-up lexer 

81 lexer_type = lexer_conf.lexer_type 

82 self.skip_lexer = False 

83 if lexer_type in ('dynamic', 'dynamic_complete'): 

84 assert lexer_conf.postlex is None 

85 self.skip_lexer = True 

86 return 

87 

88 if isinstance(lexer_type, type): 

89 assert issubclass(lexer_type, Lexer) 

90 self.lexer = _wrap_lexer(lexer_type)(lexer_conf) 

91 elif isinstance(lexer_type, str): 

92 create_lexer = { 

93 'basic': create_basic_lexer, 

94 'contextual': create_contextual_lexer, 

95 }[lexer_type] 

96 self.lexer = create_lexer(lexer_conf, self.parser, lexer_conf.postlex, options) 

97 else: 

98 raise TypeError("Bad value for lexer_type: {lexer_type}") 

99 

100 if lexer_conf.postlex: 

101 self.lexer = PostLexConnector(self.lexer, lexer_conf.postlex) 

102 

103 def _verify_start(self, start=None): 

104 if start is None: 

105 start_decls = self.parser_conf.start 

106 if len(start_decls) > 1: 

107 raise ConfigurationError("Lark initialized with more than 1 possible start rule. Must specify which start rule to parse", start_decls) 

108 start ,= start_decls 

109 elif start not in self.parser_conf.start: 

110 raise ConfigurationError("Unknown start rule %s. Must be one of %r" % (start, self.parser_conf.start)) 

111 return start 

112 

113 def _make_lexer_thread(self, text: Optional[LarkInput]) -> Union[LarkInput, LexerThread, None]: 

114 cls = (self.options and self.options._plugins.get('LexerThread')) or LexerThread 

115 if self.skip_lexer: 

116 return text 

117 if text is None: 

118 return cls(self.lexer, None) 

119 if isinstance(text, (str, bytes, TextSlice)): 

120 return cls.from_text(self.lexer, text) 

121 return cls.from_custom_input(self.lexer, text) 

122 

123 def parse(self, text: Optional[LarkInput], start=None, on_error=None): 

124 if self.lexer_conf.lexer_type in ("dynamic", "dynamic_complete"): 

125 if isinstance(text, TextSlice) and not text.is_complete_text(): 

126 raise TypeError(f"Lexer {self.lexer_conf.lexer_type} does not support text slices.") 

127 

128 chosen_start = self._verify_start(start) 

129 kw = {} if on_error is None else {'on_error': on_error} 

130 stream = self._make_lexer_thread(text) 

131 return self.parser.parse(stream, chosen_start, **kw) 

132 

133 def parse_interactive(self, text: Optional[TextOrSlice]=None, start=None): 

134 # TODO BREAK - Change text from Optional[str] to text: str = ''. 

135 # Would break behavior of exhaust_lexer(), which currently raises TypeError, and after the change would just return [] 

136 chosen_start = self._verify_start(start) 

137 if self.parser_conf.parser_type != 'lalr': 

138 raise ConfigurationError("parse_interactive() currently only works with parser='lalr' ") 

139 stream = self._make_lexer_thread(text) 

140 return self.parser.parse_interactive(stream, chosen_start) 

141 

142 

143def _validate_frontend_args(parser, lexer) -> None: 

144 assert_config(parser, ('lalr', 'earley', 'cyk')) 

145 if not isinstance(lexer, type): # not custom lexer? 

146 expected = { 

147 'lalr': ('basic', 'contextual'), 

148 'earley': ('basic', 'dynamic', 'dynamic_complete'), 

149 'cyk': ('basic', ), 

150 }[parser] 

151 assert_config(lexer, expected, 'Parser %r does not support lexer %%r, expected one of %%s' % parser) 

152 

153 

154def _get_lexer_callbacks(transformer, terminals): 

155 result = {} 

156 for terminal in terminals: 

157 callback = getattr(transformer, terminal.name, None) 

158 if callback is not None: 

159 result[terminal.name] = callback 

160 return result 

161 

162class PostLexConnector: 

163 def __init__(self, lexer, postlexer): 

164 self.lexer = lexer 

165 self.postlexer = postlexer 

166 

167 def lex(self, lexer_state, parser_state): 

168 i = self.lexer.lex(lexer_state, parser_state) 

169 return self.postlexer.process(i) 

170 

171 

172 

173def create_basic_lexer(lexer_conf, parser, postlex, options) -> BasicLexer: 

174 cls = (options and options._plugins.get('BasicLexer')) or BasicLexer 

175 return cls(lexer_conf) 

176 

177def create_contextual_lexer(lexer_conf: LexerConf, parser, postlex, options) -> ContextualLexer: 

178 cls = (options and options._plugins.get('ContextualLexer')) or ContextualLexer 

179 parse_table: ParseTableBase[int] = parser._parse_table 

180 states: Dict[int, Collection[str]] = {idx:list(t.keys()) for idx, t in parse_table.states.items()} 

181 always_accept: Collection[str] = postlex.always_accept if postlex else () 

182 return cls(lexer_conf, states, always_accept=always_accept) 

183 

184def create_lalr_parser(lexer_conf: LexerConf, parser_conf: ParserConf, options=None) -> LALR_Parser: 

185 debug = options.debug if options else False 

186 strict = options.strict if options else False 

187 cls = (options and options._plugins.get('LALR_Parser')) or LALR_Parser 

188 return cls(parser_conf, debug=debug, strict=strict) 

189 

190_parser_creators['lalr'] = create_lalr_parser 

191 

192###} 

193 

194class EarleyRegexpMatcher: 

195 def __init__(self, lexer_conf): 

196 self.regexps = {} 

197 for t in lexer_conf.terminals: 

198 regexp = t.pattern.to_regexp() 

199 try: 

200 width = get_regexp_width(regexp)[0] 

201 except ValueError: 

202 raise GrammarError("Bad regexp in token %s: %s" % (t.name, regexp)) 

203 else: 

204 if width == 0: 

205 raise GrammarError("Dynamic Earley doesn't allow zero-width regexps", t) 

206 if lexer_conf.use_bytes: 

207 regexp = regexp.encode('utf-8') 

208 

209 self.regexps[t.name] = lexer_conf.re_module.compile(regexp, lexer_conf.g_regex_flags) 

210 

211 def match(self, term, text, index=0): 

212 return self.regexps[term.name].match(text, index) 

213 

214 

215def create_earley_parser__dynamic(lexer_conf: LexerConf, parser_conf: ParserConf, **kw): 

216 if lexer_conf.callbacks: 

217 raise GrammarError("Earley's dynamic lexer doesn't support lexer_callbacks.") 

218 

219 earley_matcher = EarleyRegexpMatcher(lexer_conf) 

220 return xearley.Parser(lexer_conf, parser_conf, earley_matcher.match, **kw) 

221 

222def _match_earley_basic(term, token): 

223 return term.name == token.type 

224 

225def create_earley_parser__basic(lexer_conf: LexerConf, parser_conf: ParserConf, **kw): 

226 return earley.Parser(lexer_conf, parser_conf, _match_earley_basic, **kw) 

227 

228def create_earley_parser(lexer_conf: LexerConf, parser_conf: ParserConf, options) -> earley.Parser: 

229 resolve_ambiguity = options.ambiguity == 'resolve' 

230 debug = options.debug if options else False 

231 tree_class = options.tree_class or Tree if options.ambiguity != 'forest' else None 

232 

233 extra = {} 

234 if lexer_conf.lexer_type == 'dynamic': 

235 f = create_earley_parser__dynamic 

236 elif lexer_conf.lexer_type == 'dynamic_complete': 

237 extra['complete_lex'] = True 

238 f = create_earley_parser__dynamic 

239 else: 

240 f = create_earley_parser__basic 

241 

242 return f(lexer_conf, parser_conf, resolve_ambiguity=resolve_ambiguity, 

243 debug=debug, tree_class=tree_class, ordered_sets=options.ordered_sets, **extra) 

244 

245 

246 

247class CYK_FrontEnd: 

248 def __init__(self, lexer_conf, parser_conf, options=None): 

249 self.parser = cyk.Parser(parser_conf.rules) 

250 

251 self.callbacks = parser_conf.callbacks 

252 

253 def parse(self, lexer_thread, start): 

254 tokens = list(lexer_thread.lex(None)) 

255 tree = self.parser.parse(tokens, start) 

256 return self._transform(tree) 

257 

258 def _transform(self, tree): 

259 subtrees = list(tree.iter_subtrees()) 

260 for subtree in subtrees: 

261 subtree.children = [self._apply_callback(c) if isinstance(c, Tree) else c for c in subtree.children] 

262 

263 return self._apply_callback(tree) 

264 

265 def _apply_callback(self, tree): 

266 return self.callbacks[tree.rule](tree.children) 

267 

268 

269_parser_creators['earley'] = create_earley_parser 

270_parser_creators['cyk'] = CYK_FrontEnd 

271 

272 

273def _construct_parsing_frontend( 

274 parser_type: _ParserArgType, 

275 lexer_type: _LexerArgType, 

276 lexer_conf, 

277 parser_conf, 

278 options 

279): 

280 assert isinstance(lexer_conf, LexerConf) 

281 assert isinstance(parser_conf, ParserConf) 

282 parser_conf.parser_type = parser_type 

283 lexer_conf.lexer_type = lexer_type 

284 return ParsingFrontend(lexer_conf, parser_conf, options)