| 16 | |
| 17 | |
| 18 | class DeployChecker: |
| 19 | |
| 20 | def __init__(self): |
| 21 | self.api = HubApi() |
| 22 | |
| 23 | def check_model(self, model_id: str, model_revision=None): |
| 24 | # get model_revision & task info |
| 25 | if not model_revision: |
| 26 | model_revisions = self.api.list_model_revisions(model_id) |
| 27 | logger.info( |
| 28 | f'All model_revisions of `{model_id}`: {model_revisions}') |
| 29 | if len(model_revisions): |
| 30 | model_revision = model_revisions[0] |
| 31 | else: |
| 32 | logger.error(f'{model_id} has no revision.') |
| 33 | |
| 34 | configuration_file = model_file_download( |
| 35 | model_id=model_id, |
| 36 | file_path=ModelFile.CONFIGURATION, |
| 37 | revision=model_revision) |
| 38 | cfg = Config.from_file(configuration_file) |
| 39 | task = cfg.safe_get('task') |
| 40 | |
| 41 | # init pipeline |
| 42 | ppl = pipeline( |
| 43 | task=task, |
| 44 | model=model_id, |
| 45 | model_revision=model_revision, |
| 46 | external_engine_for_llm=True) |
| 47 | pipeline_info = get_pipeline_information_by_pipeline(ppl) |
| 48 | |
| 49 | # call pipeline |
| 50 | data = get_task_input_examples(task) |
| 51 | |
| 52 | infer_result = call_pipeline_with_json(pipeline_info, ppl, data) |
| 53 | result = pipeline_output_to_service_base64_output(task, infer_result) |
| 54 | return result |
| 55 | |
| 56 | |
| 57 | def check_deploy(models: Union[str, List], revisions: Union[str, List] = None): |
no outgoing calls
no test coverage detected
searching dependent graphs…