Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2025-05-16 19:41:09 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Fixes indentation and removes extra whitespaces

Browse Source
This commit is contained in:
Georgi Kirichkov 2016-10-18 10:59:30 +03:00
parent 3744a075c3
commit 605abfdebb
1 changed files with 134 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

268
pyHS100/pyHS100.py
View File

@ -1,4 +1,4 @@
# Parts of this code reuse code and concepts by Lubomir Stroetmann from softScheck GmbH # Parts of this code reuse code and concepts by Lubomir Stroetmann from softScheck GmbH
# licensed under the Apache License v 2.0. # licensed under the Apache License v 2.0.
# Copy of the Apache License can be found at http://www.apache.org/licenses/LICENSE-2.0 # Copy of the Apache License can be found at http://www.apache.org/licenses/LICENSE-2.0
# The code from Lubomir Stroetmann is located at http://github.com/softScheck/tplink-smartplug # The code from Lubomir Stroetmann is located at http://github.com/softScheck/tplink-smartplug
@ -55,10 +55,10 @@ class SmartPlug(object):
:param value: Future state (either ON or OFF) :param value: Future state (either ON or OFF)
""" """
if value.upper() == 'ON': if value.upper() == 'ON':
self.turn_on() self.turn_on()
elif value.upper() == 'OFF': elif value.upper() == 'OFF':
self.turn_off() self.turn_off()
else: else:
raise TypeError("State %s is not valid." % str(value)) raise TypeError("State %s is not valid." % str(value))
@ -94,84 +94,84 @@ class SmartPlug(object):
return relay_state return relay_state
def get_info(self): def get_info(self):
"""Interrogate the switch""" """Interrogate the switch"""
return self._send_command('{"system":{"get_sysinfo":{}}}') return self._send_command('{"system":{"get_sysinfo":{}}}')
def turn_on(self): def turn_on(self):
"""Turns the switch on """Turns the switch on
Return values: Return values:
True on success True on success
False on failure False on failure
""" """
response = self._send_command('{"system":{"set_relay_state":{"state":1}}}') response = self._send_command('{"system":{"set_relay_state":{"state":1}}}')
if response["system"]["set_relay_state"]["err_code"] == 0: if response["system"]["set_relay_state"]["err_code"] == 0:
return True return True
return False return False
def turn_off(self): def turn_off(self):
"""Turns the switch off """Turns the switch off
Return values: Return values:
True on success True on success
False on failure False on failure
""" """
response = self._send_command('{"system":{"set_relay_state":{"state":0}}}') response = self._send_command('{"system":{"set_relay_state":{"state":0}}}')
if response["system"]["set_relay_state"]["err_code"] == 0: if response["system"]["set_relay_state"]["err_code"] == 0:
return True return True
return False
return False
def get_emeter_realtime(self): def get_emeter_realtime(self):
"""Gets the current energy readings from the switch """Gets the current energy readings from the switch
Return values: Return values:
False if command is not successful or the switch doesn't support energy metering False if command is not successful or the switch doesn't support energy metering
Dict with the current readings Dict with the current readings
""" """
if self.model == 100: if self.model == 100:
return False return False
response = self._send_command('{"emeter":{"get_realtime":{}}}') response = self._send_command('{"emeter":{"get_realtime":{}}}')
if response["emeter"]["get_realtime"]["err_code"] != 0: if response["emeter"]["get_realtime"]["err_code"] != 0:
return False return False
response["emeter"]["get_realtime"].pop('err_code', None) response["emeter"]["get_realtime"].pop('err_code', None)
return response["emeter"]["get_realtime"] return response["emeter"]["get_realtime"]
def get_emeter_daily(self, year = datetime.datetime.now().year, month = datetime.datetime.now().month): def get_emeter_daily(self, year=datetime.datetime.now().year, month=datetime.datetime.now().month):
"""Gets daily statistics for a given month. """Gets daily statistics for a given month.
Arguments:
year (optional): The year for which to retrieve statistics, defaults to current year
month (optional): The mont for which to retrieve statistics, defaults to current month
Return values: Arguments:
False if command is not successful or the switch doesn't support energy metering year (optional): The year for which to retrieve statistics, defaults to current year
Dict where the keys represent the days, and the values are the aggregated statistics month (optional): The mont for which to retrieve statistics, defaults to current month
"""
if self.model == 100:
return False
response = self._send_command('{"emeter":{"get_daystat":{"month":' + str(month) + ',"year":' + str(year) + '}}}') Return values:
False if command is not successful or the switch doesn't support energy metering
Dict where the keys represent the days, and the values are the aggregated statistics
"""
if self.model == 100:
return False
if response["emeter"]["get_daystat"]["err_code"] != 0: response = self._send_command('{"emeter":{"get_daystat":{"month":' + str(month) + ',"year":' + str(year) + '}}}')
return False
data = dict() if response["emeter"]["get_daystat"]["err_code"] != 0:
return False
for i, j in enumerate(response["emeter"]["get_daystat"]["day_list"]): data = dict()
if j["energy"] > 0:
data[j["day"]] = j["energy"]
return data for i, j in enumerate(response["emeter"]["get_daystat"]["day_list"]):
if j["energy"] > 0:
data[j["day"]] = j["energy"]
def get_emeter_monthly(self, year = datetime.datetime.now().year): return data
"""Gets monthly statistics for a given year.
def get_emeter_monthly(self, year=datetime.datetime.now().year):
"""Gets monthly statistics for a given year.
Arguments: Arguments:
year (optional): The year for which to retrieve statistics, defaults to current year year (optional): The year for which to retrieve statistics, defaults to current year
@ -179,110 +179,110 @@ class SmartPlug(object):
Return values: Return values:
False if command is not successful or the switch doesn't support energy metering False if command is not successful or the switch doesn't support energy metering
Dict - the keys represent the months, the values are the aggregated statistics Dict - the keys represent the months, the values are the aggregated statistics
""" """
if self.model == 100: if self.model == 100:
return False return False
response = self._send_command('{"emeter":{"get_monthstat":{"year":' + str(year) + '}}}') response = self._send_command('{"emeter":{"get_monthstat":{"year":' + str(year) + '}}}')
if response["emeter"]["get_monthstat"]["err_code"] != 0: if response["emeter"]["get_monthstat"]["err_code"] != 0:
return False return False
data = dict() data = dict()
for i, j in enumerate(response["emeter"]["get_monthstat"]["month_list"]): for i, j in enumerate(response["emeter"]["get_monthstat"]["month_list"]):
if j["energy"] > 0: if j["energy"] > 0:
data[j["month"]] = j["energy"] data[j["month"]] = j["energy"]
return data
return data
def erase_emeter_stats(self): def erase_emeter_stats(self):
"""Erases all statistics. """Erases all statistics.
Return values: Return values:
True: Success True: Success
False: Failure or not supported by switch False: Failure or not supported by switch
""" """
if self.model == 100: if self.model == 100:
return False return False
response = self._send_command('{"emeter":{"erase_emeter_stat":null}}') response = self._send_command('{"emeter":{"erase_emeter_stat":null}}')
if response["emeter"]["erase_emeter_stat"]["err_code"] != 0: if response["emeter"]["erase_emeter_stat"]["err_code"] != 0:
return False return False
else: else:
return True return True
def current_consumption(self): def current_consumption(self):
"""Get the current power consumption in Watt.""" """Get the current power consumption in Watt."""
response = self.get_emeter_realtime() response = self.get_emeter_realtime()
return response["power"] return response["power"]
def _encrypt(self, string): def _encrypt(self, string):
"""Encrypts a command.""" """Encrypts a command."""
""" """
Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py
Changes: the return value is encoded in latin-1 in Python 3 and later Changes: the return value is encoded in latin-1 in Python 3 and later
""" """
key = 171 key = 171
result = "\0\0\0\0" result = "\0\0\0\0"
for i in string: for i in string:
a = key ^ ord(i) a = key ^ ord(i)
key = a key = a
result += chr(a) result += chr(a)
if sys.version_info.major > 2: if sys.version_info.major > 2:
return result.encode('latin-1') return result.encode('latin-1')
return result return result
def _decrypt(self, string): def _decrypt(self, string):
"""Decrypts a command.""" """Decrypts a command."""
""" """
Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py Taken from https://raw.githubusercontent.com/softScheck/tplink-smartplug/master/tplink-smartplug.py
Changes: the string parameter is decoded from latin-1 in Python 3 and later Changes: the string parameter is decoded from latin-1 in Python 3 and later
""" """
if sys.version_info.major > 2: if sys.version_info.major > 2:
string = string.decode('latin-1') string = string.decode('latin-1')
key = 171 key = 171
result = "" result = ""
for i in string: for i in string:
a = key ^ ord(i) a = key ^ ord(i)
key = ord(i) key = ord(i)
result += chr(a) result += chr(a)
return result return result
def _send_command(self, command): def _send_command(self, command):
"""Sends a command to the switch. """Sends a command to the switch.
Accepts one argument - the command as a string
Return values:
The decrypted JSON
"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((self.ip, self.port))
s.send(self._encrypt(command))
response = self._decrypt(s.recv(4096)[4:])
s.shutdown(1)
s.close()
return json.loads(response) Accepts one argument - the command as a string
Return values:
The decrypted JSON
"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((self.ip, self.port))
s.send(self._encrypt(command))
response = self._decrypt(s.recv(4096)[4:])
s.shutdown(1)
s.close()
return json.loads(response)
def _identify_model(self): def _identify_model(self):
"""Query sysinfo and determine model""" """Query sysinfo and determine model"""
sys_info = self.get_info() sys_info = self.get_info()
if sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS100': if sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS100':
model = 100 model = 100
elif sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS110': elif sys_info["system"]["get_sysinfo"]["model"][:5] == 'HS110':
model = 110 model = 110
return model return model