(self, n_time_in: int, n_time_out: int, n_x: int,
n_s: int, n_s_hidden: int, n_theta: int, n_mlp_units: list,
n_pool_kernel_size: int, pooling_mode: str, basis: nn.Module,
n_layers: int, batch_normalization: bool, dropout_prob: float, activation: str)
| 319 | N-BEATS block which takes a basis function as an argument. |
| 320 | """ |
| 321 | def __init__(self, n_time_in: int, n_time_out: int, n_x: int, |
| 322 | n_s: int, n_s_hidden: int, n_theta: int, n_mlp_units: list, |
| 323 | n_pool_kernel_size: int, pooling_mode: str, basis: nn.Module, |
| 324 | n_layers: int, batch_normalization: bool, dropout_prob: float, activation: str): |
| 325 | """ |
| 326 | """ |
| 327 | super().__init__() |
| 328 | |
| 329 | assert (pooling_mode in ['max','average']) |
| 330 | |
| 331 | n_time_in_pooled = int(np.ceil(n_time_in/n_pool_kernel_size)) |
| 332 | |
| 333 | if n_s == 0: |
| 334 | n_s_hidden = 0 |
| 335 | n_mlp_units = [n_time_in_pooled + (n_time_in+n_time_out)*n_x + n_s_hidden] + n_mlp_units |
| 336 | |
| 337 | self.n_time_in = n_time_in |
| 338 | self.n_time_out = n_time_out |
| 339 | self.n_s = n_s |
| 340 | self.n_s_hidden = n_s_hidden |
| 341 | self.n_x = n_x |
| 342 | self.n_pool_kernel_size = n_pool_kernel_size |
| 343 | self.batch_normalization = batch_normalization |
| 344 | self.dropout_prob = dropout_prob |
| 345 | |
| 346 | assert activation in ACTIVATIONS, f'{activation} is not in {ACTIVATIONS}' |
| 347 | activ = getattr(nn, activation)() |
| 348 | |
| 349 | if pooling_mode == 'max': |
| 350 | self.pooling_layer = nn.MaxPool1d(kernel_size=self.n_pool_kernel_size, |
| 351 | stride=self.n_pool_kernel_size, ceil_mode=True) |
| 352 | elif pooling_mode == 'average': |
| 353 | self.pooling_layer = nn.AvgPool1d(kernel_size=self.n_pool_kernel_size, |
| 354 | stride=self.n_pool_kernel_size, ceil_mode=True) |
| 355 | |
| 356 | hidden_layers = [] |
| 357 | for i in range(n_layers): |
| 358 | hidden_layers.append(nn.Linear(in_features=n_mlp_units[i], out_features=n_mlp_units[i+1])) |
| 359 | hidden_layers.append(activ) |
| 360 | |
| 361 | if self.batch_normalization: |
| 362 | hidden_layers.append(nn.BatchNorm1d(num_features=n_mlp_units[i+1])) |
| 363 | |
| 364 | if self.dropout_prob>0: |
| 365 | hidden_layers.append(nn.Dropout(p=self.dropout_prob)) |
| 366 | |
| 367 | output_layer = [nn.Linear(in_features=n_mlp_units[-1], out_features=n_theta)] |
| 368 | layers = hidden_layers + output_layer |
| 369 | |
| 370 | # n_s is computed with data, n_s_hidden is provided by user, if 0 no statics are used |
| 371 | if (self.n_s > 0) and (self.n_s_hidden > 0): |
| 372 | self.static_encoder = _StaticFeaturesEncoder(in_features=n_s, out_features=n_s_hidden) |
| 373 | self.layers = nn.Sequential(*layers) |
| 374 | self.basis = basis |
| 375 | |
| 376 | def forward(self, insample_y: t.Tensor, insample_x_t: t.Tensor, |
| 377 | outsample_x_t: t.Tensor, x_s: t.Tensor) -> Tuple[t.Tensor, t.Tensor]: |
nothing calls this directly
no test coverage detected