Feat/formulas (#99)

* breaking: remove the type enforcement with the ":" symbol

* feat: allow formulas with the "=" symbol using the simpleeval package

* fix: escape warnings

* feat: add random functions and use compound type for eval

* fix: use the method of bracket matching to avoid selection issue

* fix: wrong escape

* tests: add formulas tests without escaping characters

* fix: missing test lines (previous conflict)

* docs: changelog

* Bump version: 3.0.0-dev22 → 3.0.0-dev23

---------

Co-authored-by: github-actions <[email protected]>

docs/source/dev_documentation/changelog.md

@@ -20,6 +20,7 @@
 - `ChainSimulatorFaucetStep` and `R3D4FaucetStep` can now directly use a list from the Scenario data as targets
 - `WaitStep`
 - classes for smart values to rationalize the conversion of smart values
+- Formulas can be directly supplied as arguments within MxOps scenes
 
 
 ### Changed

mxops/errors.py

@@ -23,7 +23,7 @@
 
 class ParsingError(Exception):
     """
-    To be raise when some data could not be parsed successfuly
+    To be raised when some data could not be parsed successfuly
     """
 
     def __init__(
@@ -73,6 +73,18 @@ class MaxIterationError(Exception):
     """
 
 
+class ClosingCharNotFound(Exception):
+    """
+    To be raised when a closing character could not be found in a given string
+    """
+
+    def __init__(self, string: str, closing_char: str) -> None:
+        message = (
+            f"Could not find a closing char '{closing_char}' for string '{string}'"
+        )
+        super().__init__(message)
+
+
 #############################################################
 #
 #                   Data Management Errors

mxops/execution/smart_values.py

@@ -13,43 +13,116 @@
 
 from multiversx_sdk import Address, Token, TokenTransfer
 from multiversx_sdk.core.errors import BadAddressError
+import numpy as np
+from simpleeval import EvalWithCompoundTypes
 
 from mxops import errors
 from mxops.config.config import Config
 from mxops.data.execution_data import ScenarioData
 from mxops.execution.account import AccountsManager
 
 
-def convert_arg(arg: Any, desired_type: Optional[str]) -> Any:
+def replace_escaped_characters(s: str) -> str:
     """
-    Convert an argument to a desired type.
-    Supported type are str and int
+    Replace the escape character on symbols used by MxOps
+    ex: "45 \\% 5" -> r"42 % 5"
 
-    :param arg: argument to convert
-    :type arg: Any
-    :param desired_type: type to convert the argument to
-    :type desired_type: Optional[str]
-    :return: converted argument if a specified type was provided
+    :param s: string to modify
+    :type s: str
+    :return: modified string
+    :rtype: s
+    """
+    escaped_characters = ["%", "&", "$", "=", "{", "}"]
+    for char in escaped_characters:
+        s = re.sub(rf"\\{char}", char, s)
+    return s
+
+
+def evaluate_formula(formula_str: str) -> Any:
+    """
+    evaluate the formula provided as a string by the user.
+    Remove first the escape characters on MxOps symbols
+
+    :param formula_str: formula to evaluate
+    :type formula_str: str
+    :return: result of the formula
     :rtype: Any
     """
-    if desired_type == "str":
-        return str(arg)
-    if desired_type == "int":
-        return int(arg)
-    return arg
+    formula_str = replace_escaped_characters(formula_str)
+    return EvalWithCompoundTypes(
+        functions={
+            "int": int,
+            "str": str,
+            "float": float,
+            "rand": np.random.rand,
+            "randint": np.random.randint,
+            "choice": np.random.choice,
+        }
+    ).eval(formula_str)
+
+
+def force_bracket(s: str) -> str:
+    """
+    Add brackets if the string is using a single
+    direct MxOps symbols
+
+    :param s: string to modify
+    :type s: str
+    :return: modified string
+    :rtype: str
+    """
+    pattern = r"^([$&%=])([^\{].*)$"
+    result = re.match(pattern, s)
+    if result is not None:
+        symbol, left_over = result.groups()
+        s = f"{symbol}{{{left_over}}}"
+    return s
+
+
+def get_closing_char_position(string: str, closing_char: str) -> int:
+    """
+    Find the closing character matching the first character of the string
+    characters can be escaped with backslashes
+
+    :param string: string
+    :type string: str
+    :param closing_char: closing character ex: "}"
+    :type closing_char: str
+    :return: position of the closing character
+    :rtype: int
+    """
+    opening_char = string[0]
+    opening_counter = 0
+    i = 0
+    string_len = len(string)
+    while i < string_len:
+        c = string[i]
+        if i > 0 and string[i - 1] == "\\":
+            i += 1
+            continue
+        if c == opening_char:
+            opening_counter += 1
+        elif c == closing_char:
+            opening_counter -= 1
+        if opening_counter == 0:
+            return i
+        i += 1
+    raise errors.ClosingCharNotFound(string, closing_char)
 
 
 def retrieve_value_from_string(arg: str) -> Any:
     """
-    Check if a string argument contains an env var, a config var or a data var.
+    Check if a string argument contains an env var, a config var or a data var,
+    or a formula.
     If None of the previous apply, return the string unchanged
     Examples of formated strings:
 
-    $MY_VAR:int
-    &my-var:str
+    $MY_VAR
+    &my-var
     %KEY_1.KEY_2[0].MY_VAR
     %{composed}_var
     %{%{parametric}_${VAR}}
+    ={1 + 2 + 3}
 
     :param arg: argument to check
     :type arg: str
@@ -61,11 +134,17 @@ def retrieve_value_from_string(arg: str) -> Any:
         return base64.b64decode(base64_encoded)
     if arg.startswith("0x"):
         return bytes.fromhex(arg[2:])
-    pattern = r"(.*)([$&%])\{?([a-zA-Z0-9_\-\.\]\[]+):?([a-zA-Z]*)\}?(.*)"
-    result = re.match(pattern, arg)
-    if result is None:
+    arg = force_bracket(arg)
+    matches = list(re.finditer("[$&%=]\\{", arg))
+    if len(matches) == 0:
         return arg
-    pre_arg, symbol, inner_arg, desired_type, post_arg = result.groups()
+    match = matches[-1]  # start with the last one for correct resolution order
+    match_start = match.start()
+    match_end = match.end()
+    symbol = arg[match_start]
+    closing_pos = match_end - 1 + get_closing_char_position(arg[match_end - 1 :], "}")
+    inner_arg = arg[match_end:closing_pos]
+
     if symbol == "%":
         scenario_data = ScenarioData.get()
         retrieved_value = scenario_data.get_value(inner_arg)
@@ -74,12 +153,14 @@ def retrieve_value_from_string(arg: str) -> Any:
         retrieved_value = config.get(inner_arg.upper())
     elif symbol == "$":
         retrieved_value = os.environ[inner_arg]
+    elif symbol == "=":
+        retrieved_value = evaluate_formula(inner_arg)
     else:
         raise errors.InvalidDataFormat(f"Unknow symbol {symbol}")
-    retrieved_value = convert_arg(retrieved_value, desired_type)
 
-    if pre_arg != "" or post_arg != "":
-        retrieved_value = f"{pre_arg}{retrieved_value}{post_arg}"
+    # reconstruct the string if needed
+    if match_start > 0 or closing_pos < len(arg) - 1:
+        retrieved_value = f"{arg[:match_start]}{retrieved_value}{arg[closing_pos + 1:]}"
     return retrieved_value
 
 
@@ -176,7 +257,7 @@ def get_evaluation_string(self):
         middle_values_str = []
         previous_str = None
         for mid_val in [self.raw_value, *self.evaluated_values]:
-            mid_val_str = str(mid_val)
+            mid_val_str = replace_escaped_characters(str(mid_val))
             if mid_val_str in (evaluation_str, previous_str):
                 continue
             previous_str = mid_val_str

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "mxops"
-version = "3.0.0-dev22"
+version = "3.0.0-dev23"
 authors = [
   {name="Etienne Wallet"},
 ]

requirements.txt

@@ -6,4 +6,5 @@ multiversx-sdk~=1.0.0b3
 pandas~=2.1.1
 pyyaml~=6.0
 seaborn~=0.13.0
+simpleeval~=1.0.3
 tqdm~=4.66.1

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 3.0.0-dev22
+current_version = 3.0.0-dev23
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(-(?P<release>[^-0-9]+)(?P<build>\d+))?
 serialize = 
 	{major}.{minor}.{patch}-{release}{build}

tests/test_smart_values.py

@@ -30,15 +30,43 @@
             "%{${OWNER_NAME}_%{suffix}.identifier}",
             "BOBT-123456",
             (
-                "BOBT-123456 (%{${OWNER_NAME}_%{suffix}.identifier} -> "
-                "%{${OWNER_NAME}_token.identifier} -> %{bob_token.identifier})"
+                "BOBT-123456 (%{${OWNER_NAME}_%{suffix}.identifier} "
+                "-> %{${OWNER_NAME}_token.identifier} -> %{bob_token.identifier})"
             ),
         ),
         (
             "bytes:AQIECA==",
             b"\x01\x02\x04\x08",
             "b'\\x01\\x02\\x04\\x08' (bytes:AQIECA==)",
         ),
+        ("=123456", 123456, "123456 (=123456)"),
+        ("={123456}", 123456, "123456 (={123456})"),
+        ("={'123456'}", "123456", "123456 (={'123456'})"),
+        ("={(1+2+3) * 10}", 60, "60 (={(1+2+3) * 10})"),
+        (
+            "={int(%{my_dict.key1}) // %{my_dict.key2}}",
+            0,
+            (
+                "0 (={int(%{my_dict.key1}) // %{my_dict.key2}} "
+                "-> ={int(%{my_dict.key1}) // 2} "
+                "-> ={int(1) // 2})"
+            ),
+        ),
+        (
+            "={int(%{my_dict.key1}) * 7.0 / %{my_dict.key2}}",
+            3.5,
+            (
+                "3.5 (={int(%{my_dict.key1}) * 7.0 / %{my_dict.key2}} "
+                "-> ={int(%{my_dict.key1}) * 7.0 / 2} "
+                "-> ={int(1) * 7.0 / 2})"
+            ),
+        ),
+        (r"={42 \% 5}", 2, "2 (={42 % 5})"),
+        (r"={42 % 5}", 2, "2 (={42 % 5})"),
+        (r"={dict(a\=156)}", {"a": 156}, "{'a': 156} (={dict(a=156)})"),
+        (r"={dict(a=156)}", {"a": 156}, "{'a': 156} (={dict(a=156)})"),
+        (r"={\{'a': 156\}}", {"a": 156}, "{'a': 156} (={{'a': 156}})"),
+        (r"={{'a': 156}}", {"a": 156}, "{'a': 156} (={{'a': 156}})"),
     ],
 )
 def test_smart_value(raw_value: Any, expected_result: Any, expected_str: str):
@@ -62,7 +90,7 @@ def test_deeply_nested_smart_value():
         ["%my_list", ["%my_dict", "%my_test_contract.query_result_1"]],
     )
     scenario_data.set_value("list_name", "deep_nested_list")
-    smart_value = SmartValue("%%list_name")
+    smart_value = SmartValue("%{%{list_name}}")
 
     # When
     smart_value.evaluate()
@@ -78,7 +106,7 @@ def test_deeply_nested_smart_value():
         "['item1', 'item2', 'item3', {'item4-key1': 'e'}], "
         "[{'key1': '1', 'key2': 2, 'key3': ['x', 'y', 'z']}, "
         "[0, 1, {2: 'abc'}]]] "
-        "(%%list_name -> %deep_nested_list -> "
+        "(%{%{list_name}} -> %{deep_nested_list} -> "
         "['%my_list', ['%my_dict', '%my_test_contract.query_result_1']])"
     )
 
@@ -313,6 +341,7 @@ def test_smart_token_transfer(
 ):
     # Given
     smart_value = SmartTokenTransfer(raw_value)
+
     # When
     smart_value.evaluate()
 
@@ -346,6 +375,7 @@ def test_smart_token_transfers(
 ):
     # Given
     smart_value = SmartTokenTransfers(raw_value)
+
     # When
     smart_value.evaluate()
 
@@ -359,3 +389,53 @@ def test_smart_token_transfers(
         assert ev_tr.amount == exp_tr.amount
 
     # assert smart_value.get_evaluation_string() == expected_str # TODO: wait for str
+
+
+def test_randint():
+    # Given
+    smart_value = SmartValue("=randint(2, 12)")
+
+    # When
+    smart_value.evaluate()
+
+    # Then
+    assert isinstance(smart_value.get_evaluated_value(), int)
+    assert 2 <= smart_value.get_evaluated_value() < 12
+
+
+def test_rand():
+    # Given
+    smart_value = SmartValue("=rand()")
+
+    # When
+    smart_value.evaluate()
+
+    # Then
+    assert smart_value.is_evaluated
+    assert isinstance(smart_value.get_evaluated_value(), float)
+    assert 0 <= smart_value.get_evaluated_value() <= 1
+
+
+def test_randchoice():
+    # Given
+    smart_value = SmartValue("=choice([1, 2, 3, 4])")
+
+    # When
+    smart_value.evaluate()
+
+    # Then
+    assert smart_value.is_evaluated
+    assert smart_value.get_evaluated_value() in [1, 2, 3, 4]
+
+
+def test_randchoice_nested():
+    # Given
+    scenario_data = ScenarioData.get()
+    smart_value = SmartValue("={choice(%{my_list})}")
+
+    # When
+    smart_value.evaluate()
+
+    # Then
+    assert smart_value.is_evaluated
+    assert smart_value.get_evaluated_value() in scenario_data.get_value("my_list")