-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathaction-switch.py
executable file
·202 lines (178 loc) · 9.13 KB
/
action-switch.py
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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import configparser
from hermes_python.hermes import Hermes
from hermes_python.ffi.utils import MqttOptions
from hermes_python.ontology import *
import io
import requests
# import urllib2
import json
import jellyfish
MAX_JARO_DISTANCE = 0.4
CONFIGURATION_ENCODING_FORMAT = "utf-8"
CONFIG_INI = "config.ini"
class SnipsConfigParser(configparser.SafeConfigParser):
def to_dict(self):
return {section : {option_name : option for option_name, option in self.items(section)} for section in self.sections()}
def read_configuration_file(configuration_file):
try:
with io.open(configuration_file, encoding=CONFIGURATION_ENCODING_FORMAT) as f:
conf_parser = SnipsConfigParser()
conf_parser.readfp(f)
return conf_parser.to_dict()
except (IOError, configparser.Error) as e:
return dict()
#def subscribe_intent_callback(hermes, intentMessage):
# conf = read_configuration_file(CONFIG_INI)
# action_wrapper(hermes, intentMessage, conf)
def subscribe_intent_callback(hermes, intentMessage):
conf = read_configuration_file(CONFIG_INI)
print(conf)
print(hermes)
print(intentMessage)
# a=IntentClassifierResult(intentMessage).intent_name
# hermes.publish_end_session(intentMessage.session_id, u"Ok, das mache ich.")
hermes.publish_continue_session(intentMessage.session_id, u"Ok,",["ryanrudak:switch","ryanrudak:verschiedene"])
#hermes.publish_continue_session(intentMessage.session_id, "Ok",texte)
if len(intentMessage.slots.verschiedene) > 0:
print('---------diverse direkt----------')
action_wrapperOrdreDirect(hermes, intentMessage, conf)
else:
print('---------diverse Aktionen----------')
action_wrapperOrdre(hermes, intentMessage, conf)
def action_wrapperOrdre(hermes, intentMessage, conf):
myListSceneOrSwitch=dict()
print(" - action_wrapperOrdre - - - Scenen ermitteln")
myListSceneOrSwitch= getSceneNames(conf,myListSceneOrSwitch)
print("Schalter ermitteln")
myListSceneOrSwitch= getSwitchNames(conf,myListSceneOrSwitch)
intentSwitchActionList=BuildActionSlotList(intentMessage)
# print(" - action_wrapperOrdre intentSwitchActionList: "+intentSwitchActionList)
actionText=""
myAction = True
for intentSwitchAction in intentSwitchActionList:
#print("intentSwitchAction: "+intentSwitchAction)
Match= ActionneEntity(intentSwitchAction["Name"],intentSwitchAction["State"],myListSceneOrSwitch,conf)
#print("Match: "+Match)
DomoticzRealName=Match[1]
print("DomoticzRealName: "+DomoticzRealName)
myAction=myAction and Match[0]
if intentSwitchAction["State"]=="On":
texte = u"Einschalten"
else:
texte = u"Ausschalten"
# actionText=u'{}, {} {}'.format(actionText,texte,str(DomoticzRealName))
if myAction and len(intentSwitchActionList)>0:
hermes.publish_end_session(intentMessage.session_id, actionText)
else:
hermes.publish_end_session(intentMessage.session_id, u"Entschuldigung, ich habe es nicht verstanden.")
def getSceneNames(conf,myListSceneOrSwitch):
# response = urllib2.urlopen(global_conf.get("secret").get("hostname")+'/json?type=scenes')
# jsonresponse = json.load(response)
myURL="http://"+conf['secret'].get("username")+':'+conf.get('secret').get("passwd")+'@'+conf['secret'].get("hostname")+':'+conf.get('secret').get("port")+"/json.htm?type=scenes"
response = requests.get(myURL)
jsonresponse = response.json()
#jsonresponse = json.load(response)
for scene in jsonresponse["result"]:
myName=scene["Name"].encode('utf-8')
myListSceneOrSwitch[(scene["idx"])] = {'Type':'switchscene','Name':myName}
print('---------SceneName----------')
return myListSceneOrSwitch
def getSwitchNames(conf,myListSceneOrSwitch):
# response = urllib2.urlopen(global_conf("secret").get("hostname")+'/json?type=command¶m=getlightswitches')
# jsonresponse = json.load(response)
myURL='http://'+conf['secret'].get("username")+':'+conf.get('secret').get("passwd")+'@'+conf['secret'].get("hostname")+':'+conf['secret'].get("port")+'/json.htm?type=command¶m=getlightswitches'
response = requests.get(myURL)
jsonresponse = response.json()
# json.load(response)
for sw in jsonresponse["result"]:
myName=sw["Name"].encode('utf-8')
myListSceneOrSwitch[(sw["idx"])] = {'Type':'switchlight','Name':myName}
print('---------SwitchName----------')
return myListSceneOrSwitch
def BuildActionSlotList(intent):
intentSwitchList=list()
intentSwitchActionList=list()
intentSwitchState='None' #by default if no action
for (slot_value, slot) in intent.slots.items():
print("Slot_value: "+slot_value)
if slot_value=="action" or slot_value=="switch":
for slot_value2 in slot.all():
print("Slot_Value2: "+slot_value2.value)
print("---------------------------------")
for (slot_value, slot) in intent.slots.items():
print('Slot {} -> Raw: {} Value: {}'.format(slot_value, slot[0].raw_value, slot[0].slot_value.value.value))
# print('Slot {} -> Raw: {} Value: {}'.format(slot_value, slot[0].raw_value, slot[0].slot_value.value.value))
print(" BuildActionSlotList - action: "+slot_value)
print(" slot_value: "+slot_value)
if slot_value=="switch":
#NLU parsing does not preserve order of slot, thus it is impossible to have different action ON and OFF in the same intent=> keep only the first:
print(" - BuildAcitionSlotList - slot[0].slot_value.value.value: "+slot[0].slot_value.value.value)
if slot[0].slot_value.value.value=="TurnOn":
intentSwitchState='On'
print(" - Wenn TurnOn, dann: "+intentSwitchState)
else :
intentSwitchState='Off'
print(" - ansonsten, dann: "+intentSwitchState)
print(" - SchalterStatus: "+intentSwitchState)
elif slot_value=="action":
for slot_value2 in slot.all():
intentSwitchList.append(slot_value2.value)
print(" - Slotvalue: "+slot_value2.value)
# wenn intentSwitchState nicht 'None' enthält, dann mySwitch zusammenstellen
if not intentSwitchState=='None':
for mySwitch in intentSwitchList:
intentSwitchActionList.append({'Name':mySwitch,'State':intentSwitchState})
print(mySwitch+"------>"+intentSwitchState)
return intentSwitchActionList
def curlCmd(idx,myCmd,myParam,conf):
command_url="http://"+conf['secret'].get("username")+':'+conf.get('secret').get("passwd")+'@'+conf.get("secret").get("hostname")+':'+conf.get("secret").get("port")+'/json.htm?type=command¶m='+myParam+'&idx='+str(idx)+'&switchcmd='+myCmd
ignore_result = requests.get(command_url)
def ActionneEntity(name,action,myListSceneOrSwitch,conf):
#derived from nice work of https://github.com/iMartyn/domoticz-snips
lowest_distance = MAX_JARO_DISTANCE
lowest_idx = 65534
lowest_name = "Unknown"
MyWord=name
DomoticzRealName=""
print(" - ActionneEntity: "+MyWord)
for idx,scene in myListSceneOrSwitch.items():
print("Scene/Schalter: "+str(scene['Name'],'utf-8')+" idx: "+idx)
distance = 1-jellyfish.jaro_distance(str(scene['Name'],'utf-8'), MyWord)
print(" Distance is "+str(distance))
if distance < lowest_distance:
print(" Low enough and lowest!")
lowest_distance = distance
lowest_idx = idx
lowest_name = scene['Name']
lowest_Type= scene['Type']
if lowest_distance < MAX_JARO_DISTANCE:
print(" - ActionneEntity - lowest_Type: "+lowest_Type)
DomoticzRealName=str(lowest_name,'utf-8')
print(" - ActionneEntity - DomoticzRealName: "+DomoticzRealName)
print(" - ActionneEntity - lowest_idx: "+lowest_idx)
curlCmd(lowest_idx,action,lowest_Type,conf)
return True,DomoticzRealName
hermes.publish_end_session(intent_message.session_id, "Einschalten "+lowest_name)
else:
return False,DomoticzRealName
def action_wrapperOrdreDirect(hermes, intentMessage, conf):
myListSceneOrSwitch=dict()
myListSceneOrSwitch= getSceneNames(conf,myListSceneOrSwitch)
actionText = u"{}".format(str(intentMessage.slots.OrdreDivers.first().value))
print("actionText "+actionText)
DomoticzRealName=""
MyAction=ActionneEntity(actionText,'On',myListSceneOrSwitch,conf)
result_sentence = u"OK. Du hast {} angefordert.".format(str(MyAction[1])) # The response that will be said out loud by the TTS engine.
if MyAction[0] :
hermes.publish_end_session(intentMessage.session_id, result_sentence)
else:
print(" - action_wrapperOrdreDirect - keine Action:")
hermes.publish_end_session(intentMessage.session_id, u"Entschuldigung, es ist etwas schief gegangen.")
if __name__ == "__main__":
mqtt_opts = MqttOptions()
with Hermes(mqtt_options=mqtt_opts) as h:
h.subscribe_intent("ryanrudak:switch", subscribe_intent_callback)\
.subscribe_intent("ryanrudak:verschiedene", subscribe_intent_callback)\
.start()