Merge pull request labgrid-project#797 from initBasti/eg_pms2_network_power_driver

driver/power: Add backend driver for EG PMS2 LAN/WLAN

Author: Emantor

CHANGES.rst

@@ -9,6 +9,8 @@ New Features in 0.5.0
   considering only the closest marker.
 - The labgrid client SSH command is now able to instantiate the SSHDriver when
   there are multiple NetworkService resources available.
+- eg_pms2_network power port driver supports controlling the Energenie power
+  management series with devices like the EG_PMS2_LAN & EG_PMS2_WLAN
 
 Bug fixes in 0.5.0
 ~~~~~~~~~~~~~~~~~~

doc/configuration.rst

@@ -208,6 +208,12 @@ Currently available are:
   Controls TP-Link power strips via `python-kasa
   <https://github.com/python-kasa/python-kasa>`_.
 
+``eg_pms2_network``
+  Controls the EG_PMS2_LAN & EG_PMS2_WLAN devices, through simple HTTP POST and
+  GET requests.  The device requires a password for logging into the control
+  interface, this module deliberately uses the standard password '1' and is
+  not compatible with a different password.
+
 Used by:
   - `NetworkPowerDriver`_
 

labgrid/driver/power/eg_pms2_network.py

@@ -0,0 +1,77 @@
+import re
+import requests
+
+from ..exception import ExecutionError
+
+# This driver implementes a power port for the EG_PMS2_LAN & EG_PMS2_WLAN
+# devices, it works through a simple HTTP interface, that requires a login.
+# Driver has been tested with:
+# * EG_PMS2_LAN
+
+# The default HTTP port for usage via the SSH proxy.
+PORT = 80
+
+# The login was successful when we can locate a Log out button
+LOGIN_SUCCESS_REGEX = r'<div class="boxmenuitem"><a href="login\.html">Log Out</a></div>'
+# Search for a string similar to: "var sockstates = [1,1,1,1];" and create
+# a match group out of the numbers within the square brackets.
+SOCKSTATES_REGEX = r"var\s+sockstates\s+=\s+\[([01]),([01]),([01]),([01])\];"
+
+
+def login(base_url: str) -> None:
+    """
+    Use the default password 1, because labgrid doesn't support password
+    encryption, modifying the password doesn't secure the device as the
+    password would be stored as plain-text.
+    """
+    login_url = f"{base_url}/login.html"
+    try:
+        response = requests.post(login_url, data={'pw': 1})
+    except requests.exceptions.ConnectionError as err:
+        raise ExecutionError(
+            f"Device not found at {base_url} or the network interface of the "
+            "device is not active (press the reset button on the device)"
+        ) from err
+    if response.status_code != 200 or not re.search(LOGIN_SUCCESS_REGEX,
+                                                    response.text):
+        raise ExecutionError("Login to Energenie web interface failed")
+
+
+def logout(base_url: str) -> None:
+    """
+    After a successful login, the session is reserved for the IP address.
+    Logout explicitly to allow accessing the device from different hosts.
+    """
+    response = requests.get(f"{base_url}/login.html")
+    if response.status_code != 200:
+        raise ExecutionError("Logout from Energenie web interface failed")
+
+
+def power_set(host, port, index, value):
+    base_url = f"http://{host}:{port}"
+    index = int(index)
+    assert 1 <= index <= 4
+
+    value = 1 if value else 0
+    login(base_url=base_url)
+    response = requests.post(base_url, data={f'cte{index}': value})
+    response.raise_for_status()
+    logout(base_url=base_url)
+    if not response.status_code == 200:
+        raise ExecutionError(f"Switching socket at index {index} "
+                             f"{'on' if value else 'off'} failed")
+
+
+def power_get(host, port, index):
+    base_url = f"http://{host}:{port}"
+    index = int(index)
+    assert 1 <= index <= 4
+
+    # Fetch status
+    login(base_url=base_url)
+    response = requests.get(f"{base_url}/energenie.html")
+    response.raise_for_status()
+    logout(base_url=base_url)
+    match_group = re.search(SOCKSTATES_REGEX, response.text)
+    sockstates = [int(match_group.group(x)) for x in range(1, 5)]
+    return sockstates[index - 1] == 1

tests/test_powerdriver.py

@@ -171,6 +171,7 @@ def test_import_backends(self):
         import labgrid.driver.power.netio_kshell
         import labgrid.driver.power.rest
         import labgrid.driver.power.sentry
+        import labgrid.driver.power.eg_pms2_network
 
     def test_import_backend_siglent(self):
         pytest.importorskip("vxi11")