20781436505

Committed 07 Jan 2026 12:26PM UTC coverage: 88.906% (+0.6%) from 88.305%

Build # 20781436505

Build Type

push

github

Committed by

web-flow

Commit Message

Increase performance with custom node class (#1)

* [xml]: add write_to message for less memory usage when writing to file

* [xml]: Add custom node class and add to xml writer

* [xml]: Remove use of lxml.builder

* [writer]: change node.find to use recursion

* [writer]: fix self return

* [writer]: undo recursion change

* [xml]: add write_to method for xml writer

* [writer]: move writers into subpackage

* [grammar]: add iterator for tokens

* [misc]: formatting

* [tokens]: change match to use generator for pattern matching

* [node]: move path functions into class and fix pattern

* [writer]: Adjust writer to changes in node class

* [tokens]: change get_patterns to generator

* [statement]: change order for common data types

* [statement]: fix if statements

* [misc]: Add performance optimizations from copilot

* [writer]: Add slim xml writer

* Revert "[statement]: fix if statements"

This reverts commit d6fea2567.

* Revert "[misc]: Add performance optimizations from copilot"

This reverts commit c13e8e93b.

* [fix]: Changes by copilot

* [writers]: add slixml writer to choice

* [xml]: add fixes to slixml writer

* [slixml]: change add element to append text to current element

* [context]: change var names to be more descriptive

* [writer]: change leave_path to leave_paths

* [writer]: fix yaml writer

* [tokens]: remove unnecessary abstract functions

* [token]: change for loop to generator

* [match/when]: load match/find func before loop

* [MultiExpStatement]: fix from_string

* [node]: Remove useless iter_children function

* [node]: Add __eq__() and fix bug in get()

* [node]; switch to saving children in dict

* [node]: fix find bug

* [node]: undo changing comparison

* [node]: change find() to use filter()

* [writer]: performance changes

* [node]: localize parse func

* [node]: add function to return sorted children

* [tests]: chango comparison strategy

* Reduce c... (continued)

Run Details

1154 of 1298 relevant lines covered (88.91%)

0.89 hits per line

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

89.6

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

Hekxsler / pudding / 20781436505

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