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)