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
1from dataclasses import dataclass
2from typing import Any, Callable, Dict, Optional, Collection, Union, TYPE_CHECKING, Generic, Iterator, Tuple, TypeVar
4from .exceptions import ConfigurationError, GrammarError, LexError, UnexpectedInput, assert_config
5from .utils import get_regexp_width, Serialize, TextOrSlice, TextSlice, LarkInput
6from .lexer import LexerThread, LineCounter, Token, _TextSlice_WithLineCount, BasicLexer, ContextualLexer, Lexer
7from .parsers import earley, xearley, cyk
8from .parsers.lalr_parser import LALR_Parser
9from .tree import Tree
10from .common import LexerConf, ParserConf, _ParserArgType, _LexerArgType
12if TYPE_CHECKING:
13 from .parsers.lalr_analysis import ParseTableBase
16###{standalone
18T = TypeVar('T')
20@dataclass(frozen=True)
21class ScanMatch(Generic[T]):
22 """A non-overlapping match found by ``Lark.scan()``.
24 Attributes:
25 range: A (start, end) tuple of the indices in the input text.
26 value: The parse result. A ``Tree`` by default, or whatever the
27 ``transformer`` returns when one was supplied to Lark.
28 """
29 range: Tuple[int, int]
30 value: T
33def _wrap_lexer(lexer_class):
34 future_interface = getattr(lexer_class, '__future_interface__', 0)
35 if future_interface == 2:
36 return lexer_class
37 elif future_interface == 1:
38 class CustomLexerWrapper1(Lexer):
39 def __init__(self, lexer_conf):
40 self.lexer = lexer_class(lexer_conf)
41 def lex(self, lexer_state, parser_state):
42 if isinstance(lexer_state.text, TextSlice) and not lexer_state.text.is_complete_text():
43 raise TypeError("Interface=1 Custom Lexer don't support TextSlice")
44 lexer_state.text = lexer_state.text
45 return self.lexer.lex(lexer_state, parser_state)
46 return CustomLexerWrapper1
47 elif future_interface == 0:
48 class CustomLexerWrapper0(Lexer):
49 def __init__(self, lexer_conf):
50 self.lexer = lexer_class(lexer_conf)
52 def lex(self, lexer_state, parser_state):
53 if isinstance(lexer_state.text, TextSlice):
54 if not lexer_state.text.is_complete_text():
55 raise TypeError("Interface=0 Custom Lexer don't support TextSlice")
56 return self.lexer.lex(lexer_state.text.text)
57 return self.lexer.lex(lexer_state.text)
58 return CustomLexerWrapper0
59 else:
60 raise ValueError(f"Unknown __future_interface__ value {future_interface}, integer 0-2 expected")
63def _deserialize_parsing_frontend(data, memo, lexer_conf, callbacks, options):
64 parser_conf = ParserConf.deserialize(data['parser_conf'], memo)
65 cls = (options and options._plugins.get('LALR_Parser')) or LALR_Parser
66 parser = cls.deserialize(data['parser'], memo, callbacks, options.debug)
67 parser_conf.callbacks = callbacks
68 return ParsingFrontend(lexer_conf, parser_conf, options, parser=parser)
71_parser_creators: 'Dict[str, Callable[[LexerConf, Any, Any], Any]]' = {}
74class ParsingFrontend(Serialize):
75 __serialize_fields__ = 'lexer_conf', 'parser_conf', 'parser'
77 lexer_conf: LexerConf
78 parser_conf: ParserConf
79 options: Any
81 def __init__(self, lexer_conf: LexerConf, parser_conf: ParserConf, options, parser=None):
82 self.parser_conf = parser_conf
83 self.lexer_conf = lexer_conf
84 self.options = options
86 # Set-up parser
87 if parser: # From cache
88 self.parser = parser
89 else:
90 create_parser = _parser_creators.get(parser_conf.parser_type)
91 assert create_parser is not None, "{} is not supported in standalone mode".format(
92 parser_conf.parser_type
93 )
94 self.parser = create_parser(lexer_conf, parser_conf, options)
96 # Set-up lexer
97 lexer_type = lexer_conf.lexer_type
98 self.skip_lexer = False
99 if lexer_type in ('dynamic', 'dynamic_complete'):
100 assert lexer_conf.postlex is None
101 self.skip_lexer = True
102 return
104 if isinstance(lexer_type, type):
105 assert issubclass(lexer_type, Lexer)
106 self.lexer = _wrap_lexer(lexer_type)(lexer_conf)
107 elif isinstance(lexer_type, str):
108 create_lexer = {
109 'basic': create_basic_lexer,
110 'contextual': create_contextual_lexer,
111 }[lexer_type]
112 self.lexer = create_lexer(lexer_conf, self.parser, lexer_conf.postlex, options)
113 else:
114 raise TypeError("Bad value for lexer_type: {lexer_type}")
116 if lexer_conf.postlex:
117 self.lexer = PostLexConnector(self.lexer, lexer_conf.postlex)
119 def _verify_start(self, start=None):
120 if start is None:
121 start_decls = self.parser_conf.start
122 if len(start_decls) > 1:
123 raise ConfigurationError("Lark initialized with more than 1 possible start rule. Must specify which start rule to parse", start_decls)
124 start ,= start_decls
125 elif start not in self.parser_conf.start:
126 raise ConfigurationError("Unknown start rule %s. Must be one of %r" % (start, self.parser_conf.start))
127 return start
129 def _make_lexer_thread(self, text: Optional[LarkInput]) -> Union[LarkInput, LexerThread, None]:
130 cls = (self.options and self.options._plugins.get('LexerThread')) or LexerThread
131 if self.skip_lexer:
132 return text
133 if text is None:
134 return cls(self.lexer, None)
135 if isinstance(text, (str, bytes, TextSlice)):
136 return cls.from_text(self.lexer, text)
137 return cls.from_custom_input(self.lexer, text)
139 def parse(self, text: Optional[LarkInput], start=None, on_error=None):
140 if self.lexer_conf.lexer_type in ("dynamic", "dynamic_complete"):
141 if isinstance(text, TextSlice) and not text.is_complete_text():
142 raise TypeError(f"Lexer {self.lexer_conf.lexer_type} does not support text slices.")
144 chosen_start = self._verify_start(start)
145 kw = {} if on_error is None else {'on_error': on_error}
146 stream = self._make_lexer_thread(text)
147 return self.parser.parse(stream, chosen_start, **kw)
149 def parse_interactive(self, text: Optional[TextOrSlice]=None, start=None):
150 # TODO BREAK - Change text from Optional[str] to text: str = ''.
151 # Would break behavior of exhaust_lexer(), which currently raises TypeError, and after the change would just return []
152 chosen_start = self._verify_start(start)
153 if self.parser_conf.parser_type != 'lalr':
154 raise ConfigurationError("parse_interactive() currently only works with parser='lalr' ")
155 stream = self._make_lexer_thread(text)
156 return self.parser.parse_interactive(stream, chosen_start)
158 def scan(self, text: TextOrSlice, start: Optional[str]=None) -> Iterator[ScanMatch]:
159 """See ``Lark.scan``."""
160 if self.parser_conf.parser_type != 'lalr':
161 raise ConfigurationError("scan() requires parser='lalr'")
162 if self.skip_lexer:
163 raise ConfigurationError("scan() does not support lexer='dynamic'/'dynamic_complete'")
164 if self.lexer_conf.postlex is not None:
165 # postlex carries state across the stream (indent depth, paren nesting); mid-stream parses break it.
166 raise ConfigurationError("scan() does not support postlex")
167 if isinstance(self.lexer_conf.lexer_type, type):
168 # A custom lexer class was supplied; scan() relies on the built-in lexers' search_start().
169 raise ConfigurationError("scan() does not support custom lexers")
170 chosen_start = self._verify_start(start)
171 return self._scan(TextSlice.cast_from(text), chosen_start)
173 def _scan(self, text_slice: TextSlice, chosen_start: str) -> Iterator[ScanMatch]:
174 start_state = self.parser._parse_table.start_states[chosen_start]
175 pos = text_slice.start
176 # We count the lines here, to avoid re-counting them inside each new lexer state
177 line_ctr = LineCounter.from_text_slice(text_slice)
178 while True:
179 # Search for a plausible start
180 match_start = self.lexer.search_start(text_slice, start_state, pos)
181 if match_start is None:
182 return
183 assert text_slice.start <= match_start <= text_slice.end
185 # Parse without callbacks, to keep value-stack minimal and avoid expensive deepcopies.
186 # Aim for the longest possible match, and save the tokens we lex for later replay.
187 line_ctr.advance_to(text_slice.text, match_start)
188 text_slice_wlc = _TextSlice_WithLineCount(
189 text_slice.text, match_start, text_slice.end,
190 line_ctr.line, line_ctr.line_start_pos)
191 stunted_ip = self.parse_interactive(text_slice_wlc, start=chosen_start)
192 stunted_ip.parser_state.parse_conf.callbacks = {}
193 matched_tokens = []
194 longest_match = 0 # number of tokens in the longest accepted prefix
195 token_stream = stunted_ip.lexer_thread.lex(stunted_ip.parser_state)
196 try:
197 for token in token_stream:
198 stunted_ip.feed_token(token)
199 matched_tokens.append(token)
200 # Test if we reached a possible completed parse
201 if '$END' in stunted_ip.choices():
202 tmp_state = stunted_ip.parser_state.copy(deepcopy_values=False)
203 try:
204 tmp_state.feed_token(Token.new_borrow_pos('$END', '', token), is_end=True)
205 except UnexpectedInput:
206 continue
207 longest_match = len(matched_tokens)
208 # keep going and testing for candidates, until the parse ends or fails
209 except UnexpectedInput:
210 # Parse failed
211 pass
212 except ConfigurationError:
213 # ConfigurationError subclasses ValueError, and must not be swallowed
214 raise
215 except ValueError:
216 # A user lexer-callback raised an error
217 pass
219 if longest_match:
220 # Match found! Replay tokens with real callbacks, and yield the result
221 matched = matched_tokens[:longest_match]
222 replay_ip = self.parse_interactive(start=chosen_start)
223 for t in matched:
224 if t.start_pos is None or t.end_pos is None:
225 raise LexError(
226 f"Lexer callback for {t.type!r} did not preserve token positions; "
227 f"scan() requires source positions on every token (use Token.update() in callbacks).")
228 replay_ip.feed_token(t)
229 res = replay_ip.feed_eof(matched[-1])
230 # Range comes from the matched tokens, not match_start (the lexer may skip leading ignores).
231 yield ScanMatch((matched[0].start_pos, matched[-1].end_pos), res)
232 # Resume from end of match (no overlaps)
233 pos = matched[-1].end_pos
234 else:
235 # No match found. Scan again from next character
236 pos = match_start + 1
239def _validate_frontend_args(parser, lexer) -> None:
240 assert_config(parser, ('lalr', 'earley', 'cyk'))
241 if not isinstance(lexer, type): # not custom lexer?
242 expected = {
243 'lalr': ('basic', 'contextual'),
244 'earley': ('basic', 'dynamic', 'dynamic_complete'),
245 'cyk': ('basic', ),
246 }[parser]
247 assert_config(lexer, expected, 'Parser %r does not support lexer %%r, expected one of %%s' % parser)
250def _get_lexer_callbacks(transformer, terminals):
251 result = {}
252 for terminal in terminals:
253 callback = getattr(transformer, terminal.name, None)
254 if callback is not None:
255 result[terminal.name] = callback
256 return result
258class PostLexConnector:
259 def __init__(self, lexer, postlexer):
260 self.lexer = lexer
261 self.postlexer = postlexer
263 def lex(self, lexer_state, parser_state):
264 i = self.lexer.lex(lexer_state, parser_state)
265 return self.postlexer.process(i)
269def create_basic_lexer(lexer_conf, parser, postlex, options) -> BasicLexer:
270 cls = (options and options._plugins.get('BasicLexer')) or BasicLexer
271 return cls(lexer_conf)
273def create_contextual_lexer(lexer_conf: LexerConf, parser, postlex, options) -> ContextualLexer:
274 cls = (options and options._plugins.get('ContextualLexer')) or ContextualLexer
275 parse_table: ParseTableBase[int] = parser._parse_table
276 states: Dict[int, Collection[str]] = {idx:list(t.keys()) for idx, t in parse_table.states.items()}
277 always_accept: Collection[str] = postlex.always_accept if postlex else ()
278 return cls(lexer_conf, states, always_accept=always_accept)
280def create_lalr_parser(lexer_conf: LexerConf, parser_conf: ParserConf, options=None) -> LALR_Parser:
281 debug = options.debug if options else False
282 strict = options.strict if options else False
283 cls = (options and options._plugins.get('LALR_Parser')) or LALR_Parser
284 return cls(parser_conf, debug=debug, strict=strict)
286_parser_creators['lalr'] = create_lalr_parser
288###}
290class EarleyRegexpMatcher:
291 def __init__(self, lexer_conf):
292 self.regexps = {}
293 for t in lexer_conf.terminals:
294 regexp = t.pattern.to_regexp()
295 try:
296 width = get_regexp_width(regexp)[0]
297 except ValueError:
298 raise GrammarError("Bad regexp in token %s: %s" % (t.name, regexp))
299 else:
300 if width == 0:
301 raise GrammarError("Dynamic Earley doesn't allow zero-width regexps", t)
302 if lexer_conf.use_bytes:
303 regexp = regexp.encode('utf-8')
305 self.regexps[t.name] = lexer_conf.re_module.compile(regexp, lexer_conf.g_regex_flags)
307 def match(self, term, text, index=0):
308 return self.regexps[term.name].match(text, index)
311def create_earley_parser__dynamic(lexer_conf: LexerConf, parser_conf: ParserConf, **kw):
312 if lexer_conf.callbacks:
313 raise GrammarError("Earley's dynamic lexer doesn't support lexer_callbacks.")
315 earley_matcher = EarleyRegexpMatcher(lexer_conf)
316 return xearley.Parser(lexer_conf, parser_conf, earley_matcher.match, **kw)
318def _match_earley_basic(term, token):
319 return term.name == token.type
321def create_earley_parser__basic(lexer_conf: LexerConf, parser_conf: ParserConf, **kw):
322 return earley.Parser(lexer_conf, parser_conf, _match_earley_basic, **kw)
324def create_earley_parser(lexer_conf: LexerConf, parser_conf: ParserConf, options) -> earley.Parser:
325 resolve_ambiguity = options.ambiguity == 'resolve'
326 debug = options.debug if options else False
327 tree_class = options.tree_class or Tree if options.ambiguity != 'forest' else None
329 extra = {}
330 if lexer_conf.lexer_type == 'dynamic':
331 f = create_earley_parser__dynamic
332 elif lexer_conf.lexer_type == 'dynamic_complete':
333 extra['complete_lex'] = True
334 f = create_earley_parser__dynamic
335 else:
336 f = create_earley_parser__basic
338 return f(lexer_conf, parser_conf, resolve_ambiguity=resolve_ambiguity,
339 debug=debug, tree_class=tree_class, ordered_sets=options.ordered_sets, **extra)
343class CYK_FrontEnd:
344 def __init__(self, lexer_conf, parser_conf, options=None):
345 self.parser = cyk.Parser(parser_conf.rules)
347 self.callbacks = parser_conf.callbacks
349 def parse(self, lexer_thread, start):
350 tokens = list(lexer_thread.lex(None))
351 tree = self.parser.parse(tokens, start)
352 return self._transform(tree)
354 def _transform(self, tree):
355 subtrees = list(tree.iter_subtrees())
356 for subtree in subtrees:
357 subtree.children = [self._apply_callback(c) if isinstance(c, Tree) else c for c in subtree.children]
359 return self._apply_callback(tree)
361 def _apply_callback(self, tree):
362 return self.callbacks[tree.rule](tree.children)
365_parser_creators['earley'] = create_earley_parser
366_parser_creators['cyk'] = CYK_FrontEnd
369def _construct_parsing_frontend(
370 parser_type: _ParserArgType,
371 lexer_type: _LexerArgType,
372 lexer_conf,
373 parser_conf,
374 options
375):
376 assert isinstance(lexer_conf, LexerConf)
377 assert isinstance(parser_conf, ParserConf)
378 parser_conf.parser_type = parser_type
379 lexer_conf.lexer_type = lexer_type
380 return ParsingFrontend(lexer_conf, parser_conf, options)