add IPMIPowerPort and IPMIPowerDriver to control IPMI powered devices

this currently uses ipmi-power tool from freeipmi-tools

Author: sessl3r

debian/control

@@ -10,7 +10,7 @@ Package: labgrid
 Architecture: any
 Pre-Depends: dpkg (>= 1.16.1), python3, ${misc:Pre-Depends}
 Depends: ${python3:Depends}, ${misc:Depends}, ${shlibs:Depends}
-Recommends: openssh-client, microcom, socat, sshfs, rsync, bash-completion, python3-amt
+Recommends: openssh-client, microcom, socat, sshfs, rsync, bash-completion, python3-amt, freeipmi-tools
 Description: embedded board control python library
  Labgrid is an embedded board control python library with a focus on testing,
  development and general automation. It includes a remote control layer to

doc/configuration.rst

@@ -316,6 +316,26 @@ Arguments:
 Used by:
   - `AMTPowerDriver`_
 
+IPMIPowerPort
+++++++++++++
+A :any:`IPMIPowerPort` describes a IPMI port accessible via `ipmi-power`.
+
+.. code-block:: yaml
+
+   IPMIPowerPort:
+     host: 'ipmi-hostname'
+     username: 'admin'
+     password: 'secret-password'
+
+Arguments:
+ - host (str): hostname or ip the IPMI interface of the PC is reachable
+ - username (str): username to use for IPMI login
+ - password (str): password to use for IPMI login
+ - timeout (int): timeout to use when polling the resource
+
+Used by:
+  - `IPMIPowerDriver`_
+
 USBPowerPort
 ++++++++++++
 A :any:`USBPowerPort` describes a generic switchable USB hub as supported by

labgrid/driver/__init__.py

@@ -15,7 +15,7 @@
 from .powerdriver import ManualPowerDriver, ExternalPowerDriver, \
                          DigitalOutputPowerDriver, YKUSHPowerDriver, \
                          USBPowerDriver, SiSPMPowerDriver, NetworkPowerDriver, \
-                         PDUDaemonDriver, AMTPowerDriver
+                         PDUDaemonDriver, AMTPowerDriver, IPMIPowerDriver
 from .usbloader import MXSUSBDriver, IMXUSBDriver, BDIMXUSBDriver, RKUSBDriver, UUUDriver
 from .usbsdmuxdriver import USBSDMuxDriver
 from .usbsdwiredriver import USBSDWireDriver

labgrid/driver/powerdriver.py

@@ -1,4 +1,5 @@
 import shlex
+import subprocess
 import time
 import math
 from importlib import import_module
@@ -488,3 +489,53 @@ def get(self):
             return False
         else:
             raise ExecutionError(f"got unexpected power status {power}")
+
+
+@target_factory.reg_driver
+@attr.s(eq=False)
+class IPMIPowerDriver(Driver, PowerResetMixin, PowerProtocol):
+    """IPMIPowerDriver - Driver using an IPMI PowerPort"""
+    bindings = {"port": "IPMIPowerPort", }
+    delay = attr.ib(default=5.0, validator=attr.validators.instance_of(float))
+
+    def _ipmi_power(self, cmd):
+        runstr = f"ipmi-power -h {self.port.host} -u {self.port.username} "
+        runstr += f"--session-timeout={self.port.timeout*1000} "
+        runstr += f"-p {self.port.password} {cmd}"
+        runstr = runstr.split(' ')
+        if len(self.port.args) > 0:
+            runstr.append(self.port.args)
+        return subprocess.run(runstr, capture_output=True)
+
+    def _ipmi_cmd(self, cmd):
+        ret = self._ipmi_power(cmd)
+        stdout = ret.stdout.decode('utf-8').split(' ')
+        assert self.port.host == stdout[0][:-1]
+        assert "ok" == stdout[1][:-1]
+
+    @Driver.check_active
+    @step()
+    def on(self):
+        self._ipmi_cmd('--on')
+
+    @Driver.check_active
+    @step()
+    def off(self):
+        self._ipmi_cmd('--off')
+
+    @Driver.check_active
+    @step()
+    def cycle(self):
+        self._ipmi_cmd('--cycle')
+
+    @Driver.check_active
+    def get(self):
+        ret = self._ipmi_power('--stat')
+        stdout = ret.stdout.decode('utf-8').split(' ')
+        assert self.port.host == stdout[0][:-1]
+        power = stdout[1][:-1]
+        if power == 'off':
+            return False
+        elif power == 'on':
+            return True
+        raise ExecutionError(f"got unexpected power status {power}")

labgrid/remote/client.py

@@ -853,7 +853,7 @@ def power(self):
         target = self._get_target(place)
         from ..resource.power import NetworkPowerPort, PDUDaemonPort
         from ..resource.remote import NetworkUSBPowerPort, NetworkSiSPMPowerPort
-        from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort, AMTPowerPort
+        from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort, AMTPowerPort, IPMIPowerPort
         drv = None
         try:
             drv = target.get_driver("PowerProtocol", name=name)
@@ -875,6 +875,8 @@ def power(self):
                     drv = self._get_driver_or_new(target, "YKUSHPowerDriver", name=name)
                 elif isinstance(resource, AMTPowerPort):
                     drv = self._get_driver_or_new(target, "AMTPowerDriver", name=name)
+                elif isinstance(resource, IPMIPowerPort):
+                    drv = self._get_driver_or_new(target, "IPMIPowerDriver", name=name)
                 if drv:
                     break
 

labgrid/resource/__init__.py

@@ -5,7 +5,7 @@
 from .modbusrtu import ModbusRTU
 from .networkservice import NetworkService
 from .onewireport import OneWirePIO
-from .power import NetworkPowerPort, PDUDaemonPort, AMTPowerPort
+from .power import NetworkPowerPort, PDUDaemonPort, AMTPowerPort, IPMIPowerPort
 from .remote import RemotePlace
 from .udev import (
     AlteraUSBBlaster,

labgrid/resource/power.py

@@ -48,3 +48,20 @@ class AMTPowerPort(Resource):
     host = attr.ib(validator=attr.validators.instance_of(str))
     password = attr.ib(validator=attr.validators.instance_of(str))
     timeout = attr.ib(default=30, validator=attr.validators.instance_of(int))
+
+
+@target_factory.reg_resource
+@attr.s(eq=False)
+class IPMIPowerPort(Resource):
+    """The IPMIPowerPrt describes an IPMI controllable PC with BMC
+
+    Args:
+        host (str): hostname or ip the IPMI interface of the PC is reachable
+        username (str): username to use for IPMI login
+        password (str): password to use for IPMI login
+        timeout (int): timeout to use when polling the resource
+    """
+    host = attr.ib(validator=attr.validators.instance_of(str))
+    username = attr.ib(validator=attr.validators.instance_of(str))
+    password = attr.ib(validator=attr.validators.instance_of(str))
+    timeout = attr.ib(default=30, validator=attr.validators.instance_of(int))