client: add client commands to SysfsGPIO

Add support for the power and button protocols and add the "invert" method
of the io protocol to SysffsGPIO

Signed-off-by: Perry Melange <[email protected]>

Author: pmelange

labgrid/remote/client.py

@@ -859,7 +859,7 @@ def power(self):
         name = self.args.name
         target = self._get_target(place)
         from ..resource.power import NetworkPowerPort, PDUDaemonPort
-        from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort
+        from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort, NetworkSysfsGPIO
         from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort
 
         drv = None
@@ -881,6 +881,8 @@ def power(self):
                     drv = self._get_driver_or_new(target, "TasmotaPowerDriver", name=name)
                 elif isinstance(resource, NetworkYKUSHPowerPort):
                     drv = self._get_driver_or_new(target, "YKUSHPowerDriver", name=name)
+                elif isinstance(resource, NetworkSysfsGPIO):
+                    drv = self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name)
                 if drv:
                     break
 
@@ -892,6 +894,32 @@ def power(self):
         if action == "get":
             print(f"power{' ' + name if name else ''} for place {place.name} is {'on' if res else 'off'}")
 
+    def button(self):
+        place = self.get_acquired_place()
+        action = self.args.action
+        delay = self.args.delay
+        name = self.args.name
+        target = self._get_target(place)
+        from ..resource.remote import NetworkSysfsGPIO
+
+        drv = None
+        try:
+            drv = target.get_driver("ButtonProtocol", name=name)
+        except NoDriverFoundError:
+            for resource in target.resources:
+                if isinstance(resource, NetworkSysfsGPIO):
+                    drv = self._get_driver_or_new(target, "GpioDigitalOutputDriver", name=name)
+                if drv:
+                    break
+
+        if not drv:
+            raise UserError("target has no compatible resource available")
+        if delay is not None:
+            drv.delay = delay
+        res = getattr(drv, action)()
+        if action == "get":
+            print(f"button{' ' + name if name else ''} for place {place.name} is {'pressed' if res else 'released'}")
+
     def digital_io(self):
         place = self.get_acquired_place()
         action = self.args.action
@@ -930,6 +958,8 @@ def digital_io(self):
             drv.set(True)
         elif action == "low":
             drv.set(False)
+        elif action == "invert":
+            drv.invert()
 
     async def _console(self, place, target, timeout, *, logfile=None, loop=False, listen_only=False):
         name = self.args.name
@@ -1814,8 +1844,16 @@ def main():
     subparser.add_argument("--name", "-n", help="optional resource name")
     subparser.set_defaults(func=ClientSession.power)
 
+    subparser = subparsers.add_parser("button", help="change (or get) a place's button status")
+    subparser.add_argument("action", choices=["press", "release", "press_for", "get"])
+    subparser.add_argument(
+        "-t", "--delay", type=float, default=None, help="wait time in seconds between the press and release during press_for"
+    )
+    subparser.add_argument("--name", "-n", help="optional resource name")
+    subparser.set_defaults(func=ClientSession.button)
+
     subparser = subparsers.add_parser("io", help="change (or get) a digital IO status")
-    subparser.add_argument("action", choices=["high", "low", "get"], help="action")
+    subparser.add_argument("action", choices=["high", "low", "invert", "get"], help="action")
     subparser.add_argument("name", help="optional resource name", nargs="?")
     subparser.set_defaults(func=ClientSession.digital_io)