20368087396

Committed 19 Dec 2025 11:04AM UTC coverage: 83.669% (-4.6%) from 88.305%

Build # 20368087396

Build Type

Pull #1

github

Committed by

web-flow

Commit Message

[all]: Fix typing

Pull Request Pull Request #1: Increase performance with custom node class

Run Details

1081 of 1292 relevant lines covered (83.67%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.6

pudding/processor/processor.py

"""Module defining processor class."""

import logging

from ..compiler.compiler import Syntax
from ..reader.reader import Reader
from ..tokens.functions import grammar_call, out
from ..tokens.statements.define import Define
from ..tokens.token import Token
from ..writer import Writer
from . import PAction
from .context import Context
from .grammar import Grammar, TokenList
from .triggers import Timing, Trigger

logger = logging.getLogger(__name__)


class Processor:
    """Class processing tokens."""

    def __init__(self, context: Context, syntax: Syntax) -> None:
        """Class processing the syntax."""
        self.context = context
        self._init_syntax(syntax)

    def _init_syntax(self, syntax: Syntax) -> None:
        """Set declared variables and grammar in context.

        :param syntax: Syntax to read from.
        """

        def declare_grammar(grammar: Grammar) -> None:
            """Save a grammar to the context.

            :param grammar: Grammar to save.
            :raises SyntaxError: If grammar already exists.
            """
            exists = self.context.grammars.get(grammar.name)
            if exists is None:
                self.context.grammars[grammar.name] = grammar
                return
            msg = f"Duplicate grammar in line {grammar.lineno}."
            detail = f'Grammar "{grammar.name}" already exists in line {grammar.lineno}'
            raise SyntaxError(f"{msg}\n{detail}")

        for obj in syntax:
            match obj:
                case Define():
                    obj.execute(self.context)
                case Grammar():
                    declare_grammar(obj)
                case _:
                    raise RuntimeError(f"Unprocessed statement {obj}.")

    @property
    def reader(self) -> Reader:
        """Alias for the contexts reader."""
        return self.context.reader

    @property
    def writer(self) -> Writer:
        """Alias for the contexts writer."""
        return self.context.writer

    def convert(self) -> Writer:
        """Transform the content according to the syntax.

        :returns: The writer object with the transformed data.
        :raises RuntimeError: If no match was found.
        """
        self.execute_grammar("input")
        if self.reader.eof:
            return self.writer
        pos = self.reader.current_pos
        unmatched = repr(self.reader.content[pos : pos + 50])
        msg = f"No match found for {unmatched}..."
        raise RuntimeError(
            f"Unmatched text in line {self.reader.current_line_number}.\n{msg}"
        )

    def execute_grammar(self, name: str) -> PAction:
        """Execute a grammar by name.

        Execute tokens of a grammar with the given name. If a inherited grammar
        exists, it will be executed first.

        :param name: Name of the grammar.
        :returns: PAction.RESTART if grammar restarted at least once
            else PAction.CONTINUE.
        """
        grammar = self.context.get_grammar(name)
        logger.debug("-> Executing %s", grammar)
        action = PAction.RESTART
        restarts: int = 0
        while action == PAction.RESTART:
            restarts += 1
            if grammar.inherits:
                self.execute_grammar(grammar.inherits)
            action = self._execute_grammar(grammar)
        logger.debug("<- Leaving grammar %s", name)
        if restarts > 1:
            return PAction.RESTART
        return PAction.CONTINUE

    def execute_condition(self, token: tuple[Token, TokenList]) -> PAction:
        """Execute a condition.

        The condition token will return PAction.ENTER if sub tokens should be
        executed. Otherwise the condition is not met and sub tokens not executed.
        If all sub tokens have been executed, restart the grammar unless another
        PAction than CONTINUE is requested.

        :param token: Tuple with condition and tokens to execute.
        :returns: ProcessingAction.
        """
        condition, sub_tokens = token
        action = condition.execute(self.context)
        if not action == PAction.ENTER:
            return action
        sub_action = self._execute_tokens(sub_tokens)
        if sub_action == PAction.CONTINUE:
            return PAction.RESTART
        return sub_action

    def execute_token(self, token: Token) -> PAction:
        """Execute a token.

        Execute the token and trigger Timings before and after the execution.

        :param token: Token to execute.
        :returns: PAction of the executed token.
        """
        self.trigger(Timing.BEFORE)
        action = token.execute(self.context)
        if isinstance(token, out.Add):
            self.trigger(Timing.ON_ADD)
        self.trigger(Timing.AFTER)
        return action

    def _execute_grammar(self, grammar: Grammar) -> PAction:
        """Execute tokens of a grammar.

        Opened and entered writer paths are left at the end of the grammar.
        PAction.NEXT is treated as PAction.CONTINUE so the next token of
        the grammar is executed.

        :param syntax: Grammar to execute.
        :returns: PAction of the last executed token.
        """
        action = PAction.EXIT
        entered = 0
        for token in grammar.iter_tokens():
            logger.debug("Executing %s", token)
            match token:
                case tuple():
                    action = self.execute_condition(token)
                case grammar_call.GrammarCall():
                    action = self.execute_grammar(token.name)
                case out.Open() | out.Enter():
                    entered += 1
                    action = self.execute_token(token)
                case _:
                    action = self.execute_token(token)
            if action not in (PAction.CONTINUE, PAction.NEXT):
                break
        self.writer.leave_paths(entered)
        return action

    def _execute_tokens(self, tokens: TokenList) -> PAction:
        """Execute a TokenList.

        Opened and entered writer paths are left at the end of the grammar.
        If a token returns not PAction.CONTINUE, stop executing and return
        the last PAction.

        :param syntax: List of Tokens to execute.
        :returns: PAction of the last executed token.
        """
        action = PAction.CONTINUE
        entered = 0
        for token in tokens:
            logger.debug("Executing %s", token)
            match token:
                case tuple():
                    action = self.execute_condition(token)
                case grammar_call.GrammarCall():
                    action = self.execute_grammar(token.name)
                case out.Open() | out.Enter():
                    entered += 1
                    action = self.execute_token(token)
                case _:
                    action = self.execute_token(token)
            if action != PAction.CONTINUE:
                break
        self.writer.leave_paths(entered)
        return action

    def trigger(self, timing: Timing) -> None:
        """Test triggers of a timing.

        :param timing: The timing to trigger.
        """
        untriggered: list[Trigger] = []
        for trigger in self.context.queue.get(timing, []):
            if not self.reader.would_match(trigger.match):
                untriggered.append(trigger)
                continue
            logger.debug("Triggered trigger %s", trigger)
            self.reader.match(trigger.match)
            trigger.token.execute(self.context)
        self.context.queue[timing] = untriggered

1	"""Module defining processor class."""
2
3	import logging	1✔
4
5	from ..compiler.compiler import Syntax	1✔
6	from ..reader.reader import Reader	1✔
7	from ..tokens.functions import grammar_call, out	1✔
8	from ..tokens.statements.define import Define	1✔
9	from ..tokens.token import Token	1✔
10	from ..writer import Writer	1✔
11	from . import PAction	1✔
12	from .context import Context	1✔
13	from .grammar import Grammar, TokenList	1✔
14	from .triggers import Timing, Trigger	1✔
15
16	logger = logging.getLogger(__name__)	1✔
17
18
19	class Processor:	1✔
20	"""Class processing tokens."""
21
22	def __init__(self, context: Context, syntax: Syntax) -> None:	1✔
23	"""Class processing the syntax."""
24	self.context = context	1✔
25	self._init_syntax(syntax)	1✔
26
27	def _init_syntax(self, syntax: Syntax) -> None:	1✔
28	"""Set declared variables and grammar in context.
29
30	:param syntax: Syntax to read from.
31	"""
32
33	def declare_grammar(grammar: Grammar) -> None:	1✔
34	"""Save a grammar to the context.
35
36	:param grammar: Grammar to save.
37	:raises SyntaxError: If grammar already exists.
38	"""
39	exists = self.context.grammars.get(grammar.name)	1✔
40	if exists is None:	1✔
41	self.context.grammars[grammar.name] = grammar	1✔
42	return	1✔
43	msg = f"Duplicate grammar in line {grammar.lineno}."	×
44	detail = f'Grammar "{grammar.name}" already exists in line {grammar.lineno}'	×
45	raise SyntaxError(f"{msg}\n{detail}")	×
46
47	for obj in syntax:	1✔
48	match obj:	1✔
49	case Define():	1✔
50	obj.execute(self.context)	1✔
51	case Grammar():	1✔
52	declare_grammar(obj)	1✔
53	case _:	×
54	raise RuntimeError(f"Unprocessed statement {obj}.")	×
55
56	@property	1✔
57	def reader(self) -> Reader:	1✔
58	"""Alias for the contexts reader."""
59	return self.context.reader	1✔
60
61	@property	1✔
62	def writer(self) -> Writer:	1✔
63	"""Alias for the contexts writer."""
64	return self.context.writer	1✔
65
66	def convert(self) -> Writer:	1✔
67	"""Transform the content according to the syntax.
68
69	:returns: The writer object with the transformed data.
70	:raises RuntimeError: If no match was found.
71	"""
72	self.execute_grammar("input")	1✔
73	if self.reader.eof:	1✔
74	return self.writer	1✔
75	pos = self.reader.current_pos	×
76	unmatched = repr(self.reader.content[pos : pos + 50])	×
77	msg = f"No match found for {unmatched}..."	×
78	raise RuntimeError(	×
79	f"Unmatched text in line {self.reader.current_line_number}.\n{msg}"
80	)
81
82	def execute_grammar(self, name: str) -> PAction:	1✔
83	"""Execute a grammar by name.
84
85	Execute tokens of a grammar with the given name. If a inherited grammar
86	exists, it will be executed first.
87
88	:param name: Name of the grammar.
89	:returns: PAction.RESTART if grammar restarted at least once
90	else PAction.CONTINUE.
91	"""
92	grammar = self.context.get_grammar(name)	1✔
93	logger.debug("-> Executing %s", grammar)	1✔
94	action = PAction.RESTART	1✔
95	restarts: int = 0	1✔
96	while action == PAction.RESTART:	1✔
97	restarts += 1	1✔
98	if grammar.inherits:	1✔
99	self.execute_grammar(grammar.inherits)	1✔
100	action = self._execute_grammar(grammar)	1✔
101	logger.debug("<- Leaving grammar %s", name)	1✔
102	if restarts > 1:	1✔
103	return PAction.RESTART	1✔
104	return PAction.CONTINUE	1✔
105
106	def execute_condition(self, token: tuple[Token, TokenList]) -> PAction:	1✔
107	"""Execute a condition.
108
109	The condition token will return PAction.ENTER if sub tokens should be
110	executed. Otherwise the condition is not met and sub tokens not executed.
111	If all sub tokens have been executed, restart the grammar unless another
112	PAction than CONTINUE is requested.
113
114	:param token: Tuple with condition and tokens to execute.
115	:returns: ProcessingAction.
116	"""
117	condition, sub_tokens = token	1✔
118	action = condition.execute(self.context)	1✔
119	if not action == PAction.ENTER:	1✔
120	return action	1✔
121	sub_action = self._execute_tokens(sub_tokens)	1✔
122	if sub_action == PAction.CONTINUE:	1✔
123	return PAction.RESTART	1✔
124	return sub_action	1✔
125
126	def execute_token(self, token: Token) -> PAction:	1✔
127	"""Execute a token.
128
129	Execute the token and trigger Timings before and after the execution.
130
131	:param token: Token to execute.
132	:returns: PAction of the executed token.
133	"""
134	self.trigger(Timing.BEFORE)	1✔
135	action = token.execute(self.context)	1✔
136	if isinstance(token, out.Add):	1✔
137	self.trigger(Timing.ON_ADD)	1✔
138	self.trigger(Timing.AFTER)	1✔
139	return action	1✔
140
141	def _execute_grammar(self, grammar: Grammar) -> PAction:	1✔
142	"""Execute tokens of a grammar.
143
144	Opened and entered writer paths are left at the end of the grammar.
145	PAction.NEXT is treated as PAction.CONTINUE so the next token of
146	the grammar is executed.
147
148	:param syntax: Grammar to execute.
149	:returns: PAction of the last executed token.
150	"""
151	action = PAction.EXIT	1✔
152	entered = 0	1✔
153	for token in grammar.iter_tokens():	1✔
154	logger.debug("Executing %s", token)	1✔
155	match token:	1✔
156	case tuple():	1✔
157	action = self.execute_condition(token)	1✔
158	case grammar_call.GrammarCall():	1✔
159	action = self.execute_grammar(token.name)	×
160	case out.Open() \| out.Enter():	1✔
161	entered += 1	×
162	action = self.execute_token(token)	×
163	case _:	1✔
164	action = self.execute_token(token)	1✔
165	if action not in (PAction.CONTINUE, PAction.NEXT):	1✔
166	break	1✔
167	self.writer.leave_paths(entered)	1✔
168	return action	1✔
169
170	def _execute_tokens(self, tokens: TokenList) -> PAction:	1✔
171	"""Execute a TokenList.
172
173	Opened and entered writer paths are left at the end of the grammar.
174	If a token returns not PAction.CONTINUE, stop executing and return
175	the last PAction.
176
177	:param syntax: List of Tokens to execute.
178	:returns: PAction of the last executed token.
179	"""
180	action = PAction.CONTINUE	1✔
181	entered = 0	1✔
182	for token in tokens:	1✔
183	logger.debug("Executing %s", token)	1✔
184	match token:	1✔
185	case tuple():	1✔
186	action = self.execute_condition(token)	×
187	case grammar_call.GrammarCall():	1✔
188	action = self.execute_grammar(token.name)	1✔
189	case out.Open() \| out.Enter():	1✔
190	entered += 1	1✔
191	action = self.execute_token(token)	1✔
192	case _:	1✔
193	action = self.execute_token(token)	1✔
194	if action != PAction.CONTINUE:	1✔
195	break	1✔
196	self.writer.leave_paths(entered)	1✔
197	return action	1✔
198
199	def trigger(self, timing: Timing) -> None:	1✔
200	"""Test triggers of a timing.
201
202	:param timing: The timing to trigger.
203	"""
204	untriggered: list[Trigger] = []	1✔
205	for trigger in self.context.queue.get(timing, []):	1✔
206	if not self.reader.would_match(trigger.match):	1✔
207	untriggered.append(trigger)	1✔
208	continue	1✔
209	logger.debug("Triggered trigger %s", trigger)	1✔
210	self.reader.match(trigger.match)	1✔
211	trigger.token.execute(self.context)	1✔
212	self.context.queue[timing] = untriggered	1✔

Hekxsler / pudding / 20368087396

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous