feat: New parameter and cli arguments handling

rosbridge_library/src/rosbridge_library/capabilities/advertise.py

@@ -99,6 +99,8 @@ class Advertise(Capability):
     advertise_msg_fields = ((True, "topic", str), (True, "type", str))
     unadvertise_msg_fields = ((True, "topic", str),)
 
+    parameter_names = ("topics_glob",)
+
     topics_glob: list[str] | None = None
 
     def __init__(self, protocol: Protocol) -> None:
@@ -111,9 +113,6 @@ def __init__(self, protocol: Protocol) -> None:
 
         self._registrations: dict[str, Registration] = {}
 
-        if protocol.parameters and "unregister_timeout" in protocol.parameters:
-            manager.unregister_timeout = protocol.parameters.get("unregister_timeout")
-
     def advertise(self, message: dict[str, Any]) -> None:
         # Pull out the ID
         aid = message.get("id")
@@ -124,10 +123,10 @@ def advertise(self, message: dict[str, Any]) -> None:
         latch: bool = message.get("latch", False)
         queue_size: int = message.get("queue_size", 100)
 
-        if Advertise.topics_glob is not None and Advertise.topics_glob:
+        if self.topics_glob:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
-            for glob in Advertise.topics_glob:
+            for glob in self.topics_glob:
                 if fnmatch.fnmatch(topic, glob):
                     self.protocol.log(
                         "debug",
@@ -159,10 +158,10 @@ def unadvertise(self, message: dict[str, Any]) -> None:
         self.basic_type_check(message, self.unadvertise_msg_fields)
         topic: str = message["topic"]
 
-        if Advertise.topics_glob is not None and Advertise.topics_glob:
+        if self.topics_glob:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
-            for glob in Advertise.topics_glob:
+            for glob in self.topics_glob:
                 if fnmatch.fnmatch(topic, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/advertise_action.py

@@ -218,10 +218,12 @@ def graceful_shutdown(self) -> None:
 
 
 class AdvertiseAction(Capability):
-    actions_glob = None
-
     advertise_action_msg_fields = ((True, "action", str), (True, "type", str))
 
+    parameter_names = ("actions_glob",)
+
+    actions_glob: list[str] | None = None
+
     def __init__(self, protocol: Protocol) -> None:
         # Call superclass constructor
         Capability.__init__(self, protocol)
@@ -236,13 +238,13 @@ def advertise_action(self, message: dict) -> None:
         # parse the incoming message
         action_name: str = message["action"]
 
-        if AdvertiseAction.actions_glob is not None and AdvertiseAction.actions_glob:
+        if self.actions_glob:
             self.protocol.log(
                 "debug",
                 "Action security glob enabled, checking action: " + action_name,
             )
             match = False
-            for glob in AdvertiseAction.actions_glob:
+            for glob in self.actions_glob:
                 if fnmatch.fnmatch(action_name, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/advertise_service.py

@@ -104,10 +104,12 @@ def graceful_shutdown(self) -> None:
 
 
 class AdvertiseService(Capability):
-    services_glob: list[str] | None = None
-
     advertise_service_msg_fields = ((True, "service", str), (True, "type", str))
 
+    parameter_names = ("services_glob",)
+
+    services_glob: list[str] | None = None
+
     def __init__(self, protocol: Protocol) -> None:
         # Call superclass constructor
         Capability.__init__(self, protocol)
@@ -122,13 +124,13 @@ def advertise_service(self, message: dict[str, Any]) -> None:
         # parse the incoming message
         service_name: str = message["service"]
 
-        if AdvertiseService.services_glob is not None and AdvertiseService.services_glob:
+        if self.services_glob:
             self.protocol.log(
                 "debug",
                 "Service security glob enabled, checking service: " + service_name,
             )
             match = False
-            for glob in AdvertiseService.services_glob:
+            for glob in self.services_glob:
                 if fnmatch.fnmatch(service_name, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/call_service.py

@@ -51,25 +51,22 @@ class CallService(Capability):
         (False, "compression", str),
     )
 
+    parameter_names = (
+        "services_glob",
+        "default_call_service_timeout",
+        "call_services_in_new_thread",
+    )
+
     services_glob: list[str] | None = None
+    default_timeout: float = 5.0
+    call_services_in_new_thread: bool = True
 
     def __init__(self, protocol: Protocol) -> None:
         # Call superclass constructor
         Capability.__init__(self, protocol)
 
-        self.default_timeout = (
-            protocol.node_handle.get_parameter("default_call_service_timeout")
-            .get_parameter_value()
-            .double_value
-        )
-
         # Register the operations that this capability provides
-        call_services_in_new_thread = (
-            protocol.node_handle.get_parameter("call_services_in_new_thread")
-            .get_parameter_value()
-            .bool_value
-        )
-        if call_services_in_new_thread:
+        if self.call_services_in_new_thread:
             # Calls the service in a separate thread so multiple services can be processed simultaneously.
             protocol.node_handle.get_logger().info("Calling services in new thread")
             protocol.register_operation(
@@ -94,12 +91,12 @@ def call_service(self, message: dict[str, Any]) -> None:
         args: list | dict[str, Any] = message.get("args", [])
         timeout: float = message.get("timeout", self.default_timeout)
 
-        if CallService.services_glob is not None and CallService.services_glob:
+        if self.services_glob:
             self.protocol.log(
                 "debug", "Service security glob enabled, checking service: " + service
             )
             match = False
-            for glob in CallService.services_glob:
+            for glob in self.services_glob:
                 if fnmatch.fnmatch(service, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/defragmentation.py

@@ -26,16 +26,15 @@ class Defragment(Capability):
     # }
     lists: ClassVar[dict[str, dict]] = {}
 
-    fragment_timeout = 600
     opcode = "fragment"
 
+    parameter_names = ("fragment_timeout",)
+
+    fragment_timeout = 600
+
     def __init__(self, protocol: Protocol) -> None:
         Capability.__init__(self, protocol)
 
-        # populate parameters
-        if self.protocol.parameters is not None:
-            self.fragment_timeout = self.protocol.parameters["fragment_timeout"]
-
         protocol.register_operation(self.opcode, self.defragment)
 
         self.received_fragments = Defragment.lists

rosbridge_library/src/rosbridge_library/capabilities/publish.py

@@ -46,6 +46,8 @@
 class Publish(Capability):
     publish_msg_fields = ((True, "topic", str),)
 
+    parameter_names = ("topics_glob",)
+
     topics_glob: list[str] | None = None
 
     def __init__(self, protocol: Protocol) -> None:
@@ -58,20 +60,17 @@ def __init__(self, protocol: Protocol) -> None:
         # Save the topics that are published on for the purposes of unregistering
         self._published: dict[str, bool] = {}
 
-        if protocol.parameters and "unregister_timeout" in protocol.parameters:
-            manager.unregister_timeout = protocol.parameters.get("unregister_timeout")
-
     def publish(self, message: dict[str, Any]) -> None:
         # Do basic type checking
         self.basic_type_check(message, self.publish_msg_fields)
         topic: str = message["topic"]
         latch: bool = message.get("latch", False)
         queue_size: int = message.get("queue_size", 100)
 
-        if Publish.topics_glob is not None and Publish.topics_glob:
+        if self.topics_glob:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
-            for glob in Publish.topics_glob:
+            for glob in self.topics_glob:
                 if fnmatch.fnmatch(topic, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/send_action_goal.py

@@ -59,22 +59,21 @@ class SendActionGoal(Capability):
     )
     cancel_action_goal_msg_fields = ((True, "action", str),)
 
-    actions_glob = None
     client_handler_list: dict[str, ActionClientHandler]
 
+    parameter_names = ("actions_glob", "send_action_goals_in_new_thread")
+
+    actions_glob: list[str] | None = None
+    send_action_goals_in_new_thread: bool = False
+
     def __init__(self, protocol: Protocol) -> None:
         # Call superclass constructor
         Capability.__init__(self, protocol)
 
         self.client_handler_list = {}
 
         # Register the operations that this capability provides
-        send_action_goals_in_new_thread = (
-            protocol.node_handle.get_parameter("send_action_goals_in_new_thread")
-            .get_parameter_value()
-            .bool_value
-        )
-        if send_action_goals_in_new_thread:
+        if self.send_action_goals_in_new_thread:
             # Sends the action goal in a separate thread so multiple actions can be processed simultaneously.
             protocol.node_handle.get_logger().info("Sending action goals in new thread")
             protocol.register_operation(
@@ -106,10 +105,10 @@ def send_action_goal(self, message: dict) -> None:
         compression: str = message.get("compression", "none")
         args: list | dict[str, Any] = message.get("args", [])
 
-        if SendActionGoal.actions_glob is not None and SendActionGoal.actions_glob:
+        if self.actions_glob:
             self.protocol.log("debug", f"Action security glob enabled, checking action: {action}")
             match = False
-            for glob in SendActionGoal.actions_glob:
+            for glob in self.actions_glob:
                 if fnmatch.fnmatch(action, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/subscribe.py

@@ -248,6 +248,8 @@ class Subscribe(Capability):
     )
     unsubscribe_msg_fields = ((True, "topic", str),)
 
+    parameter_names = ("topics_glob",)
+
     topics_glob: list[str] | None = None
 
     def __init__(self, protocol: Protocol) -> None:
@@ -270,10 +272,10 @@ def subscribe(self, msg: dict[str, Any]) -> None:
         # Make the subscription
         topic: str = msg["topic"]
 
-        if Subscribe.topics_glob is not None and Subscribe.topics_glob:
+        if self.topics_glob is not None and self.topics_glob:
             self.protocol.log("debug", "Topic security glob enabled, checking topic: " + topic)
             match = False
-            for glob in Subscribe.topics_glob:
+            for glob in self.topics_glob:
                 if fnmatch.fnmatch(topic, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/unadvertise_action.py

@@ -42,6 +42,10 @@
 
 
 class UnadvertiseAction(Capability):
+    unadvertise_action_msg_fields = ((True, "action", str),)
+
+    parameter_names = ("actions_glob",)
+
     actions_glob: list[str] | None = None
 
     def __init__(self, protocol: Protocol) -> None:
@@ -52,16 +56,18 @@ def __init__(self, protocol: Protocol) -> None:
         protocol.register_operation("unadvertise_action", self.unadvertise_action)
 
     def unadvertise_action(self, message: dict[str, Any]) -> None:
+        self.basic_type_check(message, self.unadvertise_action_msg_fields)
+
         # parse the message
         action_name: str = message["action"]
 
-        if UnadvertiseAction.actions_glob is not None and UnadvertiseAction.actions_glob:
+        if self.actions_glob:
             self.protocol.log(
                 "debug",
                 f"Action security glob enabled, checking action: {action_name}",
             )
             match = False
-            for glob in UnadvertiseAction.actions_glob:
+            for glob in self.actions_glob:
                 if fnmatch.fnmatch(action_name, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capabilities/unadvertise_service.py

@@ -10,7 +10,9 @@
 
 
 class UnadvertiseService(Capability):
-    # unadvertise_service_msg_fields = [(True, "service", (str, unicode))]
+    unadvertise_service_msg_fields = ((True, "service", str),)
+
+    parameter_names = ("services_glob",)
 
     services_glob: list[str] | None = None
 
@@ -22,16 +24,18 @@ def __init__(self, protocol: Protocol) -> None:
         protocol.register_operation("unadvertise_service", self.unadvertise_service)
 
     def unadvertise_service(self, message: dict[str, Any]) -> None:
+        self.basic_type_check(message, self.unadvertise_service_msg_fields)
+
         # parse the message
         service_name: str = message["service"]
 
-        if UnadvertiseService.services_glob is not None and UnadvertiseService.services_glob:
+        if self.services_glob:
             self.protocol.log(
                 "debug",
                 "Service security glob enabled, checking service: " + service_name,
             )
             match = False
-            for glob in UnadvertiseService.services_glob:
+            for glob in self.services_glob:
                 if fnmatch.fnmatch(service_name, glob):
                     self.protocol.log(
                         "debug",

rosbridge_library/src/rosbridge_library/capability.py

@@ -58,6 +58,8 @@ class Capability:
     Protocol.send() is available to send messages back to the client.
     """
 
+    parameter_names: tuple[str, ...] | None = None
+
     def __init__(self, protocol: Protocol) -> None:
         """
         Abstract class constructor.
@@ -68,6 +70,11 @@ def __init__(self, protocol: Protocol) -> None:
         """
         self.protocol = protocol
 
+        if self.parameter_names and self.protocol.parameters:
+            for param_name, param_value in self.protocol.parameters.items():
+                if param_name in self.parameter_names:
+                    setattr(self, param_name, param_value)
+
     def handle_message(self, message: dict[str, Any]) -> None:
         """
         Handle an incoming message.