[config.git] / djavu-asus / elpy / rpc-venv / lib / python3.11 / site-packages / blib2to3 / pgen2 / driver.py

# Copyright 2004-2005 Elemental Security, Inc. All Rights Reserved.
# Licensed to PSF under a Contributor Agreement.

# Modifications:
# Copyright 2006 Google, Inc. All Rights Reserved.
# Licensed to PSF under a Contributor Agreement.

"""Parser driver.

This provides a high-level interface to parse a file into a syntax tree.

"""

__author__ = "Guido van Rossum <guido@python.org>"

__all__ = ["Driver", "load_grammar"]

# Python imports
import io
import logging
import os
import pkgutil
import sys
from contextlib import contextmanager
from dataclasses import dataclass, field
from logging import Logger
from typing import IO, Any, Iterable, Iterator, List, Optional, Tuple, Union, cast

from blib2to3.pgen2.grammar import Grammar
from blib2to3.pgen2.tokenize import GoodTokenInfo
from blib2to3.pytree import NL

# Pgen imports
from . import grammar, parse, pgen, token, tokenize

Path = Union[str, "os.PathLike[str]"]


@dataclass
class ReleaseRange:
    start: int
    end: Optional[int] = None
    tokens: List[Any] = field(default_factory=list)

    def lock(self) -> None:
        total_eaten = len(self.tokens)
        self.end = self.start + total_eaten


class TokenProxy:
    def __init__(self, generator: Any) -> None:
        self._tokens = generator
        self._counter = 0
        self._release_ranges: List[ReleaseRange] = []

    @contextmanager
    def release(self) -> Iterator["TokenProxy"]:
        release_range = ReleaseRange(self._counter)
        self._release_ranges.append(release_range)
        try:
            yield self
        finally:
            # Lock the last release range to the final position that
            # has been eaten.
            release_range.lock()

    def eat(self, point: int) -> Any:
        eaten_tokens = self._release_ranges[-1].tokens
        if point < len(eaten_tokens):
            return eaten_tokens[point]
        else:
            while point >= len(eaten_tokens):
                token = next(self._tokens)
                eaten_tokens.append(token)
            return token

    def __iter__(self) -> "TokenProxy":
        return self

    def __next__(self) -> Any:
        # If the current position is already compromised (looked up)
        # return the eaten token, if not just go further on the given
        # token producer.
        for release_range in self._release_ranges:
            assert release_range.end is not None

            start, end = release_range.start, release_range.end
            if start <= self._counter < end:
                token = release_range.tokens[self._counter - start]
                break
        else:
            token = next(self._tokens)
        self._counter += 1
        return token

    def can_advance(self, to: int) -> bool:
        # Try to eat, fail if it can't. The eat operation is cached
        # so there won't be any additional cost of eating here
        try:
            self.eat(to)
        except StopIteration:
            return False
        else:
            return True


class Driver:
    def __init__(self, grammar: Grammar, logger: Optional[Logger] = None) -> None:
        self.grammar = grammar
        if logger is None:
            logger = logging.getLogger(__name__)
        self.logger = logger

    def parse_tokens(self, tokens: Iterable[GoodTokenInfo], debug: bool = False) -> NL:
        """Parse a series of tokens and return the syntax tree."""
        # XXX Move the prefix computation into a wrapper around tokenize.
        proxy = TokenProxy(tokens)

        p = parse.Parser(self.grammar)
        p.setup(proxy=proxy)

        lineno = 1
        column = 0
        indent_columns: List[int] = []
        type = value = start = end = line_text = None
        prefix = ""

        for quintuple in proxy:
            type, value, start, end, line_text = quintuple
            if start != (lineno, column):
                assert (lineno, column) <= start, ((lineno, column), start)
                s_lineno, s_column = start
                if lineno < s_lineno:
                    prefix += "\n" * (s_lineno - lineno)
                    lineno = s_lineno
                    column = 0
                if column < s_column:
                    prefix += line_text[column:s_column]
                    column = s_column
            if type in (tokenize.COMMENT, tokenize.NL):
                prefix += value
                lineno, column = end
                if value.endswith("\n"):
                    lineno += 1
                    column = 0
                continue
            if type == token.OP:
                type = grammar.opmap[value]
            if debug:
                assert type is not None
                self.logger.debug(
                    "%s %r (prefix=%r)", token.tok_name[type], value, prefix
                )
            if type == token.INDENT:
                indent_columns.append(len(value))
                _prefix = prefix + value
                prefix = ""
                value = ""
            elif type == token.DEDENT:
                _indent_col = indent_columns.pop()
                prefix, _prefix = self._partially_consume_prefix(prefix, _indent_col)
            if p.addtoken(cast(int, type), value, (prefix, start)):
                if debug:
                    self.logger.debug("Stop.")
                break
            prefix = ""
            if type in {token.INDENT, token.DEDENT}:
                prefix = _prefix
            lineno, column = end
            if value.endswith("\n"):
                lineno += 1
                column = 0
        else:
            # We never broke out -- EOF is too soon (how can this happen???)
            assert start is not None
            raise parse.ParseError("incomplete input", type, value, (prefix, start))
        assert p.rootnode is not None
        return p.rootnode

    def parse_stream_raw(self, stream: IO[str], debug: bool = False) -> NL:
        """Parse a stream and return the syntax tree."""
        tokens = tokenize.generate_tokens(stream.readline, grammar=self.grammar)
        return self.parse_tokens(tokens, debug)

    def parse_stream(self, stream: IO[str], debug: bool = False) -> NL:
        """Parse a stream and return the syntax tree."""
        return self.parse_stream_raw(stream, debug)

    def parse_file(
        self, filename: Path, encoding: Optional[str] = None, debug: bool = False
    ) -> NL:
        """Parse a file and return the syntax tree."""
        with open(filename, encoding=encoding) as stream:
            return self.parse_stream(stream, debug)

    def parse_string(self, text: str, debug: bool = False) -> NL:
        """Parse a string and return the syntax tree."""
        tokens = tokenize.generate_tokens(
            io.StringIO(text).readline, grammar=self.grammar
        )
        return self.parse_tokens(tokens, debug)

    def _partially_consume_prefix(self, prefix: str, column: int) -> Tuple[str, str]:
        lines: List[str] = []
        current_line = ""
        current_column = 0
        wait_for_nl = False
        for char in prefix:
            current_line += char
            if wait_for_nl:
                if char == "\n":
                    if current_line.strip() and current_column < column:
                        res = "".join(lines)
                        return res, prefix[len(res) :]

                    lines.append(current_line)
                    current_line = ""
                    current_column = 0
                    wait_for_nl = False
            elif char in " \t":
                current_column += 1
            elif char == "\n":
                # unexpected empty line
                current_column = 0
            else:
                # indent is finished
                wait_for_nl = True
        return "".join(lines), current_line


def _generate_pickle_name(gt: Path, cache_dir: Optional[Path] = None) -> str:
    head, tail = os.path.splitext(gt)
    if tail == ".txt":
        tail = ""
    name = head + tail + ".".join(map(str, sys.version_info)) + ".pickle"
    if cache_dir:
        return os.path.join(cache_dir, os.path.basename(name))
    else:
        return name


def load_grammar(
    gt: str = "Grammar.txt",
    gp: Optional[str] = None,
    save: bool = True,
    force: bool = False,
    logger: Optional[Logger] = None,
) -> Grammar:
    """Load the grammar (maybe from a pickle)."""
    if logger is None:
        logger = logging.getLogger(__name__)
    gp = _generate_pickle_name(gt) if gp is None else gp
    if force or not _newer(gp, gt):
        g: grammar.Grammar = pgen.generate_grammar(gt)
        if save:
            try:
                g.dump(gp)
            except OSError:
                # Ignore error, caching is not vital.
                pass
    else:
        g = grammar.Grammar()
        g.load(gp)
    return g


def _newer(a: str, b: str) -> bool:
    """Inquire whether file a was written since file b."""
    if not os.path.exists(a):
        return False
    if not os.path.exists(b):
        return True
    return os.path.getmtime(a) >= os.path.getmtime(b)


def load_packaged_grammar(
    package: str, grammar_source: str, cache_dir: Optional[Path] = None
) -> grammar.Grammar:
    """Normally, loads a pickled grammar by doing
        pkgutil.get_data(package, pickled_grammar)
    where *pickled_grammar* is computed from *grammar_source* by adding the
    Python version and using a ``.pickle`` extension.

    However, if *grammar_source* is an extant file, load_grammar(grammar_source)
    is called instead. This facilitates using a packaged grammar file when needed
    but preserves load_grammar's automatic regeneration behavior when possible.

    """
    if os.path.isfile(grammar_source):
        gp = _generate_pickle_name(grammar_source, cache_dir) if cache_dir else None
        return load_grammar(grammar_source, gp=gp)
    pickled_name = _generate_pickle_name(os.path.basename(grammar_source), cache_dir)
    data = pkgutil.get_data(package, pickled_name)
    assert data is not None
    g = grammar.Grammar()
    g.loads(data)
    return g


def main(*args: str) -> bool:
    """Main program, when run as a script: produce grammar pickle files.

    Calls load_grammar for each argument, a path to a grammar text file.
    """
    if not args:
        args = tuple(sys.argv[1:])
    logging.basicConfig(level=logging.INFO, stream=sys.stdout, format="%(message)s")
    for gt in args:
        load_grammar(gt, save=True, force=True)
    return True


if __name__ == "__main__":
    sys.exit(int(not main()))
Commit	Line	Data
53e6db90 DC	1	# Copyright 2004-2005 Elemental Security, Inc. All Rights Reserved.
	2	# Licensed to PSF under a Contributor Agreement.
	3
	4	# Modifications:
	5	# Copyright 2006 Google, Inc. All Rights Reserved.
	6	# Licensed to PSF under a Contributor Agreement.
	7
	8	"""Parser driver.
	9
	10	This provides a high-level interface to parse a file into a syntax tree.
	11
	12	"""
	13
	14	__author__ = "Guido van Rossum <guido@python.org>"
	15
	16	__all__ = ["Driver", "load_grammar"]
	17
	18	# Python imports
	19	import io
	20	import logging
	21	import os
	22	import pkgutil
	23	import sys
	24	from contextlib import contextmanager
	25	from dataclasses import dataclass, field
	26	from logging import Logger
	27	from typing import IO, Any, Iterable, Iterator, List, Optional, Tuple, Union, cast
	28
	29	from blib2to3.pgen2.grammar import Grammar
	30	from blib2to3.pgen2.tokenize import GoodTokenInfo
	31	from blib2to3.pytree import NL
	32
	33	# Pgen imports
	34	from . import grammar, parse, pgen, token, tokenize
	35
	36	Path = Union[str, "os.PathLike[str]"]
	37
	38
	39	@dataclass
	40	class ReleaseRange:
	41	start: int
	42	end: Optional[int] = None
	43	tokens: List[Any] = field(default_factory=list)
	44
	45	def lock(self) -> None:
	46	total_eaten = len(self.tokens)
	47	self.end = self.start + total_eaten
	48
	49
	50	class TokenProxy:
	51	def __init__(self, generator: Any) -> None:
	52	self._tokens = generator
	53	self._counter = 0
	54	self._release_ranges: List[ReleaseRange] = []
	55
	56	@contextmanager
	57	def release(self) -> Iterator["TokenProxy"]:
	58	release_range = ReleaseRange(self._counter)
	59	self._release_ranges.append(release_range)
	60	try:
	61	yield self
	62	finally:
	63	# Lock the last release range to the final position that
	64	# has been eaten.
65	release_range.lock()
66
67	def eat(self, point: int) -> Any:
68	eaten_tokens = self._release_ranges[-1].tokens
69	if point < len(eaten_tokens):
70	return eaten_tokens[point]
71	else:
72	while point >= len(eaten_tokens):
73	token = next(self._tokens)
74	eaten_tokens.append(token)
75	return token
76
77	def __iter__(self) -> "TokenProxy":
78	return self
79
80	def __next__(self) -> Any:
81	# If the current position is already compromised (looked up)
82	# return the eaten token, if not just go further on the given
83	# token producer.
84	for release_range in self._release_ranges:
85	assert release_range.end is not None
86
87	start, end = release_range.start, release_range.end
88	if start <= self._counter < end:
89	token = release_range.tokens[self._counter - start]
90	break
91	else:
92	token = next(self._tokens)
93	self._counter += 1
94	return token
95
96	def can_advance(self, to: int) -> bool:
97	# Try to eat, fail if it can't. The eat operation is cached
98	# so there won't be any additional cost of eating here
99	try:
100	self.eat(to)
101	except StopIteration:
102	return False
103	else:
104	return True
105
106
107	class Driver:
108	def __init__(self, grammar: Grammar, logger: Optional[Logger] = None) -> None:
109	self.grammar = grammar
110	if logger is None:
111	logger = logging.getLogger(__name__)
112	self.logger = logger
113
114	def parse_tokens(self, tokens: Iterable[GoodTokenInfo], debug: bool = False) -> NL:
115	"""Parse a series of tokens and return the syntax tree."""
116	# XXX Move the prefix computation into a wrapper around tokenize.
117	proxy = TokenProxy(tokens)
118
119	p = parse.Parser(self.grammar)
120	p.setup(proxy=proxy)
121
122	lineno = 1
123	column = 0
124	indent_columns: List[int] = []
125	type = value = start = end = line_text = None
126	prefix = ""
127
128	for quintuple in proxy:
129	type, value, start, end, line_text = quintuple
130	if start != (lineno, column):
131	assert (lineno, column) <= start, ((lineno, column), start)
132	s_lineno, s_column = start
133	if lineno < s_lineno:
134	prefix += "\n" * (s_lineno - lineno)
135	lineno = s_lineno
136	column = 0
137	if column < s_column:
138	prefix += line_text[column:s_column]
139	column = s_column
140	if type in (tokenize.COMMENT, tokenize.NL):
141	prefix += value
142	lineno, column = end
143	if value.endswith("\n"):
144	lineno += 1
145	column = 0
146	continue
147	if type == token.OP:
148	type = grammar.opmap[value]
149	if debug:
150	assert type is not None
151	self.logger.debug(
152	"%s %r (prefix=%r)", token.tok_name[type], value, prefix
153	)
154	if type == token.INDENT:
155	indent_columns.append(len(value))
156	_prefix = prefix + value
157	prefix = ""
158	value = ""
159	elif type == token.DEDENT:
160	_indent_col = indent_columns.pop()
161	prefix, _prefix = self._partially_consume_prefix(prefix, _indent_col)
162	if p.addtoken(cast(int, type), value, (prefix, start)):
163	if debug:
164	self.logger.debug("Stop.")
165	break
166	prefix = ""
167	if type in {token.INDENT, token.DEDENT}:
168	prefix = _prefix
169	lineno, column = end
170	if value.endswith("\n"):
171	lineno += 1
172	column = 0
173	else:
174	# We never broke out -- EOF is too soon (how can this happen???)
175	assert start is not None
176	raise parse.ParseError("incomplete input", type, value, (prefix, start))
177	assert p.rootnode is not None
178	return p.rootnode
179
180	def parse_stream_raw(self, stream: IO[str], debug: bool = False) -> NL:
181	"""Parse a stream and return the syntax tree."""
182	tokens = tokenize.generate_tokens(stream.readline, grammar=self.grammar)
183	return self.parse_tokens(tokens, debug)
184
185	def parse_stream(self, stream: IO[str], debug: bool = False) -> NL:
186	"""Parse a stream and return the syntax tree."""
187	return self.parse_stream_raw(stream, debug)
188
189	def parse_file(
190	self, filename: Path, encoding: Optional[str] = None, debug: bool = False
191	) -> NL:
192	"""Parse a file and return the syntax tree."""
193	with open(filename, encoding=encoding) as stream:
194	return self.parse_stream(stream, debug)
195
196	def parse_string(self, text: str, debug: bool = False) -> NL:
197	"""Parse a string and return the syntax tree."""
198	tokens = tokenize.generate_tokens(
199	io.StringIO(text).readline, grammar=self.grammar
200	)
201	return self.parse_tokens(tokens, debug)
202
203	def _partially_consume_prefix(self, prefix: str, column: int) -> Tuple[str, str]:
204	lines: List[str] = []
205	current_line = ""
206	current_column = 0
207	wait_for_nl = False
208	for char in prefix:
209	current_line += char
210	if wait_for_nl:
211	if char == "\n":
212	if current_line.strip() and current_column < column:
213	res = "".join(lines)
214	return res, prefix[len(res) :]
215
216	lines.append(current_line)
217	current_line = ""
218	current_column = 0
219	wait_for_nl = False
220	elif char in " \t":
221	current_column += 1
222	elif char == "\n":
223	# unexpected empty line
224	current_column = 0
225	else:
226	# indent is finished
227	wait_for_nl = True
228	return "".join(lines), current_line
229
230
231	def _generate_pickle_name(gt: Path, cache_dir: Optional[Path] = None) -> str:
232	head, tail = os.path.splitext(gt)
233	if tail == ".txt":
234	tail = ""
235	name = head + tail + ".".join(map(str, sys.version_info)) + ".pickle"
236	if cache_dir:
237	return os.path.join(cache_dir, os.path.basename(name))
238	else:
239	return name
240
241
242	def load_grammar(
243	gt: str = "Grammar.txt",
244	gp: Optional[str] = None,
245	save: bool = True,
246	force: bool = False,
247	logger: Optional[Logger] = None,
248	) -> Grammar:
249	"""Load the grammar (maybe from a pickle)."""
250	if logger is None:
251	logger = logging.getLogger(__name__)
252	gp = _generate_pickle_name(gt) if gp is None else gp
253	if force or not _newer(gp, gt):
254	g: grammar.Grammar = pgen.generate_grammar(gt)
255	if save:
256	try:
257	g.dump(gp)
258	except OSError:
259	# Ignore error, caching is not vital.
260	pass
261	else:
262	g = grammar.Grammar()
263	g.load(gp)
264	return g
265
266
267	def _newer(a: str, b: str) -> bool:
268	"""Inquire whether file a was written since file b."""
269	if not os.path.exists(a):
270	return False
271	if not os.path.exists(b):
272	return True
273	return os.path.getmtime(a) >= os.path.getmtime(b)
274
275
276	def load_packaged_grammar(
277	package: str, grammar_source: str, cache_dir: Optional[Path] = None
278	) -> grammar.Grammar:
279	"""Normally, loads a pickled grammar by doing
280	pkgutil.get_data(package, pickled_grammar)
281	where pickled_grammar is computed from grammar_source by adding the
282	Python version and using a ``.pickle`` extension.
283
284	However, if grammar_source is an extant file, load_grammar(grammar_source)
285	is called instead. This facilitates using a packaged grammar file when needed
286	but preserves load_grammar's automatic regeneration behavior when possible.
287
288	"""
289	if os.path.isfile(grammar_source):
290	gp = _generate_pickle_name(grammar_source, cache_dir) if cache_dir else None
291	return load_grammar(grammar_source, gp=gp)
292	pickled_name = _generate_pickle_name(os.path.basename(grammar_source), cache_dir)
293	data = pkgutil.get_data(package, pickled_name)
294	assert data is not None
295	g = grammar.Grammar()
296	g.loads(data)
297	return g
298
299
300	def main(*args: str) -> bool:
301	"""Main program, when run as a script: produce grammar pickle files.
302
303	Calls load_grammar for each argument, a path to a grammar text file.
304	"""
305	if not args:
306	args = tuple(sys.argv[1:])
307	logging.basicConfig(level=logging.INFO, stream=sys.stdout, format="%(message)s")
308	for gt in args:
309	load_grammar(gt, save=True, force=True)
310	return True
311
312
313	if __name__ == "__main__":
314	sys.exit(int(not main()))