feat(RPS-999): activate motion groups seperatly (#11)

README.md

@@ -33,33 +33,33 @@ import asyncio
 
 
 async def main():
-    nova = Nova()
-    cell = nova.cell()
-    controller = await cell.controller("ur")
+  nova = Nova()
+  cell = nova.cell()
+  controller = await cell.controller("ur")
 
-    # Define a home position
-    home_joints = (0, -pi / 4, -pi / 4, -pi / 4, pi / 4, 0)
+  # Define a home position
+  home_joints = (0, -pi / 4, -pi / 4, -pi / 4, pi / 4, 0)
 
-    # Connect to the controller and activate motion groups
-    async with controller:
-        motion_group = controller.get_motion_group()
+  # Connect to the controller and activate motion groups
+  async with controller:
+    motion_group = controller.motion_group()
 
-        # Get current TCP pose and offset it slightly along the x-axis
-        current_pose = await motion_group.tcp_pose("Flange")
-        target_pose = current_pose @ Pose((100, 50, 0, 0, 0, 0))
+    # Get current TCP pose and offset it slightly along the x-axis
+    current_pose = await motion_group.tcp_pose("Flange")
+    target_pose = current_pose @ Pose((100, 50, 0, 0, 0, 0))
 
-        actions = [
-            jnt(home_joints),
-            ptp(target_pose),
-            ptp(target_pose @ (200, 0, 0, 0, 0, 0)),
-            jnt(home_joints),
-        ]
+    actions = [
+      jnt(home_joints),
+      ptp(target_pose),
+      ptp(target_pose @ (200, 0, 0, 0, 0, 0)),
+      jnt(home_joints),
+    ]
 
-        await motion_group.run(actions, tcp="Flange")
+    await motion_group.run(actions, tcp="Flange")
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+  asyncio.run(main())
 ```
 
 Have a look at the [examples](https://github.com/wandelbotsgmbh/wandelbots-nova/tree/main/examples) directory to see how to use the library.

examples/01_basic.py

@@ -9,7 +9,7 @@ async def main():
     controller = await cell.controller("ur")
 
     async with controller:
-        motion_group = controller.get_motion_group()
+        motion_group = controller.motion_group()
 
         # Current motion group state
         state = await motion_group.get_state("Flange")

examples/02_plan_and_execute.py

@@ -15,7 +15,7 @@ async def main():
 
     # Connect to the controller and activate motion groups
     async with controller:
-        motion_group = controller.get_motion_group()
+        motion_group = controller.motion_group()
 
         # Get current TCP pose and offset it slightly along the x-axis
         current_pose = await motion_group.tcp_pose("Flange")

examples/03_move_and_set_ios.py

@@ -16,7 +16,7 @@ async def main():
 
     # Connect to the controller and activate motion groups
     async with controller:
-        motion_group = controller.get_motion_group()
+        motion_group = controller.motion_group()
 
         # Get current TCP pose and offset it slightly along the x-axis
         current_pose = await motion_group.tcp_pose("Flange")

examples/04_move_multiple_robots.py

@@ -7,7 +7,7 @@ async def move_robot(controller: Controller):
     home_joints = (0, -pi / 4, -pi / 4, -pi / 4, pi / 4, 0)
 
     async with controller:
-        motion_group = controller.get_motion_group()
+        motion_group = controller.motion_group()
 
         current_pose = await motion_group.tcp_pose("Flange")
         target_pose = current_pose @ (100, 0, 0, 0, 0, 0)

examples/05_selection_motion_group_activation.py

@@ -0,0 +1,64 @@
+"""
+This example demonstrates how to activate specific motion groups for two robot controllers
+and execute simultaneous movements for both robots.
+
+The robots used in this example are:
+- A Universal Robots (UR) controller
+- A KUKA controller
+
+Each robot moves between a predefined home pose and a target pose sequentially.
+"""
+
+from math import pi
+from nova import Nova, ptp, MotionGroup, Pose
+import asyncio
+
+
+async def move_robot(motion_group: MotionGroup):
+    home_pose = Pose((200, 200, 600, 0, pi, 0))
+    target_pose = home_pose @ (100, 0, 0, 0, 0, 0)
+    actions = [
+        ptp(home_pose),
+        ptp(target_pose),
+        ptp(target_pose @ (0, 0, 100, 0, 0, 0)),
+        ptp(target_pose @ (0, 100, 0, 0, 0, 0)),
+        ptp(home_pose),
+    ]
+
+    await motion_group.run(actions, tcp="Flange")
+
+
+async def main():
+    nova = Nova()
+    cell = nova.cell()
+    ur = await cell.controller("ur")
+    kuka = await cell.controller("kuka")
+
+    flange_state = await ur[0].get_state("Flange")
+    print(flange_state)
+
+    # activate all motion groups
+    async with ur:
+        await move_robot(ur.motion_group(0))
+
+    # activate motion group 0
+    async with ur.motion_group(0) as mg_0:
+        await move_robot(mg_0)
+
+    # activate motion group 0
+    async with ur[0] as mg_0:
+        await move_robot(mg_0)
+
+    # activate motion group 0 from two different controllers
+    async with ur[0] as ur_0_mg, kuka[0] as kuka_0_mg:
+        await asyncio.gather(move_robot(ur_0_mg), move_robot(kuka_0_mg))
+
+    # activate motion group 0 from two different controllers
+    mg_0 = ur.motion_group(0)
+    mg_1 = kuka.motion_group(0)
+    async with mg_0, mg_1:
+        await asyncio.gather(move_robot(mg_0), move_robot(mg_1))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

nova/core/controller.py

@@ -13,7 +13,7 @@ def __init__(self, *, api_gateway: ApiGateway, cell: str, controller_host: str):
         self._motion_group_api = api_gateway.motion_group_api
         self._cell = cell
         self._controller_host = controller_host
-        self._motion_groups: dict[str, MotionGroup] = {}
+        self._activated_motion_groups: list[str] = []
 
     @final
     async def __aenter__(self):
@@ -25,22 +25,21 @@ async def __aenter__(self):
         motion_groups = activate_all_motion_groups_response.instances
         for mg in motion_groups:
             logger.info(f"Found motion group {mg.motion_group}")
-            motion_group = MotionGroup(
-                api_gateway=self._api_gateway, cell=self._cell, motion_group_id=mg.motion_group
-            )
-            self._motion_groups[motion_group.motion_group_id] = motion_group
+            self._activated_motion_groups.append(mg.motion_group)
         return self
 
     @final
     async def __aexit__(self, exc_type, exc_val, exc_tb):
-        for motion_group_id in self._motion_groups.keys():
+        for motion_group_id in self._activated_motion_groups:
             logger.info(f"Deactivating motion group {motion_group_id}")
             await self._motion_group_api.deactivate_motion_group(self._cell, motion_group_id)
 
-        await self._api_gateway.close()
-
-    def get_motion_groups(self) -> dict[str, MotionGroup]:
-        return self._motion_groups
+    def __getitem__(self, motion_group_id: int) -> MotionGroup:
+        return self.motion_group(motion_group_id)
 
-    def get_motion_group(self, motion_group_id: str = "0") -> MotionGroup | None:
-        return self._motion_groups.get(f"{motion_group_id}@{self._controller_host}", None)
+    def motion_group(self, motion_group_id: int = 0) -> MotionGroup | None:
+        return MotionGroup(
+            api_gateway=self._api_gateway,
+            cell=self._cell,
+            motion_group_id=f"{motion_group_id}@{self._controller_host}",
+        )

nova/core/motion_group.py

@@ -18,12 +18,26 @@
 
 
 class MotionGroup:
-    def __init__(self, api_gateway: ApiGateway, cell: str, motion_group_id: str):
+    def __init__(
+        self, api_gateway: ApiGateway, cell: str, motion_group_id: str, is_activated: bool = False
+    ):
         self._api_gateway = api_gateway
         self._motion_api_client = api_gateway.motion_api
         self._cell = cell
         self._motion_group_id = motion_group_id
         self._current_motion: str | None = None
+        self.is_activated = is_activated
+
+    async def __aenter__(self):
+        await self._api_gateway.motion_group_api.activate_motion_group(
+            cell=self._cell, motion_group=self._motion_group_id
+        )
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self._api_gateway.motion_group_api.deactivate_motion_group(
+            cell=self._cell, motion_group=self._motion_group_id
+        )
 
     @property
     def motion_group_id(self) -> str:

tests/core/test_controller.py

@@ -1,4 +1,3 @@
-from nova.core.controller import Controller
 import pytest
 
 import wandelbots_api_client
@@ -11,13 +10,3 @@ async def test_instance(nova_api_client):
     controllers = await controller_api.list_controllers(cell="cell")
     print(controllers)
     assert False
-
-
-@pytest.mark.asyncio
-@pytest.mark.skip
-async def test_controller(nova_api_client):
-    controller = Controller(nova_api_client, cell="cell", controller_host="ur10e")
-    async with controller:
-        motion_groups = controller.get_motion_groups()
-        print(motion_groups)
-    assert False