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

This driver adds support for the Energenie PMS2 LAN & WLAN programmable
power sockets. They are accessed over a web interface and require a
login before accessing the status of the socket or switching them on or off.
A password is required for the login in POST request and this driver
deliberately uses the default password, as the current state of labgrid
doesn't support password encryption, meaning the password would be
stored in plain text.

Signed-off-by: Sebastian Fricke <sebastian.fricke@posteo.net>
[r.czerwinski@pengutronix.de: add additional raise_for_status on responses]
Signed-off-by: Rouven Czerwinski <r.czerwinski@pengutronix.de>

Author: initBasti

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")