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

190 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 

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 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 not lexer_state.text.is_complete_text(): 

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

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

40 return CustomLexerWrapper0 

41 else: 

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

43 

44 

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

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

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

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

49 parser_conf.callbacks = callbacks 

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

51 

52 

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

54 

55 

56class ParsingFrontend(Serialize): 

57 __serialize_fields__ = 'lexer_conf', 'parser_conf', 'parser' 

58 

59 lexer_conf: LexerConf 

60 parser_conf: ParserConf 

61 options: Any 

62 

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

64 self.parser_conf = parser_conf 

65 self.lexer_conf = lexer_conf 

66 self.options = options 

67 

68 # Set-up parser 

69 if parser: # From cache 

70 self.parser = parser 

71 else: 

72 create_parser = _parser_creators.get(parser_conf.parser_type) 

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

74 parser_conf.parser_type 

75 ) 

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

77 

78 # Set-up lexer 

79 lexer_type = lexer_conf.lexer_type 

80 self.skip_lexer = False 

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

82 assert lexer_conf.postlex is None 

83 self.skip_lexer = True 

84 return 

85 

86 if isinstance(lexer_type, type): 

87 assert issubclass(lexer_type, Lexer) 

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

89 elif isinstance(lexer_type, str): 

90 create_lexer = { 

91 'basic': create_basic_lexer, 

92 'contextual': create_contextual_lexer, 

93 }[lexer_type] 

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

95 else: 

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

97 

98 if lexer_conf.postlex: 

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

100 

101 def _verify_start(self, start=None): 

102 if start is None: 

103 start_decls = self.parser_conf.start 

104 if len(start_decls) > 1: 

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

106 start ,= start_decls 

107 elif start not in self.parser_conf.start: 

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

109 return start 

110 

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

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

113 return text if self.skip_lexer else cls(self.lexer, None) if text is None else cls.from_text(self.lexer, text) 

114 

115 def parse(self, text: Optional[TextOrSlice], start=None, on_error=None): 

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

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

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

119 

120 chosen_start = self._verify_start(start) 

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

122 stream = self._make_lexer_thread(text) 

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

124 

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

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

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

128 chosen_start = self._verify_start(start) 

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

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

131 stream = self._make_lexer_thread(text) 

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

133 

134 

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

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

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

138 expected = { 

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

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

141 'cyk': ('basic', ), 

142 }[parser] 

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

144 

145 

146def _get_lexer_callbacks(transformer, terminals): 

147 result = {} 

148 for terminal in terminals: 

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

150 if callback is not None: 

151 result[terminal.name] = callback 

152 return result 

153 

154class PostLexConnector: 

155 def __init__(self, lexer, postlexer): 

156 self.lexer = lexer 

157 self.postlexer = postlexer 

158 

159 def lex(self, lexer_state, parser_state): 

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

161 return self.postlexer.process(i) 

162 

163 

164 

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

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

167 return cls(lexer_conf) 

168 

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

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

171 parse_table: ParseTableBase[int] = parser._parse_table 

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

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

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

175 

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

177 debug = options.debug if options else False 

178 strict = options.strict if options else False 

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

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

181 

182_parser_creators['lalr'] = create_lalr_parser 

183 

184###} 

185 

186class EarleyRegexpMatcher: 

187 def __init__(self, lexer_conf): 

188 self.regexps = {} 

189 for t in lexer_conf.terminals: 

190 regexp = t.pattern.to_regexp() 

191 try: 

192 width = get_regexp_width(regexp)[0] 

193 except ValueError: 

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

195 else: 

196 if width == 0: 

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

198 if lexer_conf.use_bytes: 

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

200 

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

202 

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

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

205 

206 

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

208 if lexer_conf.callbacks: 

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

210 

211 earley_matcher = EarleyRegexpMatcher(lexer_conf) 

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

213 

214def _match_earley_basic(term, token): 

215 return term.name == token.type 

216 

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

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

219 

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

221 resolve_ambiguity = options.ambiguity == 'resolve' 

222 debug = options.debug if options else False 

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

224 

225 extra = {} 

226 if lexer_conf.lexer_type == 'dynamic': 

227 f = create_earley_parser__dynamic 

228 elif lexer_conf.lexer_type == 'dynamic_complete': 

229 extra['complete_lex'] = True 

230 f = create_earley_parser__dynamic 

231 else: 

232 f = create_earley_parser__basic 

233 

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

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

236 

237 

238 

239class CYK_FrontEnd: 

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

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

242 

243 self.callbacks = parser_conf.callbacks 

244 

245 def parse(self, lexer_thread, start): 

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

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

248 return self._transform(tree) 

249 

250 def _transform(self, tree): 

251 subtrees = list(tree.iter_subtrees()) 

252 for subtree in subtrees: 

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

254 

255 return self._apply_callback(tree) 

256 

257 def _apply_callback(self, tree): 

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

259 

260 

261_parser_creators['earley'] = create_earley_parser 

262_parser_creators['cyk'] = CYK_FrontEnd 

263 

264 

265def _construct_parsing_frontend( 

266 parser_type: _ParserArgType, 

267 lexer_type: _LexerArgType, 

268 lexer_conf, 

269 parser_conf, 

270 options 

271): 

272 assert isinstance(lexer_conf, LexerConf) 

273 assert isinstance(parser_conf, ParserConf) 

274 parser_conf.parser_type = parser_type 

275 lexer_conf.lexer_type = lexer_type 

276 return ParsingFrontend(lexer_conf, parser_conf, options)