(self, input)
| 391 | self._input_has_warned = True |
| 392 | |
| 393 | def _check_output(self, input): |
| 394 | # this attribute is dynamically attached by registry |
| 395 | # when cls is registered in registry using task name |
| 396 | task_name = self.group_key |
| 397 | if task_name not in TASK_OUTPUTS: |
| 398 | if not getattr(self, '_output_has_warned', False): |
| 399 | logger.warning(f'task {task_name} output keys are missing') |
| 400 | self._output_has_warned = True |
| 401 | return |
| 402 | output_keys = TASK_OUTPUTS[task_name] |
| 403 | missing_keys = [] |
| 404 | input = input.keys() if isinstance(input, |
| 405 | (dict, ModelOutputBase)) else input |
| 406 | for k in output_keys: |
| 407 | if isinstance(k, (dict, ModelOutputBase)) and k not in input: |
| 408 | missing_keys.append(k) |
| 409 | if len(missing_keys) > 0: |
| 410 | raise ValueError(f'expected output keys are {output_keys}, ' |
| 411 | f'those {missing_keys} are missing') |
| 412 | |
| 413 | def preprocess(self, inputs: Input, **preprocess_params) -> Dict[str, Any]: |
| 414 | """ Provide default implementation based on preprocess_cfg and user can reimplement it |
no test coverage detected