(self)
| 477 | self._analyze() |
| 478 | |
| 479 | def _analyze(self): |
| 480 | input, parameters = get_pipeline_input_parameters( |
| 481 | self._source_path, self._class_name) |
| 482 | # use base pipeline __call__ if inputs and outputs are defined in modelscope lib |
| 483 | if self._task_name in TASK_INPUTS and self._task_name in TASK_OUTPUTS: |
| 484 | # delete the first default input which is defined by task. |
| 485 | if parameters is None: |
| 486 | self._parameters_schema = {} |
| 487 | else: |
| 488 | self._parameters_schema = generate_pipeline_parameters_schema( |
| 489 | parameters) |
| 490 | self._input_schema = get_input_schema(self._task_name, None) |
| 491 | self._output_schema = get_output_schema(self._task_name) |
| 492 | elif input is not None: # custom pipeline implemented it's own __call__ method |
| 493 | self._is_custom_call_method = True |
| 494 | self._input_schema = generate_pipeline_parameters_schema(input) |
| 495 | self._input_schema[ |
| 496 | 'description'] = 'For binary input such as image audio video, only url is supported.' |
| 497 | self._parameters_schema = {} |
| 498 | self._output_schema = { |
| 499 | 'type': 'object', |
| 500 | } |
| 501 | if self._task_name in TASK_OUTPUTS: |
| 502 | self._output_schema = get_output_schema(self._task_name) |
| 503 | else: |
| 504 | logger.warning( |
| 505 | 'Task: %s input is defined: %s, output is defined: %s which is not completed' |
| 506 | % (self._task_name, self._task_name |
| 507 | in TASK_INPUTS, self._task_name in TASK_OUTPUTS)) |
| 508 | self._input_schema = None |
| 509 | self._output_schema = None |
| 510 | if self._task_name in TASK_INPUTS: |
| 511 | self._input_schema = get_input_schema(self._task_name, None) |
| 512 | if self._task_name in TASK_OUTPUTS: |
| 513 | self._output_schema = get_output_schema(self._task_name) |
| 514 | self._parameters_schema = generate_pipeline_parameters_schema( |
| 515 | parameters) |
| 516 | |
| 517 | @property |
| 518 | def task_name(self): |
no test coverage detected