-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathdataset.py
116 lines (109 loc) · 4.91 KB
/
dataset.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
from torch.utils.data import Dataset
import numpy as np
class CSI_dataset(Dataset):
def __init__(self, magnitudes, label_action=None, label_people=None):
super().__init__()
self.magnitudes = magnitudes
self.label_action = label_action
self.label_people = label_people
self.num=self.magnitudes.shape[0]
if self.label_action is None:
self.label_action = [-1] * self.num
if self.label_people is None:
self.label_people = [-1] * self.num
def __len__(self):
return self.num
def __getitem__(self, index):
return self.magnitudes[index], self.label_action[index], self.label_people[index]
def load_data(data_path="./data",train_prop=None,valid_prop=None):
magnitude = np.load(data_path+"/magnitude_linear.npy").astype(np.float32)
phase = np.load(data_path+"/phase_linear.npy").astype(np.float32)
magnitude=np.concatenate([np.expand_dims(magnitude, axis=1) ,np.expand_dims(phase, axis=1)],axis=1)
people=np.load(data_path+"/people.npy").astype(np.int64)
action=np.load(data_path+"/action.npy").astype(np.int64)
if train_prop is None:
return CSI_dataset(magnitude, action, people)
else:
a = np.zeros_like(people)
num=[]
current_num=0
current_action=None
for i in range(action.shape[0]):
if action[i]==current_action:
current_num+=1
else:
current_action = action[i]
if current_action is None:
current_num+=1
else:
num.append(current_num)
current_num=0
num.append(current_num)
if valid_prop is None:
current_num=0
for i in range(len(num)):
a[current_num:current_num+int(num[i]*train_prop)]=1
current_num+=num[i]
b=1-a
a = a.astype(bool)
b = b.astype(bool)
return CSI_dataset(magnitude[a], action[a], people[a]), CSI_dataset(magnitude[b], action[b], people[b])
else:
current_num=0
b = np.zeros_like(people)
for i in range(len(num)):
a[current_num:current_num+int(num[i]*train_prop)]=1
b[current_num+int(num[i]*train_prop):current_num+int(num[i]*(train_prop+valid_prop))]=1
current_num+=num[i]
c=1-a-b
a = a.astype(bool)
b = b.astype(bool)
c = c.astype(bool)
return CSI_dataset(magnitude[a], action[a], people[a]), CSI_dataset(magnitude[b], action[b], people[b]), CSI_dataset(magnitude[c], action[c], people[c])
def load_zero_people(test_people_list, data_path="./data"):
magnitude = np.load(data_path+"/magnitude_linear.npy").astype(np.float32)
phase = np.load(data_path+"/phase_linear.npy").astype(np.float32)
magnitude=np.concatenate([np.expand_dims(magnitude, axis=1) ,np.expand_dims(phase, axis=1)],axis=1)
people=np.load(data_path+"/people.npy").astype(np.int64)
action=np.load(data_path+"/action.npy").astype(np.int64)
a = np.zeros_like(people)
b = np.zeros_like(people)
for i in range(people.shape[0]):
a[i]=(people[i] not in test_people_list)
b[i]=not a[i]
a=a.astype(bool)
b=b.astype(bool)
return CSI_dataset(magnitude[a], action[a], people[a]),CSI_dataset(magnitude[b], action[b], people[b])
def load_zero_shot(test_people_list=None, test_action_list=None, data_path="./data", func="and"):
magnitude = np.load(data_path+"/magnitude_linear.npy").astype(np.float32)
phase = np.load(data_path+"/phase_linear.npy").astype(np.float32)
magnitude=np.concatenate([np.expand_dims(magnitude, axis=1) ,np.expand_dims(phase, axis=1)],axis=1)
people=np.load(data_path+"/people.npy").astype(np.int64)
action=np.load(data_path+"/action.npy").astype(np.int64)
a = np.zeros_like(people)
b = np.zeros_like(people)
if test_action_list is None and test_people_list is None:
return CSI_dataset(magnitude, action, people)
elif test_action_list is None:
for i in range(people.shape[0]):
a[i]=(people[i] not in test_people_list)
b[i]=not a[i]
elif test_people_list is None:
for i in range(people.shape[0]):
a[i]=(action[i] not in test_action_list)
b[i]=not a[i]
else:
if func=="and":
for i in range(people.shape[0]):
a[i]=(action[i] not in test_action_list or people[i] not in test_people_list)
b[i]=not a[i]
elif func=="or":
for i in range(people.shape[0]):
a[i]=(action[i] not in test_action_list and people[i] not in test_people_list)
b[i]=not a[i]
else:
print("ERROR")
exit(-1)
a=a.astype(bool)
b=b.astype(bool)
return CSI_dataset(magnitude[a], action[a], people[a]),CSI_dataset(magnitude[b], action[b], people[b])