(self, input)
| 349 | return output_list |
| 350 | |
| 351 | def _check_input(self, input): |
| 352 | task_name = self.group_key |
| 353 | if task_name in TASK_INPUTS: |
| 354 | input_type = TASK_INPUTS[task_name] |
| 355 | |
| 356 | # if multiple input formats are defined, we first |
| 357 | # found the one that match input data and check |
| 358 | if isinstance(input_type, list): |
| 359 | matched_type = None |
| 360 | for t in input_type: |
| 361 | if isinstance(input, (dict, tuple)): |
| 362 | if type(t) == type(input): |
| 363 | matched_type = t |
| 364 | break |
| 365 | elif isinstance(t, str): |
| 366 | matched_type = t |
| 367 | break |
| 368 | if matched_type is None: |
| 369 | err_msg = 'input data format for current pipeline should be one of following: \n' |
| 370 | for t in input_type: |
| 371 | err_msg += f'{t}\n' |
| 372 | raise ValueError(err_msg) |
| 373 | else: |
| 374 | input_type = matched_type |
| 375 | |
| 376 | if isinstance(input_type, str): |
| 377 | check_input_type(input_type, input) |
| 378 | elif isinstance(input_type, tuple): |
| 379 | assert isinstance(input, tuple), 'input should be a tuple' |
| 380 | for t, input_ele in zip(input_type, input): |
| 381 | check_input_type(t, input_ele) |
| 382 | elif isinstance(input_type, dict): |
| 383 | for k in input_type.keys(): |
| 384 | # allow single input for multi-modal models |
| 385 | if isinstance(input, dict) and k in input: |
| 386 | check_input_type(input_type[k], input[k]) |
| 387 | else: |
| 388 | raise ValueError(f'invalid input_type definition {input_type}') |
| 389 | elif not getattr(self, '_input_has_warned', False): |
| 390 | logger.warning(f'task {task_name} input definition is missing') |
| 391 | self._input_has_warned = True |
| 392 | |
| 393 | def _check_output(self, input): |
| 394 | # this attribute is dynamically attached by registry |
no test coverage detected