MadPatrick
commented
1 year ago Great Job !. If you can share it it will be available for the community and we can share it
Closed ab10002 closed 1 year ago
MadPatrick
commented
1 year ago Great Job !. If you can share it it will be available for the community and we can share it
ab10002
commented
1 year ago In Domoticz v14490 it is 100%, v14495 there's a slider issue again.. It's with testing lines..:
plugin.py:
# Tahoma/Connexoon IO blind plugin
#
# Author: Nonolk, 2019-2020
# FirstFree function courtesy of @moroen https://github.com/moroen/IKEA-Tradfri-plugin
# All credits for the plugin are for Nonolk, who is the origin plugin creator
"""
<plugin key="tahomaIO" name="Somfy Tahoma or Connexoon plugin" author="MadPatrick" version="2.1.1a" externallink="https://github.com/MadPatrick/somfy">
<description>
<br/><h2>Somfy Tahoma/Connexoon plugin</h2><br/>
<ul style="list-style-type:square">
<li>version: 2.1.1a</li>
<li>This plugin require internet connection at all time.</li>
<li>It controls the Somfy for IO Blinds or Screens</li>
<li>Please provide your email and password used to connect Tahoma/Connexoon</li>
</ul>
</description>
<params>
<param field="Username" label="Username" width="200px" required="true" default=""/>
<param field="Password" label="Password" width="200px" required="true" default="" password="true"/>
<param field="Mode2" label="Refresh interval" width="75px">
<options>
<option label="20s" value="2"/>
<option label="1m" value="6"/>
<option label="5m" value="30" default="true"/>
<option label="10m" value="60"/>
<option label="15m" value="90"/>
</options>
</param>
<param field="Mode5" label="Log file location" width="300px">
<description>Enter a location for the logfile (omit final /), or leave empty to create logfile in the domoticz directory.
<br/>Default directory: '/home/user/domoticz' for raspberry pi</description>
</param>
<param field="Mode6" label="Debug" width="75px">
<options>
<option label="True" value="Debug"/>
<option label="False" value="Normal" default="true" />
</options>
</param>
</params>
</plugin>
"""
import Domoticz
import json
import sys
import logging
import exceptions
import time
import tahoma
import os
class BasePlugin:
enabled = False
def __init__(self):
self.heartbeat = False
self.devices = None
self.heartbeat_delay = 1
self.con_delay = 0
self.wait_delay = 30
self.json_data = None
self.command = False
self.refresh = True
self.actions_serialized = []
self.logger = None
self.log_filename = "somfy.log"
return
def onStart(self):
if os.path.exists(Parameters["Mode5"]):
log_dir = Parameters["Mode5"]
else:
Domoticz.Status("Location {0} does not exist, logging to default location".format(Parameters["Mode5"]))
log_dir = ""
log_fullname = os.path.join(log_dir, self.log_filename)
Domoticz.Status("Starting Tahoma blind plugin, logging to file {0}".format(log_fullname))
self.logger = logging.getLogger('root')
if Parameters["Mode6"] == "Debug":
Domoticz.Debugging(2)
DumpConfigToLog()
logging.basicConfig(format='%(asctime)s - %(levelname)-8s - %(filename)-18s - %(message)s', filename=log_fullname,level=logging.DEBUG)
else:
logging.basicConfig(format='%(asctime)s - %(levelname)-8s - %(filename)-18s - %(message)s', filename=log_fullname,level=logging.INFO)
Domoticz.Debug("os.path.exists(Parameters['Mode5']) = {}".format(os.path.exists(Parameters["Mode5"])))
logging.info("starting plugin version "+Parameters["Version"])
self.runCounter = int(Parameters['Mode2'])
logging.debug("starting to log in")
self.tahoma = tahoma.Tahoma()
try:
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
except exceptions.LoginFailure as exp:
Domoticz.Error("Failed to login: " + str(exp))
return
if self.tahoma.logged_in:
self.tahoma.register_listener()
if self.tahoma.logged_in:
self.tahoma.get_devices(Devices, firstFree())
def onStop(self):
logging.info("stopping plugin")
self.heartbeat = False
def onConnect(self, Connection, Status, Description):
logging.debug("onConnect: Connection: '"+str(Connection)+"', Status: '"+str(Status)+"', Description: '"+str(Description)+"' self.tahoma.logged_in: '"+str(self.tahoma.logged_in)+"'")
if (Status == 0 and not self.tahoma.logged_in):
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
elif (self.cookie and self.tahoma.logged_in and (not self.command)):
event_list = self.tahoma.get_events()
self.update_devices_status(event_list)
elif (self.command):
event_list = self.tahoma.tahoma_command(self.json_data)
self.update_devices_status(event_list)
self.command = False
self.heartbeat = False
self.actions_serialized = []
else:
logging.info("Failed to connect to tahoma api")
def onMessage(self, Connection, Data):
Domoticz.Error("onMessage called but not implemented")
Domoticz.Debug("onMessage data: "+str(Data))
def onCommand(self, Unit, Command, Level, Hue):
logging.debug("onCommand: Unit: '"+str(Unit)+"', Command: '"+str(Command)+"', Level: '"+str(Level)+"', Hue: '"+str(Hue)+"'")
commands_serialized = []
action = {}
commands = {}
params = []
if (str(Command) == "Off"):
commands["name"] = "open"
elif (str(Command) == "On"):
commands["name"] = "close"
elif (str(Command) == "Stop"):
commands["name"] = "stop"
elif ("Set Level" in str(Command)):
commands["name"] = "setClosure"
tmp = int(Level)
params.append(tmp)
commands["parameters"] = params
commands_serialized.append(commands)
action["deviceURL"] = Devices[Unit].DeviceID
action["commands"] = commands_serialized
self.actions_serialized.append(action)
logging.debug("preparing command: # commands: "+str(len(commands)))
logging.debug("preparing command: # actions_serialized: "+str(len(self.actions_serialized)))
data = {"label": "Domoticz - "+Devices[Unit].Name+" - "+commands["name"], "actions": self.actions_serialized}
self.json_data = json.dumps(data, indent=None, sort_keys=True)
if (not self.tahoma.logged_in):
logging.info("Not logged in, must connect")
self.command = True
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
if self.tahoma.logged_in:
self.tahoma.register_listener()
event_list = []
try:
event_list = self.tahoma.tahoma_command(self.json_data)
except (exceptions.TooManyRetries, exceptions.FailureWithErrorCode, exceptions.FailureWithoutErrorCode) as exp:
Domoticz.Error("Failed to send command: " + str(exp))
logging.error("Failed to send command: " + str(exp))
return
if event_list is not None and len(event_list) > 0:
self.update_devices_status(event_list)
self.heartbeat = False
self.actions_serialized = []
def onDisconnect(self, Connection):
return
def onHeartbeat(self):
self.runCounter = self.runCounter - 1
if self.runCounter <= 0:
logging.debug("Poll unit")
self.runCounter = int(Parameters['Mode2'])
if (self.tahoma.logged_in and (not self.tahoma.startup)):
if (not self.tahoma.logged_in):
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
if self.tahoma.logged_in:
self.tahoma.register_listener()
event_list = []
try:
event_list = self.tahoma.get_events()
except (exceptions.TooManyRetries, exceptions.FailureWithErrorCode, exceptions.FailureWithoutErrorCode) as exp:
Domoticz.Error("Failed to request data: " + str(exp))
logging.error("Failed to request data: " + str(exp))
return
if event_list is not None and len(event_list) > 0:
self.update_devices_status(event_list)
self.heartbeat = True
elif (self.heartbeat and (self.con_delay < self.wait_delay) and (not self.tahoma.logged_in)):
self.con_delay +=1
Domoticz.Status("Too many connections waiting before authenticating again")
elif (self.heartbeat and (self.con_delay == self.wait_delay) and (not self.tahoma.logged_in)):
if (not self.tahoma.logged_in):
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
if self.tahoma.logged_in:
self.tahoma.register_listener()
self.heartbeat = True
self.con_delay = 0
else:
logging.debug("Polling unit in " + str(self.runCounter) + " heartbeats.")
def update_devices_status(self, Updated_devices):
logging.debug("updating device status self.tahoma.startup = "+str(self.tahoma.startup)+"on data: "+str(Updated_devices))
for dev in Devices:
logging.debug("update_devices_status: checking Domoticz device: "+Devices[dev].Name)
for device in Updated_devices:
if (Devices[dev].DeviceID == device["deviceURL"]) and (device["deviceURL"].startswith("io://")):
level = 0
status_l = False
lumstatus_l = False
status = None
if (self.tahoma.startup):
states = device["states"]
else:
states = device["deviceStates"]
if (device["name"] != "DeviceStateChangedEvent"):
logging.debug("update_devices_status: device['name'] != DeviceStateChangedEvent: "+str(device["name"])+": breaking out")
break
for state in states:
status_l = False
lumstatus_l = False
if ((state["name"] == "core:ClosureState") or (state["name"] == "core:DeploymentState")):
level = int(state["value"])
status_l = True
elif (state["name"] == "core:LuminanceState"):
#lumlevel = int(state["value"]))
#lumlevel = 9995
lumlevel = state["value"]
lumstatus_l = True
Domoticz.Status("Updating device14:"+(state["value"]))
if status_l:
if (Devices[dev].sValue):
int_level = int(Devices[dev].sValue)
else:
int_level = 0
if (level != int_level):
Domoticz.Status("Updating device11:"+Devices[dev].Name)
Domoticz.Status("Updating device12:"+(Devices[dev].sValue))
logging.info("Updating device22:"+Devices[dev].Name)
logging.info("Test22: "+(Devices[dev].sValue))
if (level == 0):
Devices[dev].Update(0,"0")
if (level == 100):
Devices[dev].Update(1,"100")
if (level != 0 and level != 100):
Devices[dev].Update(2,str(level))
#if (lumlevel != 0 and lumlevel != 70000):
#Devices[dev].Update(3,str(lumlevel))
if lumstatus_l:
if (Devices[dev].sValue):
#int_lumlevel = int(Devices[dev].sValue)
int_lumlevel = 777
else:
int_lumlevel = 0
if (lumlevel != int_lumlevel):
Domoticz.Status("Updating device133:"+Devices[dev].Name)
#Domoticz.Status("Updating device34:"+(Devices[dev].sValue))
logging.info("Updating device144:"+Devices[dev].Name)
#logging.info("Test43: "+(Devices[dev].sValue))
if (lumlevel != 0 and lumlevel != 70000):
Devices[dev].Update(3,str(lumlevel))
return
global _plugin
_plugin = BasePlugin()
def onStart():
global _plugin
_plugin.onStart()
def onStop():
global _plugin
_plugin.onStop()
def onConnect(Connection, Status, Description):
global _plugin
_plugin.onConnect(Connection, Status, Description)
def onMessage(Connection, Data):
global _plugin
_plugin.onMessage(Connection, Data)
def onCommand(Unit, Command, Level, Hue):
global _plugin
_plugin.onCommand(Unit, Command, Level, Hue)
def onDisconnect(Connection):
global _plugin
_plugin.onDisconnect(Connection)
def onHeartbeat():
global _plugin
_plugin.onHeartbeat()
# Generic helper functions
def DumpConfigToLog():
Domoticz.Debug("Parameters count: " + str(len(Parameters)))
for x in Parameters:
if Parameters[x] != "":
Domoticz.Debug("Parameter: '" + x + "':'" + str(Parameters[x]) + "'")
Domoticz.Debug("Device count: " + str(len(Devices)))
for x in Devices:
Domoticz.Debug("Device: " + str(x) + " - " + str(Devices[x]))
return
def DumpHTTPResponseToLog(httpResp, level=0):
if (level==0): Domoticz.Debug("HTTP Details ("+str(len(httpResp))+"):")
indentStr = ""
for x in range(level):
indentStr += "----"
if isinstance(httpResp, dict):
for x in httpResp:
if not isinstance(httpResp[x], dict) and not isinstance(httpResp[x], list):
Domoticz.Debug(indentStr + ">'" + x + "':'" + str(httpResp[x]) + "'")
else:
Domoticz.Debug(indentStr + ">'" + x + "':")
DumpHTTPResponseToLog(httpResp[x], level+1)
elif isinstance(httpResp, list):
for x in httpResp:
Domoticz.Debug(indentStr + "['" + x + "']")
else:
Domoticz.Debug(indentStr + ">'" + x + "':'" + str(httpResp[x]) + "'")
def firstFree():
for num in range(1, 250):
if num not in Devices:
return num
return
tahoma.py:
import requests
import logging
import exceptions
import urllib.parse
import datetime
import Domoticz
import time
class Tahoma:
def __init__(self):
self.srvaddr = "tahomalink.com"
self.base_url = "https://tahomalink.com:443"
self.cookie = None
self.listenerId = None
self.__logged_in = False
self.startup = True
#self.heartbeat = False
self.devices = None
self.filtered_devices = None
self.events = None
self.heartbeat_delay = 1
self.con_delay = 0
self.wait_delay = 30
self.json_data = None
self.refresh = True
self.timeout = 10
self.__expiry_date = datetime.datetime.now()
self.logged_in_expiry_days = 6
@property
def logged_in(self):
logging.debug("checking logged in status: self.__logged_in = "+str(self.__logged_in)+" and self.__expiry_date >= datetime.datetime.now() = " + str(self.__expiry_date >= datetime.datetime.now()))
if self.__logged_in and (self.__expiry_date >= datetime.datetime.now()):
return True
else:
return False
def tahoma_login(self, username, password):
url = self.base_url + '/enduser-mobile-web/enduserAPI/login'
headers = { 'Host': self.srvaddr,"Connection": "keep-alive","Accept-Encoding": "gzip, deflate", "Accept": "*/*", "Content-Type": "application/x-www-form-urlencoded"}
data = "userId="+urllib.parse.quote(username)+"&userPassword="+urllib.parse.quote(password)+""
response = requests.post(url, data=data, headers=headers, timeout=self.timeout)
Status = response.status_code
Data = response.json()
logging.debug("Login respone: status_code: '"+str(Status)+"' reponse body: '"+str(Data)+"'")
if (Status == 200 and not self.__logged_in):
self.__logged_in = True
self.__expiry_date = datetime.datetime.now() + datetime.timedelta(days=self.logged_in_expiry_days)
logging.info("Tahoma authentication succeeded, login valid until " + self.__expiry_date.strftime("%Y-%m-%d %H:%M:%S"))
#self.cookie = response.cookies
self.cookie = response.headers["Set-Cookie"]
logging.debug("login: cookies: '"+ str(response.cookies)+"', headers: '"+str(response.headers)+"'")
#self.register_listener()
elif ((Status == 401) or (Status == 400)):
strData = Data["error"]
#logging.error("Tahoma error: must reconnect")
self.__logged_in = False
self.cookie = None
self.listenerId = None
if ("Too many" in strData):
logging.error("Too many connections, must wait")
#self.heartbeat = True
raise exceptions.LoginFailure("Too many connections, must wait")
elif ("Bad credentials" in strData):
logging.error("login failed: Bad credentials, please update credentials and restart plugin")
#self.heartbeat = False
raise exceptions.LoginFailure("Bad credentials, please update credentials and restart plugin")
else:
logging.error("login failed, unhandled reason: "+strData)
raise exceptions.LoginFailure("login failed, unhandled reason: "+strData)
if (not self.__logged_in):
self.tahoma_login(username, password)
return
return self.__logged_in
def tahoma_command(self, json_data):
timeout = 4
logging.debug("start command")
Headers = { 'Host': self.srvaddr, "Connection": "keep-alive","Accept-Encoding": "gzip, deflate", "Accept": "*/*", "Content-Type": "application/json", "Cookie": self.cookie}
url = self.base_url + '/enduser-mobile-web/enduserAPI/exec/apply'
logging.debug("onCommand: headers: '"+str(Headers)+"', data '"+str(json_data)+"'")
logging.info("Sending command to tahoma api")
try:
response = requests.post(url, headers=Headers, data=json_data, timeout=timeout)
except requests.exceptions.RequestException as exp:
logging.error("Send command returns RequestException: " + str(exp))
logging.debug("command response: status '" + str(response.status_code) + "' response body: '"+str(response.json())+"'")
if response.status_code != 200:
logging.error("error during command, status: " + str(response.status_code) + ", possible cause:" + str(response.json()))
self.__logged_in = False
return ""
self.executionId = response.json()['execId']
event_list = self.get_events()
return event_list
def register_listener(self):
logging.debug("start register")
Headers = { 'Host': self.srvaddr,"Connection": "keep-alive","Accept-Encoding": "gzip, deflate", "Accept": "*/*", "Content-Type": "application/json", "Cookie": self.cookie}
url = self.base_url + '/enduser-mobile-web/enduserAPI/events/register'
response = requests.post(url, headers=Headers, timeout=self.timeout)
logging.debug("register response: status '" + str(response.status_code) + "' response body: '"+str(response.json())+"'")
if response.status_code != 200:
logging.error("error during register, status: " + str(response.status_code))
return
Data = response.json()
if "id" in Data:
strData = Data["id"]
else:
logging.error("Data expected in response but not found")
return
self.listenerId = Data['id']
logging.info("Tahoma listener registred")
self.refresh = False
logging.info("Checking setup status at startup")
#self.get_devices()
def get_devices(self, Devices, firstFree):
logging.debug("start get devices")
Headers = { 'Host': self.srvaddr,"Connection": "keep-alive","Accept-Encoding": "gzip, deflate", "Accept": "*/*", "Content-Type": "application/x-www-form-urlencoded", "Cookie": self.cookie}
url = self.base_url + '/enduser-mobile-web/enduserAPI/setup/devices'
response = requests.get(url, headers=Headers, timeout=self.timeout)
logging.debug("get device response: url '" + str(response.url) + "' response headers: '"+str(response.headers)+"'")
logging.debug("get device response: status '" + str(response.status_code) + "' response body: '"+str(response.json())+"'")
if response.status_code != 200:
logging.error("get_devices: error during get devices, status: " + str(response.status_code))
return
Data = response.json()
if (not "uiClass" in response.text):
logging.error("get_devices: missing uiClass in response")
logging.debug(str(Data))
return
self.devices = Data
self.filtered_devices = list()
for device in self.devices:
logging.debug("get_devices: Device name: "+device["label"]+" Device class: "+device["uiClass"])
#if (((device["uiClass"] == "RollerShutter") or (device["uiClass"] == "ExteriorScreen") or (device["uiClass"] == "Screen") or (device["uiClass"] == "Awning") or (device["uiClass"] == "Pergola") or (device["uiClass"] == "GarageDoor") or (device["uiClass"] == "Window") or (device["uiClass"] == "VenetianBlind") or (device["uiClass"] == "ExteriorVenetianBlind")) and ((device["deviceURL"].startswith("io://")) or (device["deviceURL"].startswith("rts://")))):
if (((device["uiClass"] == "RollerShutter") or (device["uiClass"] == "LightSensor") or (device["uiClass"] == "ExteriorScreen") or (device["uiClass"] == "Screen") or (device["uiClass"] == "Awning") or (device["uiClass"] == "Pergola") or (device["uiClass"] == "GarageDoor") or (device["uiClass"] == "Window") or (device["uiClass"] == "VenetianBlind") or (device["uiClass"] == "ExteriorVenetianBlind")) and ((device["deviceURL"].startswith("io://")) or (device["deviceURL"].startswith("rts://")))):
self.filtered_devices.append(device)
logging.debug("get_devices: devices found: "+str(len(Devices))+" self.startup: "+str(self.startup))
if (len(Devices) == 0 and self.startup):
count = 1
for device in self.filtered_devices:
logging.info("get_devices: Creating device: "+device["label"])
swtype = None
if (device["deviceURL"].startswith("io://")):
if (device["uiClass"] == "RollerShutter"):
swtype = 21
devicetype = 244
subtype2 = 73
if (device["uiClass"] == "LightSensor"):
swtype = 12
devicetype = 246
subtype2 = 1
elif (device["deviceURL"].startswith("rts://")):
swtype = 6
#Domoticz.Device(Name=device["label"], Unit=count, Type=244, Subtype=73, Switchtype=swtype, DeviceID=device["deviceURL"]).Create()
Domoticz.Device(Name=device["label"], Unit=count, Type=devicetype, Subtype=subtype2, Switchtype=swtype, DeviceID=device["deviceURL"]).Create()
if not (count in Devices):
logging.error("Device creation not allowed, please allow device creation")
Domoticz.Error("Device creation not allowed, please allow device creation")
else:
logging.info("Device created: "+device["label"])
count += 1
if ((len(Devices) < len(self.filtered_devices)) and len(Devices) != 0 and self.startup):
logging.info("New device(s) detected")
found = False
for device in self.filtered_devices:
for dev in Devices:
UnitID = Devices[dev].Unit
if Devices[dev].DeviceID == device["deviceURL"]:
found = True
break
if (not found):
idx = firstFree
swtype = None
logging.debug("get_devices: Must create device: "+device["label"])
if (device["deviceURL"].startswith("io://")):
if (device["uiClass"] == "RollerShutter"):
swtype = 21
devicetype = 244
subtype2 = 73
if (device["uiClass"] == "LightSensor"):
swtype = 12
devicetype = 246
subtype2 = 1
elif (device["deviceURL"].startswith("rts://")):
swtype = 6
#Domoticz.Device(Name=device["label"], Unit=idx, Type=244, Subtype=73, Switchtype=swtype, DeviceID=device["deviceURL"]).Create()
Domoticz.Device(Name=device["label"], Unit=idx, Type=devicetype, Subtype=subtype2, Switchtype=swtype, DeviceID=device["deviceURL"]).Create()
if not (idx in Devices):
logging.error("Device creation not allowed, please allow device creation")
Domoticz.Error("Device creation not allowed, please allow device creation")
else:
logging.info("New device created: "+device["label"])
else:
found = False
self.startup = False
self.get_events()
def get_events(self):
logging.debug("start get events")
Headers = { 'Host': self.srvaddr,"Connection": "keep-alive","Accept-Encoding": "gzip, deflate", "Accept": "*/*", "Content-Type": "application/json", "Cookie": self.cookie}
url = self.base_url + '/enduser-mobile-web/enduserAPI/events/'+self.listenerId+'/fetch'
for i in range(1,4):
try:
response = requests.post(url, headers=Headers, timeout=self.timeout)
logging.debug("get events response: status '" + str(response.status_code) + "' response body: '"+str(response.json())+"'")
#logging.debug("get events: self.__logged_in = '"+str(self.__logged_in)+"' and self.heartbeat = '"+str(self.heartbeat)+"' and self.startup = '"+str(self.startup))
if response.status_code != 200:
logging.error("error during get events, status: " + str(response.status_code) + ", " + str(response.text))
self.__logged_in = False
return
elif (response.status_code == 200 and self.__logged_in and (not self.startup)):
strData = response.json()
if (not "DeviceStateChangedEvent" in response.text):
logging.debug("get_events: no DeviceStateChangedEvent found in response: " + str(strData))
return
self.events = strData
if (self.events):
filtered_events = list()
for event in self.events:
if (event["name"] == "DeviceStateChangedEvent"):
logging.debug("get_events: add event: URL: '"+event["deviceURL"]+"' num states: '"+str(len(event["deviceStates"]))+"'")
filtered_events.append(event)
return filtered_events
#self.update_devices_status(filtered_events)
# elif (response.status_code == 200 and (not self.heartbeat)):
# return
else:
logging.info("Return status"+str(response.status_code))
except requests.exceptions.RequestException as exp:
logging.error("get_events RequestException: " + str(exp))
time.sleep(i ** 3)
else:
raise exceptions.TooManyRetries
Some cleaning... working with v14490: plugin.py
# Tahoma/Connexoon IO blind plugin
#
# Author: Nonolk, 2019-2020
# FirstFree function courtesy of @moroen https://github.com/moroen/IKEA-Tradfri-plugin
# All credits for the plugin are for Nonolk, who is the origin plugin creator
"""
<plugin key="tahomaIO" name="Somfy Tahoma or Connexoon plugin" author="MadPatrick" version="2.1.1b" externallink="https://github.com/MadPatrick/somfy">
<description>
<br/><h2>Somfy Tahoma/Connexoon plugin</h2><br/>
<ul style="list-style-type:square">
<li>version: 2.1.1b</li>
<li>This plugin require internet connection at all time.</li>
<li>It controls the Somfy for IO Blinds or Screens</li>
<li>Please provide your email and password used to connect Tahoma/Connexoon</li>
</ul>
</description>
<params>
<param field="Username" label="Username" width="200px" required="true" default=""/>
<param field="Password" label="Password" width="200px" required="true" default="" password="true"/>
<param field="Mode2" label="Refresh interval" width="75px">
<options>
<option label="20s" value="2"/>
<option label="1m" value="6"/>
<option label="5m" value="30" default="true"/>
<option label="10m" value="60"/>
<option label="15m" value="90"/>
</options>
</param>
<param field="Mode5" label="Log file location" width="300px">
<description>Enter a location for the logfile (omit final /), or leave empty to create logfile in the domoticz directory.
<br/>Default directory: '/home/user/domoticz' for raspberry pi</description>
</param>
<param field="Mode6" label="Debug" width="75px">
<options>
<option label="True" value="Debug"/>
<option label="False" value="Normal" default="true" />
</options>
</param>
</params>
</plugin>
"""
import Domoticz
import json
import sys
import logging
import exceptions
import time
import tahoma
import os
class BasePlugin:
enabled = False
def __init__(self):
self.heartbeat = False
self.devices = None
self.heartbeat_delay = 1
self.con_delay = 0
self.wait_delay = 30
self.json_data = None
self.command = False
self.refresh = True
self.actions_serialized = []
self.logger = None
self.log_filename = "somfy.log"
return
def onStart(self):
if os.path.exists(Parameters["Mode5"]):
log_dir = Parameters["Mode5"]
else:
Domoticz.Status("Location {0} does not exist, logging to default location".format(Parameters["Mode5"]))
log_dir = ""
log_fullname = os.path.join(log_dir, self.log_filename)
Domoticz.Status("Starting Tahoma blind plugin, logging to file {0}".format(log_fullname))
self.logger = logging.getLogger('root')
if Parameters["Mode6"] == "Debug":
Domoticz.Debugging(2)
DumpConfigToLog()
logging.basicConfig(format='%(asctime)s - %(levelname)-8s - %(filename)-18s - %(message)s', filename=log_fullname,level=logging.DEBUG)
else:
logging.basicConfig(format='%(asctime)s - %(levelname)-8s - %(filename)-18s - %(message)s', filename=log_fullname,level=logging.INFO)
Domoticz.Debug("os.path.exists(Parameters['Mode5']) = {}".format(os.path.exists(Parameters["Mode5"])))
logging.info("starting plugin version "+Parameters["Version"])
self.runCounter = int(Parameters['Mode2'])
logging.debug("starting to log in")
self.tahoma = tahoma.Tahoma()
try:
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
except exceptions.LoginFailure as exp:
Domoticz.Error("Failed to login: " + str(exp))
return
if self.tahoma.logged_in:
self.tahoma.register_listener()
if self.tahoma.logged_in:
self.tahoma.get_devices(Devices, firstFree())
def onStop(self):
logging.info("stopping plugin")
self.heartbeat = False
def onConnect(self, Connection, Status, Description):
logging.debug("onConnect: Connection: '"+str(Connection)+"', Status: '"+str(Status)+"', Description: '"+str(Description)+"' self.tahoma.logged_in: '"+str(self.tahoma.logged_in)+"'")
if (Status == 0 and not self.tahoma.logged_in):
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
elif (self.cookie and self.tahoma.logged_in and (not self.command)):
event_list = self.tahoma.get_events()
self.update_devices_status(event_list)
elif (self.command):
event_list = self.tahoma.tahoma_command(self.json_data)
self.update_devices_status(event_list)
self.command = False
self.heartbeat = False
self.actions_serialized = []
else:
logging.info("Failed to connect to tahoma api")
def onMessage(self, Connection, Data):
Domoticz.Error("onMessage called but not implemented")
Domoticz.Debug("onMessage data: "+str(Data))
def onCommand(self, Unit, Command, Level, Hue):
logging.debug("onCommand: Unit: '"+str(Unit)+"', Command: '"+str(Command)+"', Level: '"+str(Level)+"', Hue: '"+str(Hue)+"'")
commands_serialized = []
action = {}
commands = {}
params = []
if (str(Command) == "Off"):
commands["name"] = "open"
elif (str(Command) == "On"):
commands["name"] = "close"
elif (str(Command) == "Stop"):
commands["name"] = "stop"
elif ("Set Level" in str(Command)):
commands["name"] = "setClosure"
tmp = int(Level)
params.append(tmp)
commands["parameters"] = params
commands_serialized.append(commands)
action["deviceURL"] = Devices[Unit].DeviceID
action["commands"] = commands_serialized
self.actions_serialized.append(action)
logging.debug("preparing command: # commands: "+str(len(commands)))
logging.debug("preparing command: # actions_serialized: "+str(len(self.actions_serialized)))
data = {"label": "Domoticz - "+Devices[Unit].Name+" - "+commands["name"], "actions": self.actions_serialized}
self.json_data = json.dumps(data, indent=None, sort_keys=True)
if (not self.tahoma.logged_in):
logging.info("Not logged in, must connect")
self.command = True
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
if self.tahoma.logged_in:
self.tahoma.register_listener()
event_list = []
try:
event_list = self.tahoma.tahoma_command(self.json_data)
except (exceptions.TooManyRetries, exceptions.FailureWithErrorCode, exceptions.FailureWithoutErrorCode) as exp:
Domoticz.Error("Failed to send command: " + str(exp))
logging.error("Failed to send command: " + str(exp))
return
if event_list is not None and len(event_list) > 0:
self.update_devices_status(event_list)
self.heartbeat = False
self.actions_serialized = []
def onDisconnect(self, Connection):
return
def onHeartbeat(self):
self.runCounter = self.runCounter - 1
if self.runCounter <= 0:
logging.debug("Poll unit")
self.runCounter = int(Parameters['Mode2'])
if (self.tahoma.logged_in and (not self.tahoma.startup)):
if (not self.tahoma.logged_in):
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
if self.tahoma.logged_in:
self.tahoma.register_listener()
event_list = []
try:
event_list = self.tahoma.get_events()
except (exceptions.TooManyRetries, exceptions.FailureWithErrorCode, exceptions.FailureWithoutErrorCode) as exp:
Domoticz.Error("Failed to request data: " + str(exp))
logging.error("Failed to request data: " + str(exp))
return
if event_list is not None and len(event_list) > 0:
self.update_devices_status(event_list)
self.heartbeat = True
elif (self.heartbeat and (self.con_delay < self.wait_delay) and (not self.tahoma.logged_in)):
self.con_delay +=1
Domoticz.Status("Too many connections waiting before authenticating again")
elif (self.heartbeat and (self.con_delay == self.wait_delay) and (not self.tahoma.logged_in)):
if (not self.tahoma.logged_in):
self.tahoma.tahoma_login(str(Parameters["Username"]), str(Parameters["Password"]))
if self.tahoma.logged_in:
self.tahoma.register_listener()
self.heartbeat = True
self.con_delay = 0
else:
logging.debug("Polling unit in " + str(self.runCounter) + " heartbeats.")
def update_devices_status(self, Updated_devices):
logging.debug("updating device status self.tahoma.startup = "+str(self.tahoma.startup)+"on data: "+str(Updated_devices))
for dev in Devices:
logging.debug("update_devices_status: checking Domoticz device: "+Devices[dev].Name)
for device in Updated_devices:
if (Devices[dev].DeviceID == device["deviceURL"]) and (device["deviceURL"].startswith("io://")):
level = 0
status_l = False
lumstatus_l = False
status = None
if (self.tahoma.startup):
states = device["states"]
else:
states = device["deviceStates"]
if (device["name"] != "DeviceStateChangedEvent"):
logging.debug("update_devices_status: device['name'] != DeviceStateChangedEvent: "+str(device["name"])+": breaking out")
break
for state in states:
status_l = False
lumstatus_l = False
if ((state["name"] == "core:ClosureState") or (state["name"] == "core:DeploymentState")):
level = int(state["value"])
#level = 100 - level
status_l = True
elif (state["name"] == "core:LuminanceState"):
lumlevel = state["value"]
lumstatus_l = True
if status_l:
if (Devices[dev].sValue):
int_level = int(Devices[dev].sValue)
else:
int_level = 0
if (level != int_level):
Domoticz.Status("Updating device: "+Devices[dev].Name)
logging.info("Updating device: "+Devices[dev].Name)
if (level == 0):
Devices[dev].Update(0,"0")
if (level == 100):
Devices[dev].Update(1,"100")
if (level != 0 and level != 100):
Devices[dev].Update(2,str(level))
if lumstatus_l:
if (Devices[dev].sValue):
int_lumlevel = Devices[dev].sValue
else:
int_lumlevel = 0
if (lumlevel != int_lumlevel):
Domoticz.Status("Updating device: "+Devices[dev].Name)
logging.info("Updating device: "+Devices[dev].Name)
if (lumlevel != 0 and lumlevel != 120000):
Devices[dev].Update(3,str(lumlevel))
return
global _plugin
_plugin = BasePlugin()
def onStart():
global _plugin
_plugin.onStart()
def onStop():
global _plugin
_plugin.onStop()
def onConnect(Connection, Status, Description):
global _plugin
_plugin.onConnect(Connection, Status, Description)
def onMessage(Connection, Data):
global _plugin
_plugin.onMessage(Connection, Data)
def onCommand(Unit, Command, Level, Hue):
global _plugin
_plugin.onCommand(Unit, Command, Level, Hue)
def onDisconnect(Connection):
global _plugin
_plugin.onDisconnect(Connection)
def onHeartbeat():
global _plugin
_plugin.onHeartbeat()
# Generic helper functions
def DumpConfigToLog():
Domoticz.Debug("Parameters count: " + str(len(Parameters)))
for x in Parameters:
if Parameters[x] != "":
Domoticz.Debug("Parameter: '" + x + "':'" + str(Parameters[x]) + "'")
Domoticz.Debug("Device count: " + str(len(Devices)))
for x in Devices:
Domoticz.Debug("Device: " + str(x) + " - " + str(Devices[x]))
return
def DumpHTTPResponseToLog(httpResp, level=0):
if (level==0): Domoticz.Debug("HTTP Details ("+str(len(httpResp))+"):")
indentStr = ""
for x in range(level):
indentStr += "----"
if isinstance(httpResp, dict):
for x in httpResp:
if not isinstance(httpResp[x], dict) and not isinstance(httpResp[x], list):
Domoticz.Debug(indentStr + ">'" + x + "':'" + str(httpResp[x]) + "'")
else:
Domoticz.Debug(indentStr + ">'" + x + "':")
DumpHTTPResponseToLog(httpResp[x], level+1)
elif isinstance(httpResp, list):
for x in httpResp:
Domoticz.Debug(indentStr + "['" + x + "']")
else:
Domoticz.Debug(indentStr + ">'" + x + "':'" + str(httpResp[x]) + "'")
def firstFree():
for num in range(1, 250):
if num not in Devices:
return num
return
Susis IO senors are working great!
MadPatrick
commented
1 year ago Thanks. I'll incorpated the changes
I'v got it working. If wanted i can send my file.