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
|
|
|
|