Source code for dantro.data_ops.expr_ops

"""Implements data operations that work with expressions, e.g. lambda function
definitions or symbolic math"""

import logging
import math
import re
from typing import Callable, Tuple, Union

import numpy as np

from .._import_tools import LazyLoader

log = logging.getLogger(__name__)

xr = LazyLoader("xarray")
nx = LazyLoader("networkx")
pd = LazyLoader("pandas")
scipy = LazyLoader("scipy")

# -----------------------------------------------------------------------------


[docs]def expression( expr: str, *, symbols: dict = None, evaluate: bool = True, transformations: Tuple[Callable] = None, astype: Union[type, str] = float, ): """Parses and evaluates a symbolic math expression using SymPy. For parsing, uses sympy's :py:func:`sympy.parsing.sympy_parser.parse_expr`. The ``symbols`` are provided as ``local_dict``; the ``global_dict`` is not explicitly set and subsequently uses the sympy default value, containing all basic sympy symbols and notations. .. note:: The expression given here is not Python code, but symbolic math. You cannot call arbitrary functions, but only those that are imported by ``from sympy import *``. .. hint:: When using this expression as part of the :ref:`dag_framework`, it is attached to a so-called :ref:`syntax hook <dag_op_hooks_integration>` that makes it easier to specify the ``symbols`` parameter. See :ref:`here <dag_op_hook_expression>` for more information. .. warning:: While the expression is symbolic math, be aware that smypy by default interprets the ``^`` operator as XOR, not an exponentiation! For exponentiation, use the ``**`` operator or adjust the ``transformations`` argument as specified in the sympy documentation. .. warning:: The return object of this operation will *only* contain symbolic sympy objects if ``astype is None``. Otherwise, the type cast will evaluate all symbolic objects to the numerical equivalent specified by the given ``astype``. Args: expr (str): The expression to evaluate symbols (dict, optional): The symbols to use evaluate (bool, optional): Controls whether sympy evaluates ``expr``. This *may* lead to a fully evaluated result, but does not guarantee that no sympy objects are contained in the result. For ensuring a fully numerical result, see the ``astype`` argument. transformations (Tuple[Callable], optional): The ``transformations`` argument for sympy's :py:func:`sympy.parsing.sympy_parser.parse_expr`. By default, the sympy standard transformations are performed. astype (Union[type, str], optional): If given, performs a cast to this data type, fully evaluating all symbolic expressions. Default: Python :py:class:`float`. Raises: TypeError: Upon failing ``astype`` cast, e.g. due to free symbols remaining in the evaluated expression. ValueError: When parsing of ``expr`` failed. Returns: The result of the evaluated expression. """ from sympy.parsing.sympy_parser import parse_expr as _parse_expr from sympy.parsing.sympy_parser import standard_transformations as _std_trf log.remark("Evaluating symbolic expression: %s", expr) symbols = symbols if symbols else {} parse_kwargs = dict( evaluate=evaluate, transformations=( transformations if transformations is not None else _std_trf ), ) # Now, parse the expression try: res = _parse_expr(expr, local_dict=symbols, **parse_kwargs) except Exception as exc: raise ValueError( f"Failed parsing expression '{expr}'! Got a " f"{exc.__class__.__name__}: {exc}. Check that the expression can " "be evaluated with the available symbols " f"({', '.join(symbols) if symbols else 'none specified'}) " "and inspect the chained exceptions for more information. Parse " f"arguments were: {parse_kwargs}" ) from exc # Finished here if no type cast is desired if astype is None: return res # If full evaluation is desired, do so via a numpy type cast. This works on # all sympy objects, but _importantly_ also works on numpy arrays # containing sympy objects. dtype = np.dtype(astype) log.debug("Applying %s ...", dtype) try: return dtype.type(res) except Exception as exc: raise TypeError( f"Failed casting the result of expression '{expr}' from " f"{type(res)} to {dtype}! This can also be due to free symbols " "remaining in the evaluated expression. Either specify the free " f"symbols (got: {', '.join(symbols) if symbols else 'none'}) or " "deactivate casting by specifying None as ``dtype`` argument. " f"The expression evaluated to:\n\n {res}\n" ) from exc
[docs]def generate_lambda(expr: str) -> Callable: """Generates a lambda from a string. This is useful when working with callables in other operations. The ``expr`` argument needs to be a valid Python `lambda expression <https://docs.python.org/3/reference/expressions.html#lambda>`_. Inside the lambda body, the following names are available for use: * A large part of the ``builtins`` module * Every name from the Python ``math`` module, e.g. ``sin``, ``cos``, … * These modules (and their long form): ``np``, ``xr``, ``scipy`` Internally, this uses ``eval`` but imposes the following restrictions: * The following strings may *not* appear in ``expr``: ``;``, ``__``. * There can be no nested ``lambda``, i.e. the only allowed lambda string is that in the beginning of ``expr``. * The dangerous parts from the ``builtins`` module are *not* available. Args: expr (str): The expression string to evaluate into a lambda. Returns: Callable: The generated Callable. Raises: SyntaxError: Upon failed evaluation of the given expression, invalid expression pattern, or disallowed strings in the lambda body. """ ALLOWED_BUILTINS = ( "abs", "all", "any", "callable", "chr", "divmod", "format", "hash", "hex", "id", "isinstance", "issubclass", "iter", "len", "max", "min", "next", "oct", "ord", "print", "repr", "round", "sorted", "sum", "None", "Ellipsis", "False", "True", "bool", "bytes", "bytearray", "complex", "dict", "enumerate", "filter", "float", "frozenset", "int", "list", "map", "range", "reversed", "set", "slice", "str", "tuple", "type", "zip", "open", ) DISALLOWED_STRINGS = ("lambda", ";", "__") LAMBDA_PATTERN = r"^\s*lambda\s[\w\s\,\*]+\:(.+)$" # arguments definitions : lambda body (capture group) # See also: https://regex101.com/r/OmI8NY/2/ # Check if the given expression matches this pattern pattern = re.compile(LAMBDA_PATTERN) match = pattern.match(expr) if match is None: raise SyntaxError( f"The given expression '{expr}' was not a valid lambda expression!" ) # Check if the lambda body contains disallowed strings lambda_body = match[1] if any([bad_str in lambda_body for bad_str in DISALLOWED_STRINGS]): raise SyntaxError( "Encountered one or more disallowed strings in the " f"body ('{lambda_body}') of the given lambda " f"expression ('{expr}'). Make sure none of the " "following strings appears there: " f"{DISALLOWED_STRINGS}" ) # Ok, sanitized enough now. Prepare the globals dict, restricting access to # a subset of builtins and only allowing commonly used math functionality _g = dict( __builtins__={ n: f for n, f in __builtins__.items() if n in ALLOWED_BUILTINS }, math=math, scipy=scipy, pandas=pd, numpy=np, xarray=xr, networkx=nx, np=np, xr=xr, nx=nx, pd=pd, **{n: f for n, f in math.__dict__.items() if not n.startswith("_")}, ) # Try evaluation now (with empty locals) try: f = eval(expr, _g, {}) except Exception as exc: raise SyntaxError( "Failed generating lambda object from expression " f"'{expr}'! Got a {exc.__class__.__name__}: {exc}" ) from exc return f