Domoticz

Code: Select all

#!/usr/bin/env python
# 
# TP-Link Wi-Fi Smart Plug Protocol Client
# For use with TP-Link HS110: energy monitor
# 
# Gets current power (W) and cumulative energy (kWh)
# and sends to Domoticz

import socket
import argparse
import json
import urllib
import urllib2

# ip, port, and command for HS110
ip = '10.1.1.5'
port = 9999
cmd = '{"emeter":{"get_realtime":{}}}'

# Domoticz command stub and IDx of HS110
baseURL = 'http://<10.1.1.5:8080>/json.htm?type=command&param=udevice&nvalue=0'
HSIdx = 2

# Encryption and Decryption of TP-Link Smart Home Protocol
# XOR Autokey Cipher with starting key = 171
def encrypt(string):
	key = 171
	result = "\0\0\0\0"
	for i in string: 
		a = key ^ ord(i)
		key = a
		result += chr(a)
	return result

def decrypt(string):
	key = 171 
	result = ""
	for i in string: 
		a = key ^ ord(i)
		key = ord(i) 
		result += chr(a)
	return result

def domoticzrequest (url):
   request = urllib2.Request(url)
#   request.add_header("Authorization", "Basic %s" % base64string)
   response = urllib2.urlopen(request)
   return None;

# Send command and receive reply 
try:
	sock_tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	sock_tcp.connect((ip, port))
	sock_tcp.send(encrypt(cmd))
	data = sock_tcp.recv(2048)
	sock_tcp.close()
	
#	print "Sent:     ", cmd
	result = decrypt(data[4:])
	jsonData = json.loads(result)
#	print "Received: "
#	print json.dumps(jsonData, indent=4, sort_keys=True)
	power = jsonData['emeter']['get_realtime']['power']
	total = jsonData['emeter']['get_realtime']['total'] * 1000
#	print power, total
except socket.error:
	quit("Cound not connect to host " + ip + ":" + str(port))

# Send data to Domoticz
try:
    url = baseURL + "&idx=%s&svalue=%s;%s" % (HSIdx, power, total)
    domoticzrequest(url)
except urllib2.URLError, e:
	print e.code
	

Code: Select all

./HS100.sh 192.168.0.84 9999 on
found 0 associations
found 1 connections:
     1:	flags=82<CONNECTED,PREFERRED>
	outif en0
	src 192.168.0.10 port 63787
	dst 192.168.0.84 port 9999
	rank info not available
	TCP aux info available

Connection to 192.168.0.84 port 9999 [tcp/distinct] succeeded!

Code: Select all

#!/usr/bin/python3
#
#################################################################################################
#################################################################################################
#
# TP-Link Wi-Fi Smart Plug Protocol Energy Monitoring Client
# For use with TP-Link HS-110
#
#
# Additional Information:
#
#      https://www.domoticz.com/forum/viewtopic.php?f=56&t=13290
#      https://www.softscheck.com/en/reverse-engineering-tp-link-hs110/
#
# Complete URL String to Send to Domoticz
#      http://DOMSERVER_IP:8080/json.htm?type=command&param=udevice&nvalue=0&idx=IDX&svalue=POWER;TOTAL
#
#################################################################################################
#################################################################################################

import socket
import argparse
import json
import urllib
import urllib.request

version = 0.1

# Check if IP is valid
def validIP(ip):
        try:
                socket.inet_pton(socket.AF_INET, ip)
        except socket.error:
                parser.error("Invalid IP Address.")
        return ip

# Predefined Smart Plug Commands
# For a full list of commands, consult tplink_commands.txt
commands = {'info'     : '{"system":{"get_sysinfo":{}}}',
                        'on'       : '{"system":{"set_relay_state":{"state":1}}}',
                        'off'      : '{"system":{"set_relay_state":{"state":0}}}',
                        'cloudinfo': '{"cnCloud":{"get_info":{}}}',
                        'wlanscan' : '{"netif":{"get_scaninfo":{"refresh":0}}}',
                        'time'     : '{"time":{"get_time":{}}}',
                        'schedule' : '{"schedule":{"get_rules":{}}}',
                        'countdown': '{"count_down":{"get_rules":{}}}',
                        'antitheft': '{"anti_theft":{"get_rules":{}}}',
                        'reboot'   : '{"system":{"reboot":{"delay":1}}}',
                        'reset'    : '{"system":{"reset":{"delay":1}}}',
                        'energy'    : '{"emeter":{"get_realtime":{}}}'
}

# Encryption and Decryption of TP-Link Smart Home Protocol
# XOR Autokey Cipher with starting key = 171
def encrypt(string):
    key = 171
    result = b"\0\0\0"+ chr(len(string)).encode('latin-1')
    for i in string.encode('latin-1'):
        a = key ^ i
        key = a
        result += chr(a).encode('latin-1')
    return result


def decrypt(string):
    key = 171
    result = ""
    for i in string:
        a = key ^ i
        key = i
        result += chr(a)
    return result

# Parse commandline arguments
parser = argparse.ArgumentParser(description="TP-Link Wi-Fi Smart Plug Client v" + str(version))
parser.add_argument("-t", "--target", metavar="<ip>", required=True, help="Target IP Address", type=validIP)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("-c", "--command", metavar="<command>", help="Preset command to send. Choices are: "+", ".join(commands), choices=commands)
group.add_argument("-j", "--json", metavar="<JSON string>", help="Full JSON string of command to send")
args = parser.parse_args()

# Domoticz command stub and IDx of HS110
baseURL = 'http://<DOMOTICZ IP>:8080/json.htm?type=command&param=udevice&nvalue=0'
HSIdx = <IDX from Domoticz>

# Set target IP, port and command to send
ip = args.target
port = 9999
if args.command is None:
        cmd = args.json
else:
        cmd = commands[args.command]

# Send command and receive reply
try:
        sock_tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	sock_tcp.connect((ip, port))
        sock_tcp.send(encrypt(cmd))
        data = sock_tcp.recv(2048)
        sock_tcp.close()

        cmdresult = decrypt(data[4:])
        jsonData = json.loads(cmdresult)
        power = jsonData['emeter']['get_realtime']['power_mw']
        total = jsonData['emeter']['get_realtime']['total_wh'] * 1000
        url = baseURL + "&idx=%s&svalue=%s;%s" % (HSIdx, power, total)
        domreq = urllib.request.Request(url)

# Troubleshooting Print Commands
#       print (json.dumps(jsonData, indent=4,sort_keys=True))
#       print(power,total)
#       print(url)
#       print("Sent:     ", cmd)
#       print("Received: ", decrypt(data[4:]))

except socket.error:
        quit("Cound not connect to host " + ip + ":" + str(port))

try:
        urllib.request.urlopen(domreq)
except Exception as e:
        print(str(e))

Code: Select all

#!/usr/bin/env python3
#
# TP-Link Wi-Fi Smart Plug Protocol Client
# For use with TP-Link HS-100 or HS-110
#  
# by Lubomir Stroetmann
# Copyright 2016 softScheck GmbH 
# 
# modified by Kawa
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# 
#      http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# 
#
import socket
import argparse

version = 0.5

# Check if IP is valid
def validIP(ip):
	try:
		socket.inet_pton(socket.AF_INET, ip)
	except socket.error:
		parser.error("Invalid IP Address.")
	return ip 

# Predefined Smart Plug Commands
# For a full list of commands, consult tplink_commands.txt
commands = {'info'     : '{"system":{"get_sysinfo":{}}}',
			'on'       : '{"system":{"set_relay_state":{"state":1}}}',
			'off'      : '{"system":{"set_relay_state":{"state":0}}}',
			'cloudinfo': '{"cnCloud":{"get_info":{}}}',
			'wlanscan' : '{"netif":{"get_scaninfo":{"refresh":0}}}',
			'time'     : '{"time":{"get_time":{}}}',
			'schedule' : '{"schedule":{"get_rules":{}}}',
			'countdown': '{"count_down":{"get_rules":{}}}',
			'antitheft': '{"anti_theft":{"get_rules":{}}}',
			'reboot'   : '{"system":{"reboot":{"delay":1}}}',
			'reset'    : '{"system":{"reset":{"delay":1}}}'
}

# Encryption and Decryption of TP-Link Smart Home Protocol
# XOR Autokey Cipher with starting key = 171
def encrypt(string):
	key = 171
	result = b"\0\0\0"+ chr(len(string)).encode('utf-8')
	for i in string.encode('utf-8'):
		a = key ^ i
		key = a
		result += chr(a).encode('utf-8')
	return result

def decrypt(string):
	key = 171 
	result = ""
	for i in string: 
		a = key ^ i
		key = i 
		result += chr(a)
	return result

# Parse commandline arguments
parser = argparse.ArgumentParser(description="TP-Link Wi-Fi Smart Plug Client v" + str(version))
parser.add_argument("-t", "--target", metavar="<ip>", required=True, help="Target IP Address", type=validIP)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("-c", "--command", metavar="<command>", help="Preset command to send. Choices are: "+", ".join(commands), choices=commands) 
group.add_argument("-j", "--json", metavar="<JSON string>", help="Full JSON string of command to send")
args = parser.parse_args()

# Set target IP, port and command to send
ip = args.target
port = 9999
if args.command is None:
	cmd = args.json
else:
	cmd = commands[args.command]



# Send command and receive reply 
try:
	sock_tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	sock_tcp.connect((ip, port))
	sock_tcp.send(encrypt(cmd))
	data = sock_tcp.recv(2048)
	sock_tcp.close()
	
#	print ("Sent:     ", cmd)
#	print ("Received: ", decrypt(data[4:]))
except socket.error:
	quit("Cound not connect to host " + ip + ":" + str(port))

Code: Select all

script://tplink-smartplug.py -t 192.168.1.32 -c on

Code: Select all

script://tplink-smartplug.py -t 192.168.1.32 -c off

Code: Select all

#!/usr/bin/env python3
#
# TP-Link Wi-Fi Smart Plug Protocol Client
# For use with TP-Link HS100
#
# modified by Kawa
#
# Gets current switch status
# and sends to Domoticz

import socket
import argparse
import json
import urllib.request, urllib.parse, urllib.error
import urllib.request, urllib.error, urllib.parse


# Check if IP is valid
def validIP(ip):
	try:
		socket.inet_pton(socket.AF_INET, ip)
	except socket.error:
		parser.error("Invalid IP Address.")
	return ip 


parser = argparse.ArgumentParser(description="TP-Link Wi-Fi Smart Plug Client")
parser.add_argument("-t", "--target", metavar="<ip>", required=True, help="Target IP Address", type=validIP)
parser.add_argument("-i", "--idx", metavar="<idx>", help="Which Domoticz IDX")
parser.add_argument("-d", "--domoticz", metavar="<Domoticz IP>", help="Full IP address with port number for Domoticz")
args = parser.parse_args()


# ip, port, and command for HS100
ip = args.target
port = 9999
cmd = '{"system":{"get_sysinfo":{}}}'
idx = args.idx
domoticzIP = args.domoticz

# Domoticz command stub and IDx of HS100
baseURL = 'http://' + domoticzIP + '/json.htm?type=command&param=udevice&idx=' + str(idx) + '&nvalue='

# Encryption and Decryption of TP-Link Smart Home Protocol
# XOR Autokey Cipher with starting key = 171
def encrypt(string):
	key = 171
	result = b"\0\0\0"+ chr(len(string)).encode('latin-1')
	for i in string.encode('latin-1'):
		a = key ^ i
		key = a
		result += chr(a).encode('latin-1')
	return result

def decrypt(string):
	key = 171 
	result = ""
	for i in string:
		a = key ^ i
		key = i 
		result += chr(a)
	return result

def domoticzrequest (url):
   request = urllib.request.Request(url)
   response = urllib.request.urlopen(request)
   result = response.read()
   return result;

def getcurrentstate():
   url = 'http://' + domoticzIP + '/json.htm?type=devices&rid=' + str(idx)
   response = domoticzrequest(url)
   jsonData = json.loads(response)
   status = jsonData['result'][0]['Status']
   lampState = 0
   if status == "On":
      lampState = 1

   return lampState;

# Send command and receive reply
try:
   sock_tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
   sock_tcp.connect((ip, port))
   sock_tcp.send(encrypt(cmd))
   data = sock_tcp.recv(2048)
   sock_tcp.close()

   result = decrypt(data[4:])
   jsonData = json.loads(result)
   relay_state = jsonData['system']['get_sysinfo']['relay_state']
except socket.error:
   quit("Cound not connect to host " + ip + ":" + str(port))

# Send data to Domoticz
try:
   currentState = getcurrentstate()
   if currentState != relay_state:
      url = baseURL + str(relay_state)
      domoticzrequest(url)
      
except urllib.error.URLError as e:
   print((e.code))

Code: Select all

*       *       *       *       *       /config/scripts/tplink-status.py -t 192.168.1.32 -i 5 -d 192.168.1.50:8080 > /var/log/tplink.log

Code: Select all

./tplink-smartplug.py -t 192.168.1.165 -j '{"emeter":{"get_realtime":{}}}'

Code: Select all

Sent:      {"emeter":{"get_realtime":{}}}
Received:

Code: Select all

./tplink-smartplug.py -t 192.168.1.165 -c on

Code: Select all

Sent:      {"system":{"set_relay_state":{"state":1}}}
Received:

Code: Select all

./tplink-status.py -t 192.168.1.165 -i 1891 -d 192.168.1.150:8080

Code: Select all

Traceback (most recent call last):
  File "./tplink-status.py", line 97, in <module>
    currentState = getcurrentstate()
  File "./tplink-status.py", line 73, in getcurrentstate
    jsonData = json.loads(response)
  File "/usr/lib/python3.5/json/__init__.py", line 312, in loads
    s.__class__.__name__))
TypeError: the JSON object must be str, not 'bytes'

Code: Select all

"""
Support for TPLink HS100/HS110/HS200 smart switch.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/switch.tplink/
"""
import logging
import time

import voluptuous as vol

from homeassistant.components.switch import (
    SwitchDevice, PLATFORM_SCHEMA, ATTR_CURRENT_POWER_W, ATTR_TODAY_ENERGY_KWH)
from homeassistant.const import (CONF_HOST, CONF_NAME, ATTR_VOLTAGE)
import homeassistant.helpers.config_validation as cv

REQUIREMENTS = ['pyHS100==0.3.2']

_LOGGER = logging.getLogger(__name__)

ATTR_TOTAL_ENERGY_KWH = 'total_energy_kwh'
ATTR_CURRENT_A = 'current_a'

CONF_LEDS = 'enable_leds'

DEFAULT_NAME = 'TP-Link Switch'

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_HOST): cv.string,
    vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
    vol.Optional(CONF_LEDS): cv.boolean,
})


# pylint: disable=unused-argument
def setup_platform(hass, config, add_devices, discovery_info=None):
    """Set up the TPLink switch platform."""
    from pyHS100 import SmartPlug
    host = config.get(CONF_HOST)
    name = config.get(CONF_NAME)
    leds_on = config.get(CONF_LEDS)

    add_devices([SmartPlugSwitch(SmartPlug(host), name, leds_on)], True)


class SmartPlugSwitch(SwitchDevice):
    """Representation of a TPLink Smart Plug switch."""

    def __init__(self, smartplug, name, leds_on):
        """Initialize the switch."""
        self.smartplug = smartplug
        self._name = name
        if leds_on is not None:
            self.smartplug.led = leds_on
        self._state = None
        self._available = True
        # Set up emeter cache
        self._emeter_params = {}

    @property
    def name(self):
        """Return the name of the Smart Plug, if any."""
        return self._name

    @property
    def available(self) -> bool:
        """Return if switch is available."""
        return self._available

    @property
    def is_on(self):
        """Return true if switch is on."""
        return self._state

    def turn_on(self, **kwargs):
        """Turn the switch on."""
        self.smartplug.turn_on()

    def turn_off(self):
        """Turn the switch off."""
        self.smartplug.turn_off()

    @property
    def device_state_attributes(self):
        """Return the state attributes of the device."""
        return self._emeter_params

    def update(self):
        """Update the TP-Link switch's state."""
        from pyHS100 import SmartDeviceException
        try:
            self._available = True
            self._state = self.smartplug.state == \
                self.smartplug.SWITCH_STATE_ON

            if self._name is None:
                self._name = self.smartplug.alias

            if self.smartplug.has_emeter:
                emeter_readings = self.smartplug.get_emeter_realtime()
                for value in emeter_readings :
                  try :
                      name, unit = value.split("_",1)
                      div = 1000
                  except ValueError:
                      name = value.split("_",1)[0]
                      div = 1

                  if name == "power":
                      self._emeter_params[ATTR_CURRENT_POWER_W] \
                          = "%.1f W" % (float(emeter_readings[value])/div)
                  if name == "total":
                      self._emeter_params[ATTR_TODAY_ENERGY_KWH] \
                          = "%.2f kW" % (float(emeter_readings[value])/div)
                  if name == "voltage":
                      self._emeter_params[ATTR_VOLTAGE] \
                          = "%.2f V" % (float(emeter_readings[value])/div)
                  if name == "current":
                      self._emeter_params[ATTR_CURRENT_A] \
                          = "%.1f A" % (float(emeter_readings[value])/div)

                try:
                    emeter_statics = self.smartplug.get_emeter_daily()
                    self._emeter_params[ATTR_TODAY_ENERGY_KWH] \
                        = "%.2f kW" % (float(emeter_statics[int(time.strftime("%e"))]/1000))
                except KeyError:
                    # Device returned no daily history
                    pass

        except (SmartDeviceException, OSError) as ex:
            _LOGGER.warning("Could not read state for %s: %s", self.name, ex)
            self._available = False