| # |
| # 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. |
| |
| # This implementation of {@link TokenStream} loads tokens from a |
| # {@link TokenSource} on-demand, and places the tokens in a buffer to provide |
| # access to any previous token by index. |
| # |
| # <p> |
| # This token stream ignores the value of {@link Token#getChannel}. If your |
| # parser requires the token stream filter tokens to only those on a particular |
| # channel, such as {@link Token#DEFAULT_CHANNEL} or |
| # {@link Token#HIDDEN_CHANNEL}, use a filtering token stream such a |
| # {@link CommonTokenStream}.</p> |
| from io import StringIO |
| from antlr4.Token import Token |
| from antlr4.error.Errors import IllegalStateException |
| |
| # need forward declaration |
| Lexer = None |
| |
| # this is just to keep meaningful parameter types to Parser |
| class TokenStream(object): |
| |
| pass |
| |
| |
| class BufferedTokenStream(TokenStream): |
| __slots__ = ('tokenSource', 'tokens', 'index', 'fetchedEOF') |
| |
| def __init__(self, tokenSource:Lexer): |
| # The {@link TokenSource} from which tokens for this stream are fetched. |
| self.tokenSource = tokenSource |
| |
| # A collection of all tokens fetched from the token source. The list is |
| # considered a complete view of the input once {@link #fetchedEOF} is set |
| # to {@code true}. |
| self.tokens = [] |
| |
| # The index into {@link #tokens} of the current token (next token to |
| # {@link #consume}). {@link #tokens}{@code [}{@link #p}{@code ]} should be |
| # {@link #LT LT(1)}. |
| # |
| # <p>This field is set to -1 when the stream is first constructed or when |
| # {@link #setTokenSource} is called, indicating that the first token has |
| # not yet been fetched from the token source. For additional information, |
| # see the documentation of {@link IntStream} for a description of |
| # Initializing Methods.</p> |
| self.index = -1 |
| |
| # Indicates whether the {@link Token#EOF} token has been fetched from |
| # {@link #tokenSource} and added to {@link #tokens}. This field improves |
| # performance for the following cases: |
| # |
| # <ul> |
| # <li>{@link #consume}: The lookahead check in {@link #consume} to prevent |
| # consuming the EOF symbol is optimized by checking the values of |
| # {@link #fetchedEOF} and {@link #p} instead of calling {@link #LA}.</li> |
| # <li>{@link #fetch}: The check to prevent adding multiple EOF symbols into |
| # {@link #tokens} is trivial with this field.</li> |
| # <ul> |
| self.fetchedEOF = False |
| |
| def mark(self): |
| return 0 |
| |
| def release(self, marker:int): |
| # no resources to release |
| pass |
| |
| def reset(self): |
| self.seek(0) |
| |
| def seek(self, index:int): |
| self.lazyInit() |
| self.index = self.adjustSeekIndex(index) |
| |
| def get(self, index:int): |
| self.lazyInit() |
| return self.tokens[index] |
| |
| def consume(self): |
| skipEofCheck = False |
| if self.index >= 0: |
| if self.fetchedEOF: |
| # the last token in tokens is EOF. skip check if p indexes any |
| # fetched token except the last. |
| skipEofCheck = self.index < len(self.tokens) - 1 |
| else: |
| # no EOF token in tokens. skip check if p indexes a fetched token. |
| skipEofCheck = self.index < len(self.tokens) |
| else: |
| # not yet initialized |
| skipEofCheck = False |
| |
| if not skipEofCheck and self.LA(1) == Token.EOF: |
| raise IllegalStateException("cannot consume EOF") |
| |
| if self.sync(self.index + 1): |
| self.index = self.adjustSeekIndex(self.index + 1) |
| |
| # Make sure index {@code i} in tokens has a token. |
| # |
| # @return {@code true} if a token is located at index {@code i}, otherwise |
| # {@code false}. |
| # @see #get(int i) |
| #/ |
| def sync(self, i:int): |
| n = i - len(self.tokens) + 1 # how many more elements we need? |
| if n > 0 : |
| fetched = self.fetch(n) |
| return fetched >= n |
| return True |
| |
| # Add {@code n} elements to buffer. |
| # |
| # @return The actual number of elements added to the buffer. |
| #/ |
| def fetch(self, n:int): |
| if self.fetchedEOF: |
| return 0 |
| for i in range(0, n): |
| t = self.tokenSource.nextToken() |
| t.tokenIndex = len(self.tokens) |
| self.tokens.append(t) |
| if t.type==Token.EOF: |
| self.fetchedEOF = True |
| return i + 1 |
| return n |
| |
| |
| # Get all tokens from start..stop inclusively#/ |
| def getTokens(self, start:int, stop:int, types:set=None): |
| if start<0 or stop<0: |
| return None |
| self.lazyInit() |
| subset = [] |
| if stop >= len(self.tokens): |
| stop = len(self.tokens)-1 |
| for i in range(start, stop): |
| t = self.tokens[i] |
| if t.type==Token.EOF: |
| break |
| if types is None or t.type in types: |
| subset.append(t) |
| return subset |
| |
| def LA(self, i:int): |
| return self.LT(i).type |
| |
| def LB(self, k:int): |
| if (self.index-k) < 0: |
| return None |
| return self.tokens[self.index-k] |
| |
| def LT(self, k:int): |
| self.lazyInit() |
| if k==0: |
| return None |
| if k < 0: |
| return self.LB(-k) |
| i = self.index + k - 1 |
| self.sync(i) |
| if i >= len(self.tokens): # return EOF token |
| # EOF must be last token |
| return self.tokens[len(self.tokens)-1] |
| return self.tokens[i] |
| |
| # Allowed derived classes to modify the behavior of operations which change |
| # the current stream position by adjusting the target token index of a seek |
| # operation. The default implementation simply returns {@code i}. If an |
| # exception is thrown in this method, the current stream index should not be |
| # changed. |
| # |
| # <p>For example, {@link CommonTokenStream} overrides this method to ensure that |
| # the seek target is always an on-channel token.</p> |
| # |
| # @param i The target token index. |
| # @return The adjusted target token index. |
| |
| def adjustSeekIndex(self, i:int): |
| return i |
| |
| def lazyInit(self): |
| if self.index == -1: |
| self.setup() |
| |
| def setup(self): |
| self.sync(0) |
| self.index = self.adjustSeekIndex(0) |
| |
| # Reset this token stream by setting its token source.#/ |
| def setTokenSource(self, tokenSource:Lexer): |
| self.tokenSource = tokenSource |
| self.tokens = [] |
| self.index = -1 |
| self.fetchedEOF = False |
| |
| |
| # Given a starting index, return the index of the next token on channel. |
| # Return i if tokens[i] is on channel. Return the index of the EOF token |
| # if there are no tokens on channel between i and EOF. |
| #/ |
| def nextTokenOnChannel(self, i:int, channel:int): |
| self.sync(i) |
| if i>=len(self.tokens): |
| return len(self.tokens) - 1 |
| token = self.tokens[i] |
| while token.channel!=channel: |
| if token.type==Token.EOF: |
| return i |
| i += 1 |
| self.sync(i) |
| token = self.tokens[i] |
| return i |
| |
| # Given a starting index, return the index of the previous token on channel. |
| # Return i if tokens[i] is on channel. Return -1 if there are no tokens |
| # on channel between i and 0. |
| def previousTokenOnChannel(self, i:int, channel:int): |
| while i>=0 and self.tokens[i].channel!=channel: |
| i -= 1 |
| return i |
| |
| # Collect all tokens on specified channel to the right of |
| # the current token up until we see a token on DEFAULT_TOKEN_CHANNEL or |
| # EOF. If channel is -1, find any non default channel token. |
| def getHiddenTokensToRight(self, tokenIndex:int, channel:int=-1): |
| self.lazyInit() |
| if tokenIndex<0 or tokenIndex>=len(self.tokens): |
| raise Exception(str(tokenIndex) + " not in 0.." + str(len(self.tokens)-1)) |
| from antlr4.Lexer import Lexer |
| nextOnChannel = self.nextTokenOnChannel(tokenIndex + 1, Lexer.DEFAULT_TOKEN_CHANNEL) |
| from_ = tokenIndex+1 |
| # if none onchannel to right, nextOnChannel=-1 so set to = last token |
| to = (len(self.tokens)-1) if nextOnChannel==-1 else nextOnChannel |
| return self.filterForChannel(from_, to, channel) |
| |
| |
| # Collect all tokens on specified channel to the left of |
| # the current token up until we see a token on DEFAULT_TOKEN_CHANNEL. |
| # If channel is -1, find any non default channel token. |
| def getHiddenTokensToLeft(self, tokenIndex:int, channel:int=-1): |
| self.lazyInit() |
| if tokenIndex<0 or tokenIndex>=len(self.tokens): |
| raise Exception(str(tokenIndex) + " not in 0.." + str(len(self.tokens)-1)) |
| from antlr4.Lexer import Lexer |
| prevOnChannel = self.previousTokenOnChannel(tokenIndex - 1, Lexer.DEFAULT_TOKEN_CHANNEL) |
| if prevOnChannel == tokenIndex - 1: |
| return None |
| # if none on channel to left, prevOnChannel=-1 then from=0 |
| from_ = prevOnChannel+1 |
| to = tokenIndex-1 |
| return self.filterForChannel(from_, to, channel) |
| |
| |
| def filterForChannel(self, left:int, right:int, channel:int): |
| hidden = [] |
| for i in range(left, right+1): |
| t = self.tokens[i] |
| if channel==-1: |
| from antlr4.Lexer import Lexer |
| if t.channel!= Lexer.DEFAULT_TOKEN_CHANNEL: |
| hidden.append(t) |
| elif t.channel==channel: |
| hidden.append(t) |
| if len(hidden)==0: |
| return None |
| return hidden |
| |
| def getSourceName(self): |
| return self.tokenSource.getSourceName() |
| |
| # Get the text of all tokens in this buffer.#/ |
| def getText(self, start:int=None, stop:int=None): |
| self.lazyInit() |
| self.fill() |
| if isinstance(start, Token): |
| start = start.tokenIndex |
| elif start is None: |
| start = 0 |
| if isinstance(stop, Token): |
| stop = stop.tokenIndex |
| elif stop is None or stop >= len(self.tokens): |
| stop = len(self.tokens) - 1 |
| if start < 0 or stop < 0 or stop < start: |
| return "" |
| with StringIO() as buf: |
| for i in range(start, stop+1): |
| t = self.tokens[i] |
| if t.type==Token.EOF: |
| break |
| buf.write(t.text) |
| return buf.getvalue() |
| |
| |
| # Get all tokens from lexer until EOF#/ |
| def fill(self): |
| self.lazyInit() |
| while self.fetch(1000)==1000: |
| pass |
