From e858ef03e65180b80ad93302dab472d3b8c97f0d Mon Sep 17 00:00:00 2001 From: Trevor Gamblin Date: Wed, 2 Jul 2025 20:03:09 -0400 Subject: [PATCH 1/4] labgrid: tplink: support 'plug' devices - differentiate between 'strip' and 'plug' devices, using simpler logic for controlling power on the latter - create new _power_get() async function to match _power_set() Signed-off-by: Trevor Gamblin --- labgrid/driver/power/tplink.py | 50 ++++++++++++++++++++++------------ 1 file changed, 33 insertions(+), 17 deletions(-) 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)) From f14d75df68fc9ea4e1cc4c449c7b64150c3614da Mon Sep 17 00:00:00 2001 From: Trevor Gamblin Date: Fri, 11 Jul 2025 13:16:56 -0400 Subject: [PATCH 2/4] NetworkPowerPort: make index optional (default 0) Smart plugs and similar single-port network-controlled devices shouldn't require the exporter config to include the 'index' argument. Make it optional by including a default of 0 in NetworkPowerPort. Signed-off-by: Trevor Gamblin --- labgrid/resource/power.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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))) From e135f9d81f7fcc1c9b9f1c3ccc209f6f6820c6db Mon Sep 17 00:00:00 2001 From: Trevor Gamblin Date: Fri, 11 Jul 2025 13:24:29 -0400 Subject: [PATCH 3/4] doc/configuration: update NetworkPowerPort - Document that 'index' is now optional - Change 'tplink' section to indicate it controls both smart plugs and power strips Signed-off-by: Trevor Gamblin --- doc/configuration.rst | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/doc/configuration.rst b/doc/configuration.rst index c85cc7a0c..0fcedb896 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`` From fe83e9552268ee1d39b470b44867af6c6215ba30 Mon Sep 17 00:00:00 2001 From: Trevor Gamblin Date: Fri, 11 Jul 2025 14:42:11 -0400 Subject: [PATCH 4/4] TestNetworkPowerDriver: add test_default_index This is similar test_create, but it doesn't pass an index value, testing that the default index of the NetworkPowerPort object is '0' and also that the NetworkPowerDriver can still be instantiated correctly. Signed-off-by: Trevor Gamblin --- tests/test_powerdriver.py | 6 ++++++ 1 file changed, 6 insertions(+) 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',