|
| 1 | +import time |
| 2 | +from abc import abstractmethod |
| 3 | +from typing import Optional, List, Union |
| 4 | + |
| 5 | +from antlr4 import CommonTokenStream, InputStream, Lexer, Parser, ParserATNSimulator, ParserRuleContext, \ |
| 6 | + PredictionContextCache, Recognizer, Token, TokenStream |
| 7 | +from antlr4.error.ErrorListener import ErrorListener |
| 8 | +from pylasu.model import walk, Source, Position, Node, Point |
| 9 | +from pylasu.model.processing import assign_parents |
| 10 | +from pylasu.parsing.parse_tree import token_end_point |
| 11 | +from pylasu.parsing.results import ParsingResultWithFirstStage, FirstStageParsingResult |
| 12 | +from pylasu.validation import Issue, IssueType |
| 13 | + |
| 14 | + |
| 15 | +class PylasuANTLRParser: |
| 16 | + """ A complete description of a multi-stage ANTLR-based parser, from source code to AST. |
| 17 | +
|
| 18 | + You should extend this class to implement the parts that are specific to your language. |
| 19 | +
|
| 20 | + Note: instances of this class are thread-safe and they're meant to be reused. Do not create a new PylasuANTLRParser |
| 21 | + instance every time you need to parse some source code, or performance may suffer.""" |
| 22 | + def __init__(self): |
| 23 | + self.prediction_context_cache = PredictionContextCache() |
| 24 | + |
| 25 | + def parse(self, input_stream: Union[InputStream, str], consider_range: bool = True, |
| 26 | + measure_lexing_time: bool = False, source: Optional[Source] = None): |
| 27 | + """Parses source code, returning a result that includes an AST and a collection of parse issues |
| 28 | + (errors, warnings). |
| 29 | + The parsing is done in accordance to the StarLasu methodology i.e. a first-stage parser builds a parse tree |
| 30 | + which is then mapped onto a higher-level tree called the AST. |
| 31 | + @param inputStream the source code. |
| 32 | + @param charset the character set in which the input is encoded. |
| 33 | + @param considerPosition if true (the default), parsed AST nodes record their position in the input text. |
| 34 | + @param measureLexingTime if true, the result will include a measurement of the time spent in lexing i.e. |
| 35 | + breaking the input stream into tokens.""" |
| 36 | + start = time.time_ns() |
| 37 | + if type(input_stream) is str: |
| 38 | + input_stream = InputStream(input_stream) |
| 39 | + first_stage = self.parse_first_stage(input_stream, measure_lexing_time) |
| 40 | + issues = first_stage.issues |
| 41 | + ast = self.parse_tree_to_ast(first_stage.root, consider_range, issues, source) |
| 42 | + self.assign_parents(ast) |
| 43 | + ast = self.post_process_ast(ast, issues) if ast else ast |
| 44 | + if ast and not consider_range: |
| 45 | + # Remove parseTreeNodes because they cause the range to be computed |
| 46 | + for node in walk(ast): |
| 47 | + node.origin = None |
| 48 | + now = time.time_ns() |
| 49 | + return ParsingResultWithFirstStage( |
| 50 | + issues, |
| 51 | + ast, |
| 52 | + input_stream.getText(0, input_stream.index + 1), |
| 53 | + (now - start) // 1_000_000, |
| 54 | + first_stage, |
| 55 | + source, |
| 56 | + ) |
| 57 | + |
| 58 | + def parse_first_stage( |
| 59 | + self, input_stream: InputStream, measure_lexing_time: bool = False, source: Source = None |
| 60 | + ) -> FirstStageParsingResult: |
| 61 | + """Executes only the first stage of the parser, i.e., the production of a parse tree. Usually, you'll want to |
| 62 | + use the [parse] method, that returns an AST which is simpler to use and query.""" |
| 63 | + issues = [] |
| 64 | + lexing_time: Optional[int] = None |
| 65 | + total_time: int = time.time_ns() |
| 66 | + parser = self.create_parser(input_stream, issues) |
| 67 | + if measure_lexing_time: |
| 68 | + token_stream = parser.getInputStream() |
| 69 | + if isinstance(token_stream, CommonTokenStream): |
| 70 | + lexing_time = time.time_ns() |
| 71 | + token_stream.fill() |
| 72 | + token_stream.seek(0) |
| 73 | + lexing_time = (time.time_ns() - lexing_time) // 1_000_000 |
| 74 | + root = self.invoke_root_rule(parser) |
| 75 | + if root: |
| 76 | + self.verify_parse_tree(parser, issues, root) |
| 77 | + total_time = (time.time_ns() - total_time) // 1_000_000 |
| 78 | + return FirstStageParsingResult(issues, root, None, total_time, lexing_time, source) |
| 79 | + |
| 80 | + def create_parser(self, input_stream: InputStream, issues: List[Issue]) -> Parser: |
| 81 | + """Creates the first-stage parser.""" |
| 82 | + lexer = self.create_antlr_lexer(input_stream) |
| 83 | + self.attach_listeners(lexer, issues) |
| 84 | + token_stream = self.create_token_stream(lexer) |
| 85 | + parser = self.create_antlr_parser(token_stream) |
| 86 | + # Assign interpreter to avoid caching DFA states indefinitely across executions |
| 87 | + parser._interp = \ |
| 88 | + ParserATNSimulator(parser, parser.atn, parser._interp.decisionToDFA, self.prediction_context_cache) |
| 89 | + self.attach_listeners(parser, issues) |
| 90 | + return parser |
| 91 | + |
| 92 | + def invoke_root_rule(self, parser: Parser): |
| 93 | + """Invokes the parser's root rule, i.e., the method which is responsible for parsing the entire input. |
| 94 | + Usually this is the topmost rule, the one with index 0 (as also assumed by other libraries such as antlr4-c3), |
| 95 | + so this method invokes that rule. If your grammar/parser is structured differently, or if you're using this to |
| 96 | + parse only a portion of the input or a subset of the language, you have to override this method to invoke the |
| 97 | + correct entry point.""" |
| 98 | + return getattr(parser, parser.ruleNames[0])() |
| 99 | + |
| 100 | + def verify_parse_tree(self, parser: Parser, issues: List[Issue], root: ParserRuleContext): |
| 101 | + """Checks the parse tree for correctness. |
| 102 | + If you're concerned about performance, you may want to override this to do nothing.""" |
| 103 | + last_token: Token = parser.getTokenStream().get(parser.getTokenStream().index) |
| 104 | + if last_token.type != Token.EOF: |
| 105 | + issues.append( |
| 106 | + Issue( |
| 107 | + IssueType.SYNTACTIC, |
| 108 | + "The whole input was not consumed", |
| 109 | + position=Position(token_end_point(last_token), token_end_point(last_token)) |
| 110 | + ) |
| 111 | + ) |
| 112 | + # TODO Kolasu also traverses the parse tree searching for exceptions |
| 113 | + |
| 114 | + def assign_parents(self, ast): |
| 115 | + if ast: |
| 116 | + assign_parents(ast) |
| 117 | + |
| 118 | + def post_process_ast(self, ast, issues): |
| 119 | + return ast |
| 120 | + |
| 121 | + def create_token_stream(self, lexer: Lexer) -> TokenStream: |
| 122 | + return CommonTokenStream(lexer) |
| 123 | + |
| 124 | + @abstractmethod |
| 125 | + def create_antlr_lexer(self, input_stream: InputStream) -> Lexer: |
| 126 | + """Creates the lexer.""" |
| 127 | + pass |
| 128 | + |
| 129 | + @abstractmethod |
| 130 | + def create_antlr_parser(self, token_stream: TokenStream) -> Parser: |
| 131 | + """Creates the first-stage parser.""" |
| 132 | + pass |
| 133 | + |
| 134 | + @abstractmethod |
| 135 | + def parse_tree_to_ast(self, root, consider_range: bool, issues: List[Issue], source: Source) -> Optional[Node]: |
| 136 | + pass |
| 137 | + |
| 138 | + def attach_listeners(self, recognizer: Recognizer, issues: List[Issue]): |
| 139 | + recognizer.removeErrorListeners() |
| 140 | + recognizer.addErrorListener(ParserErrorListener(issues)) |
| 141 | + |
| 142 | + |
| 143 | +class ParserErrorListener(ErrorListener): |
| 144 | + def __init__(self, issues: List[Issue]): |
| 145 | + self.issues = issues |
| 146 | + |
| 147 | + def syntaxError(self, recognizer, offending_symbol, line, column, msg, e): |
| 148 | + start_point = Point(line, column) |
| 149 | + end_point = start_point |
| 150 | + if isinstance(offending_symbol, Token): |
| 151 | + end_point = token_end_point(offending_symbol) |
| 152 | + msg = (msg or "unspecified").capitalize() |
| 153 | + self.issues.append(Issue(IssueType.SYNTACTIC, msg, position=Position(start_point, end_point))) |
0 commit comments