-
Notifications
You must be signed in to change notification settings - Fork 8
/
Copy pathmain.py
228 lines (202 loc) · 9.47 KB
/
main.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
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
#!/usr/bin/env python3
import json
import time
import requests
from flask import Flask
from flask import request, Response
from requests.exceptions import ReadTimeout, ConnectionError, RequestException
from config import *
app = Flask(__name__)
def getstate(entity_id):
headers = {
"x-ha-access": password,
"Content-Type": "application/json"
}
url = 'http://{}:{}/api/states/{}'.format(host, port, entity_id)
try:
response = requests.get(url, headers=headers, timeout=1)
response_result = response.json()
app.logger.debug('getstate url: {}, result: {}'.format(url, json.dumps(response_result)))
except ReadTimeout:
response_result = "连接超时"
except ConnectionError:
response_result = "连接错误"
except RequestException:
response_result = "发生未知错误"
return response_result
def setstate(entity_id, domain, service, **kwargs):
headers = {
"x-ha-access": password,
"Content-Type": "application/json"
}
postdata = {'entity_id': entity_id}
postdata.update(kwargs)
postdata = json.dumps(postdata)
url = 'http://{}:{}/api/services/{}/{}'.format(host, port, domain, service)
try:
response = requests.post(url, headers=headers,
data=postdata, timeout=1)
response_result = response.status_code
app.logger.debug('setstate url: {}, data: {}, result: {}'.format(url, postdata, response_result))
except ReadTimeout:
response_result = "连接超时"
except ConnectionError:
response_result = "连接错误"
except RequestException:
response_result = "发生未知错误"
return response_result
def has_any(query, keywords):
if type(keywords) == list:
for kw in keywords:
if kw in query:
return (True, kw)
elif type(keywords) == dict:
for k,v in keywords.items():
if k in query:
return (True, (k, v))
return (False, None)
def get_tts_text_by_setstate(entity_id, domain, service, action_name, entity_name, **kwargs):
call_result = setstate(entity_id, domain, service, **kwargs)
if call_result == 200:
return '好的，为你' + action_name + entity_name
else:
return '与控制中心' + call_result
tts_word = ""
open_mic = True
is_session_end = False
@app.route(route_path, methods=['POST'])
def index():
datax = request.get_json()
app.logger.debug(datax)
if datax["request"]["type"] == 0:
open_mic = True
is_session_end = False
tts_word = "您好，我能为你做什么呢？"
elif datax["request"]["type"] == 2:
open_mic = False
is_session_end = True
tts_word = "再见，我在这里等你哦!"
elif datax["request"]["type"] == 1:
# user hasn't action for a while
if 'no_response' in datax["request"]:
open_mic = True
is_session_end = False
tts_word = '主人，您还在吗？'
else:
query = datax.get('query', '')
open_mic = True
is_session_end = False
tts_word = '对不起, {}暂时不支持{}这种操作'.format(skill_name, query)
if datax["request"]["intent"]["is_direct_wakeup"] == True:
open_mic = False
is_session_end = True
while True:
# match cover setpos
find, kw = has_any(query, cmd_words_cover_setpos)
find2, item = has_any(query, covers)
if find and find2:
app.logger.debug('match cmd_words_cover_setpos, the query is: {}'.format(query))
if kw == '一半':
tts_word = get_tts_text_by_setstate(item[1], "cover", "set_cover_position", '打开'+kw, item[0], position=50)
break
tts_word = '{}还不会处理这个请求'.format(skill_name)
break
elif find:
tts_word = '对不起,然而帘子并不存在'
break
# match cover open
find, kw = has_any(query, cmd_words_cover_open)
find2, item = has_any(query, covers)
if find and find2:
app.logger.debug('match cmd_words_cover_open, the query is: {}'.format(query))
cover_state = getstate(item[1])
if type(cover_state) == str:
tts_word = '与控制中心' + cover_state
break
if cover_state.get('state','') == 'open':
tts_word = '已经打开了'
else:
tts_word = get_tts_text_by_setstate(item[1], "cover", "open_cover", kw, item[0])
break
# match cover close
find, kw = has_any(query, cmd_words_cover_close)
find2, item = has_any(query, covers)
if find and find2:
app.logger.debug('match cmd_words_cover_close, the query is: {}'.format(query))
cover_state = getstate(item[1])
if type(cover_state) == str:
tts_word = '与控制中心' + cover_state
break
if cover_state.get('state','') == 'closed':
tts_word = '已经关闭了'
else:
tts_word = get_tts_text_by_setstate(item[1], "cover", "close_cover", kw, item[0])
break
# match cover stop
find, kw = has_any(query, cmd_words_cover_stop)
find2, item = has_any(query, covers)
if find and find2:
app.logger.debug('match cmd_words_cover_stop, the query is: {}'.format(query))
tts_word = get_tts_text_by_setstate(item[1], "cover", "stop_cover", kw, item[0])
break
elif find:
tts_word = '对不起,然而帘子并不存在'
break
# match turn on
find, kw = has_any(query, cmd_words_turnon)
find2, item = has_any(query, onoffs)
if find and find2:
app.logger.debug('match cmd_words_turnon, the query is: {}'.format(query))
entity_state = getstate(item[1])
if type(entity_state) == str:
tts_word = '与控制中心' + entity_state
break
if entity_state.get('state','') == 'on':
tts_word = '已经打开了'
else:
tts_word = get_tts_text_by_setstate(item[1], "homeassistant", "turn_on", kw, item[0])
break
elif find:
tts_word = '对不起,您想要打开的神器不存在'
break
# match turn off
find, kw = has_any(query, cmd_words_turnoff)
find2, item = has_any(query, onoffs)
if find and find2:
app.logger.debug('match cmd_words_turnoff, the query is: {}'.format(query))
entity_state = getstate(item[1])
if type(entity_state) == str:
tts_word = '与控制中心' + entity_state
break
if entity_state.get('state','') == 'off':
tts_word = '已经关闭了'
else:
tts_word = get_tts_text_by_setstate(item[1], "homeassistant", "turn_off", kw, item[0])
break
elif find:
tts_word = '对不起,您想要关闭的神器不存在'
break
# query read-only variables
find, kw = has_any(query, cmd_words_query)
find2, item = has_any(query, read_onlys)
if find and find2:
app.logger.debug('match cmd_words_query, the query is: {}'.format(query))
entity_state = getstate(item[1][0])
if type(entity_state) == str:
tts_word = '与控制中心' + entity_state
break
tts_word = '当前'+ item[0] +'为' + entity_state['state'] + item[1][1]
break
elif find:
tts_word = '对不起,您要查询的项目没有找到'
break
# nothing matched
# tts_word the default answer
break # break the while True
say_text = {'version': '1.0', 'session_attributes': {'sessi_id': '12345'}, 'response': {'open_mic': open_mic, 'to_speak': {
'type': 0, 'text': tts_word}, 'to_display': {'type': 0, 'text': tts_word}}, 'directives': [], 'is_session_end': is_session_end}
return Response(json.dumps(say_text), mimetype='application/json')
if ssl_fullchain_pem_path and ssl_private_pem_path:
app.run('0.0.0.0', debug=flask_debug, port=serve_port, ssl_context=(ssl_fullchain_pem_path, ssl_private_pem_path))
else:
app.run('0.0.0.0', debug=flask_debug, port=serve_port)