Files
SystemSimulationApp/app/parameter_expression.py
T

409 lines
14 KiB
Python

from __future__ import annotations
from dataclasses import dataclass
import math
import re
from typing import Callable
MAX_INPUT_LENGTH = 512
MAX_TOKEN_COUNT = 256
MAX_OPERATION_COUNT = 256
MAX_NESTING_DEPTH = 32
MAX_FUNCTION_ARGUMENTS = 16
_UNSIGNED_NUMBER_PREFIX = re.compile(
r"(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?"
)
class ParameterExpressionError(ValueError):
"""Raised when an editor parameter expression cannot be resolved safely."""
@dataclass(frozen=True)
class _Token:
kind: str
text: str
position: int
value: float | None = None
def evaluate_parameter_expression(expression: str) -> float:
"""Evaluate the same bounded arithmetic subset accepted by the frontend.
The parser never executes Python code and cannot access names other than
the constants ``pi`` and ``e`` or the explicitly supported functions.
"""
source = expression.strip()
if source.startswith("="):
source = source[1:].strip()
if not source:
raise ParameterExpressionError("expression must not be empty")
if len(source) > MAX_INPUT_LENGTH:
raise ParameterExpressionError(
f"expression must not exceed {MAX_INPUT_LENGTH} characters"
)
return _ParameterExpressionParser(_tokenize(source)).parse()
def expression_value_to_base_unit(
value: float,
*,
quantity: str,
selected_unit: str,
) -> float:
"""Convert an expression result from its editor unit to the SI contract.
Plain numeric JSON values are already stored in SI and must not pass
through this function. Only expression results use the selected display
unit, matching the existing frontend behavior.
"""
conversions = _UNIT_CONVERSIONS.get(quantity)
if not conversions:
return _ensure_finite(value, "expression result")
conversion = conversions.get(selected_unit)
if conversion is None:
# The frontend falls back to the first (base) unit for an unknown or
# absent selection. Keep the execution boundary behavior identical.
conversion = next(iter(conversions.values()))
scale, offset = conversion
return _ensure_finite(value * scale + offset, "converted expression result")
def _tokenize(source: str) -> tuple[_Token, ...]:
tokens: list[_Token] = []
position = 0
def append(token: _Token) -> None:
tokens.append(token)
if len(tokens) > MAX_TOKEN_COUNT:
raise ParameterExpressionError(
f"expression must not exceed {MAX_TOKEN_COUNT} tokens"
)
while position < len(source):
character = source[position]
if character.isspace():
position += 1
continue
if character.isdigit() or (
character == "."
and position + 1 < len(source)
and source[position + 1].isdigit()
):
match = _UNSIGNED_NUMBER_PREFIX.match(source, position)
if match is None:
raise ParameterExpressionError(
f"invalid number near character {position + 1}"
)
text = match.group(0)
value = _ensure_finite(float(text), f"number {text!r}")
append(_Token("number", text, position, value))
position = match.end()
continue
if character.isascii() and (character.isalpha() or character == "_"):
end = position + 1
while end < len(source):
candidate = source[end]
if not candidate.isascii() or not (
candidate.isalnum() or candidate == "_"
):
break
end += 1
append(_Token("identifier", source[position:end], position))
position = end
continue
if character == "*" and source[position : position + 2] == "**":
append(_Token("operator", "**", position))
position += 2
continue
if character in "+-*/^":
append(_Token("operator", character, position))
position += 1
continue
if character == "(":
append(_Token("left_parenthesis", character, position))
position += 1
continue
if character == ")":
append(_Token("right_parenthesis", character, position))
position += 1
continue
if character == ",":
append(_Token("comma", character, position))
position += 1
continue
raise ParameterExpressionError(
f"unsupported symbol {character!r} at character {position + 1}"
)
tokens.append(_Token("end", "", len(source)))
return tuple(tokens)
class _ParameterExpressionParser:
def __init__(self, tokens: tuple[_Token, ...]) -> None:
self._tokens = tokens
self._index = 0
self._operation_count = 0
def parse(self) -> float:
value = self._parse_additive(0)
trailing = self._current()
if trailing.kind != "end":
raise ParameterExpressionError(
f"unexpected content {trailing.text!r} near character "
f"{trailing.position + 1}"
)
return _ensure_finite(value, "expression result")
def _parse_additive(self, depth: int) -> float:
value = self._parse_multiplicative(depth)
while self._is_operator("+") or self._is_operator("-"):
operator = self._advance().text
right = self._parse_multiplicative(depth)
self._count_operation()
value = _safe_operation(
lambda: value + right if operator == "+" else value - right,
f"operation {operator!r}",
)
return value
def _parse_multiplicative(self, depth: int) -> float:
value = self._parse_unary(depth)
while self._is_operator("*") or self._is_operator("/"):
operator = self._advance().text
right = self._parse_unary(depth)
self._count_operation()
if operator == "/" and right == 0:
raise ParameterExpressionError("division by zero is not allowed")
value = _safe_operation(
lambda: value * right if operator == "*" else value / right,
f"operation {operator!r}",
)
return value
def _parse_unary(self, depth: int) -> float:
self._assert_depth(depth)
if self._is_operator("+") or self._is_operator("-"):
operator = self._advance().text
self._count_operation()
operand = self._parse_unary(depth + 1)
return _ensure_finite(
operand if operator == "+" else -operand,
f"unary operation {operator!r}",
)
return self._parse_power(depth)
def _parse_power(self, depth: int) -> float:
self._assert_depth(depth)
base = self._parse_primary(depth)
if not self._is_operator("^") and not self._is_operator("**"):
return base
operator = self._advance().text
exponent = self._parse_unary(depth + 1)
self._count_operation()
return _safe_operation(
lambda: math.pow(base, exponent),
f"operation {operator!r}",
)
def _parse_primary(self, depth: int) -> float:
self._assert_depth(depth)
token = self._current()
if token.kind == "number":
self._advance()
return _ensure_finite(
token.value if token.value is not None else math.nan,
f"number {token.text!r}",
)
if token.kind == "identifier":
self._advance()
normalized_name = token.text.casefold()
if self._current().kind == "left_parenthesis":
return self._parse_function_call(
normalized_name,
token.text,
depth + 1,
)
if normalized_name == "pi":
return math.pi
if normalized_name == "e":
return math.e
raise ParameterExpressionError(f"unknown identifier {token.text!r}")
if token.kind == "left_parenthesis":
self._advance()
value = self._parse_additive(depth + 1)
self._expect("right_parenthesis", "missing closing parenthesis")
return value
if token.kind == "end":
raise ParameterExpressionError(
"expression ends before a number, constant, or function"
)
raise ParameterExpressionError(
f"expected a number, constant, or function near character "
f"{token.position + 1}"
)
def _parse_function_call(
self,
normalized_name: str,
source_name: str,
depth: int,
) -> float:
self._assert_depth(depth)
self._expect(
"left_parenthesis",
f"function {source_name} is missing an opening parenthesis",
)
arguments: list[float] = []
if self._current().kind != "right_parenthesis":
while True:
if len(arguments) >= MAX_FUNCTION_ARGUMENTS:
raise ParameterExpressionError(
f"function {source_name} accepts at most "
f"{MAX_FUNCTION_ARGUMENTS} arguments"
)
arguments.append(self._parse_additive(depth))
if self._current().kind != "comma":
break
self._advance()
if self._current().kind == "right_parenthesis":
raise ParameterExpressionError(
f"function {source_name} has no argument after its comma"
)
self._expect(
"right_parenthesis",
f"function {source_name} is missing a closing parenthesis",
)
self._count_operation()
return _evaluate_function(normalized_name, source_name, arguments)
def _current(self) -> _Token:
return self._tokens[min(self._index, len(self._tokens) - 1)]
def _advance(self) -> _Token:
token = self._current()
if token.kind != "end":
self._index += 1
return token
def _expect(self, kind: str, message: str) -> _Token:
if self._current().kind != kind:
raise ParameterExpressionError(message)
return self._advance()
def _is_operator(self, operator: str) -> bool:
token = self._current()
return token.kind == "operator" and token.text == operator
def _assert_depth(self, depth: int) -> None:
if depth > MAX_NESTING_DEPTH:
raise ParameterExpressionError(
f"expression nesting must not exceed {MAX_NESTING_DEPTH} levels"
)
def _count_operation(self) -> None:
self._operation_count += 1
if self._operation_count > MAX_OPERATION_COUNT:
raise ParameterExpressionError(
f"expression must not exceed {MAX_OPERATION_COUNT} operations"
)
def _evaluate_function(
normalized_name: str,
source_name: str,
arguments: list[float],
) -> float:
def require_count(expected: int) -> None:
if len(arguments) != expected:
raise ParameterExpressionError(
f"function {source_name} requires {expected} arguments, "
f"received {len(arguments)}"
)
if normalized_name == "sqrt":
require_count(1)
if arguments[0] < 0:
raise ParameterExpressionError("sqrt argument must not be negative")
operation = lambda: math.sqrt(arguments[0])
elif normalized_name == "abs":
require_count(1)
operation = lambda: abs(arguments[0])
elif normalized_name in {"sin", "cos", "tan", "asin", "acos", "atan"}:
require_count(1)
if normalized_name in {"asin", "acos"} and not -1 <= arguments[0] <= 1:
raise ParameterExpressionError(
f"{source_name} argument must be between -1 and 1"
)
function = getattr(math, normalized_name)
operation = lambda: function(arguments[0])
elif normalized_name == "exp":
require_count(1)
operation = lambda: math.exp(arguments[0])
elif normalized_name in {"ln", "log"}:
require_count(1)
if arguments[0] <= 0:
raise ParameterExpressionError(f"{source_name} argument must be positive")
operation = lambda: math.log(arguments[0])
elif normalized_name == "log10":
require_count(1)
if arguments[0] <= 0:
raise ParameterExpressionError("log10 argument must be positive")
operation = lambda: math.log10(arguments[0])
elif normalized_name in {"min", "max"}:
if not arguments:
raise ParameterExpressionError(
f"function {source_name} requires at least one argument"
)
function = min if normalized_name == "min" else max
operation = lambda: float(function(arguments))
elif normalized_name == "pow":
require_count(2)
operation = lambda: math.pow(arguments[0], arguments[1])
else:
raise ParameterExpressionError(f"unsupported function {source_name!r}")
return _safe_operation(operation, f"function {source_name}")
def _safe_operation(operation: Callable[[], float], context: str) -> float:
try:
value = operation()
except (ArithmeticError, ValueError) as exc:
raise ParameterExpressionError(f"{context} has no finite real result") from exc
return _ensure_finite(float(value), context)
def _ensure_finite(value: float, context: str) -> float:
if not math.isfinite(value):
raise ParameterExpressionError(f"{context} is not finite")
return value
# Ordered exactly like the editor's unit selector. The first entry is the
# fallback SI unit when a persisted selection is absent or unknown.
_UNIT_CONVERSIONS: dict[str, dict[str, tuple[float, float]]] = {
"area": {"m2": (1.0, 0.0), "cm2": (1.0e-4, 0.0), "mm2": (1.0e-6, 0.0)},
"heat_transfer_coefficient": {"W/(m2*K)": (1.0, 0.0)},
"pressure": {
"Pa": (1.0, 0.0),
"kPa": (1.0e3, 0.0),
"MPa": (1.0e6, 0.0),
"bar": (1.0e5, 0.0),
},
"volume": {"m3": (1.0, 0.0), "L": (1.0e-3, 0.0), "mL": (1.0e-6, 0.0)},
"temperature": {"K": (1.0, 0.0), "degC": (1.0, 273.15)},
"length": {"m": (1.0, 0.0), "cm": (1.0e-2, 0.0), "mm": (1.0e-3, 0.0)},
}