Duplicate global features at all layers
| 209 | |
| 210 | |
| 211 | class DuplicateFeaturesTransform(OnlyLayersAreNodesTransforms): |
| 212 | """ |
| 213 | Duplicate global features at all layers |
| 214 | """ |
| 215 | |
| 216 | def __init__(self, |
| 217 | level_policy: str = "drop", |
| 218 | drop_last_level: bool = True, |
| 219 | use_level_features: bool = True, |
| 220 | drop_node_encoding: bool = True, |
| 221 | node_encoding_len: int = 3, |
| 222 | *args, **kwargs |
| 223 | ): |
| 224 | """ |
| 225 | Duplicate global (and optionally level) features across all layers |
| 226 | |
| 227 | Args: |
| 228 | level_policy: If 'drop', level features are simply dropped, |
| 229 | If 'concat', all levels but one are concatenated to its adjacent layer |
| 230 | drop_last_level: Only used if level_policy is concat, i.e. all levels but one are concatenated to the layers. |
| 231 | """ |
| 232 | super().__init__(*args, **kwargs) |
| 233 | self.level_policy = level_policy |
| 234 | self.drop_last_level = drop_last_level |
| 235 | self.use_level_features = use_level_features |
| 236 | self.drop_node_encoding = drop_node_encoding |
| 237 | self.node_encoding_len = node_encoding_len |
| 238 | |
| 239 | if self.use_level_features: |
| 240 | self._out_dim = sum([self.input_dim[GLOBALS], self.input_dim[LAYERS], self.input_dim[LEVELS]]) |
| 241 | else: |
| 242 | self.log.info(' Dropping level features!') |
| 243 | self._out_dim = sum([self.input_dim[GLOBALS], self.input_dim[LAYERS]]) |
| 244 | |
| 245 | if self.drop_node_encoding: |
| 246 | self._out_dim -= 2 * self.node_encoding_len if not self.use_level_features else 3 * self.node_encoding_len |
| 247 | |
| 248 | def transform(self, X: Dict[str, np.ndarray]) -> np.ndarray: |
| 249 | """ |
| 250 | Returns: |
| 251 | A (b, #layers, d) array, where d = #layer-feats + #global-feats (+ #levels-feats, if not drop_levels) |
| 252 | """ |
| 253 | global_node, levels, layers = X[GLOBALS], X[LEVELS], X[LAYERS] |
| 254 | if self.drop_node_encoding: |
| 255 | global_node = global_node[:, :-self.node_encoding_len] |
| 256 | layers = layers[:, :, :-self.node_encoding_len] |
| 257 | levels = levels[:, :, :-self.node_encoding_len] |
| 258 | data_size, n_layers, n_layer_feats = layers.shape |
| 259 | n_global_feats, n_level_feats = global_node.shape[-1], levels.shape[-1] |
| 260 | n_layers_feats_with_globals = n_layer_feats + n_global_feats |
| 261 | |
| 262 | if self.use_level_features: |
| 263 | n_layers_feats_with_globals += n_level_feats |
| 264 | |
| 265 | all_data = np.zeros((data_size, n_layers, n_layers_feats_with_globals)) |
| 266 | all_data[:, :, :n_layer_feats] = layers |
| 267 | all_data[:, :, n_layer_feats:n_layer_feats + n_global_feats] = global_node[:, None, :] |
| 268 | if self.use_level_features: |
nothing calls this directly
no outgoing calls
no test coverage detected