migrate siglent power backend vxi11 -> pyvisa, add power show/config options

labgrid/driver/power/siglent.py

@@ -1,35 +1,62 @@
-"""tested with Siglent SPD3303X-E, and should be compatible with all SPD3000X series modules"""
+""" tested with Siglent SPD3303X-E, SPD1168X and should be compatible with all SPD3000X series modules """
 
-import warnings
-
-import vxi11
+import pyvisa
 
+def _get_psu(host):
+    """Helper to initialize the raw network socket resource session via PyVISA."""
+    rm = pyvisa.ResourceManager("@py")
+    resource_string = f"TCPIP0::{host}::5025::SOCKET"
+    psu = rm.open_resource(resource_string)
+    # Siglent scopes and PSUs require explicit newline termination on raw sockets
+    psu.read_termination = "\n"
+    psu.write_termination = "\n"
+    psu.timeout = 5000  # 5 second safety timeout
+    return psu
 
 def power_set(host, port, index, value):
-    warnings.warn(
-        "siglent power backend uses vxi11 module using deprecated xdrlib module, see https://github.com/labgrid-project/labgrid/issues/1507",
-        DeprecationWarning,
-    )
-
     assert port is None
     index = int(index)
     assert 1 <= index <= 2
     value = "ON" if value else "OFF"
-    psu = vxi11.Instrument(host)
-    psu.write(f"OUTPUT CH{index},{value}")
-
+    with _get_psu(host) as psu:
+        psu.write(f"OUTPUT CH{index},{value}")
 
 def power_get(host, port, index):
-    warnings.warn(
-        "siglent power backend uses vxi11 module using deprecated xdrlib module, see https://github.com/labgrid-project/labgrid/issues/1507",
-        DeprecationWarning,
-    )
-
     assert port is None
     index = int(index)
     assert 1 <= index <= 2
-    psu = vxi11.Instrument(host)
-    state = psu.ask("SYSTEM:STATUS?")
+    with _get_psu(host) as psu:
+        state = psu.query("SYSTEM:STATUS?")
     state = int(state, 16)
     bitmask = 1 << (index + 3)
     return bool(state & bitmask)
+
+def power_show(host, port, index):
+    assert port is None
+    index = int(index)
+    assert 1 <= index <= 2
+    with _get_psu(host) as psu:
+        v_measured = psu.query(f"MEAS:VOLT? CH{index}")
+        a_measured = psu.query(f"MEAS:CURR? CH{index}")
+        v_set = psu.query(f"CH{index}:VOLT?")
+        a_set = psu.query(f"CH{index}:CURR?")
+    return {
+        "voltage": float(v_measured),
+        "amps": float(a_measured),
+        "v_limit": float(v_set),
+        "a_limit": float(a_set)
+    }
+
+def power_voltage(host, port, index, voltage):
+    assert port is None
+    index = int(index)
+    assert 1 <= index <= 2
+    with _get_psu(host) as psu:
+        psu.write(f"CH{index}:VOLT {voltage}")
+
+def power_amps(host, port, index, amps):
+    assert port is None
+    index = int(index)
+    with _get_psu(host) as psu:
+        psu.write(f"CH{index}:CURR {amps}")
+

labgrid/driver/powerdriver.py

@@ -7,7 +7,7 @@
 
 from ..exceptions import InvalidConfigError
 from ..factory import target_factory
-from ..protocol import PowerProtocol, DigitalOutputProtocol, ResetProtocol
+from ..protocol import PowerProtocol, DigitalOutputProtocol, ResetProtocol, ProgrammablePowerProtocol
 from ..resource import NetworkPowerPort
 from ..step import step
 from ..util.proxy import proxymanager
@@ -233,6 +233,21 @@ def cycle(self):
     def get(self):
         return self.backend.power_get(self._host, self._port, self.port.index)
 
+    @Driver.check_active
+    @step()
+    def show(self):
+        return self.backend.power_show(self._host, self._port, self.port.index)
+
+    @Driver.check_active
+    @step(args=['voltage'])
+    def voltage(self, voltage):
+        return self.backend.power_voltage(self._host, self._port, self.port.index, voltage)
+
+    @Driver.check_active
+    @step(args=['amps'])
+    def amps(self, amps):
+        return self.backend.power_amps(self._host, self._port, self.port.index, amps)
+
 @target_factory.reg_driver
 @attr.s(eq=False)
 class DigitalOutputPowerDriver(Driver, PowerResetMixin, PowerProtocol):

labgrid/protocol/__init__.py

@@ -2,7 +2,7 @@
 from .commandprotocol import CommandProtocol
 from .consoleprotocol import ConsoleProtocol
 from .linuxbootprotocol import LinuxBootProtocol
-from .powerprotocol import PowerProtocol
+from .powerprotocol import PowerProtocol, ProgrammablePowerProtocol
 from .filetransferprotocol import FileTransferProtocol
 from .infoprotocol import InfoProtocol
 from .digitaloutputprotocol import DigitalOutputProtocol

labgrid/protocol/powerprotocol.py

@@ -13,3 +13,17 @@ def off(self):
     @abc.abstractmethod
     def cycle(self):
         raise NotImplementedError
+
+
+class ProgrammablePowerProtocol(PowerProtocol):
+    @abc.abstractmethod
+    def show(self, index):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def voltage(self, index, voltage):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def amps(self, index, amps):
+        raise NotImplementedError

labgrid/remote/client.py

@@ -941,6 +941,7 @@ def power(self):
         from ..resource.power import NetworkPowerPort, PDUDaemonPort
         from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort, NetworkSysfsGPIO
         from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort
+        from ..protocol import ProgrammablePowerProtocol
 
         drv = None
         try:
@@ -971,9 +972,26 @@ def power(self):
             raise UserError("target has no compatible resource available")
         if delay is not None:
             drv.delay = delay
+
+        if action == "config":
+            if self.args.voltage is None and self.args.amps is None:
+                raise UserError("Option config requires --voltage <float> and/or --amps <float>")
+
+            if self.args.voltage is not None:
+                drv.voltage(self.args.voltage)
+                print(f"set voltage to {self.args.voltage}V")
+            if self.args.amps is not None:
+                drv.amps(self.args.amps)
+                print(f"set amps to {self.args.amps}A")
+            return
+
         res = getattr(drv, action)()
         if action == "get":
             print(f"power{' ' + name if name else ''} for place {place.name} is {'on' if res else 'off'}")
+        elif action == "show":
+            res = drv.show()
+            print(f"Voltage: {res['voltage']}V [{res['v_limit']}V], Current: {res['amps']}A [{res['a_limit']}A]")
+
 
     def digital_io(self):
         place = self.get_acquired_place()
@@ -1982,10 +2000,16 @@ def get_parser(auto_doc_mode=False) -> "argparse.ArgumentParser | AutoProgramArg
     subparser.set_defaults(func=ClientSession.print_env)
 
     subparser = subparsers.add_parser("power", aliases=("pw",), help="change (or get) a place's power status")
-    subparser.add_argument("action", choices=["on", "off", "cycle", "get"])
+    subparser.add_argument("action", choices=["on", "off", "cycle", "get", "show", "config"])
     subparser.add_argument(
         "-t", "--delay", type=float, default=None, help="wait time in seconds between off and on during cycle"
     )
+    subparser.add_argument(
+        "-v", "--voltage", type=float, default=None, help="voltage value to be configured"
+    )
+    subparser.add_argument(
+        "-a", "--amps", type=float, default=None, help="amps value to be configured"
+    )
     subparser.add_argument("--name", "-n", help="optional resource name")
     subparser.set_defaults(func=ClientSession.power)