tplink: add support for kasa smart plugs

doc/configuration.rst

@@ -146,7 +146,8 @@ The example describes port 0 on the remote power switch
 Arguments:
   - model (str): model of the power switch
   - host (str): hostname of the power switch
-  - index (int): number of the port to switch
+  - index (int, default=0): optional, number of the port to switch. This
+    can be omitted if the device only has a single output.
 
 The ``model`` property selects one of several `backend implementations
 <https://github.com/labgrid-project/labgrid/tree/master/labgrid/driver/power>`_.
@@ -259,7 +260,7 @@ Currently available are:
   See the `documentation <https://docs.tinycontrol.pl/en/tcpdu/api/commands/>`__
 
 ``tplink``
-  Controls *TP-Link power strips* via `python-kasa
+  Controls *TP-Link smart plugs and power strips* via `python-kasa
   <https://github.com/python-kasa/python-kasa>`_.
 
 ``ubus``

labgrid/driver/power/tplink.py

@@ -1,34 +1,50 @@
 """ Tested with TP Link KP303, and should be compatible with any strip supported by kasa """
 
 import asyncio
+from kasa import DeviceType, Discover
 from kasa.iot import IotStrip
 
 
 async def _power_set(host, port, index, value):
     """We embed the coroutines in an `async` function to minimise calls to `asyncio.run`"""
     assert port is None
     index = int(index)
-    strip = IotStrip(host)
-    await strip.update()
-    assert (
-        len(strip.children) > index
-    ), "Trying to access non-existant plug socket on strip"
-    if value is True:
-        await strip.children[index].turn_on()
-    elif value is False:
-        await strip.children[index].turn_off()
+    dev = await Discover.discover_single(host)
+    if dev.device_type == DeviceType.Strip:
+        iotstrip = IotStrip(host)
+        await iotstrip.update()
+        assert (
+            len(iotstrip.children) > index
+        ), "Trying to access non-existant plug socket on strip"
+        if value:
+            await iotstrip.children[index].turn_on()
+        else:
+            await iotstrip.children[index].turn_off()
+    elif dev.device_type == DeviceType.Plug:
+        await dev.update()
+        if value:
+            await dev.turn_on()
+        else:
+            await dev.turn_off()
 
 
 def power_set(host, port, index, value):
     asyncio.run(_power_set(host, port, index, value))
 
-
-def power_get(host, port, index):
+async def _power_get(host, port, index):
     assert port is None
     index = int(index)
-    strip = IotStrip(host)
-    asyncio.run(strip.update())
-    assert (
-        len(strip.children) > index
-    ), "Trying to access non-existant plug socket on strip"
-    return strip.children[index].is_on
+    dev = await Discover.discover_single(host)
+    if dev.device_type == DeviceType.Strip:
+        iotstrip = IotStrip(host)
+        await iotstrip.update()
+        assert (
+            len(iotstrip.children) > index
+        ), "Trying to access non-existant plug socket on strip"
+        return iotstrip.children[index].is_on
+    elif dev.device_type == DeviceType.Plug:
+        await dev.update()
+        return dev.is_on
+
+def power_get(host, port, index):
+    return asyncio.run(_power_get(host, port, index))

labgrid/resource/power.py

@@ -12,11 +12,11 @@ class NetworkPowerPort(Resource):
     Args:
         model (str): model of the external power switch
         host (str): host to connect to
-        index (str): index of the power port on the external switch
+        index (str): optional, index of the power port on the external switch
     """
     model = attr.ib(validator=attr.validators.instance_of(str))
     host = attr.ib(validator=attr.validators.instance_of(str))
-    index = attr.ib(validator=attr.validators.instance_of(str),
+    index = attr.ib(default=0, validator=attr.validators.instance_of(str),
                     converter=lambda x: str(int(x)))
 
 

tests/test_powerdriver.py

@@ -173,6 +173,12 @@ def test_create(self, target):
         d = NetworkPowerDriver(target, 'power')
         assert isinstance(d, NetworkPowerDriver)
 
+    def test_default_index(self, target):
+        r = NetworkPowerPort(target, 'power', model='netio', host='dummy')
+        d = NetworkPowerDriver(target, 'power')
+        assert r.index == '0'
+        assert isinstance(d, NetworkPowerDriver)
+
     @pytest.mark.parametrize('backend', ('rest', 'simplerest'))
     @pytest.mark.parametrize(
         'host',