Add powerdrivers for AMT and IPMI powered PC's

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
+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

@@ -310,6 +310,45 @@ NetworkYKUSHPowerPort
 A :any:`NetworkYKUSHPowerPort` describes a `YKUSHPowerPort`_ available on a
 remote computer.
 
+AMTPowerPort
+++++++++++++
+A :any:`AMTPowerPort` describes a AMT port accessible via `amtctrl`.
+
+.. code-block:: yaml
+
+   AMTPowerPort:
+     host: 'amt-hostname'
+     password: 'secret-password'
+
+Arguments:
+ - host (str): hostname or ip the AMT interface of the PC is reachable
+ - password (str): password to use for AMT login
+ - timeout (int): timeout to use when polling the resource
+
+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
+ - args (str): extra args to prepend the command with
+
+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
+                         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
@@ -440,3 +441,101 @@ def cycle(self):
     @Driver.check_active
     def get(self):
         raise NotImplementedError("pdudaemon does not support retrieving the port's state")
+
+
+@target_factory.reg_driver
+@attr.s(eq=False)
+class AMTPowerDriver(Driver, PowerResetMixin, PowerProtocol):
+    """AMTPowerDriver - Driver using an Intel AMT PowerPort"""
+    bindings = {"port": "AMTPowerPort", }
+    delay = attr.ib(default=5.0, validator=attr.validators.instance_of(float))
+
+    def __attrs_post_init__(self):
+        super().__attrs_post_init__()
+
+    def _amt_power(self, cmd):
+        runstr = f"amtctrl -p {self.port.host} {cmd}"
+        p = subprocess.Popen(runstr.split(' '), text=True,
+                             stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                             stdin=subprocess.PIPE)
+        p.stdin.write(self.port.password)
+        p.stdin.close()
+        p.wait(self.port.timeout)
+        out = p.stdout.read() + p.stderr.read()
+        assert p.returncode == 0, "Failed to execute AMT command {cmd}, ret: {p.returncode}, {out}"
+        return out.strip()
+
+    @Driver.check_active
+    @step()
+    def on(self):
+        self._amt_power("on")
+
+    @Driver.check_active
+    @step()
+    def off(self):
+        self._amt_power("off")
+
+    @Driver.check_active
+    @step()
+    def cycle(self):
+        self._amt_power("reboot")
+
+    @Driver.check_active
+    def get(self):
+        power = self._amt_power("status")
+        if power == "on":
+            return True
+        elif power == "off":
+            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.extend(self.port.args.split(' '))
+        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

@@ -867,8 +867,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
-
+        from ..resource import TasmotaPowerPort, NetworkYKUSHPowerPort, AMTPowerPort, IPMIPowerPort
         drv = None
         try:
             drv = target.get_driver("PowerProtocol", name=name)
@@ -888,6 +887,10 @@ 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, 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
+from .power import NetworkPowerPort, PDUDaemonPort, AMTPowerPort, IPMIPowerPort
 from .remote import RemotePlace
 from .udev import (
     AlteraUSBBlaster,

labgrid/resource/power.py

@@ -33,3 +33,37 @@ class PDUDaemonPort(Resource):
     host = attr.ib(validator=attr.validators.instance_of(str))
     pdu = attr.ib(validator=attr.validators.instance_of(str))
     index = attr.ib(validator=attr.validators.instance_of(int), converter=int)
+
+
+@target_factory.reg_resource
+@attr.s(eq=False)
+class AMTPowerPort(Resource):
+    """The AMTPowerPort describes an Intel AMT power controllable PC with BMC
+
+    Args:
+        host (str): hostname or ip the AMT interface of the PC is reachable
+        password (str): password to use for AMT login
+        timeout (int): timeout to use when polling the 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
+        args (str): extra args to prepend the command with
+    """
+    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))
+    args = attr.ib(default="", validator=attr.validators.instance_of(str))