|
1
|
|
|
import json |
|
2
|
|
|
import os |
|
3
|
|
|
import logging |
|
4
|
|
|
import numpy as np |
|
5
|
|
|
import matplotlib.image as mimg |
|
6
|
|
|
import matplotlib.pyplot as plt |
|
7
|
|
|
import matplotlib.patches as patches |
|
8
|
|
|
from ..utils.LabeledLine import LabeledLine |
|
9
|
|
|
|
|
10
|
|
|
logger = logging.getLogger(__name__) |
|
11
|
|
|
|
|
12
|
|
|
|
|
13
|
|
|
# noinspection PyPackageRequirements |
|
14
|
|
|
class CASASHome: |
|
15
|
|
|
"""Load Home Data Structure from JSON file |
|
16
|
|
|
|
|
17
|
|
|
Attributes: |
|
18
|
|
|
data_dict (:obj:`dict`): A dictionary contains information about smart home. |
|
19
|
|
|
directory (:obj:`str`): Directory that stores CASAS smart home data |
|
20
|
|
|
|
|
21
|
|
|
Parameters: |
|
22
|
|
|
directory (:obj:`str`): Directory that stores CASAS smart home data |
|
23
|
|
|
""" |
|
24
|
|
|
def __init__(self, directory): |
|
25
|
|
|
self.directory = directory |
|
26
|
|
|
dataset_json_fname = directory + '/dataset.json' |
|
27
|
|
|
if os.path.exists(dataset_json_fname): |
|
28
|
|
|
f = open(dataset_json_fname, 'r') |
|
29
|
|
|
self.data_dict = json.load(f) |
|
30
|
|
|
else: |
|
31
|
|
|
logger.error('Smart home metadata file %s does not exist. Create an empty CASASHome Structure' |
|
32
|
|
|
% dataset_json_fname) |
|
33
|
|
|
raise FileNotFoundError('File %s not found.' % dataset_json_fname) |
|
34
|
|
|
# self.data_dict = { |
|
35
|
|
|
# 'name': '', |
|
36
|
|
|
# 'floorplan': '', |
|
37
|
|
|
# 'sensors': [], |
|
38
|
|
|
# 'activities': [], |
|
39
|
|
|
# 'residents': [] |
|
40
|
|
|
# } |
|
41
|
|
|
|
|
42
|
|
|
def get_name(self): |
|
43
|
|
|
"""Get the smart home name |
|
44
|
|
|
|
|
45
|
|
|
Returns: |
|
46
|
|
|
:obj:`str`: smart home name |
|
47
|
|
|
""" |
|
48
|
|
|
return self.data_dict['name'] |
|
49
|
|
|
|
|
50
|
|
|
def get_all_activities(self): |
|
51
|
|
|
"""Get All Activities |
|
52
|
|
|
|
|
53
|
|
|
Returns: |
|
54
|
|
|
:obj:`list` of :obj:`str`: list of activity names |
|
55
|
|
|
""" |
|
56
|
|
|
names = [activity['name'] for activity in self.data_dict['activities']] |
|
57
|
|
|
return names |
|
58
|
|
|
|
|
59
|
|
|
def get_activity(self, label): |
|
60
|
|
|
"""Find the information about the activity |
|
61
|
|
|
|
|
62
|
|
|
Parameters: |
|
63
|
|
|
label (:obj:`str`): activity label |
|
64
|
|
|
|
|
65
|
|
|
Returns: |
|
66
|
|
|
:obj:`dict`: A dictionary containing activity information |
|
67
|
|
|
""" |
|
68
|
|
|
for activity in self.data_dict['activities']: |
|
69
|
|
|
if activity['name'] == label: |
|
70
|
|
|
return activity |
|
71
|
|
|
return None |
|
72
|
|
|
|
|
73
|
|
|
def get_activity_color(self, label): |
|
74
|
|
|
"""Find the color string of the activity |
|
75
|
|
|
|
|
76
|
|
|
Parameters: |
|
77
|
|
|
label (:obj:`str`): activity label |
|
78
|
|
|
|
|
79
|
|
|
Returns: |
|
80
|
|
|
:obj:`str`: RGB color string |
|
81
|
|
|
""" |
|
82
|
|
|
activity = self.get_activity(label) |
|
83
|
|
|
if activity is not None: |
|
84
|
|
|
return "#" + activity['color'][3:9] |
|
85
|
|
|
else: |
|
86
|
|
|
raise ValueError('Activity %s Not Found' % label) |
|
87
|
|
|
|
|
88
|
|
|
def get_sensor(self, name): |
|
89
|
|
|
"""Get the information about the sensor |
|
90
|
|
|
|
|
91
|
|
|
Parameters: |
|
92
|
|
|
name (:obj:`str`): name of the sensor |
|
93
|
|
|
|
|
94
|
|
|
Returns: |
|
95
|
|
|
:obj:`dict`: A dictionary that stores sensor information |
|
96
|
|
|
""" |
|
97
|
|
|
for sensor in self.data_dict['sensors']: |
|
98
|
|
|
if sensor['name'] == name: |
|
99
|
|
|
return sensor |
|
100
|
|
|
return None |
|
101
|
|
|
|
|
102
|
|
|
def get_all_sensors(self): |
|
103
|
|
|
"""Get All Sensor Names |
|
104
|
|
|
|
|
105
|
|
|
Returns: |
|
106
|
|
|
:obj:`list` of :obj:`str`: a list of sensor names |
|
107
|
|
|
""" |
|
108
|
|
|
names = [sensor['name'] for sensor in self.data_dict['sensors']] |
|
109
|
|
|
return names |
|
110
|
|
|
|
|
111
|
|
|
def get_resident(self, name): |
|
112
|
|
|
"""Get Information about the resident |
|
113
|
|
|
|
|
114
|
|
|
Parameters: |
|
115
|
|
|
name (:obj:`str`): name of the resident |
|
116
|
|
|
|
|
117
|
|
|
Returns: |
|
118
|
|
|
:obj:`dict`: A Dictionary that stores resident information |
|
119
|
|
|
""" |
|
120
|
|
|
for resident in self.data_dict['residents']: |
|
121
|
|
|
if resident['name'] == name: |
|
122
|
|
|
return resident |
|
123
|
|
|
return None |
|
124
|
|
|
|
|
125
|
|
|
def get_resident_color(self, name): |
|
126
|
|
|
"""Get the color string for the resident |
|
127
|
|
|
|
|
128
|
|
|
Parameters: |
|
129
|
|
|
name (:obj:`str`): name of the resident |
|
130
|
|
|
|
|
131
|
|
|
Returns: |
|
132
|
|
|
:obj:`str`: RGB color string representing the resident |
|
133
|
|
|
""" |
|
134
|
|
|
resident = self.get_resident(name) |
|
135
|
|
|
if resident is not None: |
|
136
|
|
|
return "#" + resident['color'][3:9] |
|
137
|
|
|
else: |
|
138
|
|
|
raise ValueError('Resident %s Not Found' % name) |
|
139
|
|
|
|
|
140
|
|
|
def get_all_residents(self): |
|
141
|
|
|
"""Get All Resident Names |
|
142
|
|
|
|
|
143
|
|
|
Returns: |
|
144
|
|
|
:obj:`list` of :obj:`str`: A list of resident names |
|
145
|
|
|
""" |
|
146
|
|
|
names = [resident['name'] for resident in self.data_dict['residents']] |
|
147
|
|
|
return names |
|
148
|
|
|
|
|
149
|
|
|
def _prepare_floorplan(self): |
|
150
|
|
|
"""Prepare the floorplan for drawing |
|
151
|
|
|
|
|
152
|
|
|
Returns: |
|
153
|
|
|
:obj:`dict`: A dictionary contains all the pieces needed to draw the floorplan |
|
154
|
|
|
""" |
|
155
|
|
|
floorplan_dict = {} |
|
156
|
|
|
img = mimg.imread(os.path.join(self.directory, self.data_dict['floorplan'])) |
|
157
|
|
|
img_x = img.shape[1] |
|
158
|
|
|
img_y = img.shape[0] |
|
159
|
|
|
# Create Sensor List/Patches |
|
160
|
|
|
sensor_boxes = {} |
|
161
|
|
|
sensor_texts = {} |
|
162
|
|
|
sensor_centers = {} |
|
163
|
|
|
# Check Bias |
|
164
|
|
|
for sensor in self.data_dict['sensors']: |
|
165
|
|
|
loc_x = sensor['locX'] * img_x |
|
166
|
|
|
loc_y = sensor['locY'] * img_y |
|
167
|
|
|
size_x = sensor['sizeX'] * img_x |
|
168
|
|
|
size_y = sensor['sizeY'] * img_y |
|
169
|
|
|
sensor_center_x = loc_x + size_x / 2 |
|
170
|
|
|
sensor_center_y = loc_y + size_y / 2 |
|
171
|
|
|
sensor_boxes[sensor['name']] = \ |
|
172
|
|
|
patches.Rectangle((loc_x, loc_y), size_x, size_y, |
|
173
|
|
|
edgecolor='grey', facecolor='orange', linewidth=1, |
|
174
|
|
|
zorder=2) |
|
175
|
|
|
sensor_texts[sensor['name']] = (loc_x + size_x / 2, loc_y + size_y / 2, sensor['name']) |
|
176
|
|
|
sensor_centers[sensor['name']] = (sensor_center_x, sensor_center_y) |
|
177
|
|
|
# Populate dictionary |
|
178
|
|
|
floorplan_dict['img'] = img |
|
179
|
|
|
floorplan_dict['width'] = img_x |
|
180
|
|
|
floorplan_dict['height'] = img_y |
|
181
|
|
|
floorplan_dict['sensor_centers'] = sensor_centers |
|
182
|
|
|
floorplan_dict['sensor_boxes'] = sensor_boxes |
|
183
|
|
|
floorplan_dict['sensor_texts'] = sensor_texts |
|
184
|
|
|
return floorplan_dict |
|
185
|
|
|
|
|
186
|
|
|
def draw_floorplan(self, filename=None): |
|
187
|
|
|
"""Draw the floorplan of the house, save it to file or display it on screen |
|
188
|
|
|
|
|
189
|
|
|
Args: |
|
190
|
|
|
filename (:obj:`str`): Name of the file to save the floorplan to |
|
191
|
|
|
""" |
|
192
|
|
|
floorplan_dict = self._prepare_floorplan() |
|
193
|
|
|
self._plot_floorplan(floorplan_dict, filename) |
|
194
|
|
|
|
|
195
|
|
|
@staticmethod |
|
196
|
|
|
def _plot_floorplan(floorplan_dict, filename=None): |
|
197
|
|
|
fig, (ax) = plt.subplots(1, 1) |
|
198
|
|
|
fig.set_size_inches(18, 18) |
|
199
|
|
|
ax.imshow(floorplan_dict['img']) |
|
200
|
|
|
# Draw Sensor block patches |
|
201
|
|
|
for key, patch in floorplan_dict['sensor_boxes'].items(): |
|
202
|
|
|
ax.add_patch(patch) |
|
203
|
|
|
# Draw Sensor name |
|
204
|
|
|
for key, text in floorplan_dict['sensor_texts'].items(): |
|
205
|
|
|
ax.text(*text, color='black', |
|
206
|
|
|
horizontalalignment='center', verticalalignment='center', |
|
207
|
|
|
zorder=3) |
|
208
|
|
|
if floorplan_dict.get('sensor_lines', None) is not None: |
|
209
|
|
|
for key, line in floorplan_dict['sensor_lines'].items(): |
|
210
|
|
|
ax.add_line(line) |
|
211
|
|
|
if filename is None: |
|
212
|
|
|
# Show image |
|
213
|
|
|
fig.show() |
|
214
|
|
|
else: |
|
215
|
|
|
fig.savefig(filename) |
|
216
|
|
|
plt.close(fig) |
|
217
|
|
|
|
|
218
|
|
|
def plot_sensor_distance(self, sensor_name, distance_matrix, max_sensors=None, filename=None): |
|
219
|
|
|
"""Plot distance in distance_matrix |
|
220
|
|
|
""" |
|
221
|
|
|
sensor_index = self.get_all_sensors().index(sensor_name) |
|
222
|
|
|
num_sensors = len(self.data_dict['sensors']) |
|
223
|
|
|
floorplan_dict = self._prepare_floorplan() |
|
224
|
|
|
x1 = floorplan_dict['sensor_centers'][sensor_name][0] |
|
225
|
|
|
y1 = floorplan_dict['sensor_centers'][sensor_name][1] |
|
226
|
|
|
# Draw Lines, and Set alpha for each sensor box |
|
227
|
|
|
sensor_lines ={} |
|
228
|
|
|
for i in range(num_sensors): |
|
229
|
|
|
sensor = self.data_dict['sensors'][i] |
|
230
|
|
|
if sensor_name != sensor['name']: |
|
231
|
|
|
x2 = floorplan_dict['sensor_centers'][sensor['name']][0] |
|
232
|
|
|
y2 = floorplan_dict['sensor_centers'][sensor['name']][1] |
|
233
|
|
|
line = LabeledLine([x1, x2], [y1, y2], linewidth=1, |
|
234
|
|
|
linestyle='--', color='b', zorder=10, |
|
235
|
|
|
label='%.5f' % distance_matrix[sensor_index, i], |
|
236
|
|
|
alpha=(1 - distance_matrix[sensor_index, i]) * 0.9 + 0.1) |
|
237
|
|
|
sensor_lines[sensor['name']] = line |
|
238
|
|
|
floorplan_dict['sensor_boxes'][sensor['name']].set_alpha(1 - distance_matrix[sensor_index, i]) |
|
239
|
|
|
# Only show up to `max_lines` of sensors |
|
240
|
|
|
if max_sensors is not None and max_sensors < num_sensors: |
|
241
|
|
|
sorted_index = np.argsort(distance_matrix[sensor_index, :]) |
|
242
|
|
|
for i in range(max_sensors + 1, num_sensors): |
|
243
|
|
|
sensor_lines.pop(self.data_dict['sensors'][sorted_index[i]]['name'], None) |
|
244
|
|
|
floorplan_dict['sensor_lines'] = sensor_lines |
|
245
|
|
|
self._plot_floorplan(floorplan_dict, filename) |
|
246
|
|
|
|