Fixes in services in database_server

Author: fcladera

mocha_core/mocha_core/database_server.py

@@ -67,86 +67,82 @@ def __init__(self, robot_configs, topic_configs, robot_name, node=None):
 
         self.msg_types = du.msg_types(self.robot_configs, self.topic_configs)
 
-    def add_update_db_service_cb(self, req):
+    def add_update_db_service_cb(self, request, response):
         try:
-            if not isinstance(req.topic_id, int) or req.topic_id is None:
+            if not isinstance(request.topic_id, int) or request.topic_id is None:
                 self.logger.error("topic_id empty")
-                return mocha_core.srv.AddUpdateDB.Response()
-            if len(req.msg_content) == 0:
+                return response
+            if len(request.msg_content) == 0:
                 self.logger.error("msg_content empty")
-                return mocha_core.srv.AddUpdateDB.Response()
-            if req.topic_id >= len(self.topic_list):  # Changed > to >= for proper bounds check
-                self.logger.error(f"topic_id {req.topic_id} not in topic_list (length: {len(self.topic_list)})")
-                return mocha_core.srv.AddUpdateDB.Response()
+                return response
+            if request.topic_id >= len(self.topic_list):  # Changed > to >= for proper bounds check
+                self.logger.error(f"topic_id {request.topic_id} not in topic_list (length: {len(self.topic_list)})")
+                return response
 
-            topic = self.topic_list[req.topic_id]
+            topic = self.topic_list[request.topic_id]
             priority = du.get_priority_number(topic["msg_priority"])
-            ts = req.timestamp
+            ts = request.timestamp
             # ROS2 builtin_interfaces/Time uses 'sec' and 'nanosec' fields
             ts = rclpy.time.Time(seconds=ts.sec, nanoseconds=ts.nanosec)
 
             # Convert array to bytes if needed (ROS2 service messages use array)
-            msg_data = req.msg_content
+            msg_data = request.msg_content
             if hasattr(msg_data, 'tobytes'):
                 msg_data = msg_data.tobytes()
             elif isinstance(msg_data, (list, tuple)):
                 msg_data = bytes(msg_data)
 
             dbm = database.DBMessage(self.robot_number,
-                                     req.topic_id,
-                                     dtype=self.msg_types[self.robot_number][req.topic_id]["dtype"],
+                                     request.topic_id,
+                                     dtype=self.msg_types[self.robot_number][request.topic_id]["dtype"],
                                      priority=priority,
                                      ts=ts,
                                      data=msg_data)
 
             header = self.dbl.add_modify_data(dbm)
-            response = mocha_core.srv.AddUpdateDB.Response()
             response.new_header = header
             return response
         except Exception as e:
             self.logger.error(f"Exception in add_update_db_service_cb: {e}")
-            return mocha_core.srv.AddUpdateDB.Response()
+            return response
 
-    def get_data_hash_db_service_cb(self, req):
+    def get_data_hash_db_service_cb(self, request, response):
         try:
-            if req.msg_header is None or len(req.msg_header) == 0:
+            if request.msg_header is None or len(request.msg_header) == 0:
                 self.logger.error("msg_header empty")
-                return mocha_core.srv.GetDataHeaderDB.Response()
+                return response
 
             # Convert array to bytes if needed (ROS2 service messages use array)
-            header_data = req.msg_header
+            header_data = request.msg_header
             if hasattr(header_data, 'tobytes'):
                 header_data = header_data.tobytes()
             elif isinstance(header_data, (list, tuple)):
                 header_data = bytes(header_data)
 
             dbm = self.dbl.find_header(header_data)
 
-            response = mocha_core.srv.GetDataHeaderDB.Response()
             response.robot_id = dbm.robot_id
             response.topic_id = dbm.topic_id
             response.timestamp = dbm.ts
             response.msg_content = dbm.data
             return response
         except Exception as e:
             self.logger.error(f"Exception in get_data_hash_db_service_cb: {e}")
-            return mocha_core.srv.GetDataHeaderDB.Response()
+            return response
 
-    def select_db_service_cb(self, req):
+    def select_db_service_cb(self, request, response):
         try:
             # TODO Implement filtering
-            response = mocha_core.srv.SelectDB.Response()
 
             # Note: robot_id and topic_id are uint8 in ROS2, so they can't be None
             # We can add range validation if needed, but for now just proceed
 
-            list_headers = self.dbl.get_header_list(req.robot_id)
+            list_headers = self.dbl.get_header_list(request.robot_id)
 
             response.headers = du.serialize_headers(list_headers)
             return response
         except Exception as e:
             self.logger.error(f"Exception in select_db_service_cb: {e}")
-            response = mocha_core.srv.SelectDB.Response()
             response.headers = b''
             return response
 

mocha_core/mocha_core/integrate_database.py

@@ -163,8 +163,9 @@ def shutdown(self, reason):
             channel.comm_node.terminate()
             self.all_channels.remove(channel)
         self.logger.warning(f"{self.this_robot} - Integrate - " + "Killed Channels")
-        self.DBServer.shutdown()
-        self.logger.warning(f"{self.this_robot} - Integrate - " + "Killed DB")
+        if hasattr(self, 'DBServer') and self.DBServer is not None:
+            self.DBServer.shutdown()
+            self.logger.warning(f"{self.this_robot} - Integrate - " + "Killed DB")
         if rclpy.ok():
             rclpy.shutdown()
         self.logger.warning(f"{self.this_robot} - Integrate - " + "Shutting down")
@@ -180,6 +181,9 @@ def rssi_cb(self, data, comm_node):
                 traceback.print_exception(*sys.exc_info())
 
 
-if __name__ == "__main__":
+def main():
     # Start the node
     IntegrateDatabase()
+
+if __name__ == "__main__":
+    main()

mocha_core/test/test_database_server.py

@@ -74,7 +74,8 @@ def test_add_retrieve_single_msg(self):
             req.topic_id = tid
             req.timestamp = rclpy.clock.Clock().now().to_msg()
             req.msg_content = serialized_msg
-            answ = self.dbs.add_update_db_service_cb(req)
+            response = mocha_core.srv.AddUpdateDB.Response()
+            answ = self.dbs.add_update_db_service_cb(req, response)
             answ_header = answ.new_header
         except Exception as exc:
             print("Service did not process request: " + str(exc))
@@ -87,7 +88,8 @@ def test_add_retrieve_single_msg(self):
             # Create request and call service method directly
             req = mocha_core.srv.GetDataHeaderDB.Request()
             req.msg_header = answ_header
-            answ = self.dbs.get_data_hash_db_service_cb(req)
+            response = mocha_core.srv.GetDataHeaderDB.Response()
+            answ = self.dbs.get_data_hash_db_service_cb(req, response)
         except Exception as exc:
             print("Service did not process request: " + str(exc))
             self.assertTrue(False)
@@ -125,7 +127,8 @@ def test_add_select_robot(self):
                 req.topic_id = tid
                 req.timestamp = rclpy.clock.Clock().now().to_msg()
                 req.msg_content = serialized_msg
-                answ = self.dbs.add_update_db_service_cb(req)
+                response = mocha_core.srv.AddUpdateDB.Response()
+                answ = self.dbs.add_update_db_service_cb(req, response)
                 answ_header = answ.new_header
             except Exception as exc:
                 print("Service did not process request: " + str(exc))
@@ -140,7 +143,8 @@ def test_add_select_robot(self):
             req.robot_id = robot_id
             req.topic_id = 0  # Changed from None to 0 since it's uint8
             req.ts_limit = rclpy.clock.Clock().now().to_msg()
-            answ = self.dbs.select_db_service_cb(req)
+            response = mocha_core.srv.SelectDB.Response()
+            answ = self.dbs.select_db_service_cb(req, response)
         except Exception as exc:
             print("Service did not process request: " + str(exc))
             self.assertTrue(False)
@@ -185,7 +189,8 @@ def recorder_thread():
                     req.topic_id = tid
                     req.timestamp = timestamp
                     req.msg_content = serialized_msg
-                    _ = self.dbs.add_update_db_service_cb(req)
+                    response = mocha_core.srv.AddUpdateDB.Response()
+                    _ = self.dbs.add_update_db_service_cb(req, response)
                 except Exception as exc:
                     print(f"Service did not process request: {exc}")
 
@@ -209,7 +214,8 @@ def recorder_thread():
             req.robot_id = robot_id
             req.topic_id = 0  # Changed from None to 0 since it's uint8
             req.ts_limit = rclpy.clock.Clock().now().to_msg()
-            answ = self.dbs.select_db_service_cb(req)
+            response = mocha_core.srv.SelectDB.Response()
+            answ = self.dbs.select_db_service_cb(req, response)
         except Exception as exc:
             print("Service did not process request: " + str(exc))
             self.assertTrue(False)