Merge pull request #1324 from DanielKriz/feature/aten_pe6216_power_driver

driver/pe6216: add support for Aten PE6216 PDU

Author: Bastian-Krause

doc/configuration.rst

@@ -208,6 +208,11 @@ Currently available are:
 ``netio_kshell``
   Controls *NETIO 4C PDUs* via a Telnet interface.
 
+``pe6216``
+  Controls an Aten PE6216 PDU via a simple HTTP API.
+  This backend deliberately uses the standard username ``administrator`` and the default password
+  ``password`` and is not compatible with other credentials.
+
 ``poe_mib``
   Controls PoE switches using the PoE SNMP administration MiBs.
 

labgrid/driver/power/pe6216.py

@@ -0,0 +1,55 @@
+"""Tested with Aten PE6216.
+
+HTTP API is defined by in the aten PDU PE6216 specification:
+https://assets.aten.com/product/manual/Restful-API-Guide-for-PDU_2022-11-18.pdf
+"""
+
+import re
+import requests
+
+from ..exception import ExecutionError
+
+PORT = 80
+
+MIN_OUTLET_INDEX = 1
+MAX_OUTLET_INDEX = 16
+HEADERS = {"Content-Type": "application/x-www-form-urlencoded"}
+
+
+def power_set(host, port, index, value):
+    index = int(index)
+    assert MIN_OUTLET_INDEX <= index <= MAX_OUTLET_INDEX
+    value = "on" if value else "off"
+
+    response = requests.post(
+        f"http://{host}:{port}/api/outlet/relay",
+        headers=HEADERS,
+        data={
+            "usr": "administrator",
+            "pwd": "password",
+            "index": index,
+            "method": value,
+        },
+    )
+    response.raise_for_status()
+
+
+def power_get(host, port, index):
+    index = int(index)
+    assert MIN_OUTLET_INDEX <= index <= MAX_OUTLET_INDEX
+
+    response = requests.get(
+        f"http://{host}:{port}/api/outlet/relay",
+        headers=HEADERS,
+        params={
+            "usr": "administrator",
+            "pwd": "password",
+            "index": index,
+        },
+    )
+    response.raise_for_status()
+
+    m = re.search(r"<\d+>(?P<state>(ON|OFF|PENDING))<.*", response.text)
+    if m is None:
+        raise ExecutionError("PE6216: could not match response")
+    return m.group("state") == "ON"

pyproject.toml

@@ -222,6 +222,7 @@ include = [
   "labgrid/driver/httpvideodriver.py",
   "labgrid/driver/manualswitchdriver.py",
   "labgrid/driver/power/gude8031.py",
+  "labgrid/driver/power/pe6216.py",
   "labgrid/driver/power/shelly_gen2.py",
   "labgrid/driver/rawnetworkinterfacedriver.py",
   "labgrid/protocol/**/*.py",

tests/test_powerdriver.py

@@ -291,6 +291,7 @@ def test_import_backends(self):
         import labgrid.driver.power.gude24
         import labgrid.driver.power.netio
         import labgrid.driver.power.netio_kshell
+        import labgrid.driver.power.pe6216
         import labgrid.driver.power.rest
         import labgrid.driver.power.sentry
         import labgrid.driver.power.eg_pms2_network