From aea05752d885a04c5b348de2435a9feb2c78960f Mon Sep 17 00:00:00 2001 From: anbukannadhasan <154507930+Anbukannadhasan@users.noreply.github.com> Date: Thu, 6 Nov 2025 09:54:49 +0000 Subject: [PATCH 1/2] gh #192 updated the tapo multi power switch changes --- framework/core/powerModules/tapoControl.py | 93 +++++++++++++--------- requirements.txt | 6 +- 2 files changed, 59 insertions(+), 40 deletions(-) diff --git a/framework/core/powerModules/tapoControl.py b/framework/core/powerModules/tapoControl.py index d16866e..022280d 100644 --- a/framework/core/powerModules/tapoControl.py +++ b/framework/core/powerModules/tapoControl.py @@ -48,7 +48,6 @@ import json import re import subprocess -import time from framework.core.logModule import logModule from framework.core.powerModules.abstractPowerModule import PowerModuleInterface @@ -75,7 +74,7 @@ def __init__( self, log:logModule, ip:str, outlet:str = None, **kwargs ): self.ip = ip self._username = kwargs.get("username", None) self._password = kwargs.get("password", None) - if outlet: + if outlet is not None: self._outlet=str(outlet) self._device_type = None self._encryption_type = None @@ -111,7 +110,7 @@ def _performCommand(self, command, json = False, append_args:list = []): command_list.append("--encrypt-type") command_list.append(self._encryption_type) else: - if self._outlet: + if self._outlet is not None: command_list.append("--type") command_list.append("strip") else: @@ -133,16 +132,16 @@ def powerOff(self): bool: True if the operation is successful, False otherwise. """ self._get_state() - if self.is_off: + if not self._is_on: return True - if self._outlet: + if self._outlet is not None: self._performCommand("off", append_args=["--index", str(self._outlet)]) else: self._performCommand("off") self._get_state() - if self.is_off == False: + if self._is_on: self._log.error(" Power Off Failed") - return self.is_off + return not self._is_on def powerOn(self): """ @@ -152,21 +151,33 @@ def powerOn(self): bool: True if the operation is successful, False otherwise. """ self._get_state() - if self.is_on: + if self._is_on: return True - if self._outlet: + if self._outlet is not None: self._performCommand("on", append_args=["--index", str(self._outlet)]) - self._performCommand("on") + else: + self._performCommand("on") self._get_state() - if self.is_on == False: + if self._is_on == False: self._log.error(" Power On Failed") - return self.is_on + return self._is_on def _get_state(self): """Get the state of the device. """ result = self._performCommand("state") - if self._outlet: + if self._outlet is not None: + # == Children == + # + # == Smart Plug 1 (P304M) == + # == Primary features == + # State(state): True + if result.find('Children') > 1: # smart extension plug with multiple outlets + all_states = re.findall(r"^\s*State\s*\(state\)\s*:\s*(True|False)\s*$", + result, flags=re.IGNORECASE | re.MULTILINE) + self._is_on = all_states[int(self._outlet)] == 'True' + self._log.debug(f"Slot state: {'ON' if self._is_on else 'OFF'}") + return # We have a strip look at the status of the strip, and check the index and the device state #Device state: ON #== Plugs == @@ -233,6 +244,12 @@ def _discover_device(self): self._device_type = info.get("mic_type", "UNKNOWN") else: self._device_type = "UNKNOWN" + elif result.get("get_child_device_list", {}).get('child_device_list', []): + child_devices = result.get("get_child_device_list", {}).get('child_device_list', []) + if len(child_devices) >= int(self._outlet) + 1: + self._device_type = child_devices[int(self._outlet)].get("type", "UNKNOWN") + elif result.get('get_device_info'): + self._device_type = result.get("get_device_info").get("type", "UNKNOWN") else: self._device_type = "UNKNOWN" self._encryption_type = self._get_encryption_type() @@ -249,28 +266,30 @@ def _get_encryption_type(self): return None def getPowerLevel(self): - if self._outlet: - # TODO: implement this for a powerstrip - # result = self._performCommand("emeter", - # json=True, - # append_args=["--index", str(self._outlet)]) - raise RuntimeError("Power monitoring is not yet supported for Tapo strips") + if self._outlet is not None: + args = [ + "--module", 'energy', 'get_current_power', + "--index", self._outlet + ] else: - result = self._performCommand("emeter", json=True) - - if not result: - raise ValueError("Received empty response from Tapo device for power monitoring") - - try: - result = json.loads(result) - except json.JSONDecodeError as e: - raise ValueError(f"Failed to parse JSON from Tapo device response: {e}") - - millewatt = result.get('power_mw') - if millewatt: - try: - power = int(millewatt) / 1000 - return power - except: - raise ValueError(f"Invalid value for power_mw: {millewatt}") - raise KeyError("The dictionary returned by the Tapo device does not contain a valid 'power_mw' value.") + args = [ + "--module", 'energy', 'get_current_power' + ] + result = self._performCommand("command", + json=True, + append_args=args + ) + + if not result: + raise ValueError("Received empty response from Tapo device for power monitoring") + + try: + result = json.loads(result) + except json.JSONDecodeError as e: + raise ValueError(f"Failed to parse JSON from Tapo device response: {e}") + + watt = result.get("get_current_power", {}).get('current_power', None) + if watt is not None: + return watt + + raise KeyError("The dictionary returned by the Tapo device does not contain a valid 'power_mw' value.") diff --git a/requirements.txt b/requirements.txt index a16f9b5..6bf9776 100644 --- a/requirements.txt +++ b/requirements.txt @@ -34,7 +34,7 @@ jmespath==1.0.1 marshmallow==3.21.1 multidict==6.0.5 netifaces==0.11.0 -numpy==1.26.4 +numpy>=2.0.0,<2.3.0 opencv-python==4.9.0.80 outcome==1.3.0.post0 packaging==24.0 @@ -49,7 +49,7 @@ pyserial==3.5 PySocks==1.7.1 pytesseract==0.3.10 python-dateutil==2.9.0.post0 -python-kasa==0.6.2.1 +python-kasa==0.7.7 PyYAML==6.0.1 requests==2.31.0 requests-toolbelt==1.0.0 @@ -61,7 +61,7 @@ sortedcontainers==2.4.0 soupsieve==2.5 trio==0.25.0 trio-websocket==0.11.1 -typing_extensions==4.10.0 +typing_extensions==4.12.2 urllib3==1.26.18 wcwidth==0.2.14 wrapt==1.16.0 From f8f7d9f8a0615b1c6e8634c11c5bf35d5ad58a66 Mon Sep 17 00:00:00 2001 From: anbukannadhasan <154507930+Anbukannadhasan@users.noreply.github.com> Date: Tue, 11 Nov 2025 09:16:08 +0000 Subject: [PATCH 2/2] gh #192 updated the co pilot review suggestions --- framework/core/powerModules/tapoControl.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/framework/core/powerModules/tapoControl.py b/framework/core/powerModules/tapoControl.py index 022280d..d81d2b2 100644 --- a/framework/core/powerModules/tapoControl.py +++ b/framework/core/powerModules/tapoControl.py @@ -158,7 +158,7 @@ def powerOn(self): else: self._performCommand("on") self._get_state() - if self._is_on == False: + if not self._is_on: self._log.error(" Power On Failed") return self._is_on @@ -172,7 +172,7 @@ def _get_state(self): # == Smart Plug 1 (P304M) == # == Primary features == # State(state): True - if result.find('Children') > 1: # smart extension plug with multiple outlets + if 'Children' in result: # smart extension plug with multiple outlets all_states = re.findall(r"^\s*State\s*\(state\)\s*:\s*(True|False)\s*$", result, flags=re.IGNORECASE | re.MULTILINE) self._is_on = all_states[int(self._outlet)] == 'True' @@ -246,7 +246,7 @@ def _discover_device(self): self._device_type = "UNKNOWN" elif result.get("get_child_device_list", {}).get('child_device_list', []): child_devices = result.get("get_child_device_list", {}).get('child_device_list', []) - if len(child_devices) >= int(self._outlet) + 1: + if len(child_devices) > int(self._outlet): self._device_type = child_devices[int(self._outlet)].get("type", "UNKNOWN") elif result.get('get_device_info'): self._device_type = result.get("get_device_info").get("type", "UNKNOWN") @@ -292,4 +292,4 @@ def getPowerLevel(self): if watt is not None: return watt - raise KeyError("The dictionary returned by the Tapo device does not contain a valid 'power_mw' value.") + raise KeyError("The dictionary returned by the Tapo device does not contain a valid 'current_power' value.")