16361817426

Committed 18 Jul 2025 03:57AM UTC coverage: 52.861%. First build

Build # 16361817426

Build Type

push

github

Committed by

feihoo87

Commit Message

Add mollifier function and update version to 2.2.0

- Introduced a new mollifier function for smooth transitions in waveforms.
- Added support for the mollifier in the Waveform class and its associated formatting.
- Updated version number to 2.2.0 to reflect the new feature addition.

Run Details

14 of 51 new or added lines in 3 files covered. (27.45%)

1321 of 2499 relevant lines covered (52.86%)

6.34 hits per line

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

73.53

/waveforms/waveform_parser.py

import os
import subprocess
from ast import literal_eval
from functools import lru_cache
from pathlib import Path

from antlr4 import *
from antlr4.error.ErrorListener import ErrorListener

from . import multy_drag, waveform


class WaveformParseError(Exception):
    """Custom exception for waveform parsing errors."""
    pass


class WaveformErrorListener(ErrorListener):
    """Custom error listener for ANTLR parser."""

    def syntaxError(self, recognizer, offendingSymbol, line, column, msg, e):
        raise WaveformParseError(
            f"Syntax error at line {line}, column {column}: {msg}")


class WaveformVisitor:
    """Visitor class to evaluate waveform expressions."""

    def __init__(self):
        self.functions = [
            'D', 'chirp', 'const', 'cos', 'cosh', 'coshPulse', 'cosPulse',
            'cut', 'drag', 'drag_sin', 'drag_sinx', 'exp', 'gaussian',
            'general_cosine', 'hanning', 'interp', 'mixing', 'mollifier',
            'one', 'poly', 'samplingPoints', 'sign', 'sin', 'sinc', 'sinh',
            'square', 'step', 't', 'zero'
        ]
        self.constants = {
            'pi': waveform.pi,
            'e': waveform.e,
            'inf': waveform.inf
        }

    def get_function(self, name):
        """Get function from waveform or multy_drag modules."""
        for mod in [waveform, multy_drag]:
            try:
                return getattr(mod, name)
            except AttributeError:
                continue
        raise WaveformParseError(f"Unknown function '{name}'")

    def visit(self, ctx):
        """Visit a parse tree node."""
        method_name = f'visit{type(ctx).__name__}'
        visitor = getattr(self, method_name, self.generic_visit)
        return visitor(ctx)

    def generic_visit(self, ctx):
        """Default visitor method."""
        if hasattr(ctx, 'children') and ctx.children:
            if len(ctx.children) == 1:
                return self.visit(ctx.children[0])
            else:
                return [self.visit(child) for child in ctx.children]
        return ctx.getText()

    def visitExprContext(self, ctx):
        """Visit expression context."""
        if ctx.assignment():
            return self.visit(ctx.assignment())
        else:
            return self.visit(ctx.expression())

    def visitAssignmentContext(self, ctx):
        """Visit assignment context - not supported in waveform expressions."""
        raise WaveformParseError("Assignment expressions are not supported")

    def visitPowerExpressionContext(self, ctx):
        """Handle power expressions (** or ^)."""
        left = self.visit(ctx.expression(0))
        right = self.visit(ctx.expression(1))
        return left**right

    def visitMultiplyDivideExpressionContext(self, ctx):
        """Handle multiplication and division."""
        left = self.visit(ctx.expression(0))
        right = self.visit(ctx.expression(1))
        op = ctx.getChild(1).getText()  # Get operator text
        if op == '*':
            return left * right
        else:  # op == '/'
            return left / right

    def visitAddSubtractExpressionContext(self, ctx):
        """Handle addition and subtraction."""
        left = self.visit(ctx.expression(0))
        right = self.visit(ctx.expression(1))
        op = ctx.getChild(1).getText()  # Get operator text
        if op == '+':
            return left + right
        else:  # op == '-'
            return left - right

    def visitShiftExpressionContext(self, ctx):
        """Handle shift expressions (<< and >>)."""
        left = self.visit(ctx.expression(0))
        right = self.visit(ctx.expression(1))
        op = ctx.getChild(1).getText()  # Get operator text
        if op == '<<':
            return left << right
        else:  # op == '>>'
            return left >> right

    def visitParenthesesExpressionContext(self, ctx):
        """Handle parentheses expressions."""
        return self.visit(ctx.expression())

    def visitUnaryMinusExpressionContext(self, ctx):
        """Handle unary minus expressions."""
        return -self.visit(ctx.expression())

    def visitFunctionCallExpressionContext(self, ctx):
        """Handle function call expressions."""
        return self.visit(ctx.functionCall())

    def visitConstantExpressionContext(self, ctx):
        """Handle constant expressions."""
        const_name = ctx.CONSTANT().getText()
        return self.constants[const_name]

    def visitNumberExpressionContext(self, ctx):
        """Handle number expressions."""
        return literal_eval(ctx.NUMBER().getText())

    def visitStringExpressionContext(self, ctx):
        """Handle string expressions."""
        return literal_eval(ctx.STRING().getText())

    def visitListExpressionContext(self, ctx):
        """Handle list expressions."""
        return self.visit(ctx.list_())

    def visitTupleExpressionContext(self, ctx):
        """Handle tuple expressions."""
        return self.visit(ctx.tuple_())

    def visitIdentifierExpressionContext(self, ctx):
        """Handle identifier expressions."""
        # This could be a variable reference, but for now we'll raise an error
        # as the original implementation doesn't support variables
        var_name = ctx.ID().getText()
        raise WaveformParseError(f"Unknown identifier '{var_name}'")

    def visitNoArgFunctionContext(self, ctx):
        """Handle function calls with no arguments."""
        func_name = ctx.ID().getText()
        func = self.get_function(func_name)
        return func()

    def visitArgsFunctionContext(self, ctx):
        """Handle function calls with positional arguments."""
        func_name = ctx.ID().getText()
        func = self.get_function(func_name)
        args = self.visit(ctx.args())
        return func(*args)

    def visitKwargsFunctionContext(self, ctx):
        """Handle function calls with keyword arguments."""
        func_name = ctx.ID().getText()
        func = self.get_function(func_name)
        kwargs = self.visit(ctx.kwargs())
        return func(**kwargs)

    def visitArgsKwargsFunctionContext(self, ctx):
        """Handle function calls with both positional and keyword arguments."""
        func_name = ctx.ID().getText()
        func = self.get_function(func_name)
        args = self.visit(ctx.args())
        kwargs = self.visit(ctx.kwargs())
        return func(*args, **kwargs)

    def visitArgsContext(self, ctx):
        """Handle argument lists."""
        return [self.visit(expr) for expr in ctx.expression()]

    def visitKwargsContext(self, ctx):
        """Handle keyword argument lists."""
        kwargs = {}
        for kwarg in ctx.kwarg():
            key, value = self.visit(kwarg)
            kwargs[key] = value
        return kwargs

    def visitKwargContext(self, ctx):
        """Handle individual keyword arguments."""
        key = ctx.ID().getText()
        value = self.visit(ctx.expression())
        return key, value

    def visitListContext(self, ctx):
        """Handle list literals."""
        if ctx.expression():
            return [self.visit(expr) for expr in ctx.expression()]
        return []

    def visitTupleContext(self, ctx):
        """Handle tuple literals."""
        expressions = ctx.expression()
        if len(expressions) == 1:
            return (self.visit(expressions[0]), )
        return tuple(self.visit(expr) for expr in expressions)


def _generate_antlr_parser():
    """Generate ANTLR parser files if needed."""
    current_dir = Path(__file__).parent
    grammar_file = current_dir / "Waveform.g4"
    lexer_file = current_dir / "WaveformLexer.py"
    parser_file = current_dir / "WaveformParser.py"

    # Check if parser files exist and are newer than grammar file
    if (lexer_file.exists() and parser_file.exists()
            and lexer_file.stat().st_mtime > grammar_file.stat().st_mtime
            and parser_file.stat().st_mtime > grammar_file.stat().st_mtime):
        return

    # Generate ANTLR files
    try:
        result = subprocess.run(
            ["antlr4", "-Dlanguage=Python3",
             str(grammar_file)],
            cwd=str(current_dir),
            capture_output=True,
            text=True,
            check=True)
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        # Fall back to java command if antlr4 command is not available
        try:
            antlr_jar = os.environ.get('ANTLR_JAR',
                                       'antlr-4.11.1-complete.jar')
            result = subprocess.run([
                "java", "-jar", antlr_jar, "-Dlanguage=Python3",
                str(grammar_file)
            ],
                                    cwd=str(current_dir),
                                    capture_output=True,
                                    text=True,
                                    check=True)
        except (subprocess.CalledProcessError, FileNotFoundError):
            raise WaveformParseError(
                "Failed to generate ANTLR parser. Please install ANTLR4 or set ANTLR_JAR environment variable."
            )


def parse_waveform_expression(expr: str) -> waveform.Waveform:
    """Parse a waveform expression using ANTLR4."""
    try:
        # Generate parser files if they don't exist
        # _generate_antlr_parser()

        # Import generated ANTLR classes
        from .WaveformLexer import WaveformLexer
        from .WaveformParser import WaveformParser

        # Create lexer and parser
        input_stream = InputStream(expr)
        lexer = WaveformLexer(input_stream)
        stream = CommonTokenStream(lexer)
        parser = WaveformParser(stream)

        # Add error listener
        error_listener = WaveformErrorListener()
        parser.removeErrorListeners()
        parser.addErrorListener(error_listener)

        # Parse expression
        tree = parser.expr()

        # Visit tree and evaluate
        visitor = WaveformVisitor()
        result = visitor.visit(tree)

        # Convert numeric results to waveforms
        if isinstance(result, (int, float, complex)):
            result = waveform.const(result)

        return result.simplify()

    except Exception as e:
        if isinstance(e, WaveformParseError):
            raise
        raise WaveformParseError(
            f"Failed to parse expression '{expr}': {str(e)}")


@lru_cache(maxsize=1024)
def wave_eval(expr: str) -> "waveform.Waveform":
    """
    Parse and evaluate a waveform expression using ANTLR 4.
    
    Args:
        expr: The expression string to parse
        
    Returns:
        A Waveform object representing the parsed expression
        
    Raises:
        SyntaxError: If the expression cannot be parsed
    """
    try:
        return parse_waveform_expression(expr)
    except WaveformParseError as e:
        raise SyntaxError(f"Failed to parse expression '{expr}': {str(e)}")
    except Exception as e:
        raise SyntaxError(f"Failed to parse expression '{expr}': {str(e)}")

1	import os	12✔
2	import subprocess	12✔
3	from ast import literal_eval	12✔
4	from functools import lru_cache	12✔
5	from pathlib import Path	12✔
6
7	from antlr4 import *	12✔
8	from antlr4.error.ErrorListener import ErrorListener	12✔
9
10	from . import multy_drag, waveform	12✔
11
12
13	class WaveformParseError(Exception):	12✔
14	"""Custom exception for waveform parsing errors."""
15	pass	12✔
16
17
18	class WaveformErrorListener(ErrorListener):	12✔
19	"""Custom error listener for ANTLR parser."""
20
21	def syntaxError(self, recognizer, offendingSymbol, line, column, msg, e):	12✔
22	raise WaveformParseError(	×
23	f"Syntax error at line {line}, column {column}: {msg}")
24
25
26	class WaveformVisitor:	12✔
27	"""Visitor class to evaluate waveform expressions."""
28
29	def __init__(self):	12✔
30	self.functions = [	12✔
31	'D', 'chirp', 'const', 'cos', 'cosh', 'coshPulse', 'cosPulse',
32	'cut', 'drag', 'drag_sin', 'drag_sinx', 'exp', 'gaussian',
33	'general_cosine', 'hanning', 'interp', 'mixing', 'mollifier',
34	'one', 'poly', 'samplingPoints', 'sign', 'sin', 'sinc', 'sinh',
35	'square', 'step', 't', 'zero'
36	]
37	self.constants = {	12✔
38	'pi': waveform.pi,
39	'e': waveform.e,
40	'inf': waveform.inf
41	}
42
43	def get_function(self, name):	12✔
44	"""Get function from waveform or multy_drag modules."""
45	for mod in [waveform, multy_drag]:	12✔
46	try:	12✔
47	return getattr(mod, name)	12✔
48	except AttributeError:	×
49	continue	×
50	raise WaveformParseError(f"Unknown function '{name}'")	×
51
52	def visit(self, ctx):	12✔
53	"""Visit a parse tree node."""
54	method_name = f'visit{type(ctx).__name__}'	12✔
55	visitor = getattr(self, method_name, self.generic_visit)	12✔
56	return visitor(ctx)	12✔
57
58	def generic_visit(self, ctx):	12✔
59	"""Default visitor method."""
60	if hasattr(ctx, 'children') and ctx.children:	×
61	if len(ctx.children) == 1:	×
62	return self.visit(ctx.children[0])	×
63	else:
64	return [self.visit(child) for child in ctx.children]	×
65	return ctx.getText()	×
66
67	def visitExprContext(self, ctx):	12✔
68	"""Visit expression context."""
69	if ctx.assignment():	12✔
70	return self.visit(ctx.assignment())	×
71	else:
72	return self.visit(ctx.expression())	12✔
73
74	def visitAssignmentContext(self, ctx):	12✔
75	"""Visit assignment context - not supported in waveform expressions."""
76	raise WaveformParseError("Assignment expressions are not supported")	×
77
78	def visitPowerExpressionContext(self, ctx):	12✔
79	"""Handle power expressions (** or ^)."""
80	left = self.visit(ctx.expression(0))	×
81	right = self.visit(ctx.expression(1))	×
82	return left**right	×
83
84	def visitMultiplyDivideExpressionContext(self, ctx):	12✔
85	"""Handle multiplication and division."""
86	left = self.visit(ctx.expression(0))	12✔
87	right = self.visit(ctx.expression(1))	12✔
88	op = ctx.getChild(1).getText() # Get operator text	12✔
89	if op == '*':	12✔
90	return left * right	12✔
91	else: # op == '/'
92	return left / right	12✔
93
94	def visitAddSubtractExpressionContext(self, ctx):	12✔
95	"""Handle addition and subtraction."""
96	left = self.visit(ctx.expression(0))	12✔
97	right = self.visit(ctx.expression(1))	12✔
98	op = ctx.getChild(1).getText() # Get operator text	12✔
99	if op == '+':	12✔
100	return left + right	12✔
101	else: # op == '-'
102	return left - right	×
103
104	def visitShiftExpressionContext(self, ctx):	12✔
105	"""Handle shift expressions (<< and >>)."""
106	left = self.visit(ctx.expression(0))	12✔
107	right = self.visit(ctx.expression(1))	12✔
108	op = ctx.getChild(1).getText() # Get operator text	12✔
109	if op == '<<':	12✔
110	return left << right	12✔
111	else: # op == '>>'
112	return left >> right	12✔
113
114	def visitParenthesesExpressionContext(self, ctx):	12✔
115	"""Handle parentheses expressions."""
116	return self.visit(ctx.expression())	12✔
117
118	def visitUnaryMinusExpressionContext(self, ctx):	12✔
119	"""Handle unary minus expressions."""
120	return -self.visit(ctx.expression())	12✔
121
122	def visitFunctionCallExpressionContext(self, ctx):	12✔
123	"""Handle function call expressions."""
124	return self.visit(ctx.functionCall())	12✔
125
126	def visitConstantExpressionContext(self, ctx):	12✔
127	"""Handle constant expressions."""
128	const_name = ctx.CONSTANT().getText()	12✔
129	return self.constants[const_name]	12✔
130
131	def visitNumberExpressionContext(self, ctx):	12✔
132	"""Handle number expressions."""
133	return literal_eval(ctx.NUMBER().getText())	12✔
134
135	def visitStringExpressionContext(self, ctx):	12✔
136	"""Handle string expressions."""
137	return literal_eval(ctx.STRING().getText())	12✔
138
139	def visitListExpressionContext(self, ctx):	12✔
140	"""Handle list expressions."""
141	return self.visit(ctx.list_())	12✔
142
143	def visitTupleExpressionContext(self, ctx):	12✔
144	"""Handle tuple expressions."""
145	return self.visit(ctx.tuple_())	12✔
146
147	def visitIdentifierExpressionContext(self, ctx):	12✔
148	"""Handle identifier expressions."""
149	# This could be a variable reference, but for now we'll raise an error
150	# as the original implementation doesn't support variables
151	var_name = ctx.ID().getText()	×
152	raise WaveformParseError(f"Unknown identifier '{var_name}'")	×
153
154	def visitNoArgFunctionContext(self, ctx):	12✔
155	"""Handle function calls with no arguments."""
156	func_name = ctx.ID().getText()	12✔
157	func = self.get_function(func_name)	12✔
158	return func()	12✔
159
160	def visitArgsFunctionContext(self, ctx):	12✔
161	"""Handle function calls with positional arguments."""
162	func_name = ctx.ID().getText()	12✔
163	func = self.get_function(func_name)	12✔
164	args = self.visit(ctx.args())	12✔
165	return func(*args)	12✔
166
167	def visitKwargsFunctionContext(self, ctx):	12✔
168	"""Handle function calls with keyword arguments."""
169	func_name = ctx.ID().getText()	×
170	func = self.get_function(func_name)	×
171	kwargs = self.visit(ctx.kwargs())	×
172	return func(**kwargs)	×
173
174	def visitArgsKwargsFunctionContext(self, ctx):	12✔
175	"""Handle function calls with both positional and keyword arguments."""
176	func_name = ctx.ID().getText()	12✔
177	func = self.get_function(func_name)	12✔
178	args = self.visit(ctx.args())	12✔
179	kwargs = self.visit(ctx.kwargs())	12✔
180	return func(args, *kwargs)	12✔
181
182	def visitArgsContext(self, ctx):	12✔
183	"""Handle argument lists."""
184	return [self.visit(expr) for expr in ctx.expression()]	12✔
185
186	def visitKwargsContext(self, ctx):	12✔
187	"""Handle keyword argument lists."""
188	kwargs = {}	12✔
189	for kwarg in ctx.kwarg():	12✔
190	key, value = self.visit(kwarg)	12✔
191	kwargs[key] = value	12✔
192	return kwargs	12✔
193
194	def visitKwargContext(self, ctx):	12✔
195	"""Handle individual keyword arguments."""
196	key = ctx.ID().getText()	12✔
197	value = self.visit(ctx.expression())	12✔
198	return key, value	12✔
199
200	def visitListContext(self, ctx):	12✔
201	"""Handle list literals."""
202	if ctx.expression():	12✔
203	return [self.visit(expr) for expr in ctx.expression()]	12✔
204	return []	×
205
206	def visitTupleContext(self, ctx):	12✔
207	"""Handle tuple literals."""
208	expressions = ctx.expression()	12✔
209	if len(expressions) == 1:	12✔
210	return (self.visit(expressions[0]), )	×
211	return tuple(self.visit(expr) for expr in expressions)	12✔
212
213
214	def _generate_antlr_parser():	12✔
215	"""Generate ANTLR parser files if needed."""
216	current_dir = Path(__file__).parent	×
217	grammar_file = current_dir / "Waveform.g4"	×
218	lexer_file = current_dir / "WaveformLexer.py"	×
219	parser_file = current_dir / "WaveformParser.py"	×
220
221	# Check if parser files exist and are newer than grammar file
222	if (lexer_file.exists() and parser_file.exists()	×
223	and lexer_file.stat().st_mtime > grammar_file.stat().st_mtime
224	and parser_file.stat().st_mtime > grammar_file.stat().st_mtime):
225	return	×
226
227	# Generate ANTLR files
228	try:	×
NEW 229	result = subprocess.run(	×
230	["antlr4", "-Dlanguage=Python3",
231	str(grammar_file)],
232	cwd=str(current_dir),
233	capture_output=True,
234	text=True,
235	check=True)
236	except (subprocess.CalledProcessError, FileNotFoundError) as e:	×
237	# Fall back to java command if antlr4 command is not available
238	try:	×
239	antlr_jar = os.environ.get('ANTLR_JAR',	×
240	'antlr-4.11.1-complete.jar')
241	result = subprocess.run([	×
242	"java", "-jar", antlr_jar, "-Dlanguage=Python3",
243	str(grammar_file)
244	],
245	cwd=str(current_dir),
246	capture_output=True,
247	text=True,
248	check=True)
249	except (subprocess.CalledProcessError, FileNotFoundError):	×
250	raise WaveformParseError(	×
251	"Failed to generate ANTLR parser. Please install ANTLR4 or set ANTLR_JAR environment variable."
252	)
253
254
255	def parse_waveform_expression(expr: str) -> waveform.Waveform:	12✔
256	"""Parse a waveform expression using ANTLR4."""
257	try:	12✔
258	# Generate parser files if they don't exist
259	# _generate_antlr_parser()
260
261	# Import generated ANTLR classes
262	from .WaveformLexer import WaveformLexer	12✔
263	from .WaveformParser import WaveformParser	12✔
264
265	# Create lexer and parser
266	input_stream = InputStream(expr)	12✔
267	lexer = WaveformLexer(input_stream)	12✔
268	stream = CommonTokenStream(lexer)	12✔
269	parser = WaveformParser(stream)	12✔
270
271	# Add error listener
272	error_listener = WaveformErrorListener()	12✔
273	parser.removeErrorListeners()	12✔
274	parser.addErrorListener(error_listener)	12✔
275
276	# Parse expression
277	tree = parser.expr()	12✔
278
279	# Visit tree and evaluate
280	visitor = WaveformVisitor()	12✔
281	result = visitor.visit(tree)	12✔
282
283	# Convert numeric results to waveforms
284	if isinstance(result, (int, float, complex)):	12✔
285	result = waveform.const(result)	12✔
286
287	return result.simplify()	12✔
288
289	except Exception as e:	×
290	if isinstance(e, WaveformParseError):	×
291	raise	×
292	raise WaveformParseError(	×
293	f"Failed to parse expression '{expr}': {str(e)}")
294
295
296	@lru_cache(maxsize=1024)	12✔
297	def wave_eval(expr: str) -> "waveform.Waveform":	12✔
298	"""
299	Parse and evaluate a waveform expression using ANTLR 4.
300
301	Args:
302	expr: The expression string to parse
303
304	Returns:
305	A Waveform object representing the parsed expression
306
307	Raises:
308	SyntaxError: If the expression cannot be parsed
309	"""
310	try:	12✔
311	return parse_waveform_expression(expr)	12✔
312	except WaveformParseError as e:	×
313	raise SyntaxError(f"Failed to parse expression '{expr}': {str(e)}")	×
314	except Exception as e:	×
315	raise SyntaxError(f"Failed to parse expression '{expr}': {str(e)}")	×

feihoo87 / waveforms / 16361817426

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