-
-
Notifications
You must be signed in to change notification settings - Fork 82
/
utils.py
206 lines (163 loc) · 7.26 KB
/
utils.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
"""
Data utils.
"""
# Created by Wenjie Du <[email protected]>
# License: BSD-3-Clause
from typing import Union
import numpy as np
import torch
def turn_data_into_specified_dtype(
data: Union[np.ndarray, torch.Tensor, list],
dtype: str = "tensor",
) -> Union[np.ndarray, torch.Tensor]:
"""Turn the given data into the specified data type."""
if isinstance(data, torch.Tensor):
data = data if dtype == "tensor" else data.numpy()
elif isinstance(data, list):
data = torch.tensor(data) if dtype == "tensor" else np.asarray(data)
elif isinstance(data, np.ndarray):
data = torch.from_numpy(data) if dtype == "tensor" else data
else:
raise TypeError(
f"data should be an instance of list/np.ndarray/torch.Tensor, but got {type(data)}"
)
return data
def _parse_delta_torch(missing_mask: torch.Tensor) -> torch.Tensor:
"""Generate the time-gap matrix (i.e. the delta metrix) from the missing mask.
Please refer to :cite:`che2018GRUD` for its math definition.
Parameters
----------
missing_mask : shape of [n_steps, n_features] or [n_samples, n_steps, n_features]
Binary masks indicate missing data (0 means missing values, 1 means observed values).
Returns
-------
delta :
The delta matrix indicates the time gaps between observed values.
With the same shape of missing_mask.
References
----------
.. [1] `Che, Zhengping, Sanjay Purushotham, Kyunghyun Cho, David Sontag, and Yan Liu.
"Recurrent neural networks for multivariate time series with missing values."
Scientific reports 8, no. 1 (2018): 6085.
<https://www.nature.com/articles/s41598-018-24271-9.pdf>`_
"""
def cal_delta_for_single_sample(mask: torch.Tensor) -> torch.Tensor:
"""calculate single sample's delta. The sample's shape is [n_steps, n_features]."""
# the first step in the delta matrix is all 0
d = [torch.zeros(1, n_features, device=device)]
for step in range(1, n_steps):
d.append(
torch.ones(1, n_features, device=device) + (1 - mask[step - 1]) * d[-1]
)
d = torch.concat(d, dim=0)
return d
device = missing_mask.device
if len(missing_mask.shape) == 2:
n_steps, n_features = missing_mask.shape
delta = cal_delta_for_single_sample(missing_mask)
else:
n_samples, n_steps, n_features = missing_mask.shape
delta_collector = []
for m_mask in missing_mask:
delta = cal_delta_for_single_sample(m_mask)
delta_collector.append(delta.unsqueeze(0))
delta = torch.concat(delta_collector, dim=0)
return delta
def _parse_delta_numpy(missing_mask: np.ndarray) -> np.ndarray:
"""Generate the time-gap matrix (i.e. the delta metrix) from the missing mask.
Please refer to :cite:`che2018GRUD` for its math definition.
Parameters
----------
missing_mask : shape of [n_steps, n_features] or [n_samples, n_steps, n_features]
Binary masks indicate missing data (0 means missing values, 1 means observed values).
Returns
-------
delta :
The delta matrix indicates the time gaps between observed values.
With the same shape of missing_mask.
References
----------
.. [1] `Che, Zhengping, Sanjay Purushotham, Kyunghyun Cho, David Sontag, and Yan Liu.
"Recurrent neural networks for multivariate time series with missing values."
Scientific reports 8, no. 1 (2018): 6085.
<https://www.nature.com/articles/s41598-018-24271-9.pdf>`_
"""
def cal_delta_for_single_sample(mask: np.ndarray) -> np.ndarray:
"""calculate single sample's delta. The sample's shape is [n_steps, n_features]."""
# the first step in the delta matrix is all 0
d = [np.zeros(n_features)]
for step in range(1, seq_len):
d.append(np.ones(n_features) + (1 - mask[step - 1]) * d[-1])
d = np.asarray(d)
return d
if len(missing_mask.shape) == 2:
seq_len, n_features = missing_mask.shape
delta = cal_delta_for_single_sample(missing_mask)
else:
n_samples, seq_len, n_features = missing_mask.shape
delta_collector = []
for m_mask in missing_mask:
delta = cal_delta_for_single_sample(m_mask)
delta_collector.append(delta)
delta = np.asarray(delta_collector)
return delta
def parse_delta(
missing_mask: Union[np.ndarray, torch.Tensor]
) -> Union[np.ndarray, torch.Tensor]:
"""Generate the time-gap matrix (i.e. the delta metrix) from the missing mask.
Please refer to :cite:`che2018GRUD` for its math definition.
Parameters
----------
missing_mask :
Binary masks indicate missing data (0 means missing values, 1 means observed values).
Shape of [n_steps, n_features] or [n_samples, n_steps, n_features].
Returns
-------
delta :
The delta matrix indicates the time gaps between observed values.
With the same shape of missing_mask.
References
----------
.. [1] `Che, Zhengping, Sanjay Purushotham, Kyunghyun Cho, David Sontag, and Yan Liu.
"Recurrent neural networks for multivariate time series with missing values."
Scientific reports 8, no. 1 (2018): 6085.
<https://www.nature.com/articles/s41598-018-24271-9.pdf>`_
"""
if isinstance(missing_mask, np.ndarray):
delta = _parse_delta_numpy(missing_mask)
elif isinstance(missing_mask, torch.Tensor):
delta = _parse_delta_torch(missing_mask)
else:
raise RuntimeError
return delta
def sliding_window(time_series, window_len, sliding_len=None):
"""Generate time series samples with sliding window method, truncating windows from time-series data
with a given sequence length.
Given a time series of shape [seq_len, n_features] (seq_len is the total sequence length of the time series), this
sliding_window function will generate time-series samples from this given time series with sliding window method.
The number of generated samples is seq_len//sliding_len. And the final returned numpy ndarray has a shape
[seq_len//sliding_len, n_steps, n_features].
Parameters
----------
time_series : np.ndarray,
time series data, len(shape)=2, [total_length, feature_num]
window_len : int,
The length of the sliding window, i.e. the number of time steps in the generated data samples.
sliding_len : int, default = None,
The sliding length of the window for each moving step. It will be set as the same with n_steps if None.
Returns
-------
samples : np.ndarray,
The generated time-series data samples of shape [seq_len//sliding_len, n_steps, n_features].
"""
sliding_len = window_len if sliding_len is None else sliding_len
total_len = time_series.shape[0]
start_indices = np.asarray(range(total_len // sliding_len)) * sliding_len
# remove the last one if left length is not enough
if total_len - start_indices[-1] * sliding_len < window_len:
start_indices = start_indices[:-1]
sample_collector = []
for idx in start_indices:
sample_collector.append(time_series[idx : idx + window_len])
samples = np.asarray(sample_collector).astype("float32")
return samples