Changed to implicit mastership - Tested RWS2 tests and added testing app - updated readme.

Author: mattiskoh

README.md

@@ -1,10 +1,12 @@
-# abb_robot_client
+# abb_robot_client 
+[![Python](https://img.shields.io/badge/python-3.8+-blue.svg)](https://github.com/rpiRobotics/abb_robot_client)
+
+This package is forked from https://github.com/rpiRobotics/abb_robot_client and has been modified and modernized.
+
 
-[![Python](https://img.shields.io/badge/python-3.6+-blue.svg)](https://github.com/rpiRobotics/abb_robot_client)
-![PyPI](https://img.shields.io/pypi/v/abb-robot-client)
 
 Python package providing clients for ABB robots using RWS (Robot Web Services) and Externally Guided Motion (EGM). 
-This package currently supports IRC5 controllers running RobotWare 6.xx. It does not support RobotWare 7+.
+This package supports IRC5 controllers running RobotWare 6.xx as well as Omnicore controllers running RobotWare 7+.
 
 This package is typically used with [abb-motion-program-exec](https://pypi.org/project/abb-motion-program-exec/),
 which provides a higher level interface to generate motion programs. `abb-motion-program-exec` includes the ability
@@ -21,59 +23,12 @@ Documentation can be found at: https://abb_robot_client.readthedocs.org
 
 ## Installation
 
-`abb-robot-client` is avaliable on PyPi. Use the `[aio]` option to include support for asyncio:
-
-```
-pip install abb-robot-client[aio]
-```
-
-## Examples
-
-See the `examples/` directory for examples using the modules.
-
-## Robot Raconteur Driver
-
-The Robot Raconteur driver provides access to the features of RWS and EGM, along with standard Robot Raconteur
-data structures such as `RobotInfo` and `RobotState`. See `examples/robotraconteur` for examples using
-the Robot Raconteur driver. See `src/abb_robot_client/robotraconteur/experimental.abb_robot.rws.robdef` and
-`src/abb_robot_client/robotraconteur/experimental.abb_robot.egm.robdef` service definitions
-for full information on the objects and data types provided by the service.
-
-### Installation
-
-Install the `abb-robot-client` with the `[robotraconteur]` feature:
-
-```bash
-python -m pip install abb-robot-client[robotraconteur]
-```
-
-### Start the Driver
-
-Start the driver:
-
 ```
-abb-robot-client-robotraconteur --robot-info-file=config\abb_1200_5_90_rws_default_config.yml --robot-url=http://127.0.0.1:80
+uv add https://github.com/Mesh-ch/abb_robot_client.git
 ```
 
-or
-
-```
-python -m pip abb_robot_client.robotraconteur --robot-info-file=config\abb_1200_5_90_rws_default_config.yml --robot-url=http://127.0.0.1:80
-```
-
-Change the `--robot-info-file=` to the appropriate Robot Raconteur format yaml file for your robot,
-and change `--robot-url=` to the URL of the IRC5 robot controller.
-
-If EGM is used, the EGM must be configured on the robot controller to point to the IP address of the
-computer running the driver, on port 6510.
-
-### Connection Info
+## Tests
 
-- URL: `rr+tcp://localhost:59926?service=robot`
-- Node Name: `experimental.abb_rws_robot.robot`
-- Device Name: `abb_robot_rws`
-- Service Name: `robot`
-- Root Object Type: `experimental.abb_robot.rws.ABBRWSRobot`
 
 ## License
 

pyproject.toml

@@ -25,6 +25,7 @@ build-backend = "hatchling.build"
 [dependency-groups]
 dev = [
     "pytest>=8.4.2",
+    "textual>=6.1.0",
 ]
 
 [project.urls]
@@ -35,4 +36,4 @@ dev = [
 [tool.pytest.ini_options]
 testpaths = [
     "tests"
-]
+]

src/abb_robot_client/rws2.py

@@ -190,36 +190,36 @@ def start(self, cycle: Optional[str] = 'asis', tasks: Optional[list[str]] = ["T_
                     self.deactivate_task(rob_task.name)
 
         payload={"regain": "continue", "execmode": "continue", "cycle": cycle, "condition": "none", "stopatbp": "disabled", "alltaskbytsp": "true"}
-        res=self._do_post("rw/rapid/execution/start", payload)
-        
+        res=self._do_post("rw/rapid/execution/start?mastership=implicit", payload)
+
     def activate_task(self, task: str):
         """
         Activate a RAPID task
 
         :param task: The name of the task to activate
         """
-        self._do_post("rw/rapid/tasks/activate", data={"task": task})
-        
-    def deactivate_task(self, task: str) -> None: 
+        self._do_post("rw/rapid/tasks/activate?mastership=implicit", data={"task": task})
+
+    def deactivate_task(self, task: str) -> None:
         """
         Deactivate a RAPID task
 
         :param task: The name of the task to deactivate
         """
-        self._do_post(f"rw/rapid/tasks/{task}/deactivate", data={"task": task})
+        self._do_post(f"rw/rapid/tasks/{task}/deactivate?mastership=implicit", data={"task": task})
 
     def stop(self):
         """
         Stop RAPID execution of normal tasks
         """
         payload={"stopmode": "stop"}
-        res=self._do_post("rw/rapid/execution/stop", payload)
+        res=self._do_post("rw/rapid/execution/stop?mastership=implicit", payload)
 
     def resetpp(self):
         """
         Reset RAPID program pointer to main in normal tasks
         """
-        res=self._do_post("rw/rapid/execution/resetpp")
+        res=self._do_post("rw/rapid/execution/resetpp?mastership=implicit")
 
     def get_ramdisk_path(self) -> str:
         """
@@ -257,9 +257,7 @@ def get_controller_state(self) -> str:
     def set_controller_state(self, ctrl_state): 
         """Possible ctrl-states to set are `motoroff` or `motoron`"""
         payload = {"ctrl-state": ctrl_state}
-        self.request_mastership()
-        res=self._do_post("rw/panel/ctrl-state", payload)
-        self.release_mastership()
+        res=self._do_post("rw/panel/ctrl-state?mastership=implicit", payload)
 
     def set_motors_on(self):
         """
@@ -326,7 +324,7 @@ def set_digital_io(self, signal: str, value: bool|int, network: str='Local', uni
         """
         lvalue = '1' if bool(value) else '0'
         payload={'lvalue': lvalue}
-        res=self._do_post(f"rw/iosystem/signals/{network}/{unit}/{signal}", payload)
+        res=self._do_post(f"rw/iosystem/signals/{network}/{unit}/{signal}?mastership=implicit", payload)
 
     def get_analog_io(self, signal: str, network: str='Local', unit: str='DRV_1') -> float:
         """
@@ -383,7 +381,7 @@ def get_rapid_variable(self, var: str, task: str = "T_ROB1") -> str:
         :param task: The task containing the pers variable
         :return: The pers variable encoded as a string
         """
-        res_json = self._do_get(f"rw/rapid/symbol/RAPID/{task}/{var}/data")
+        res_json = self._do_get(f"rw/rapid/symbol/RAPID/{task}/{var}/data?mastership=implicit")
         state = res_json["state"][0]["value"]
         return state
 
@@ -401,14 +399,15 @@ def set_rapid_variable(self, var: str, value: str, task: str = "T_ROB1"):
         else:
             var1 = var
         self.request_mastership()
-        res=self._do_post(f"rw/rapid/symbol/RAPID/{var1}/data/", payload)
+        res=self._do_post(f"rw/rapid/symbol/RAPID/{var1}/data?mastership=implicit", payload)
         self.release_mastership()
 
     def read_file(self, filename: str, directory: str = "$HOME") -> bytes:
         """
         Read a file off the controller
 
         :param filename: The relative path to the filename to read, e.g.: $HOME/...
+        :param directory: The directory to read the file from, e.g. $HOME
         :return: The file bytes
         """
         res_json = self._do_get_raw(f"fileservice/{directory}/{filename}")
@@ -426,6 +425,7 @@ def upload_file(self, filename: str, contents: str, directory: str = "$HOME") ->
 
         :param filename: The filename to write
         :param contents: The file content as str
+        :param directory: The directory to write the file to, e.g. $HOME
         """
         url=f"{self.base_url}/fileservice/{directory}/{filename}"
         header = {"Content-Type": "text/plain;v=2.0"}
@@ -692,7 +692,7 @@ def set_speedratio(self, speedratio: float):
         :param speedratio: The new speed ratio between 0% - 100%
         """
         payload = {"speed-ratio": str(speedratio)}
-        self._do_post(f"rw/panel/speedratio?mastership=implicit", payload)
+        self._do_post("rw/panel/speedratio?mastership=implicit", payload)
 
 
     # def send_ipc_message(self, target_queue: str, data: str, queue_name: str, cmd: int=111, userdef: int=1, msgtype: int=1 ):

tests/RWS2_test_app.py

@@ -0,0 +1,59 @@
+"""
+A textualize-based interactive CLI to run test functions from test_RWS2_POST.py.
+"""
+import asyncio
+from textual.app import App, ComposeResult
+from textual.widgets import Header, Footer, Button, Static, ListView, ListItem
+from textual.containers import Container
+import importlib
+import inspect
+
+
+# Import the test module and get test functions
+test_module = importlib.import_module("test_RWS2_POST")
+
+from abb_robot_client.rws2 import RWS2
+
+def get_test_functions():
+    return [
+        (name, func)
+        for name, func in inspect.getmembers(test_module, inspect.isfunction)
+        if name.startswith("test_")
+    ]
+
+class TestResult(Static):
+    pass
+
+class TestRWS2POST(App):
+    CSS_PATH = None
+    BINDINGS = [ ("q", "quit", "Quit") ]
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.client = RWS2()
+
+    def compose(self) -> ComposeResult:
+        yield Header()
+        yield Container(
+            ListView(*[ListItem(Button(name, id=name)) for name, _ in get_test_functions()], id="test-list"),
+            TestResult("Select a test to run.", id="result"),
+        )
+        yield Footer()
+
+    async def on_button_pressed(self, event: Button.Pressed) -> None:
+        test_name = event.button.id
+        test_func = dict(get_test_functions())[test_name]
+        result_widget = self.query_one("#result", TestResult)
+        try:
+            sig = inspect.signature(test_func)
+            if len(sig.parameters) == 1:
+                await asyncio.to_thread(test_func, self.client)
+            else:
+                await asyncio.to_thread(test_func)
+            result_widget.update(f"[green]Test {test_name} passed![/green]")
+        except Exception as e:
+            result_widget.update(f"[red]Test {test_name} failed: {e}[/red]")
+
+if __name__ == "__main__":
+    app = TestRWS2POST()
+    app.run()

tests/test_RWS2_GET.py

@@ -64,7 +64,7 @@ def test_read_event_log(client):
 def test_get_robtarget(client):
     robtarget = client.get_robtarget()
     assert isinstance(robtarget, RobTarget), f"Unexpected robtarget type: {type(robtarget)}"
-    
+    print(robtarget)
 
 def test_get_speedratio(client):
     speedratio = client.get_speedratio()
@@ -87,20 +87,20 @@ def test_get_speedratio(client):
 if __name__ == "__main__":
     rws = RWS2()
     #### GET REQUESTS ####
-    test_get_tasks(rws)
+    # test_get_tasks(rws)
     test_get_jointtargets(rws)
-    test_get_execution_state(rws)
-    test_get_ramdisk_paths(rws)
-    test_get_controller_state(rws)
-    test_get_operation_mode(rws)
-    test_get_digital_io(rws)
-    test_get_analog_io(rws)
-    test_get_rapid_variable(rws)
-    test_get_rapid_variable_num(rws)
-    test_read_file(rws)
-    test_read_event_log(rws)
-    test_read_event_log(rws)
+    # test_get_execution_state(rws)
+    # test_get_ramdisk_paths(rws)
+    # test_get_controller_state(rws)
+    # test_get_operation_mode(rws)
+    # test_get_digital_io(rws)
+    # test_get_analog_io(rws)
+    # test_get_rapid_variable(rws)
+    # test_get_rapid_variable_num(rws)
+    # test_read_file(rws)
+    # test_read_event_log(rws)
+    # test_read_event_log(rws)
     test_get_robtarget(rws)
-    test_get_speedratio(rws)
+    # test_get_speedratio(rws)
     # test_get_ipc_queue(rws)
     # test_ipc_message(rws)

tests/test_RWS2_POST.py

@@ -1,4 +1,5 @@
 import pytest
+import time
 from abb_robot_client.rws2 import RWS2, EventLogEntry, JointTarget, RAPIDExecutionState, RobTarget
 
 @pytest.fixture(scope="module")
@@ -13,6 +14,10 @@ def client():
 #     for msg in messages:
 #         print("IPC Message:", msg)
 
+def test_resetpp(client):
+    client.resetpp()
+    print("Program pointer reset.")
+
 def test_set_speedratio(client):
     original_speedratio = client.get_speedratio()
     new_speedratio = 50 if original_speedratio != 50 else 60
@@ -22,6 +27,17 @@ def test_set_speedratio(client):
     # Restore original speed ratio
     client.set_speedratio(100)
     assert client.get_speedratio() == 100
+    
+def test_set_digital_io(client):
+    io_state = client.get_digital_io("motion_program_error")
+    new_state = '1' if io_state == '0' else '0'
+    client.set_digital_io("motion_program_error", new_state)
+    updated_state = client.get_digital_io("motion_program_error")
+    assert updated_state == int(new_state)
+    time.sleep(0.5)
+    # Restore original state
+    client.set_digital_io("motion_program_error", io_state)
+    assert client.get_digital_io("motion_program_error") == int(io_state)
 
 def test_start(client):
     client.start()
@@ -65,13 +81,11 @@ def test_toggle_motors(client):
     else:
         raise Exception(f"Controller in unexpected state: {motor_state}")
 
-if __name__ == "__main__":
-    rws = RWS2()
-    # rws.request_mastership()
-    #### POST REQUESTS ####
-    # test_set_speedratio(rws)
-    # test_start(rws)
-    # test_upload_delete_file(rws)
-    # test_set_rapid_variable_num(rws)
-    
-    
+def test_request_mastership(client):
+    client.request_mastership()
+
+def test_release_mastership(client):
+    result = client.request_mastership()
+    time.sleep(1)  # Simulate some operations while holding mastership
+    result = client.release_mastership()
+    print(result)