20849001152

Committed 09 Jan 2026 10:30AM UTC coverage: 89.189% (+0.3%) from 88.906%

Build # 20849001152

Build Type

Pull #2

github

Committed by

web-flow

Commit Message

Merge d9ef120c7 into 26faf08cc

Pull Request Pull Request #2: Documentation

Run Details

1155 of 1295 relevant lines covered (89.19%)

0.89 hits per line

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

91.8

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

Hekxsler / pudding / 20849001152

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