-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapi.py
160 lines (135 loc) · 6.87 KB
/
api.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
from flask import Flask
from flask_restful import Api, Resource, reqparse, fields, marshal
import shelve
import functools
import copy
import control_hue_lights
import control_chromecast_audio
app = Flask(__name__)
api = Api(app)
api.app.config['DATABASE'] = 'database_shelve'
device_attributes = ['device_id', 'device_name', 'device_type', 'device_controller_address', 'device_data']
def shelve_db_decorator(func):
@functools.wraps(func)
def inner(*args, **kwargs):
global shelve_db # global is necessary, otherwise func can't see shelve_db
# opens and closes shelve database automatically even if an exception is raised
with shelve.open(api.app.config['DATABASE'], writeback=True) as shelve_db:
rv = func(*args, **kwargs)
return rv
return inner
class DeviceListAPI(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('device_id', type=str, required=True,
help='No/invalid device_id provided', location='json')
self.reqparse.add_argument('device_name', type=str, required=True,
help='No/invalid device_name provided', location='json')
self.reqparse.add_argument('device_type', type=str, required=True,
help='No/invalid device_type provided', location='json')
self.reqparse.add_argument('device_controller_address', type=str, required=True,
help='No/invalid device_controller_address provided', location='json')
self.reqparse.add_argument('device_data', type=dict, required=True,
help='No/invalid device_data provided', location='json')
super(DeviceListAPI, self).__init__()
@shelve_db_decorator
def get(self):
for device in shelve_db['devices']:
# if needed: call respective API to get new device data
if 'Hue' in device['device_type']:
new_lights_info = control_hue_lights.get_light_info(device['device_controller_address'])
for key, val in new_lights_info['state'].items():
device['device_data'][key] = val
return shelve_db['devices'], 200
@shelve_db_decorator
def post(self):
args = self.reqparse.parse_args()
if [device for device in shelve_db['devices'] if device['device_id'] == args['device_id']]:
return 'Device with device_id "{}" already exists'.format(args["device_id"]), 409
new_device = {
'device_id': args['device_id'],
'device_name': args['device_name'],
'device_type': args['device_type'],
'device_controller_address': args['device_controller_address'],
'device_data': args['device_data']
}
shelve_db['devices'].append(new_device)
return marshal(new_device, device_fields), 200
device_fields = {
'uri': fields.Url('device', absolute=True),
'device_id': fields.String,
'device_name': fields.String,
'device_type': fields.String,
'device_controller_address': fields.String,
'device_data': fields.Raw
}
class DeviceAPI(Resource):
def __init__(self):
self.reqparse = reqparse.RequestParser()
self.reqparse.add_argument('device_id', type=str, location='json')
self.reqparse.add_argument('device_name', type=str, location='json')
self.reqparse.add_argument('device_type', type=str, location='json')
self.reqparse.add_argument('device_controller_address', type=str, location='json')
self.reqparse.add_argument('device_data', type=dict, location='json')
super(DeviceAPI, self).__init__()
@shelve_db_decorator
def get(self, device_id):
device = [device for device in shelve_db['devices'] if device['device_id'] == device_id]
if len(device) == 0:
# device not found
return 'There is no device with device_id "{}"'.format(device_id), 404
if len(device) == 1:
# device found
device = device[0]
# if needed: call respective API to get new device data
if 'Hue' in device['device_type']:
new_lights_info = control_hue_lights.get_light_info(device['device_controller_address'])
for key, val in new_lights_info['state'].items():
device['device_data'][key] = val
return marshal(device, device_fields), 200
if len(device) > 1:
return 'Found {} devices with the same device_id'.format(len(device)), 500
@shelve_db_decorator
def patch(self, device_id):
args = self.reqparse.parse_args()
# get a copy of the device
device = [device for device in enumerate(shelve_db['devices']) if device[1]['device_id'] == device_id]
device_index = device[0][0]
device = copy.deepcopy(device[0][1])
for key, val in args.items():
if val is not None:
if key != 'device_data':
# change 'device_id', 'device_name', 'device_type', 'device_controller_address'
device[key] = val
else:
# change device data (e.g. color of a light or the music station being played)
if 'Hue' in device['device_type']:
control_hue_lights.set_state(device['device_controller_address'], val)
elif device['device_type'] == 'Chromecast Audio':
for subkey, subval in val.items():
device['device_data'][subkey] = subval
control_chromecast_audio.update(device, val)
else:
return 'Device type "{}" not implemented yet'.format(device["device_type"]), 400
# if needed: call respective API to get new device data
if 'Hue' in device['device_type']:
new_lights_info = control_hue_lights.get_light_info(device['device_controller_address'])
for key, val in new_lights_info['state'].items():
device['device_data'][key] = val
# write back edited copy
shelve_db['devices'][device_index] = device
return marshal(device, device_fields), 200
@shelve_db_decorator
def delete(self, device_id):
device = [device for device in enumerate(shelve_db['devices']) if device[1]['device_id'] == device_id]
if device:
device_index = device[0][0]
device = device[0][1]
del shelve_db['devices'][device_index]
return 'Deleted device {}'.format(device), 200
else:
return 'There is no device with device_id "{}"'.format(device_id), 404
api.add_resource(DeviceListAPI, '/devices', endpoint='devices')
api.add_resource(DeviceAPI, '/devices/<device_id>', endpoint='device')
if __name__ == '__main__':
app.run(host='::', port='80', debug=True)