(self, model_id: str, model_revision=None)
| 19 | self.api = HubApi() |
| 20 | |
| 21 | def test_single(self, model_id: str, model_revision=None): |
| 22 | # get model_revision & task info |
| 23 | configuration_file = os.path.join( |
| 24 | get_model_cache_dir(model_id), ModelFile.CONFIGURATION) |
| 25 | if not model_revision: |
| 26 | model_revision = self.api.list_model_revisions( |
| 27 | model_id=model_id)[0] |
| 28 | if not os.path.exists(configuration_file): |
| 29 | |
| 30 | configuration_file = model_file_download( |
| 31 | model_id=model_id, |
| 32 | file_path=ModelFile.CONFIGURATION, |
| 33 | revision=model_revision) |
| 34 | cfg = Config.from_file(configuration_file) |
| 35 | task = cfg.safe_get('task') |
| 36 | |
| 37 | # init pipeline |
| 38 | ppl = pipeline( |
| 39 | task=task, |
| 40 | model=model_id, |
| 41 | model_revision=model_revision, |
| 42 | external_engine_for_llm=True) |
| 43 | pipeline_info = get_pipeline_information_by_pipeline(ppl) |
| 44 | |
| 45 | # call pipeline |
| 46 | data = get_task_input_examples(task) |
| 47 | |
| 48 | infer_result = call_pipeline_with_json(pipeline_info, ppl, data) |
| 49 | result = pipeline_output_to_service_base64_output(task, infer_result) |
| 50 | return result |
| 51 | |
| 52 | |
| 53 | if __name__ == '__main__': |
no test coverage detected