18468262725

Committed 13 Oct 2025 02:00PM UTC coverage: 85.586% (+2.8%) from 82.779%

Build # 18468262725

Build Type

push

github

Committed by

web-flow

Commit Message

[writer]: fix value being None as a string

Coverage Stats

1045 of 1221 relevant lines covered (85.59%)

0.86 hits per line

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

90.65

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 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.test_trigger(trigger):
                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 grammars own and inherited statements.
84
85	:param name: Name of the grammar.
86	:returns: ProcessingAction of the executed syntax.
87	"""
88	action = PAction.RESTART	1✔
89	grammar = self.context.get_grammar(name)	1✔
90	logger.debug("-> Executing %s", grammar)	1✔
91	matched = False	1✔
92	while action == PAction.RESTART:	1✔
93	if grammar.inherits:	1✔
94	self.execute_grammar(grammar.inherits)	1✔
95	action = self.execute_tokens(grammar.tokens)	1✔
96	if action == PAction.RESTART:	1✔
97	matched = True	1✔
98	logger.debug("<- Leaving grammar %s", name)	1✔
99	if matched:	1✔
100	return PAction.RESTART	1✔
101	return action	1✔
102
103	def execute_tokens(self, syntax: TokenList) -> PAction:	1✔
104	"""Execute a given syntax.
105
106	:param syntax: List of Tokens to execute.
107	:returns: ProcessingAction of the last executed token.
108	"""
109
110	def execute_token(token: Token) -> PAction:	1✔
111	"""Execute a token.
112
113	:param token: Token to execute.
114	:returns: ProcessingAction of the executed token.
115	"""
116	self.trigger(Timing.BEFORE)	1✔
117	action = token.execute(self.context)	1✔
118	if isinstance(token, out.Add):	1✔
119	self.trigger(Timing.ON_ADD)	1✔
120	self.trigger(Timing.AFTER)	1✔
121	return action	1✔
122
123	action = PAction.CONTINUE	1✔
124	entered = 0	1✔
125	for token in syntax:	1✔
126	logger.debug("Executing %s", token)	1✔
127	if isinstance(token, tuple):	1✔
128	action = self.execute_condition(token)	1✔
129	elif isinstance(token, grammar_call.GrammarCall):	1✔
130	action = self.execute_grammar(token.name)	1✔
131	if action == PAction.EXIT:	1✔
132	# continue current grammar, if called grammar is exited
133	action = PAction.CONTINUE	×
134	else:
135	if isinstance(token, (out.Open, out.Enter)):	1✔
136	entered += 1	1✔
137	action = execute_token(token)	1✔
138	if not action == PAction.CONTINUE:	1✔
139	break	1✔
140	for _ in range(entered):	1✔
141	self.writer.leave_path()	1✔
142	return action	1✔
143
144	def execute_condition(self, token: tuple[Token, TokenList]) -> PAction:	1✔
145	"""Execute a condition.
146
147	:param token: Tuple with condition and tokens to execute.
148	:returns: ProcessingAction of the executed condition.
149	"""
150	condition, sub_tokens = token	1✔
151	action = condition.execute(self.context)	1✔
152	if not action == PAction.RESTART:	1✔
153	return action	1✔
154	sub_action = self.execute_tokens(sub_tokens)	1✔
155	if sub_action == PAction.EXIT:	1✔
156	return sub_action	1✔
157	return action	1✔
158
159	def trigger(self, timing: Timing) -> None:	1✔
160	"""Test triggers of a timing.
161
162	:param timing: The timing to trigger.
163	"""
164	untriggered: list[Trigger] = []	1✔
165	for trigger in self.context.queue.get(timing, []):	1✔
166	if not self.reader.test_trigger(trigger):	1✔
167	untriggered.append(trigger)	1✔
168	continue	1✔
169	logger.debug("Triggered trigger %s", trigger)	1✔
170	self.reader.match(trigger.match)	1✔
171	trigger.token.execute(self.context)	1✔
172	self.context.queue[timing] = untriggered	1✔

Hekxsler / pudding / 18468262725

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