lib/python3.9/site-packages/antlr4/Lexer.py

# Copyright (c) 2012-2017 The ANTLR Project. All rights reserved.
# Use of this file is governed by the BSD 3-clause license that
# can be found in the LICENSE.txt file in the project root.
#/

# A lexer is recognizer that draws input symbols from a character stream.
#  lexer grammars result in a subclass of self object. A Lexer object
#  uses simplified match() and error recovery mechanisms in the interest
#  of speed.
#/
from io import StringIO

import sys
if sys.version_info[1] > 5:
    from typing import TextIO
else:
    from typing.io import TextIO
from antlr4.CommonTokenFactory import CommonTokenFactory
from antlr4.atn.LexerATNSimulator import LexerATNSimulator
from antlr4.InputStream import InputStream
from antlr4.Recognizer import Recognizer
from antlr4.Token import Token
from antlr4.error.Errors import IllegalStateException, LexerNoViableAltException, RecognitionException

class TokenSource(object):

    pass


class Lexer(Recognizer, TokenSource):
    __slots__ = (
        '_input', '_output', '_factory', '_tokenFactorySourcePair', '_token',
        '_tokenStartCharIndex', '_tokenStartLine', '_tokenStartColumn',
        '_hitEOF', '_channel', '_type', '_modeStack', '_mode', '_text'
    )

    DEFAULT_MODE = 0
    MORE = -2
    SKIP = -3

    DEFAULT_TOKEN_CHANNEL = Token.DEFAULT_CHANNEL
    HIDDEN = Token.HIDDEN_CHANNEL
    MIN_CHAR_VALUE = 0x0000
    MAX_CHAR_VALUE = 0x10FFFF

    def __init__(self, input:InputStream, output:TextIO = sys.stdout):
        super().__init__()
        self._input = input
        self._output = output
        self._factory = CommonTokenFactory.DEFAULT
        self._tokenFactorySourcePair = (self, input)

        self._interp = None # child classes must populate this

        # The goal of all lexer rules/methods is to create a token object.
        #  self is an instance variable as multiple rules may collaborate to
        #  create a single token.  nextToken will return self object after
        #  matching lexer rule(s).  If you subclass to allow multiple token
        #  emissions, then set self to the last token to be matched or
        #  something nonnull so that the auto token emit mechanism will not
        #  emit another token.
        self._token = None

        # What character index in the stream did the current token start at?
        #  Needed, for example, to get the text for current token.  Set at
        #  the start of nextToken.
        self._tokenStartCharIndex = -1

        # The line on which the first character of the token resides#/
        self._tokenStartLine = -1

        # The character position of first character within the line#/
        self._tokenStartColumn = -1

        # Once we see EOF on char stream, next token will be EOF.
        #  If you have DONE : EOF ; then you see DONE EOF.
        self._hitEOF = False

        # The channel number for the current token#/
        self._channel = Token.DEFAULT_CHANNEL

        # The token type for the current token#/
        self._type = Token.INVALID_TYPE

        self._modeStack = []
        self._mode = self.DEFAULT_MODE

        # You can set the text for the current token to override what is in
        #  the input char buffer.  Use setText() or can set self instance var.
        #/
        self._text = None


    def reset(self):
        # wack Lexer state variables
        if self._input is not None:
            self._input.seek(0) # rewind the input
        self._token = None
        self._type = Token.INVALID_TYPE
        self._channel = Token.DEFAULT_CHANNEL
        self._tokenStartCharIndex = -1
        self._tokenStartColumn = -1
        self._tokenStartLine = -1
        self._text = None

        self._hitEOF = False
        self._mode = Lexer.DEFAULT_MODE
        self._modeStack = []

        self._interp.reset()

    # Return a token from self source; i.e., match a token on the char
    #  stream.
    def nextToken(self):
        if self._input is None:
            raise IllegalStateException("nextToken requires a non-null input stream.")

        # Mark start location in char stream so unbuffered streams are
        # guaranteed at least have text of current token
        tokenStartMarker = self._input.mark()
        try:
            while True:
                if self._hitEOF:
                    self.emitEOF()
                    return self._token
                self._token = None
                self._channel = Token.DEFAULT_CHANNEL
                self._tokenStartCharIndex = self._input.index
                self._tokenStartColumn = self._interp.column
                self._tokenStartLine = self._interp.line
                self._text = None
                continueOuter = False
                while True:
                    self._type = Token.INVALID_TYPE
                    ttype = self.SKIP
                    try:
                        ttype = self._interp.match(self._input, self._mode)
                    except LexerNoViableAltException as e:
                        self.notifyListeners(e)		# report error
                        self.recover(e)
                    if self._input.LA(1)==Token.EOF:
                        self._hitEOF = True
                    if self._type == Token.INVALID_TYPE:
                        self._type = ttype
                    if self._type == self.SKIP:
                        continueOuter = True
                        break
                    if self._type!=self.MORE:
                        break
                if continueOuter:
                    continue
                if self._token is None:
                    self.emit()
                return self._token
        finally:
            # make sure we release marker after match or
            # unbuffered char stream will keep buffering
            self._input.release(tokenStartMarker)

    # Instruct the lexer to skip creating a token for current lexer rule
    #  and look for another token.  nextToken() knows to keep looking when
    #  a lexer rule finishes with token set to SKIP_TOKEN.  Recall that
    #  if token==null at end of any token rule, it creates one for you
    #  and emits it.
    #/
    def skip(self):
        self._type = self.SKIP

    def more(self):
        self._type = self.MORE

    def mode(self, m:int):
        self._mode = m

    def pushMode(self, m:int):
        if self._interp.debug:
            print("pushMode " + str(m), file=self._output)
        self._modeStack.append(self._mode)
        self.mode(m)

    def popMode(self):
        if len(self._modeStack)==0:
            raise Exception("Empty Stack")
        if self._interp.debug:
            print("popMode back to "+ self._modeStack[:-1], file=self._output)
        self.mode( self._modeStack.pop() )
        return self._mode

    # Set the char stream and reset the lexer#/
    @property
    def inputStream(self):
        return self._input

    @inputStream.setter
    def inputStream(self, input:InputStream):
        self._input = None
        self._tokenFactorySourcePair = (self, self._input)
        self.reset()
        self._input = input
        self._tokenFactorySourcePair = (self, self._input)

    @property
    def sourceName(self):
        return self._input.sourceName

    # By default does not support multiple emits per nextToken invocation
    #  for efficiency reasons.  Subclass and override self method, nextToken,
    #  and getToken (to push tokens into a list and pull from that list
    #  rather than a single variable as self implementation does).
    #/
    def emitToken(self, token:Token):
        self._token = token

    # The standard method called to automatically emit a token at the
    #  outermost lexical rule.  The token object should point into the
    #  char buffer start..stop.  If there is a text override in 'text',
    #  use that to set the token's text.  Override self method to emit
    #  custom Token objects or provide a new factory.
    #/
    def emit(self):
        t = self._factory.create(self._tokenFactorySourcePair, self._type, self._text, self._channel, self._tokenStartCharIndex,
                                 self.getCharIndex()-1, self._tokenStartLine, self._tokenStartColumn)
        self.emitToken(t)
        return t

    def emitEOF(self):
        cpos = self.column
        lpos = self.line
        eof = self._factory.create(self._tokenFactorySourcePair, Token.EOF, None, Token.DEFAULT_CHANNEL, self._input.index,
                                   self._input.index-1, lpos, cpos)
        self.emitToken(eof)
        return eof

    @property
    def type(self):
        return self._type

    @type.setter
    def type(self, type:int):
        self._type = type

    @property
    def line(self):
        return self._interp.line

    @line.setter
    def line(self, line:int):
        self._interp.line = line

    @property
    def column(self):
        return self._interp.column

    @column.setter
    def column(self, column:int):
        self._interp.column = column

    # What is the index of the current character of lookahead?#/
    def getCharIndex(self):
        return self._input.index

    # Return the text matched so far for the current token or any
    #  text override.
    @property
    def text(self):
        if self._text is not None:
            return self._text
        else:
            return self._interp.getText(self._input)

    # Set the complete text of self token; it wipes any previous
    #  changes to the text.
    @text.setter
    def text(self, txt:str):
        self._text = txt

    # Return a list of all Token objects in input char stream.
    #  Forces load of all tokens. Does not include EOF token.
    #/
    def getAllTokens(self):
        tokens = []
        t = self.nextToken()
        while t.type!=Token.EOF:
            tokens.append(t)
            t = self.nextToken()
        return tokens

    def notifyListeners(self, e:LexerNoViableAltException):
        start = self._tokenStartCharIndex
        stop = self._input.index
        text = self._input.getText(start, stop)
        msg = "token recognition error at: '" + self.getErrorDisplay(text) + "'"
        listener = self.getErrorListenerDispatch()
        listener.syntaxError(self, None, self._tokenStartLine, self._tokenStartColumn, msg, e)

    def getErrorDisplay(self, s:str):
        with StringIO() as buf:
            for c in s:
                buf.write(self.getErrorDisplayForChar(c))
            return buf.getvalue()

    def getErrorDisplayForChar(self, c:str):
        if ord(c[0])==Token.EOF:
            return "<EOF>"
        elif c=='\n':
            return "\\n"
        elif c=='\t':
            return "\\t"
        elif c=='\r':
            return "\\r"
        else:
            return c

    def getCharErrorDisplay(self, c:str):
        return "'" + self.getErrorDisplayForChar(c) + "'"

    # Lexers can normally match any char in it's vocabulary after matching
    #  a token, so do the easy thing and just kill a character and hope
    #  it all works out.  You can instead use the rule invocation stack
    #  to do sophisticated error recovery if you are in a fragment rule.
    #/
    def recover(self, re:RecognitionException):
        if self._input.LA(1) != Token.EOF:
            if isinstance(re, LexerNoViableAltException):
                    # skip a char and try again
                    self._interp.consume(self._input)
            else:
                # TODO: Do we lose character or line position information?
                self._input.consume()