Lang verhaal kort; ik heb nieuwe zonneschermen, met een Somfy tahoma Switch. Hier is nog geen domoticz plugin voor gemaakt, dus ik dacht, ik doe een poging om een begin te maken. Wat inmiddels een steeds lastiger doel lijkt te worden voor iemand met precies 0 programmeerervaring.
Tis een week of twee nu constant twee stappen vooruit, één stap terug, maar ben volgens mij al best een eind gekomen. Heb de API's aan de praat, token aangevraagd, eerste responses terug via Python script. Dus dat stuk loopt best aardig. Maar dan de vervolgstappen.
Wat ik nu probeer is om iets aan output te checken vanuit de API's. Heb daarvoor de 'oude plugin' misbruikt. De code en response lijkt redelijk overeen te komen.
Goed, het script zover:
Ik probeer de 'update_devices_status' aan te roepen, aangezien dat voor zover ik kan bepalen de functie is die wordt aangeroepen 'on heartbeat', dus op gezette tijden.
Mijn eerste stap was om te proberen daarvan de resultaten op het scherm te krijgen. Maar ik zit te rommelen met het definiëren van velden. Vooral nondefined fields, die voor zover ik kan bepalen toch al eerder in het script gedefinieerd worden.
Probeerde dat via:
Maar krijg dan:
NameError: name 'self' is not defined
met wat zoekwerk, zou 'self' toch een terugverwijzing naar eerder in hetzelfde script moeten zijn?
Is er iemand die me een zetje in de goede richting kan geven?
Tis een week of twee nu constant twee stappen vooruit, één stap terug, maar ben volgens mij al best een eind gekomen. Heb de API's aan de praat, token aangevraagd, eerste responses terug via Python script. Dus dat stuk loopt best aardig. Maar dan de vervolgstappen.
Wat ik nu probeer is om iets aan output te checken vanuit de API's. Heb daarvoor de 'oude plugin' misbruikt. De code en response lijkt redelijk overeen te komen.
Goed, het script zover:
Python:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
| # Credits to # Nonolk: https://github.com/nonolk/domoticz_tahoma_blind # Madpatrick: https://github.com/MadPatrick/somfy #Variables: pin = "2017-8871-5701" port = "8443" baseUrl = f"https://gateway-{pin}.local:{port}/enduser-mobile-web/1/enduserAPI" token = "..." ###API URLS #Events fetchevents = f"{baseUrl}/events/:listenerId/fetch" #Post unregisterlistener = f"{baseUrl}/events/:listenerId/unregister" #Post registerlistener = f"{baseUrl}/events/register" #Post #Setup getdevice = f"{baseUrl}/setup/devices/:deviceURL" #Get getdevices = f"{baseUrl}/setup/devices" #Get getdevicescontrol = f"{baseUrl}/setup/devices/controllables/:controllableName" #Get getsetup = f"{baseUrl}/setup" #Get getgateway = f"{baseUrl}/setup/gateways" #Get #Exec cancelexecutions = f"{baseUrl}/exec/current/setup" #Del cancelspecificexec = f"{baseUrl}/exec/current/setup/:executionId" #Del getexecutions = f"{baseUrl}/exec/current" #Get getexecution = f"{baseUrl}/exec/current/:executionId" #Get executeaction = f"{baseUrl}/exec/apply" #Post #Api version getapiversion = f"{baseUrl}/apiVersion" #Get ### API Urls payload={} headers = { 'Accept': 'application/json', 'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json' } #Print variable output #print(f"Pin = {pin}") #print(f"Port = {port}") #print(f"URL = {baseUrl}") #print(f"Token = {token}") #print(f"Headers = {headers}") import requests import json #import Domoticz #import DomoticzEx as Domoticz import sys import tahoma import logging import time import os #Test api's #Get verion #response = requests.request("GET", getapiversion, headers=headers, data=payload, verify=False) #Get devices response = requests.request("GET", getdevices, headers=headers, data=payload, verify=False) #Get setup #response = requests.request("GET", getsetup, headers=headers, data=payload, verify=False) #Get gateway #response = requests.request("GET", getgateway, headers=headers, data=payload, verify=False) #print (response.text) #print (type(response.text)) #print (isinstance(response.text, str)) output_file = json.loads(response.text) #print (json.dumps(output_file, indent=4)) with open('response.json', 'w', encoding='utf-8') as outfile: json.dump(output_file, outfile, indent=4) class BasePlugin: 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" self.headers = headers 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"] = "close" elif (str(Command) == "On"): commands["name"] = "open" elif ("Set Level" in str(Command)): commands["name"] = "setClosure" tmp = 100 - 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 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 if ((state["name"] == "core:ClosureState") or (state["name"] == "core:DeploymentState")): level = int(state["value"]) level = 100 - level status_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)) return # Test entry in tahoma.py def test_print(self, headers=None): self.tahoma = tahoma.Tahoma() headers = self.tahoma.headers print(headers) def test_update(self, Updated_devices=None): update_devices_status(self) print(device) |
Ik probeer de 'update_devices_status' aan te roepen, aangezien dat voor zover ik kan bepalen de functie is die wordt aangeroepen 'on heartbeat', dus op gezette tijden.
Mijn eerste stap was om te proberen daarvan de resultaten op het scherm te krijgen. Maar ik zit te rommelen met het definiëren van velden. Vooral nondefined fields, die voor zover ik kan bepalen toch al eerder in het script gedefinieerd worden.
Probeerde dat via:
Python:
1
2
3
| def test_update(self, Updated_devices=None): update_devices_status(self) print(Updated_devices) |
Maar krijg dan:
NameError: name 'self' is not defined
met wat zoekwerk, zou 'self' toch een terugverwijzing naar eerder in hetzelfde script moeten zijn?
Is er iemand die me een zetje in de goede richting kan geven?
[ Voor 2% gewijzigd door Creepy op 15-08-2022 16:29 . Reden: quotes om code heen verwijderd ]