add AMTPowerPort and AMTPowerDriver to control AMT powered devices

This currently uses amtctrl as the backend as it is simpler then
importing the library of the same tool. If more flexibility is needed
the python amt library could be used.

Signed-off-by: Tobias Binkowski <[email protected]>

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
+Recommends: openssh-client, microcom, socat, sshfs, rsync, bash-completion, python3-amt
 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,24 @@ 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`_
+
 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
 from .usbloader import MXSUSBDriver, IMXUSBDriver, BDIMXUSBDriver, RKUSBDriver, UUUDriver
 from .usbsdmuxdriver import USBSDMuxDriver
 from .usbsdwiredriver import USBSDWireDriver

labgrid/driver/powerdriver.py

@@ -440,3 +440,51 @@ 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}")

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
         drv = None
         try:
             drv = target.get_driver("PowerProtocol", name=name)
@@ -888,6 +887,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, AMTPowerPort):
+                    drv = self._get_driver_or_new(target, "AMTPowerDriver", 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
 from .remote import RemotePlace
 from .udev import (
     AlteraUSBBlaster,

labgrid/resource/power.py

@@ -33,3 +33,18 @@ 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))