18497564652

Committed 14 Oct 2025 01:08PM UTC coverage: 86.918% (+1.3%) from 85.586%

Build # 18497564652

Build Type

push

github

Committed by

web-flow

Commit Message

[tests]: update expected yaml

Coverage Stats

1083 of 1246 relevant lines covered (86.92%)

0.87 hits per line

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

88.98

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.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.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
        for _ in range(entered):
            self.writer.leave_path()
        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
        for _ in range(entered):
            self.writer.leave_path()
        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.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.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	for _ in range(entered):	1✔
168	self.writer.leave_path()	×
169	return action	1✔
170
171	def _execute_tokens(self, tokens: TokenList) -> PAction:	1✔
172	"""Execute a TokenList.
173
174	Opened and entered writer paths are left at the end of the grammar.
175	If a token returns not PAction.CONTINUE, stop executing and return
176	the last PAction.
177
178	:param syntax: List of Tokens to execute.
179	:returns: PAction of the last executed token.
180	"""
181	action = PAction.CONTINUE	1✔
182	entered = 0	1✔
183	for token in tokens:	1✔
184	logger.debug("Executing %s", token)	1✔
185	match token:	1✔
186	case tuple():	1✔
187	action = self.execute_condition(token)	×
188	case grammar_call.GrammarCall():	1✔
189	action = self.execute_grammar(token.name)	1✔
190	case out.Open() \| out.Enter():	1✔
191	entered += 1	1✔
192	action = self.execute_token(token)	1✔
193	case _:	1✔
194	action = self.execute_token(token)	1✔
195	if action != PAction.CONTINUE:	1✔
196	break	1✔
197	for _ in range(entered):	1✔
198	self.writer.leave_path()	1✔
199	return action	1✔
200
201	def trigger(self, timing: Timing) -> None:	1✔
202	"""Test triggers of a timing.
203
204	:param timing: The timing to trigger.
205	"""
206	untriggered: list[Trigger] = []	1✔
207	for trigger in self.context.queue.get(timing, []):	1✔
208	if not self.reader.would_match(trigger.match):	1✔
209	untriggered.append(trigger)	1✔
210	continue	1✔
211	logger.debug("Triggered trigger %s", trigger)	1✔
212	self.reader.match(trigger.match)	1✔
213	trigger.token.execute(self.context)	1✔
214	self.context.queue[timing] = untriggered	1✔

Hekxsler / pudding / 18497564652

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