mirror of
https://github.com/modelscope/modelscope.git
synced 2025-12-21 10:39:24 +01:00
a1688f5775
1. build preprocessor for pipeline automatically if preprocessor is configed in configuration.json 2. refactor scrfd_detect.py as a standard cv model code 3. refacotr card_detection_pipeline face_detection_pipeline as standard pipeline code Link: https://code.alibaba-inc.com/Ali-MaaS/MaaS-lib/codereview/11057557
120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
# Copyright (c) Alibaba, Inc. and its affiliates.
|
|
|
|
import os
|
|
import unittest
|
|
from typing import Any, Dict, Union
|
|
|
|
import numpy as np
|
|
from PIL import Image
|
|
|
|
from modelscope.fileio import io
|
|
from modelscope.outputs import OutputKeys
|
|
from modelscope.pipelines import Pipeline, pipeline
|
|
from modelscope.pipelines.builder import PIPELINES, add_default_pipeline_info
|
|
from modelscope.utils.constant import (ConfigFields, Frameworks, ModelFile,
|
|
Tasks)
|
|
from modelscope.utils.logger import get_logger
|
|
|
|
logger = get_logger()
|
|
|
|
Input = Union[str, 'PIL.Image', 'numpy.ndarray']
|
|
|
|
|
|
class CustomPipelineTest(unittest.TestCase):
|
|
|
|
def setUp(self) -> None:
|
|
self.model_dir = '/tmp/custom-image'
|
|
self.prepare_dir(self.model_dir, 'custom-image')
|
|
|
|
def prepare_dir(self, dirname, pipeline_name):
|
|
if not os.path.exists(dirname):
|
|
os.makedirs(dirname)
|
|
cfg_file = os.path.join(dirname, ModelFile.CONFIGURATION)
|
|
cfg = {
|
|
ConfigFields.framework: 'dummy',
|
|
ConfigFields.task: 'dummy-task',
|
|
ConfigFields.pipeline: {
|
|
'type': pipeline_name,
|
|
}
|
|
}
|
|
io.dump(cfg, cfg_file)
|
|
|
|
def test_abstract(self):
|
|
|
|
@PIPELINES.register_module()
|
|
class CustomPipeline1(Pipeline):
|
|
|
|
def __init__(self,
|
|
config_file: str = None,
|
|
model=None,
|
|
preprocessor=None,
|
|
**kwargs):
|
|
super().__init__(config_file, model, preprocessor, **kwargs)
|
|
|
|
with self.assertRaises(TypeError):
|
|
CustomPipeline1()
|
|
|
|
def test_custom(self):
|
|
dummy_task = 'dummy-task'
|
|
|
|
@PIPELINES.register_module(
|
|
group_key=dummy_task, module_name='custom-image')
|
|
class CustomImagePipeline(Pipeline):
|
|
|
|
def __init__(self,
|
|
config_file: str = None,
|
|
model=None,
|
|
preprocessor=None,
|
|
**kwargs):
|
|
super().__init__(config_file, model, preprocessor, **kwargs)
|
|
|
|
def preprocess(self, input: Union[str,
|
|
'PIL.Image']) -> Dict[str, Any]:
|
|
""" Provide default implementation based on preprocess_cfg and user can reimplement it
|
|
|
|
"""
|
|
if not isinstance(input, Image.Image):
|
|
from modelscope.preprocessors import load_image
|
|
data_dict = {'img': load_image(input), 'url': input}
|
|
else:
|
|
data_dict = {'img': input}
|
|
return data_dict
|
|
|
|
def forward(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
|
|
""" Provide default implementation using self.model and user can reimplement it
|
|
"""
|
|
outputs = {}
|
|
if 'url' in inputs:
|
|
outputs['filename'] = inputs['url']
|
|
img = inputs['img']
|
|
new_image = img.resize((img.width // 2, img.height // 2))
|
|
outputs[OutputKeys.OUTPUT_IMG] = np.array(new_image)
|
|
return outputs
|
|
|
|
def postprocess(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
|
|
return inputs
|
|
|
|
self.assertTrue('custom-image' in PIPELINES.modules[dummy_task])
|
|
add_default_pipeline_info(dummy_task, 'custom-image', overwrite=True)
|
|
pipe = pipeline(
|
|
task=dummy_task,
|
|
pipeline_name='custom-image',
|
|
model=self.model_dir)
|
|
pipe2 = pipeline(dummy_task, model=self.model_dir)
|
|
self.assertTrue(type(pipe) is type(pipe2))
|
|
|
|
img_url = 'data/test/images/dogs.jpg'
|
|
output = pipe(img_url)
|
|
self.assertEqual(output['filename'], img_url)
|
|
self.assertEqual(output[OutputKeys.OUTPUT_IMG].shape, (318, 512, 3))
|
|
|
|
outputs = pipe([img_url for i in range(4)])
|
|
self.assertEqual(len(outputs), 4)
|
|
for out in outputs:
|
|
self.assertEqual(out['filename'], img_url)
|
|
self.assertEqual(out[OutputKeys.OUTPUT_IMG].shape, (318, 512, 3))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|