Fetches data from callback or provided with feed_input.
(self, epoch_idx)
| 234 | self._feed_inputs.put((data, kwargs)) |
| 235 | |
| 236 | def _fetch(self, epoch_idx): |
| 237 | """Fetches data from callback or provided with feed_input.""" |
| 238 | |
| 239 | def to_data_node_debug(data): |
| 240 | data = _transform_data_to_tensorlist(data, self._batch_size, layout, self._device_id) |
| 241 | |
| 242 | if self._device == "gpu" and isinstance(data, _tensors.TensorListCPU): |
| 243 | data = data._as_gpu() |
| 244 | elif self._device == "cpu" and isinstance(data, _tensors.TensorListGPU): |
| 245 | data = data.as_cpu() |
| 246 | warnings.warn( |
| 247 | "Loading GPU-originated data into CPU ExternalSource operator is " |
| 248 | "discouraged and might be inefficient", |
| 249 | Warning, |
| 250 | ) |
| 251 | return DataNodeDebug(data, self._name, self._device, self._source_desc) |
| 252 | |
| 253 | if self._callback is not None: |
| 254 | callback_out = self._get_batch(epoch_idx) |
| 255 | layout = self._layout |
| 256 | if self._num_outputs is not None: |
| 257 | raw_data = [] |
| 258 | for idx in range(self._num_outputs): |
| 259 | if self._batch: |
| 260 | raw_data.append(callback_out[idx]) |
| 261 | else: |
| 262 | raw_data.append([callback_out[i][idx] for i in range(self._batch_size)]) |
| 263 | else: |
| 264 | raw_data = callback_out |
| 265 | else: |
| 266 | raw_data, feed_input_params = self._feed_inputs.get() |
| 267 | layout = feed_input_params.get("layout", self._layout) |
| 268 | |
| 269 | if self._num_outputs is not None: |
| 270 | return [to_data_node_debug(data) for data in raw_data] |
| 271 | |
| 272 | return to_data_node_debug(raw_data) |
| 273 | |
| 274 | |
| 275 | class _IterBatchInfo: |
no test coverage detected