27359646861

Committed 11 Jun 2026 03:53PM UTC coverage: 85.177% (+1.4%) from 83.791%

Build # 27359646861

Build Type

push

github

Committed by

jcubic

Commit Message

add Array operators

Coverage Stats

48 of 52 new or added lines in 2 files covered. (92.31%)

385 of 452 relevant lines covered (85.18%)

0.85 hits per line

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

86.59

/expression/__init__.py

# Copyright (C) 2026 Jakub T. Jankiewicz <https://jcu.bi/>
#
# This file is part of expression.py.
#
# expression.py is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# expression.py is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with expression.py. If not, see <https://www.gnu.org/licenses/>.

import math

from expression.parser import ExpressionParser as _GeneratedParser
from expression.tokenizer import ExpressionTokenizer
from expression.helpers import (
    with_type, is_typed, is_string_type, is_array_type,
    validate_number, validate_types, maybe_regex, to_array,
)


class ExpressionParserExt(_GeneratedParser):
    KEYWORDS = ('true', 'false', 'null', 'in')
    SOFT_KEYWORDS = ()

    def __init__(self, tokenizer, variables, constants, functions, source_text):
        super().__init__(tokenizer, verbose=False)
        self.variables = variables
        self.constants = constants
        self.functions = functions
        self._source_text = source_text

    def regex_literal(self):
        tok = self._tokenizer.peek()
        if tok.type == ExpressionTokenizer.REGEX_TOKEN_TYPE:
            return self._tokenizer.getnext()
        return None

    def function_assignment(self):
        mark = self._mark()
        name_tok = self.name()
        if name_tok is None:
            self._reset(mark)
            return None
        if not self.expect('('):
            self._reset(mark)
            return None
        params = []
        rparen = self.expect(')')
        if not rparen:
            p = self.name()
            if p is None:
                self._reset(mark)
                return None
            params.append(p.string)
            while self.expect(','):
                p = self.name()
                if p is None:
                    self._reset(mark)
                    return None
                params.append(p.string)
            rparen = self.expect(')')
            if not rparen:
                self._reset(mark)
                return None
        eq = self.expect('=')
        if not eq:
            self._reset(mark)
            return None
        next_tok = self._tokenizer.peek()
        if next_tok.string in ('=', '~'):
            self._reset(mark)
            return None
        body_start = next_tok.start[1]
        body = self._source_text[body_start:].rstrip(';').strip()
        while self._tokenizer.peek().string != '' and self._tokenizer.peek().type != 0:
            self._tokenizer.getnext()
        return self._func_assign(name_tok.string, params, body)

    def _func_assign(self, name, params, body=None):
        if body is None:
            body_start = self._tokenizer.peek().start[1]
            body = self._source_text[body_start:].rstrip(';').strip()

        captured_params = list(params)
        captured_body = body

        def func(*args):
            expr = Expression()
            for i, p in enumerate(captured_params):
                expr.variables[p] = args[i]
            return expr.evaluate(captured_body)

        self.functions[name] = func
        return with_type(True)

    def _var_assign(self, name, value):
        if name in self.constants:
            raise Exception(f"Can't assign value to constant '{name}'")
        self.variables[name] = value
        return value

    def _resolve_var(self, name):
        if name in ('true', 'false', 'null'):
            return None
        if name in self.constants:
            return with_type(self.constants[name])
        if name in self.variables:
            return with_type(self.variables[name])
        raise Exception(f"Variable '{name}' not found")

    def _parse_string(self, tok):
        raw = tok.string
        if raw.startswith('"'):
            value = raw[1:-1].replace('\\"', '"').replace('\\\\', '\\')
        else:
            value = raw[1:-1].replace("\\'", "'").replace('\\\\', '\\')
        return maybe_regex(value)

    def _parse_number(self, tok):
        s = tok.string
        if s.startswith('0x') or s.startswith('0X'):
            return with_type(int(s, 16))
        if s.startswith('0b') or s.startswith('0B'):
            return with_type(int(s, 2))
        if '.' in s or 'e' in s.lower():
            return with_type(float(s))
        return with_type(int(s))

    def _compare_op(self, op, a, b, fn):
        validate_types(['integer', 'double', 'boolean'], op, a)
        validate_types(['integer', 'double', 'boolean'], op, b)
        return with_type(fn(a['value'], b['value']))

    def _shift_op(self, op, a, b, fn):
        validate_number(op, a)
        validate_number(op, b)
        return with_type(fn(int(a['value']), int(b['value'])))

    def _plus(self, a, b):
        if is_array_type(a) or is_array_type(b):
            return with_type(to_array(a) + to_array(b), 'array')
        if is_string_type(b):
            return with_type(str(a['value']) + b['value'])
        validate_number('+', b)
        validate_number('+', a)
        return with_type(a['value'] + b['value'])

    def _minus(self, a, b):
        if is_array_type(a) or is_array_type(b):
            right = to_array(b)
            return with_type([x for x in to_array(a) if x not in right], 'array')
        validate_number('-', b)
        validate_number('-', a)
        return with_type(a['value'] - b['value'])

    def _mul(self, a, b):
        if is_array_type(a) or is_array_type(b):
            if is_array_type(a) and is_string_type(b):
                return with_type(b['value'].join(str(x) for x in a['value']))
            if is_array_type(b) and is_string_type(a):
                return with_type(a['value'].join(str(x) for x in b['value']))
            arr = a if is_array_type(a) else b
            num = b if is_array_type(a) else a
            validate_number('*', num)
            return with_type(list(arr['value']) * int(num['value']), 'array')
        validate_number('*', a)
        validate_number('*', b)
        return with_type(a['value'] * b['value'])

    def _union(self, a, b):
        result = []
        for x in to_array(a) + to_array(b):
            if x not in result:
                result.append(x)
        return with_type(result, 'array')

    def _intersect(self, a, b):
        if not is_array_type(a) and not is_array_type(b):
            validate_number('&', a)
            validate_number('&', b)
            return with_type(int(a['value']) & int(b['value']))
        right = to_array(b)
        result = []
        for x in to_array(a):
            if x in right and x not in result:
                result.append(x)
        return with_type(result, 'array')

    def _lshift(self, a, b):
        if is_array_type(a):
            a['value'].append(b['value'])
            return a
        return self._shift_op('<<', a, b, lambda x, y: x << y)

    def _in(self, a, b):
        right = b['value'] if is_array_type(b) else [b['value']]
        return with_type(a['value'] in right)

    def _spaceship(self, a, b):
        x = a['value']
        y = b['value']
        if x < y:
            return with_type(-1)
        if x > y:
            return with_type(1)
        return with_type(0)

    def _div(self, a, b):
        validate_number('/', a)
        validate_number('/', b)
        if b['value'] == 0:
            raise ZeroDivisionError("Division by zero")
        return with_type(a['value'] / b['value'])

    def _mod(self, a, b):
        validate_number('%', a)
        validate_number('%', b)
        return with_type(a['value'] % b['value'])

    def _property_access(self, obj, prop):
        validate_types(['array'], '[', obj)
        validate_types(['string', 'double', 'integer', 'boolean'], '[', prop)
        return with_type(obj['value'][prop['value']])

    def _implicit_mul(self, a, b):
        validate_number('[*]', a)
        validate_number('[*]', b)
        return with_type(a['value'] * b['value'])

    def _unary_minus(self, a):
        validate_number('-', a)
        return with_type(a['value'] * -1)

    def _unary_plus(self, a):
        if is_string_type(a):
            return with_type(float(a['value']))
        return a

    def _call_function(self, name, args):
        BUILTIN_MAP = {
            'arcsin': 'asin', 'arcsinh': 'asinh',
            'arccos': 'acos', 'arccosh': 'acosh',
            'arctan': 'atan', 'arctanh': 'atanh',
            'ln': 'log',
        }
        BUILTIN_FUNCTIONS = [
            'sin', 'sinh', 'asin', 'asinh',
            'cos', 'cosh', 'acos', 'acosh',
            'tan', 'tanh', 'atan', 'atanh',
            'sqrt', 'abs', 'ln', 'log',
        ]
        resolved = BUILTIN_MAP.get(name, name)
        is_builtin = resolved in BUILTIN_FUNCTIONS
        is_custom = name in self.functions
        if not is_builtin and not is_custom:
            raise Exception(f"function '{name}' doesn't exist")
        arg_values = [a['value'] for a in args]
        if is_builtin:
            if resolved == 'abs':
                func = abs
            else:
                func = getattr(math, resolved)
            result = func(*arg_values)
        else:
            result = self.functions[name](*arg_values)
        return with_type(result)

    def _build_json_obj(self, key_tok, value, rest):
        key = self._json_string_val(key_tok)
        d = {key: value}
        d.update(rest)
        return d

    def _make_obj(self, key_tok, value):
        key = self._json_string_val(key_tok)
        return {key: value}

    def _json_string_val(self, tok):
        raw = tok.string
        if raw.startswith('"'):
            return raw[1:-1].replace('\\"', '"').replace('\\\\', '\\')
        return raw[1:-1].replace("\\'", "'").replace('\\\\', '\\')

    def _json_number_val(self, tok):
        s = tok.string
        if '.' in s or 'e' in s.lower():
            return float(s)
        return int(s)

    def _merge_obj(self, key_tok, value, rest):
        key = self._json_string_val(key_tok)
        d = {key: value}
        d.update(rest)
        return d


class Expression:
    def __init__(self):
        self.constants = {"e": math.e, "pi": math.pi}
        self.variables = {}
        self.functions = {}
        self.suppress_errors = False
        self.last_error = ""

    def evaluate(self, expr):
        if not expr or not expr.strip():
            return None
        text = expr.strip()
        tokenizer = ExpressionTokenizer(text)
        parser = ExpressionParserExt(
            tokenizer, self.variables, self.constants, self.functions, text
        )
        try:
            result = parser.start()
            if result is None:
                raise Exception(f"invalid syntax: {expr}")
            self.variables = parser.variables
            self.functions = parser.functions
            if is_typed(result):
                return result['value']
            return result
        except ZeroDivisionError:
            raise
        except Exception as e:
            self.last_error = str(e) + f" in expression: {expr}"
            if not self.suppress_errors:
                raise
            return None

1	# Copyright (C) 2026 Jakub T. Jankiewicz <https://jcu.bi/>
2	#
3	# This file is part of expression.py.
4	#
5	# expression.py is free software: you can redistribute it and/or modify
6	# it under the terms of the GNU General Public License as published by
7	# the Free Software Foundation, either version 3 of the License, or
8	# (at your option) any later version.
9	#
10	# expression.py is distributed in the hope that it will be useful,
11	# but WITHOUT ANY WARRANTY; without even the implied warranty of
12	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13	# GNU General Public License for more details.
14	#
15	# You should have received a copy of the GNU General Public License
16	# along with expression.py. If not, see <https://www.gnu.org/licenses/>.
17
18	import math	1✔
19
20	from expression.parser import ExpressionParser as _GeneratedParser	1✔
21	from expression.tokenizer import ExpressionTokenizer	1✔
22	from expression.helpers import (	1✔
23	with_type, is_typed, is_string_type, is_array_type,
24	validate_number, validate_types, maybe_regex, to_array,
25	)
26
27
28	class ExpressionParserExt(_GeneratedParser):	1✔
29	KEYWORDS = ('true', 'false', 'null', 'in')	1✔
30	SOFT_KEYWORDS = ()	1✔
31
32	def __init__(self, tokenizer, variables, constants, functions, source_text):	1✔
33	super().__init__(tokenizer, verbose=False)	1✔
34	self.variables = variables	1✔
35	self.constants = constants	1✔
36	self.functions = functions	1✔
37	self._source_text = source_text	1✔
38
39	def regex_literal(self):	1✔
40	tok = self._tokenizer.peek()	1✔
41	if tok.type == ExpressionTokenizer.REGEX_TOKEN_TYPE:	1✔
42	return self._tokenizer.getnext()	1✔
43	return None	1✔
44
45	def function_assignment(self):	1✔
46	mark = self._mark()	1✔
47	name_tok = self.name()	1✔
48	if name_tok is None:	1✔
49	self._reset(mark)	1✔
50	return None	1✔
51	if not self.expect('('):	1✔
52	self._reset(mark)	1✔
53	return None	1✔
54	params = []	1✔
55	rparen = self.expect(')')	1✔
56	if not rparen:	1✔
57	p = self.name()	1✔
58	if p is None:	1✔
59	self._reset(mark)	1✔
60	return None	1✔
61	params.append(p.string)	1✔
62	while self.expect(','):	1✔
63	p = self.name()	1✔
64	if p is None:	1✔
65	self._reset(mark)	×
66	return None	×
67	params.append(p.string)	1✔
68	rparen = self.expect(')')	1✔
69	if not rparen:	1✔
70	self._reset(mark)	1✔
71	return None	1✔
72	eq = self.expect('=')	1✔
73	if not eq:	1✔
74	self._reset(mark)	1✔
75	return None	1✔
76	next_tok = self._tokenizer.peek()	1✔
77	if next_tok.string in ('=', '~'):	1✔
78	self._reset(mark)	×
79	return None	×
80	body_start = next_tok.start[1]	1✔
81	body = self._source_text[body_start:].rstrip(';').strip()	1✔
82	while self._tokenizer.peek().string != '' and self._tokenizer.peek().type != 0:	1✔
83	self._tokenizer.getnext()	1✔
84	return self._func_assign(name_tok.string, params, body)	1✔
85
86	def _func_assign(self, name, params, body=None):	1✔
87	if body is None:	1✔
88	body_start = self._tokenizer.peek().start[1]	×
89	body = self._source_text[body_start:].rstrip(';').strip()	×
90
91	captured_params = list(params)	1✔
92	captured_body = body	1✔
93
94	def func(*args):	1✔
95	expr = Expression()	1✔
96	for i, p in enumerate(captured_params):	1✔
97	expr.variables[p] = args[i]	1✔
98	return expr.evaluate(captured_body)	1✔
99
100	self.functions[name] = func	1✔
101	return with_type(True)	1✔
102
103	def _var_assign(self, name, value):	1✔
104	if name in self.constants:	1✔
105	raise Exception(f"Can't assign value to constant '{name}'")	×
106	self.variables[name] = value	1✔
107	return value	1✔
108
109	def _resolve_var(self, name):	1✔
110	if name in ('true', 'false', 'null'):	1✔
111	return None	×
112	if name in self.constants:	1✔
113	return with_type(self.constants[name])	×
114	if name in self.variables:	1✔
115	return with_type(self.variables[name])	1✔
116	raise Exception(f"Variable '{name}' not found")	×
117
118	def _parse_string(self, tok):	1✔
119	raw = tok.string	1✔
120	if raw.startswith('"'):	1✔
121	value = raw[1:-1].replace('\\"', '"').replace('\\\\', '\\')	1✔
122	else:
123	value = raw[1:-1].replace("\\'", "'").replace('\\\\', '\\')	1✔
124	return maybe_regex(value)	1✔
125
126	def _parse_number(self, tok):	1✔
127	s = tok.string	1✔
128	if s.startswith('0x') or s.startswith('0X'):	1✔
129	return with_type(int(s, 16))	1✔
130	if s.startswith('0b') or s.startswith('0B'):	1✔
131	return with_type(int(s, 2))	1✔
132	if '.' in s or 'e' in s.lower():	1✔
133	return with_type(float(s))	1✔
134	return with_type(int(s))	1✔
135
136	def _compare_op(self, op, a, b, fn):	1✔
137	validate_types(['integer', 'double', 'boolean'], op, a)	1✔
138	validate_types(['integer', 'double', 'boolean'], op, b)	1✔
139	return with_type(fn(a['value'], b['value']))	1✔
140
141	def _shift_op(self, op, a, b, fn):	1✔
142	validate_number(op, a)	1✔
143	validate_number(op, b)	1✔
144	return with_type(fn(int(a['value']), int(b['value'])))	1✔
145
146	def _plus(self, a, b):	1✔
147	if is_array_type(a) or is_array_type(b):	1✔
148	return with_type(to_array(a) + to_array(b), 'array')	1✔
149	if is_string_type(b):	1✔
150	return with_type(str(a['value']) + b['value'])	1✔
151	validate_number('+', b)	1✔
152	validate_number('+', a)	1✔
153	return with_type(a['value'] + b['value'])	1✔
154
155	def _minus(self, a, b):	1✔
156	if is_array_type(a) or is_array_type(b):	1✔
157	right = to_array(b)	1✔
158	return with_type([x for x in to_array(a) if x not in right], 'array')	1✔
159	validate_number('-', b)	1✔
160	validate_number('-', a)	1✔
161	return with_type(a['value'] - b['value'])	1✔
162
163	def _mul(self, a, b):	1✔
164	if is_array_type(a) or is_array_type(b):	1✔
165	if is_array_type(a) and is_string_type(b):	1✔
166	return with_type(b['value'].join(str(x) for x in a['value']))	1✔
167	if is_array_type(b) and is_string_type(a):	1✔
NEW 168	return with_type(a['value'].join(str(x) for x in b['value']))	×
169	arr = a if is_array_type(a) else b	1✔
170	num = b if is_array_type(a) else a	1✔
171	validate_number('*', num)	1✔
172	return with_type(list(arr['value']) * int(num['value']), 'array')	1✔
173	validate_number('*', a)	1✔
174	validate_number('*', b)	1✔
175	return with_type(a['value'] * b['value'])	1✔
176
177	def _union(self, a, b):	1✔
178	result = []	1✔
179	for x in to_array(a) + to_array(b):	1✔
180	if x not in result:	1✔
181	result.append(x)	1✔
182	return with_type(result, 'array')	1✔
183
184	def _intersect(self, a, b):	1✔
185	if not is_array_type(a) and not is_array_type(b):	1✔
NEW 186	validate_number('&', a)	×
NEW 187	validate_number('&', b)	×
NEW 188	return with_type(int(a['value']) & int(b['value']))	×
189	right = to_array(b)	1✔
190	result = []	1✔
191	for x in to_array(a):	1✔
192	if x in right and x not in result:	1✔
193	result.append(x)	1✔
194	return with_type(result, 'array')	1✔
195
196	def _lshift(self, a, b):	1✔
197	if is_array_type(a):	1✔
198	a['value'].append(b['value'])	1✔
199	return a	1✔
200	return self._shift_op('<<', a, b, lambda x, y: x << y)	1✔
201
202	def _in(self, a, b):	1✔
203	right = b['value'] if is_array_type(b) else [b['value']]	1✔
204	return with_type(a['value'] in right)	1✔
205
206	def _spaceship(self, a, b):	1✔
207	x = a['value']	1✔
208	y = b['value']	1✔
209	if x < y:	1✔
210	return with_type(-1)	1✔
211	if x > y:	1✔
212	return with_type(1)	1✔
213	return with_type(0)	1✔
214
215	def _div(self, a, b):	1✔
216	validate_number('/', a)	1✔
217	validate_number('/', b)	1✔
218	if b['value'] == 0:	1✔
219	raise ZeroDivisionError("Division by zero")	1✔
220	return with_type(a['value'] / b['value'])	1✔
221
222	def _mod(self, a, b):	1✔
223	validate_number('%', a)	1✔
224	validate_number('%', b)	1✔
225	return with_type(a['value'] % b['value'])	1✔
226
227	def _property_access(self, obj, prop):	1✔
228	validate_types(['array'], '[', obj)	1✔
229	validate_types(['string', 'double', 'integer', 'boolean'], '[', prop)	1✔
230	return with_type(obj['value'][prop['value']])	1✔
231
232	def _implicit_mul(self, a, b):	1✔
233	validate_number('[*]', a)	1✔
234	validate_number('[*]', b)	1✔
235	return with_type(a['value'] * b['value'])	1✔
236
237	def _unary_minus(self, a):	1✔
238	validate_number('-', a)	1✔
239	return with_type(a['value'] * -1)	1✔
240
241	def _unary_plus(self, a):	1✔
242	if is_string_type(a):	×
243	return with_type(float(a['value']))	×
244	return a	×
245
246	def _call_function(self, name, args):	1✔
247	BUILTIN_MAP = {	1✔
248	'arcsin': 'asin', 'arcsinh': 'asinh',
249	'arccos': 'acos', 'arccosh': 'acosh',
250	'arctan': 'atan', 'arctanh': 'atanh',
251	'ln': 'log',
252	}
253	BUILTIN_FUNCTIONS = [	1✔
254	'sin', 'sinh', 'asin', 'asinh',
255	'cos', 'cosh', 'acos', 'acosh',
256	'tan', 'tanh', 'atan', 'atanh',
257	'sqrt', 'abs', 'ln', 'log',
258	]
259	resolved = BUILTIN_MAP.get(name, name)	1✔
260	is_builtin = resolved in BUILTIN_FUNCTIONS	1✔
261	is_custom = name in self.functions	1✔
262	if not is_builtin and not is_custom:	1✔
263	raise Exception(f"function '{name}' doesn't exist")	×
264	arg_values = [a['value'] for a in args]	1✔
265	if is_builtin:	1✔
266	if resolved == 'abs':	1✔
267	func = abs	×
268	else:
269	func = getattr(math, resolved)	1✔
270	result = func(*arg_values)	1✔
271	else:
272	result = self.functions[name](*arg_values)	1✔
273	return with_type(result)	1✔
274
275	def _build_json_obj(self, key_tok, value, rest):	1✔
276	key = self._json_string_val(key_tok)	×
277	d = {key: value}	×
278	d.update(rest)	×
279	return d	×
280
281	def _make_obj(self, key_tok, value):	1✔
282	key = self._json_string_val(key_tok)	1✔
283	return {key: value}	1✔
284
285	def _json_string_val(self, tok):	1✔
286	raw = tok.string	1✔
287	if raw.startswith('"'):	1✔
288	return raw[1:-1].replace('\\"', '"').replace('\\\\', '\\')	1✔
289	return raw[1:-1].replace("\\'", "'").replace('\\\\', '\\')	1✔
290
291	def _json_number_val(self, tok):	1✔
292	s = tok.string	1✔
293	if '.' in s or 'e' in s.lower():	1✔
294	return float(s)	×
295	return int(s)	1✔
296
297	def _merge_obj(self, key_tok, value, rest):	1✔
298	key = self._json_string_val(key_tok)	×
299	d = {key: value}	×
300	d.update(rest)	×
301	return d	×
302
303
304	class Expression:	1✔
305	def __init__(self):	1✔
306	self.constants = {"e": math.e, "pi": math.pi}	1✔
307	self.variables = {}	1✔
308	self.functions = {}	1✔
309	self.suppress_errors = False	1✔
310	self.last_error = ""	1✔
311
312	def evaluate(self, expr):	1✔
313	if not expr or not expr.strip():	1✔
314	return None	1✔
315	text = expr.strip()	1✔
316	tokenizer = ExpressionTokenizer(text)	1✔
317	parser = ExpressionParserExt(	1✔
318	tokenizer, self.variables, self.constants, self.functions, text
319	)
320	try:	1✔
321	result = parser.start()	1✔
322	if result is None:	1✔
323	raise Exception(f"invalid syntax: {expr}")	×
324	self.variables = parser.variables	1✔
325	self.functions = parser.functions	1✔
326	if is_typed(result):	1✔
327	return result['value']	1✔
328	return result	×
329	except ZeroDivisionError:	1✔
330	raise	1✔
331	except Exception as e:	×
332	self.last_error = str(e) + f" in expression: {expr}"	×
333	if not self.suppress_errors:	×
334	raise	×
335	return None	×

jcubic / expression.py / 27359646861

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