Simplifying API

Signed-off-by: Aaron Chong <aaronchongth@gmail.com>

free_fleet_adapter/free_fleet_adapter/fleet_adapter.py

@@ -172,7 +172,7 @@ def update_loop():
             # Update all the robots in parallel using a thread pool
             update_jobs = []
             for robot in robots.values():
-                update_jobs.append(update_robot(robot, tf_buffer))
+                update_jobs.append(update_robot(robot))
 
             asyncio.get_event_loop().run_until_complete(
                 asyncio.wait(update_jobs)
@@ -211,7 +211,7 @@ def run_in_parallel(*args, **kwargs):
 
 
 @parallel
-def update_robot(robot: Nav2RobotAdapter, tf_buffer: Buffer):
+def update_robot(robot: Nav2RobotAdapter):
     transform = robot.tf_handler.get_transform()
     if transform is None:
         robot.node.get_logger().info(

free_fleet_adapter/free_fleet_adapter/robot_adapter.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 Open Source Robotics Foundation, Inc.
+#
+# 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 abc import ABC, abstractmethod
+from typing import Annotated
+
+import rmf_adapter.easy_full_control as rmf_easy
+from rmf_adapter.robot_update_handle import ActivityIdentifier
+
+
+class RobotAdapter(ABC):
+
+    '''
+    This method returns the battery state of charge as a float, with value
+    between 0 and 1.0.
+    '''
+    @abstractmethod
+    def battery_soc(self) -> float:
+        ...
+
+    '''
+    This method returns the last known 2D position in meters and orientation
+    (yaw) of the robot in radians as a list of 3 floats, in the form of
+    [x, y, yaw]. If the last known position of the robot is not available,
+    returns None. 
+    '''
+    @abstractmethod
+    def pose(self) -> Annotated[list[float], 3] | None:
+        ...
+
+    '''
+    This method is called to update RMF with the latest robot state.
+    '''
+    @abstractmethod
+    def update(self, state: rmf_easy.RobotState):
+        ...
+
+    '''
+    This method is called to send a navigation command to the robot.
+    '''
+    @abstractmethod
+    def navigate(
+        self,
+        destination: rmf_easy.Destination,
+        execution: rmf_easy.CommandExecution
+    ):
+        ...
+
+    '''
+    This method is called to stop the execution/continuation of the provided
+    activity.
+    '''
+    @abstractmethod
+    def stop(self, activity):
+        ...
+
+    '''
+    This method is called to send a custom action command to the robot.
+    '''
+    @abstractmethod
+    def execute_action(
+        self,
+        category: str,
+        description: dict,
+        execution: ActivityIdentifier
+    ):
+        ...