26163860904

Committed 20 May 2026 12:51PM UTC coverage: 92.385% (-0.9%) from 93.254%

Build # 26163860904

Build Type

Pull #5

github

Committed by

web-flow

Commit Message

[datatypes]: Adjust error message for invalid type

Pull Request Pull Request #5: Feature/specifiy output

Coverage Stats

1286 of 1392 relevant lines covered (92.39%)

0.92 hits per line

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

91.87

pudding/processor/processor.py

"""Module defining processor class."""

import logging

from pudding.exceptions import RuntimeError

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 BaseToken
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:
            """Set a grammar in the context.

            :param grammar: Grammar to declare.
            """
            exists = self.context.grammars.get(grammar.name)
            if exists is not None:
                logger.warning(
                    "Duplicate grammar %s in line %s already exists in line %s.",
                    repr(grammar.name),
                    grammar.lineno,
                    exists.lineno,
                )
            self.context.grammars[grammar.name] = grammar

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

Hekxsler / pudding / 26163860904

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