diff --git a/doc/configuration.rst b/doc/configuration.rst index d93e58c90..ed7e00828 100644 --- a/doc/configuration.rst +++ b/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 `_. @@ -259,7 +260,7 @@ Currently available are: See the `documentation `__ ``tplink`` - Controls *TP-Link power strips* via `python-kasa + Controls *TP-Link smart plugs and power strips* via `python-kasa `_. ``ubus`` diff --git a/labgrid/driver/power/tplink.py b/labgrid/driver/power/tplink.py index e556a8732..c61583c2d 100644 --- a/labgrid/driver/power/tplink.py +++ b/labgrid/driver/power/tplink.py @@ -1,6 +1,7 @@ """ 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 @@ -8,27 +9,42 @@ 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)) diff --git a/labgrid/resource/power.py b/labgrid/resource/power.py index c64861106..84c9a9427 100644 --- a/labgrid/resource/power.py +++ b/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))) diff --git a/tests/test_powerdriver.py b/tests/test_powerdriver.py index 2ae8783b2..1e7b9033a 100644 --- a/tests/test_powerdriver.py +++ b/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',