Handle more complex expressions and add missing ROS actions functiona…

…lities (#32)

Signed-off-by: Marco Lampacrescia <[email protected]>
Signed-off-by: Christian Henkel <[email protected]>

as2fm_common/pyproject.toml

@@ -17,7 +17,7 @@ classifiers = [
 ]
 keywords = []
 dependencies = [
-
+    "js2py"
 ]
 requires-python = ">=3.10"
 

as2fm_common/src/as2fm_common/common.py

@@ -17,7 +17,7 @@
 Common functionalities used throughout the toolchain.
 """
 
-from typing import get_args, MutableSequence, Union, Type
+from typing import get_args, get_origin, MutableSequence, Union, Type
 from array import array
 
 """
@@ -96,8 +96,36 @@ def value_to_string(value: ValidTypes) -> str:
         raise ValueError(f"Unsupported value type {type(value)}.")
 
 
+def string_to_value(value_str: str, value_type: Type[ValidTypes]) -> ValidTypes:
+    """Convert a string to a value of the desired type."""
+    value_str = value_str.strip()
+    assert isinstance(value_str, str), \
+        f"Error: provided value is of type {type(value_str)}, expected a string."
+    assert len(value_str) > 0, "Error: provided value is an empty string, cannot convert."
+    is_array_value = value_str.startswith('[') and value_str.endswith(']')
+    if not is_array_value:
+        assert value_type in (bool, int, float), \
+            f"Error: the value {value_str} shall be converted to a base type."
+        return value_type(value_str)
+    else:
+        str_entries = value_str.strip('[]').split(',')
+        if str_entries == ['']:
+            str_entries = []
+        if value_type is MutableSequence[int]:
+            return array('i', [int(v) for v in str_entries])
+        elif value_type is MutableSequence[float]:
+            return array('d', [float(v) for v in str_entries])
+        else:
+            raise ValueError(f"Unsupported value type {value_type}.")
+
+
 def check_value_type_compatible(value: ValidTypes, field_type: Type[ValidTypes]) -> bool:
     """Check if the value is compatible with the field type."""
     if field_type is float:
         return isinstance(value, (int, float))
     return isinstance(value, field_type)
+
+
+def is_array_type(field_type: Type[ValidTypes]) -> bool:
+    """Check if the field type is an array type."""
+    return get_origin(field_type) == get_origin(MutableSequence)

as2fm_common/src/as2fm_common/ecmascript_interpretation.py

@@ -51,7 +51,11 @@ def interpret_ecma_script_expr(
         return expr_result
     elif isinstance(expr_result, js2py.base.JsObjectWrapper):
         if isinstance(expr_result._obj, js2py.base.PyJsArray):
-            return expr_result.to_list()
+            res_as_list = expr_result.to_list()
+            if all(isinstance(x, int) for x in res_as_list):
+                return array("i", res_as_list)
+            else:
+                return array('d', res_as_list)
         else:
             raise ValueError(f"Expected expr. {expr} to be of type {BASIC_JS_TYPES} or "
                              f"an array, got '{type(expr_result._obj)}'")

as2fm_common/test/test_unittest_ecmascript_interpretation.py

@@ -17,6 +17,7 @@
 
 import unittest
 import pytest
+from array import array
 
 from as2fm_common.ecmascript_interpretation import interpret_ecma_script_expr
 
@@ -34,7 +35,7 @@ def test_ecmascript_types(self):
         self.assertEqual(interpret_ecma_script_expr("1.1"), 1.1)
         self.assertEqual(interpret_ecma_script_expr("true"), True)
         self.assertEqual(interpret_ecma_script_expr("false"), False)
-        self.assertEqual(interpret_ecma_script_expr("[1,2,3]"), [1, 2, 3])
+        self.assertEqual(interpret_ecma_script_expr("[1,2,3]"), array('i', [1, 2, 3]))
 
     def test_ecmascript_unsupported(self):
         """

jani_generator/src/jani_generator/jani_entries/jani_convince_expression_expansion.py

@@ -63,6 +63,7 @@
     "=>": "⇒",
     "aa": "aa",
     "ac": "ac",
+    "av": "av",
 }
 
 

jani_generator/src/jani_generator/jani_entries/jani_expression.py

@@ -22,7 +22,7 @@
 
 from jani_generator.jani_entries import JaniValue
 
-SupportedExp = Union[str, int, float, bool, dict]
+SupportedExp = Union[str, int, float, bool, dict, list]
 
 
 class JaniExpressionType(Enum):
@@ -118,6 +118,9 @@ def _get_operands(self, expression_dict: dict) -> Dict[str, 'JaniExpression']:
             return {
                 "exp": JaniExpression(expression_dict["exp"]),
                 "index": JaniExpression(expression_dict["index"])}
+        if (self.op == "av"):
+            return {
+                "elements": JaniExpression(expression_dict["elements"])}
         # Convince specific expressions
         if (self.op in ("norm2d")):
             return {

jani_generator/src/jani_generator/jani_entries/jani_expression_generator.py

@@ -143,3 +143,12 @@ def array_access_operator(exp, index) -> JaniExpression:
     :param index: The index to access on exp
     """
     return JaniExpression({"op": "aa", "exp": exp, "index": index})
+
+
+def array_value_operator(elements) -> JaniExpression:
+    """
+    Generate an array value expression
+
+    :param elements: The elements of the array
+    """
+    return JaniExpression({"op": "av", "elements": elements})

jani_generator/src/jani_generator/jani_entries/jani_utils.py

@@ -0,0 +1,90 @@
+# Copyright (c) 2024 - for information on the respective copyright owner
+# see the NOTICE file
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Collection of various utilities for Jani entries."""
+
+from typing import Any, Dict, Optional, MutableSequence, Tuple, Type, get_args
+from as2fm_common.common import get_default_expression_for_type, is_array_type
+from jani_generator.jani_entries import JaniAutomaton
+
+
+def get_variable_type(jani_automaton: JaniAutomaton, variable_name: Optional[str]) -> type:
+    """
+    Retrieve the variable type from the Jani automaton.
+
+    :param jani_automaton: The Jani automaton to check the variable in.
+    :param variable_name: The name of the variable to check.
+    :return: The retrieved type.
+    """
+    assert variable_name is not None, "Variable name must be provided."
+    variable = jani_automaton.get_variables().get(variable_name)
+    assert variable is not None, \
+        f"Variable {variable_name} not found in {jani_automaton.get_variables()}."
+    return variable.get_type()
+
+
+def is_variable_array(jani_automaton: JaniAutomaton, variable_name: Optional[str]) -> bool:
+    """Check if a variable is an array.
+
+    :param jani_automaton: The Jani automaton to check the variable in.
+    :param variable_name: The name of the variable to check.
+    :return: True if the variable is an array, False otherwise.
+    """
+    return is_array_type(get_variable_type(jani_automaton, variable_name))
+
+
+def get_array_type_and_size(jani_automaton: JaniAutomaton, var_name: str) -> Tuple[Type, int]:
+    """
+    Generate the ArrayInfo obj. related to the provided variable.
+
+    :param jani_automaton: The Jani automaton to get the variable from.
+    :param var_name: The name of the variable to generate the info from.
+    :return: The ArrayInfo obj. with array type and max size.
+    """
+    assert var_name is not None, "Variable name must be provided."
+    variable = jani_automaton.get_variables().get(var_name)
+    var_type = variable.get_type()
+    assert is_array_type(var_type), f"Variable {var_name} not an array, cannot extract array info."
+    array_type = get_args(var_type)[0]
+    assert array_type in (int, float), f"Array type {array_type} not supported."
+    init_operator = variable.get_init_expr().as_operator()
+    assert init_operator is not None, f"Expected init expr of {var_name} to be an operator expr."
+    if init_operator[0] == "av":
+        max_size = len(init_operator[1]['elements'].as_literal().value())
+    elif init_operator[0] == "ac":
+        max_size = init_operator[1]['length'].as_literal().value()
+    else:
+        raise ValueError(f"Unexpected operator {init_operator[0]} for {var_name} init expr.")
+    return (array_type, max_size)
+
+
+def get_all_variables_and_instantiations(jani_automaton: JaniAutomaton) -> Dict[str, Any]:
+    """
+    Retrieve all variables and their instantiations from the Jani automaton.
+
+    :param jani_automaton: The Jani automaton to retrieve the variables from.
+    :return: A dictionary mapping each variable to a dummy value
+    """
+    variables: Dict[str, Any] = {}
+    for n, v in jani_automaton.get_variables().items():
+        variables[n] = get_default_expression_for_type(v.get_type())
+        # Hack to solve issue for expressions with explicit access to array entries
+        if isinstance(variables[n], MutableSequence):
+            for _ in range(50):
+                variables[n].append(0)
+        # Another hack, since javascript interprets 0.0 as int...
+        if isinstance(variables[n], float):
+            variables[n] = 0.1
+    return variables

jani_generator/src/jani_generator/jani_entries/jani_value.py

@@ -32,7 +32,8 @@ def is_valid(self) -> bool:
                 assert self._value["constant"] in ("e", "π"), \
                     f"Unknown constant value {self._value['constant']}. Only 'e' and 'π' supported."
                 return True
-            return False
+        elif isinstance(self._value, list):
+            return all(JaniValue(v).is_valid() for v in self._value)
         return isinstance(self._value, (int, float, bool))
 
     def value(self) -> Union[int, float, bool]:
@@ -45,6 +46,6 @@ def value(self) -> Union[int, float, bool]:
                 return pi
         return self._value
 
-    def as_dict(self) -> Union[dict, int, float, bool]:
+    def as_dict(self) -> Union[dict, int, float, bool, list]:
         # Note: this might be a value or a dictionary
         return self._value

jani_generator/src/jani_generator/jani_entries/jani_variable.py

@@ -84,10 +84,14 @@ def name(self):
         """Get name."""
         return self._name
 
-    def get_type(self):
+    def get_type(self) -> Type[ValidTypes]:
         """Get type."""
         return self._type
 
+    def get_init_expr(self) -> Optional[JaniExpression]:
+        """Get initial expression.  if available. None otherwise."""
+        return self._init_expr
+
     def as_dict(self):
         """Return the variable as a dictionary."""
         d = {

jani_generator/src/jani_generator/ros_helpers/ros_action_handler.py

@@ -22,7 +22,7 @@
 from jani_generator.ros_helpers.ros_communication_handler import RosCommunicationHandler
 
 from scxml_converter.scxml_entries import (
-    ScxmlAssign, ScxmlDataModel, ScxmlIf, ScxmlParam, ScxmlRoot, ScxmlSend, ScxmlState,
+    ScxmlAssign, ScxmlData, ScxmlDataModel, ScxmlIf, ScxmlParam, ScxmlRoot, ScxmlSend, ScxmlState,
     ScxmlTransition)
 from scxml_converter.scxml_entries.ros_utils import (
     get_action_type_params,
@@ -33,6 +33,9 @@
     generate_action_result_event, generate_action_result_handle_event,
     get_action_goal_id_definition, sanitize_ros_interface_name)
 
+from scxml_converter.scxml_entries.utils import (
+    PLAIN_SCXML_EVENT_PREFIX, PLAIN_FIELD_EVENT_PREFIX, ROS_FIELD_PREFIX)
+
 
 class RosActionHandler(RosCommunicationHandler):
     """
@@ -60,9 +63,10 @@ def _generate_goal_request_transition(
         send_params = [ScxmlParam(goal_id_name, expr=str(goal_id))]
         for field_name in req_params:
             # Add preliminary assignments (part of the hack mentioned in self.to_scxml())
+            field_w_pref = ROS_FIELD_PREFIX + field_name
             goal_req_transition.append_body_executable_entry(
-                ScxmlAssign(field_name, f"_event.{field_name}"))
-            send_params.append(ScxmlParam(field_name, expr=field_name))
+                ScxmlAssign(field_w_pref, PLAIN_FIELD_EVENT_PREFIX + field_name))
+            send_params.append(ScxmlParam(field_w_pref, expr=field_w_pref))
         # Add the send to the server
         goal_req_transition.append_body_executable_entry(
             ScxmlSend(action_srv_handle_event, send_params))
@@ -71,25 +75,32 @@ def _generate_goal_request_transition(
     def _generate_srv_event_transition(
             self, goal_state: ScxmlState, client_to_goal_id: List[Tuple[str, int]],
             event_fields: Dict[str, str], srv_event_function: Callable[[str], str],
-            client_event_function: Callable[[str, str], str]) -> ScxmlTransition:
+            client_event_function: Callable[[str, str], str],
+            additional_data: List[str]) -> ScxmlTransition:
         """
         Generate a scxml transition that triggers the client related to the input event's goal_id.
 
         :param client_to_goal_id: List of tuples (client_id, goal_id) relating clients to goal ids.
         :param event_fields: Dictionary of the parameters of the event.
         :param srv_event_function: Function to generate the server (input) event name.
         :param client_event_function: Function to generate the client (output) event name.
+        :param additional_fields: List of additional fields to be added to the event.
         """
         goal_id_name = get_action_goal_id_definition()[0]
+        extra_entries = additional_data + [goal_id_name]
         srv_event_name = srv_event_function(self._interface_name)
         scxml_transition = ScxmlTransition(goal_state.get_id(), [srv_event_name])
-        scxml_transition.append_body_executable_entry(
-            ScxmlAssign(goal_id_name, f"_event.{goal_id_name}"))
+        for entry_name in extra_entries:
+            scxml_transition.append_body_executable_entry(
+                ScxmlAssign(entry_name, PLAIN_SCXML_EVENT_PREFIX + entry_name))
         out_params: List[ScxmlParam] = []
+        for entry_name in additional_data:
+            out_params.append(ScxmlParam(entry_name, expr=entry_name))
         for field_name in event_fields:
+            field_w_pref = ROS_FIELD_PREFIX + field_name
             scxml_transition.append_body_executable_entry(
-                ScxmlAssign(field_name, f"_event.{field_name}"))
-            out_params.append(ScxmlParam(field_name, expr=field_name))
+                ScxmlAssign(field_w_pref, PLAIN_FIELD_EVENT_PREFIX + field_name))
+            out_params.append(ScxmlParam(field_w_pref, expr=field_w_pref))
         condition_send_pairs: List[Tuple[str, List[ScxmlSend]]] = []
         for client_id, goal_id in client_to_goal_id:
             client_event = client_event_function(self._interface_name, client_id)
@@ -108,7 +119,7 @@ def _generate_goal_accept_transition(
         """
         return self._generate_srv_event_transition(
             goal_state, client_to_goal_id, {}, generate_action_goal_accepted_event,
-            generate_action_goal_handle_accepted_event)
+            generate_action_goal_handle_accepted_event, [])
 
     def _generate_goal_reject_transition(
             self, goal_state: ScxmlState, client_to_goal_id: List[Tuple[str, int]]
@@ -120,7 +131,7 @@ def _generate_goal_reject_transition(
         """
         return self._generate_srv_event_transition(
             goal_state, client_to_goal_id, {}, generate_action_goal_rejected_event,
-            generate_action_goal_handle_rejected_event)
+            generate_action_goal_handle_rejected_event, [])
 
     def _generate_feedback_response_transition(
             self, goal_state: ScxmlState, client_to_goal_id: List[Tuple[str, int]],
@@ -133,7 +144,7 @@ def _generate_feedback_response_transition(
         """
         return self._generate_srv_event_transition(
             goal_state, client_to_goal_id, feedback_params, generate_action_feedback_event,
-            generate_action_feedback_handle_event)
+            generate_action_feedback_handle_event, [])
 
     def _generate_result_response_transition(
             self, goal_state: ScxmlState, client_to_goal_id: List[Tuple[str, int]],
@@ -146,7 +157,7 @@ def _generate_result_response_transition(
         """
         return self._generate_srv_event_transition(
             goal_state, client_to_goal_id, result_params, generate_action_result_event,
-            generate_action_result_handle_event)
+            generate_action_result_handle_event, ["code"])
 
     def to_scxml(self) -> ScxmlRoot:
         """
@@ -168,8 +179,10 @@ def to_scxml(self) -> ScxmlRoot:
 
         # Hack: Using support variables in the data model to avoid having _event in send params
         goal_id_def = get_action_goal_id_definition()
-        req_fields_as_data = self._generate_datamodel_from_ros_fields(
-            goal_params | feedback_params | result_params | {goal_id_def[0]: goal_id_def[1]})
+        action_fields_as_data = self._generate_datamodel_from_ros_fields(
+            goal_params | feedback_params | result_params)
+        action_fields_as_data.append(ScxmlData(goal_id_def[0], "0", goal_id_def[1]))
+        action_fields_as_data.append(ScxmlData("code", "0", "int32"))
         # Make sure the service name has no slashes and spaces
         scxml_root_name = \
             self.get_interface_prefix() + sanitize_ros_interface_name(self._interface_name)
@@ -188,7 +201,7 @@ def to_scxml(self) -> ScxmlRoot:
         wait_state.add_transition(self._generate_result_response_transition(
             wait_state, client_to_goal_id, result_params))
         scxml_root = ScxmlRoot(scxml_root_name)
-        scxml_root.set_data_model(ScxmlDataModel(req_fields_as_data))
+        scxml_root.set_data_model(ScxmlDataModel(action_fields_as_data))
         scxml_root.add_state(wait_state, initial=True)
         scxml_root.add_state(goal_requested_state)
         assert scxml_root.is_plain_scxml(), "Generated SCXML for srv sync is not plain SCXML."