parse rossystem, add qos and parameters

ros2model/api/model_generator/component_generator.py

@@ -17,17 +17,26 @@
 from pathlib import Path
 
 from ros2model.core.generator.generator_core import GeneratorCore
-from ros2model.core.generator.file_generator import FileGenerator
 import typing as t
+from ament_index_python import get_package_share_directory
 
-Template_Folder = Path(__file__).parent.resolve() / "templates"
+Template_Folder = Path(__file__).parent.parent.parent.parent.resolve() / "templates"
 Template = Path(Template_Folder / "component.ros2.j2")
 
+Template_Folder_ROS = Path(get_package_share_directory("ros2model") + "/templates")
+Template_ROS = Path(Template_Folder_ROS / "component.ros2.j2")
+
 
 class ComponentGenerator(GeneratorCore):
     def __init__(self, template_path=None) -> None:
         if template_path != None:
             self.template_path = Path(template_path).resolve()
-        else:
+        elif Template_ROS.is_file():
             self.template_path = Template
+        elif Template.is_file():
+            self.template_path = Template
+        else:
+            raise FileNotFoundError(
+                f"Can't find template either from {Template.absolute().as_posix()} or {Template_ROS.absolute().as_posix()}"
+            )
         super().__init__(self.template_path, ".ros2")

ros2model/api/model_generator/system_generator.py

@@ -0,0 +1,42 @@
+#
+# Copyright 2023 Fraunhofer IPA
+#
+# 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.
+#
+
+from pathlib import Path
+
+from ros2model.core.generator.generator_core import GeneratorCore
+import typing as t
+from ament_index_python import get_package_share_directory
+
+Template_Folder = Path(__file__).parent.parent.parent.parent.resolve() / "templates"
+Template = Path(Template_Folder / "rossystem.rossystem.j2")
+
+Template_Folder_ROS = Path(get_package_share_directory("ros2model") + "/templates")
+Template_ROS = Path(Template_Folder_ROS / "rossystem.rossystem.j2")
+
+
+class SystemGenerator(GeneratorCore):
+    def __init__(self, template_path=None) -> None:
+        if template_path != None:
+            self.template_path = Path(template_path).resolve()
+        elif Template_ROS.is_file():
+            self.template_path = Template
+        elif Template.is_file():
+            self.template_path = Template
+        else:
+            raise FileNotFoundError(
+                f"Can't find template either from {Template.absolute().as_posix()} or {Template_ROS.absolute().as_posix()}"
+            )
+        super().__init__(self.template_path, ".rossystem")

ros2model/api/runtime_parser/rosmodel_runtime_parser.py

@@ -1,3 +1,5 @@
+import re
+import subprocess
 import ros2model.core.metamodels.metamodel_ros as ROSModel
 from ros2model.core.metamodels.metamodel_ros import set_full_name
 import ros2node.api as ROS2Node
@@ -7,7 +9,13 @@
 from pathlib import Path
 from rclpy.topic_endpoint_info import TopicEndpointInfo
 import rclpy
+from rcl_interfaces.msg import ParameterType
+
 from collections import namedtuple
+from ros2param.api import (
+    call_list_parameters,
+    call_describe_parameters,
+)
 
 Topic_BlackList = ["/parameter_events", "/rosout"]
 Service_BlackList = [
@@ -21,6 +29,24 @@
 
 TopicInfoVerbose = namedtuple("TopicInfoVerbose", ("full_name", "info"))
 
+ParameterReg = "^qos_overrides\..*$"
+
+
+def get_parameter_type_string(parameter_type):
+    mapping = {
+        ParameterType.PARAMETER_BOOL: "Boolean",
+        ParameterType.PARAMETER_INTEGER: "Integer",
+        ParameterType.PARAMETER_DOUBLE: "Double",
+        ParameterType.PARAMETER_STRING: "String",
+        ParameterType.PARAMETER_BYTE_ARRAY: "Array: Byte",
+        ParameterType.PARAMETER_BOOL_ARRAY: "Array: Boolean",
+        ParameterType.PARAMETER_INTEGER_ARRAY: "Array: Integer",
+        ParameterType.PARAMETER_DOUBLE_ARRAY: "Array: Double",
+        ParameterType.PARAMETER_STRING_ARRAY: "Array: String",
+        ParameterType.PARAMETER_NOT_SET: "Any",
+    }
+    return mapping[parameter_type]
+
 
 def get_interface_name(node_namespace: str, interface_name: str):
     """get interface relative name
@@ -71,14 +97,24 @@ def parse_interface(
     getattr(node, typee).append(interface)
 
 
+def set_name(namespace: str, full_name: str) -> str:
+    if namespace == "/":
+        name = full_name[1:]
+    else:
+        name = full_name[len(namespace) + 1 :]
+    return name
+
+
 def parse_interface_verbose(
     node: ROSModel.Node, typee: str, topic_endpoint_info: TopicInfoVerbose
 ):
     if topic_endpoint_info.full_name not in Topic_BlackList:
-        print(f"topic name again: {topic_endpoint_info.full_name}\n")
+        print(
+            f"topic name again: {topic_endpoint_info.full_name[len(node.name.namespace) :]}\n"
+        )
         interface = ROSModel.InterfaceTypeImpl(
             namespace=node.name.namespace,
-            name=topic_endpoint_info.full_name[len(node.name.namespace) :],
+            name=set_name(node.name.namespace, topic_endpoint_info.full_name),
             type=topic_endpoint_info.info.topic_type,
             qos=ROSModel.QualityOfService(
                 deadline=topic_endpoint_info.info.qos_profile.deadline.nanoseconds,
@@ -214,6 +250,53 @@ def get_action_clients(self, node, include_hidden=False):
         except AttributeError:
             pass
 
+    def get_parameters(self, node, include_hidden_nodes=False):
+        # for humble
+        new_proc = subprocess.Popen(["rosversion", "-d"], stdout=subprocess.PIPE)
+        version_str = new_proc.communicate()[0].decode("utf-8").rstrip("\n")
+
+        sorted_names = []
+        if version_str == "humble":
+            from rcl_interfaces.srv import ListParameters
+            from ros2service.api import get_service_names
+
+            service_names = get_service_names(
+                node=node, include_hidden_services=include_hidden_nodes
+            )
+            service_name = f"{self.name.full_name}/list_parameters"
+
+            if service_name in service_names:
+                client = node.create_client(ListParameters, service_name)
+                if client.service_is_ready():
+                    request = ListParameters.Request()
+                    future = client.call_async(request)
+                    rclpy.spin_until_future_complete(node, future, timeout_sec=1.0)
+                    if future.result() != None:
+                        response = future.result()
+                        sorted_names = sorted(response.result.names)
+
+        if version_str == "rolling":
+            response = call_list_parameters(node=node, node_name=self.name.full_name)
+            if response.result() != None:
+                param_names = response.result().result.names
+                sorted_names = sorted(param_names)
+
+        regex_filter = re.compile(ParameterReg)
+        sorted_names = [name for name in sorted_names if not regex_filter.match(name)]
+        des_resp = call_describe_parameters(
+            node=node,
+            node_name=self.name.full_name,
+            parameter_names=sorted_names,
+        )
+        for descriptor in des_resp.descriptors:
+            self.parameter.append(
+                ROSModel.Parameter(
+                    name=descriptor.name,
+                    namespace=self.name.namespace,
+                    type=get_parameter_type_string(descriptor.type),
+                )
+            )
+
 
 def get_node_graph_names():
     args = NodeArgs(node_name="get_all_nodes")
@@ -243,6 +326,7 @@ def parse(
         parsed_node.get_service_clients(node, include_hidden=include_hidden_interfaces)
         parsed_node.get_action_clients(node, include_hidden=include_hidden_interfaces)
         parsed_node.get_action_servers(node, include_hidden=include_hidden_interfaces)
+        parsed_node.get_parameters(node)
         return parsed_node
 
 

ros2model/api/runtime_parser/rossystem_runtime_parser.py

@@ -0,0 +1,47 @@
+from ros2model.api.runtime_parser.rosmodel_runtime_parser import (
+    save_to_file,
+    get_node_graph_names,
+    parse,
+)
+import ros2model.core.metamodels.metamodel_ros as ROSModel
+from pathlib import Path
+
+
+class RuntimeRossystem(ROSModel.Rossystem):
+    pass
+
+    def __init__(self, *, system: ROSModel.Rossystem, **data):
+        super().__init__(name=system.name, **data)
+
+    def get_nodes(self):
+        nodes = get_node_graph_names()
+        for n in nodes:
+            print("node name: ", n.namespace, n.name, n.full_name)
+            parsed_node = parse(
+                ROSModel.GraphName(
+                    name=n.name, namespace=n.namespace, full_name=n.full_name
+                )
+            )
+            self.nodes.append(parsed_node)
+
+
+def name_system_file(grapg_name: ROSModel.GraphName):
+    n = grapg_name.name.replace("/", "_")
+    file_name = f"{grapg_name.namespace}__{n}"
+    return file_name
+
+
+def main(result_file_path):
+    system_name = Path(result_file_path).name
+    system = RuntimeRossystem(system=ROSModel.Rossystem(name=system_name))
+    system.get_nodes()
+    save_to_file(
+        result_path,
+        system_name,
+        system,
+    )
+
+
+if __name__ == "__main__":
+    result_path = "./test.rossystem"
+    main(result_path)

ros2model/core/metamodels/metamodel_ros.py

@@ -206,8 +206,7 @@ class ParameterType(str, Enum):
 
 
 class Parameter(InterfaceTypeImpl):
-    type: str
-    value: str
+    value: Optional[str] = None
 
 
 class Node(BaseModel):
@@ -272,5 +271,41 @@ class AmentPackage(Package):
     pass
 
 
+class Process(BaseModel):
+    name: str
+    threads: Optional[int]
+    components: List[str] = Field(default_factory=list)
+
+
+class RosInterface(BaseModel):
+    name: str
+    reference: str
+
+
+class RosParameter(BaseModel):
+    name: str
+    value: str
+    reffrom: str
+
+
+class RosNode(BaseModel):
+    name: str
+    reffrom: str
+    rosinterfaces: List[RosInterface] = Field(default_factory=list)
+    rosparameters: List[RosParameter] = Field(default_factory=list)
+
+
+class Connection(BaseModel):
+    reffrom: str
+    to: str
+
+
+class Rossystem(BaseModel):
+    name: str
+    proccesses: List[Process] = Field(default_factory=list)
+    nodes: List[Node] = Field(default_factory=list)
+    parameters: List[Parameter] = Field(default_factory=list)
+
+
 # serialized = dump(builtin_interfaces.model_dump()).strip()
 # print(serialized)

ros2model/verb/runtime_system.py

@@ -0,0 +1,56 @@
+from pathlib import Path
+
+from ros2cli.node.strategy import add_arguments
+
+from ros2model.verb import VerbExtension
+from ros2model.api.model_generator.system_generator import SystemGenerator
+import ros2model.api.runtime_parser.rossystem_runtime_parser as RuntimeParser
+import ros2model.core.metamodels.metamodel_ros as ROSModel
+
+
+class RuntimeVerb(VerbExtension):
+    """Create .rossystem for a runtime system"""
+
+    def add_arguments(self, parser, cli_name):
+        add_arguments(parser)
+
+        parser.add_argument(
+            "-o",
+            "--output_file",
+            default=".",
+            required=False,
+            help="The system model file path.",
+        )
+
+        parser.add_argument(
+            "--include_hidden_nodes",
+            action="store_true",
+            required=False,
+            help="Consider hidden nodes.",
+        )
+
+        parser.add_argument(
+            "--include_hidden_interfaces",
+            action="store_true",
+            required=False,
+            help="Consider hidden topics, services or actions.",
+        )
+
+    def main(self, *, args):
+        output_file = Path(args.output_file)
+        system_name = output_file.name
+        if output_file.is_absolute() != True:
+            output_file = output_file.resolve()
+
+        generator = SystemGenerator()
+
+        system = RuntimeParser.RuntimeRossystem(
+            system=ROSModel.Rossystem(name=system_name)
+        )
+        system.get_nodes()
+
+        generator.generate_a_file(
+            model=system,
+            output_dir=output_file.parent,
+            filename=system_name,
+        )

setup.py

@@ -27,6 +27,10 @@
     data_files=[
         ("share/" + package_name, ["package.xml"]),
         ("share/ament_index/resource_index/packages", ["resource/" + package_name]),
+        (
+            "share/" + package_name + "/templates",
+            ["templates/component.ros2.j2", "templates/rossystem.rossystem.j2"],
+        ),
     ],
     install_requires=[
         "jinja2",
@@ -54,6 +58,7 @@
         ],
         "ros2model.verb": [
             "node = ros2model.verb.runtime_node:RuntimeNodeVerb",
+            "system = ros2model.verb.runtime_system:RuntimeVerb",
         ],
     },
 )