18459430922

Committed 13 Oct 2025 08:15AM UTC coverage: 82.779% (+0.4%) from 82.393%

Build # 18459430922

Build Type

push

github

Committed by

web-flow

Commit Message

Linting and typing

Coverage Stats

995 of 1202 relevant lines covered (82.78%)

0.83 hits per line

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

87.16

pudding/processor/processor.py

"""Module defining processor class."""

import logging

from ..tokens.statements.import_ import Import

from ..tokens.statements.define import Define

from ..compiler.compiler import Syntax
from ..reader.reader import Reader
from ..tokens.datatypes import Data
from ..tokens.functions import grammar_call, out
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}")

        def set_var(name: str, value: Data) -> None:
            """Set a variable with a value in the context.

            :param name: Name of the variable.
            :param value: Value as a DataType object.
            """
            self.context.variables[name] = value

        for obj in syntax:
            match obj:
                case Define():
                    set_var(obj.values[0].value, obj.values[1])
                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 grammars own and inherited statements.

        :param name: Name of the grammar.
        :returns: ProcessingAction of the executed syntax.
        """
        action = PAction.RESTART
        grammar = self.context.get_grammar(name)
        logger.debug("-> Executing %s", grammar)
        matched = False
        while action == PAction.RESTART:
            if grammar.inherits:
                self.execute_grammar(grammar.inherits)
            action = self.execute_tokens(grammar.tokens)
            if action == PAction.RESTART:
                matched = True
        logger.debug("<- Leaving grammar %s", name)
        if matched:
            return PAction.RESTART
        return action

    def execute_tokens(self, syntax: TokenList) -> PAction:
        """Execute a given syntax.

        :param syntax: List of Tokens to execute.
        :returns: ProcessingAction of the last executed token.
        """

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

            :param token: Token to execute.
            :returns: ProcessingAction 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

        action = PAction.CONTINUE
        entered = 0
        for token in syntax:
            logger.debug("Executing %s", token)
            if isinstance(token, tuple):
                action = self.execute_condition(token)
            elif isinstance(token, grammar_call.GrammarCall):
                action = self.execute_grammar(token.name)
                if action == PAction.EXIT:
                    # continue current grammar, if called grammar is exited
                    action = PAction.CONTINUE
            else:
                if isinstance(token, (out.Open, out.Enter)):
                    entered += 1
                action = execute_token(token)
            if not action == PAction.CONTINUE:
                break
        for _ in range(entered):
            self.writer.leave_path()
        return action

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

        :param token: Tuple with condition and tokens to execute.
        :returns: ProcessingAction of the executed condition.
        """
        condition, sub_tokens = token
        action = condition.execute(self.context)
        if not action == PAction.RESTART:
            return action
        sub_action = self.execute_tokens(sub_tokens)
        if sub_action == PAction.EXIT:
            return sub_action
        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.match(trigger.match):
                untriggered.append(trigger)
                continue
            trigger.token.execute(self)
        self.context.queue[timing] = untriggered

1	"""Module defining processor class."""
2
3	import logging	1✔
4
5	from ..tokens.statements.import_ import Import	1✔
6
7	from ..tokens.statements.define import Define	1✔
8
9	from ..compiler.compiler import Syntax	1✔
10	from ..reader.reader import Reader	1✔
11	from ..tokens.datatypes import Data	1✔
12	from ..tokens.functions import grammar_call, out	1✔
13	from ..tokens.token import Token	1✔
14	from ..writer.writer import Writer	1✔
15	from . import PAction	1✔
16	from .context import Context	1✔
17	from .grammar import Grammar, TokenList	1✔
18	from .triggers import Timing, Trigger	1✔
19
20	logger = logging.getLogger(__name__)	1✔
21
22
23	class Processor:	1✔
24	"""Class processing tokens."""
25
26	def __init__(self, context: Context, syntax: Syntax) -> None:	1✔
27	"""Class processing the syntax."""
28	self.context = context	1✔
29	self._init_syntax(syntax)	1✔
30
31	def _init_syntax(self, syntax: Syntax) -> None:	1✔
32	"""Set declared variables and grammar in context.
33
34	:param syntax: Syntax to read from.
35	"""
36
37	def declare_grammar(grammar: Grammar) -> None:	1✔
38	"""Save a grammar to the context.
39
40	:param grammar: Grammar to save.
41	:raises SyntaxError: If grammar already exists.
42	"""
43	exists = self.context.grammars.get(grammar.name)	1✔
44	if exists is None:	1✔
45	self.context.grammars[grammar.name] = grammar	1✔
46	return	1✔
47	msg = f"Duplicate grammar in line {grammar.lineno}."	×
48	detail = f'Grammar "{grammar.name}" already exists in line {grammar.lineno}'	×
49	raise SyntaxError(f"{msg}\n{detail}")	×
50
51	def set_var(name: str, value: Data) -> None:	1✔
52	"""Set a variable with a value in the context.
53
54	:param name: Name of the variable.
55	:param value: Value as a DataType object.
56	"""
57	self.context.variables[name] = value	1✔
58
59	for obj in syntax:	1✔
60	match obj:	1✔
61	case Define():	1✔
62	set_var(obj.values[0].value, obj.values[1])	1✔
63	case Grammar():	1✔
64	declare_grammar(obj)	1✔
65	case _:	×
66	raise RuntimeError(f"Unprocessed statement {obj}.")	×
67
68	@property	1✔
69	def reader(self) -> Reader:	1✔
70	"""Alias for the contexts reader."""
71	return self.context.reader	1✔
72
73	@property	1✔
74	def writer(self) -> Writer:	1✔
75	"""Alias for the contexts writer."""
76	return self.context.writer	1✔
77
78	def convert(self) -> Writer:	1✔
79	"""Transform the content according to the syntax.
80
81	:returns: The writer object with the transformed data.
82	:raises RuntimeError: If no match was found.
83	"""
84	self.execute_grammar("input")	1✔
85	if self.reader.eof:	1✔
86	return self.writer	1✔
87	pos = self.reader.current_pos	×
88	unmatched = repr(self.reader.content[pos : pos + 50])	×
89	msg = f"No match found for {unmatched}..."	×
90	raise RuntimeError(	×
91	f"Unmatched text in line {self.reader.current_line_number}.\n{msg}"
92	)
93
94	def execute_grammar(self, name: str) -> PAction:	1✔
95	"""Execute a grammars own and inherited statements.
96
97	:param name: Name of the grammar.
98	:returns: ProcessingAction of the executed syntax.
99	"""
100	action = PAction.RESTART	1✔
101	grammar = self.context.get_grammar(name)	1✔
102	logger.debug("-> Executing %s", grammar)	1✔
103	matched = False	1✔
104	while action == PAction.RESTART:	1✔
105	if grammar.inherits:	1✔
106	self.execute_grammar(grammar.inherits)	1✔
107	action = self.execute_tokens(grammar.tokens)	1✔
108	if action == PAction.RESTART:	1✔
109	matched = True	1✔
110	logger.debug("<- Leaving grammar %s", name)	1✔
111	if matched:	1✔
112	return PAction.RESTART	1✔
113	return action	1✔
114
115	def execute_tokens(self, syntax: TokenList) -> PAction:	1✔
116	"""Execute a given syntax.
117
118	:param syntax: List of Tokens to execute.
119	:returns: ProcessingAction of the last executed token.
120	"""
121
122	def execute_token(token: Token) -> PAction:	1✔
123	"""Execute a token.
124
125	:param token: Token to execute.
126	:returns: ProcessingAction of the executed token.
127	"""
128	self.trigger(Timing.BEFORE)	1✔
129	action = token.execute(self.context)	1✔
130	if isinstance(token, out.Add):	1✔
131	self.trigger(Timing.ON_ADD)	1✔
132	self.trigger(Timing.AFTER)	1✔
133	return action	1✔
134
135	action = PAction.CONTINUE	1✔
136	entered = 0	1✔
137	for token in syntax:	1✔
138	logger.debug("Executing %s", token)	1✔
139	if isinstance(token, tuple):	1✔
140	action = self.execute_condition(token)	1✔
141	elif isinstance(token, grammar_call.GrammarCall):	1✔
142	action = self.execute_grammar(token.name)	1✔
143	if action == PAction.EXIT:	1✔
144	# continue current grammar, if called grammar is exited
145	action = PAction.CONTINUE	×
146	else:
147	if isinstance(token, (out.Open, out.Enter)):	1✔
148	entered += 1	1✔
149	action = execute_token(token)	1✔
150	if not action == PAction.CONTINUE:	1✔
151	break	1✔
152	for _ in range(entered):	1✔
153	self.writer.leave_path()	1✔
154	return action	1✔
155
156	def execute_condition(self, token: tuple[Token, TokenList]) -> PAction:	1✔
157	"""Execute a condition.
158
159	:param token: Tuple with condition and tokens to execute.
160	:returns: ProcessingAction of the executed condition.
161	"""
162	condition, sub_tokens = token	1✔
163	action = condition.execute(self.context)	1✔
164	if not action == PAction.RESTART:	1✔
165	return action	1✔
166	sub_action = self.execute_tokens(sub_tokens)	1✔
167	if sub_action == PAction.EXIT:	1✔
168	return sub_action	1✔
169	return action	1✔
170
171	def trigger(self, timing: Timing) -> None:	1✔
172	"""Test triggers of a timing.
173
174	:param timing: The timing to trigger.
175	"""
176	untriggered: list[Trigger] = []	1✔
177	for trigger in self.context.queue.get(timing, []):	1✔
178	if not self.reader.match(trigger.match):	×
179	untriggered.append(trigger)	×
180	continue	×
181	trigger.token.execute(self)	×
182	self.context.queue[timing] = untriggered	1✔

Hekxsler / pudding / 18459430922

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