diff --git a/doc/configuration.rst b/doc/configuration.rst index eabb44155..4b91d68f1 100644 --- a/doc/configuration.rst +++ b/doc/configuration.rst @@ -239,6 +239,13 @@ Currently available are: `__ for details. +``tapoplug`` + Controls *Tapo Smart Plugs* via `python-kasa + `_. + Requires *KASA_USERNAME* and *KASA_USERNAME* environment variables to be set. + Tested on *Tapo P100* smart plug. + See the `list of supported TAPO devices `_. + ``tplink`` Controls *TP-Link power strips* via `python-kasa `_. diff --git a/labgrid/driver/power/tapoplug.py b/labgrid/driver/power/tapoplug.py new file mode 100644 index 000000000..0897cb7ff --- /dev/null +++ b/labgrid/driver/power/tapoplug.py @@ -0,0 +1,77 @@ +"""Tested with TAPO P100, and should be compatible with any TAPO smart plug supported by kasa + +There is a list of supported devices in python-kasa package official documentation. +https://python-kasa.readthedocs.io/en/stable/SUPPORTED.html#tapo-devices + +""" + +import asyncio +import os + +from kasa import ( + Credentials, + DeviceConfig, + DeviceConnectionParameters, + DeviceEncryptionType, + DeviceFamily, + SmartProtocol, + smart, + transports, +) + +connection_type = DeviceConnectionParameters( + device_family=DeviceFamily.SmartTapoPlug, encryption_type=DeviceEncryptionType.Klap, login_version=2 +) + + +def get_credentials(): + username = os.environ.get("KASA_USERNAME") + password = os.environ.get("KASA_PASSWORD") + if username is None or password is None: + raise ValueError( + "Username and password cannot be None.Set KASA_USERNAME and KASA_PASSWORD environment variables" + ) + return Credentials(username=username, password=password) + + +async def _power_set(host, port, index, value): + assert port is None + config = DeviceConfig(host=host, credentials=get_credentials(), connection_type=connection_type) + protocol = SmartProtocol(transport=transports.KlapTransportV2(config=config)) + + device = smart.SmartDevice(host=host, config=config, protocol=protocol) + + try: + await device.update() + + if value: + await device.turn_on() + else: + await device.turn_off() + except Exception as e: + print(f"An error occurred while setting power: {e}") + finally: + await device.disconnect() + + +def power_set(host, port, index, value): + asyncio.run(_power_set(host, port, index, value)) + + +async def _power_get(host, port, index): + assert port is None + config = DeviceConfig(host=host, credentials=get_credentials(), connection_type=connection_type) + protocol = SmartProtocol(transport=transports.KlapTransportV2(config=config)) + + device = smart.SmartDevice(host=host, config=config, protocol=protocol) + try: + await device.update() + return device.is_on + except Exception as e: + print(f"An error occurred while getting power: {e}") + finally: + await device.disconnect() + + +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..77e1bc4eb 100644 --- a/labgrid/resource/power.py +++ b/labgrid/resource/power.py @@ -16,7 +16,7 @@ class NetworkPowerPort(Resource): """ 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 0dfff1547..5ac665d36 100644 --- a/tests/test_powerdriver.py +++ b/tests/test_powerdriver.py @@ -299,6 +299,10 @@ def test_import_backend_tplink(self): pytest.importorskip("kasa") import labgrid.driver.power.tplink + def test_import_backend_tapoplug(self): + pytest.importorskip("kasa") + import labgrid.driver.power.tapoplug + def test_import_backend_siglent(self): pytest.importorskip("vxi11") import labgrid.driver.power.siglent