AI Earth如何接入外部三方模型

我爱免费

作者：L兄自有格局
大规模遥感数据处理通常面临两个困难：一是需要加载GDAL、fiona等专业库来处理栅格和矢量数据；二是因为数据量太大很难将影像直接传递给深度学习模型。因此，对于没有遥感、AI和分布式计算专业背景的用户来说，将深度学习模型应用于大规模遥感影像高效分析处理，是一件非常具有挑战性的事情。
为解决其中的一些挑战，AI Earth云平台推出了AI Earth Predict SDK，将时空数据和AI模型进行相结合，用于进行大规模遥感数据模型推理开发，且支持并行处理、本地运行，同时支持对接外部的三方模型，将三方模型快速地部署到AI Earth云平台，和云平台已有的Engine、Deeplearning、Data等SDK共同组成云平台的SDK系列家族，将集成后的能力透出给用户。



AI Earth Predict SDK做到了代码级开源，目前已经推到了Github上：aiearth-damo/predict，在使用过程中如果遇到问题或者建议可以在Github创建相关 Issues ，如果您想贡献代码，欢迎checkout分支添加代码之后提交 Pull requests，更多详细使用文档请参考predict docs。

三方模型示例

以图像分割任务里面的全景分割子任务为例，介绍一下对于一个具体的任务，如何快速集成三方模型，并将遥感影像解译pipeline部署到云平台的详细流程。
图像分割任务需要根据不同的语义信息区分并聚集起对应相同语义的像素点，大体上图像分割任务可以分为三个子任务：实例分割 (instance segmentation) 、语义分割 (semantic segmentation) 、全景分割 (panoptic segmentation)。其中全景分割是将图像划分为语义区域和对象实例的任务，而且不同的部分间不存在重叠。
明确了具体任务之后就可以去开源模型社区选择相应任务的模型文件，比如HuggingFace和ModelScope，这里以ModelScope模型库为例，依次选择计算机视觉->视觉分割->通用图像分割，然后选择按照下载量排序，我们就选择排名第一个的Mask2Former-SwinL全景分割模型。



Mask2Former是一种能够解决任何图像分割任务（全景、实例或语义）的新架构。它包含了一个masked attention结构，通过将交叉注意力计算内来提取局部特征。
安装ModelScope依赖：pip install torch -i https://mirrors.aliyun.com/pypi/simple/pip install "modelscope[cv]" -f https://modelscope.oss-cn-beijing.aliyuncs.com/releases/repo.htmlpip install openmim && mim install mmcv-full测试跑通ModelScope代码范例：from modelscope.pipelines import pipelinefrom modelscope.utils.constant import Tasksfrom modelscope.utils.cv.image_utils import panoptic_seg_masks_to_imagefrom modelscope.outputs import OutputKeysimport cv2
segmentor = pipeline(Tasks.image_segmentation, model='damo/cv_swinL_panoptic-segmentation_cocopan')input_url = 'https://modelscope.oss-cn-beijing.aliyuncs.com/test/images/image_panoptic_segmentation.jpg'result = segmentor(input_url)draw_img = panoptic_seg_masks_to_image(result[OutputKeys.MASKS])cv2.imwrite('result.jpg', draw_img)

平台集成三方模型流程

集成时建议先在本地测试模型任务Pipeline，本地测试跑通之后可以部署到云平台GUI模式工具箱，利用云平台的算力提交遥感AI解译任务。

一. 本地模型pipeline测试

1. 安装predict SDK

pip install aiearth-predict

2. 实现predictor

Predictor是执行推理的包装类，需要从Checkpoint加载模型来进行实例化。对于自定义的Predictor，需要继承 Predictor ，并实现 __init__ 、 _predict_numpy 接口：

__init__: 从checkpoint加载模型对predictor进行延迟实例化_predict_numpy: 推理调用入口，返回类型 pandas.DataFrame

import typing as timport numpy as npimport pandas as pdfrom aiearth.predict.predictors.predictor import Predictorfrom aiearth.predict.checkpoint import ModelCheckpoint
# coco全景数据集label，添加0为背景COCO_PANOPTIC_CLASSES = [ "background", "person", "bicycle", "car", "motorcycle", "airplane", "bus", "train", "truck", "boat", "traffic light", "fire hydrant", "stop sign", "parking meter", "bench", "bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe", "backpack", "umbrella", "handbag", "tie", "suitcase", "frisbee", "skis", "snowboard", "sports ball", "kite", "baseball bat", "baseball glove", "skateboard", "surfboard", "tennis racket", "bottle", "wine glass", "cup", "fork", "knife", "spoon", "bowl", "banana", "apple", "sandwich", "orange", "broccoli", "carrot", "hot dog", "pizza", "donut", "cake", "chair", "couch", "potted plant", "bed", "dining table", "toilet", "tv", "laptop", "mouse", "remote", "keyboard", "cell phone", "microwave", "oven", "toaster", "sink", "refrigerator", "book", "clock", "vase", "scissors", "teddy bear", "hair drier", "toothbrush", "banner", "blanket", "bridge", "cardboard", "counter", "curtain", "door-stuff", "floor-wood", "flower", "fruit", "gravel", "house", "light", "mirror-stuff", "net", "pillow", "platform", "playingfield", "railroad", "river", "road", "roof", "sand", "sea", "shelf", "snow", "stairs", "tent", "towel", "wall-brick", "wall-stone", "wall-tile", "wall-wood", "water-other", "window-blind", "window-other", "tree-merged", "fence-merged", "ceiling-merged", "sky-other-merged", "cabinet-merged", "table-merged", "floor-other-merged", "pavement-merged", "mountain-merged", "grass-merged", "dirt-merged", "paper-merged", "food-other-merged", "building-other-merged", "rock-merged", "wall-other-merged", "rug-merged", ]

class MSImageSegmentationPredictor(Predictor):    def init(self, ckpt: ModelCheckpoint):        # 通过ModelScope pipeline构造初始化模型        from modelscope.pipelines import pipeline        from modelscope.utils.constant import Tasks
        cfg = ckpt.get_cfg()
        self.classes = cfg.params.additional_args["classes"]        self.segmentor = pipeline(Tasks.image_segmentation, model=cfg.path)
    def _predict_numpy(        self,        data: t.Union[np.ndarray, t.Dict[str, np.ndarray]],        dtype: t.Optional[np.dtype] = None,    ) -> t.Dict[str, np.ndarray]:        # 调用模型推理        predict_outputs = self.predict_udf(data["image"])
        return pd.DataFrame({"image": predict_outputs})
    def predict_udf(self, data: np.ndarray):        data = list(data)
        h, w, _ = data[0].shape        segments = self.segmentor(data)
        predict_outputs = []        # 将模型输出结果转换为对应label的index        for seg in segments:            predict_output = np.zeros((h, w), dtype=np.uint8)            for mask, label in zip(seg["masks"], seg["labels"]):                if label in self.classes:                    label_idx = COCO_PANOPTIC_CLASSES.index(label)                    predict_output[mask == 1] = label_idx                    predict_outputs.append(predict_output)
                    return predict_outputs

3. 构造pipeline

Pipeline是对复杂任务流的更高层次的封装，每个Pipeline表示一种任务类型：

model_checkpoint：模型checkpointpredictor_cls：模型推理类，通过model_checkpoint进行延迟初始化，可根据需求自定义predictorpre_processors：模型前处理，参数类型为List，可传递多个处理算子，在pipeline里面会将算子组合成一个调用链post_processors：模型后处理，参数类型为List，可传递多个处理算子，在pipeline里面会将算子组合成一个调用链

from aiearth.predict.checkpoint import ModelCheckpointfrom aiearth.predict.processors import Chw2HwcProcessor, SqueezeNdimProcessorfrom aiearth.predict.pipelines import GeoSegmentationPredictPipeline
from aiearth.predict.logging import root_logger as logger
# 只从全景分割里面提取道路classes = ["road"]# 从ModelScope构造Checkpointckpt = ModelCheckpoint.from_model_scope(    "damo/cv_swinL_panoptic-segmentation_cocopan",    classes=classes,)
# 构造分割任务Pipelinepipe = GeoSegmentationPredictPipeline(    model_checkpoint=ckpt,    predictor_cls=MSImageSegmentationPredictor,    pre_processors=[Chw2HwcProcessor(["image"])],    post_processors=[SqueezeNdimProcessor(["image"])],)本地运行测试：# 运行Pipelinelogger.info("run pipeline")pipe(    "/path/to/your/tiff",)运行的时候需要将/path/to/your/tiff路径替换为本地tiff影像的绝对路径，运行完成之后会在当前目录下面生成result.shp图斑文件。

二. 云平台部署

模型pipeline可部署到云平台工具箱模式下个人的AI解译工具，并通过云平台的工具箱模式从页面提交AI解译任务。

1. 获取授权

在与云平台进行交互时，需要先进行云平台的授权，获取用户个人token才可使用。from aiearth.core import Authenticate
Authenticate()

2. 创建项目空间

项目空间和模型任务一一对应，表示一类模型任务，并通过 project_id 进行索引。from aiearth.predict.deploy.sdk import Project
project = Project.create("道路提取")print(project)

3. 创建模型版本

每个项目空间下面，可以有多个模型版本，模型版本命名规则请遵循 x.y.z，xyz均为数字，如1.0.1，并通过 version_id 进行索引。from aiearth.predict.deploy.sdk import Version
version = project.create_version("1.0.1")print(version)

4. 模型pipeline

部署到云平台时，从ModelScope构建的本地模型pipeline进行少许改动，进行云平台工具箱模式的参数适配，主要改动为：

pipeline运行时的输入参数：本地运行时pipeline的输入参数可直接通过接口指定，部署到云平台时pipeline的输入参数需要通过parse_input_args进行解析得到运行时参数

修改之后的模型pipeline代码为：import typing as timport numpy as npimport pandas as pdfrom aiearth.predict.predictors.predictor import Predictorfrom aiearth.predict.checkpoint import ModelCheckpoint
COCO_PANOPTIC_CLASSES = [ "background", "person", "bicycle", "car", "motorcycle", "airplane", "bus", "train", "truck", "boat", "traffic light", "fire hydrant", "stop sign", "parking meter", "bench", "bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe", "backpack", "umbrella", "handbag", "tie", "suitcase", "frisbee", "skis", "snowboard", "sports ball", "kite", "baseball bat", "baseball glove", "skateboard", "surfboard", "tennis racket", "bottle", "wine glass", "cup", "fork", "knife", "spoon", "bowl", "banana", "apple", "sandwich", "orange", "broccoli", "carrot", "hot dog", "pizza", "donut", "cake", "chair", "couch", "potted plant", "bed", "dining table", "toilet", "tv", "laptop", "mouse", "remote", "keyboard", "cell phone", "microwave", "oven", "toaster", "sink", "refrigerator", "book", "clock", "vase", "scissors", "teddy bear", "hair drier", "toothbrush", "banner", "blanket", "bridge", "cardboard", "counter", "curtain", "door-stuff", "floor-wood", "flower", "fruit", "gravel", "house", "light", "mirror-stuff", "net", "pillow", "platform", "playingfield", "railroad", "river", "road", "roof", "sand", "sea", "shelf", "snow", "stairs", "tent", "towel", "wall-brick", "wall-stone", "wall-tile", "wall-wood", "water-other", "window-blind", "window-other", "tree-merged", "fence-merged", "ceiling-merged", "sky-other-merged", "cabinet-merged", "table-merged", "floor-other-merged", "pavement-merged", "mountain-merged", "grass-merged", "dirt-merged", "paper-merged", "food-other-merged", "building-other-merged", "rock-merged", "wall-other-merged", "rug-merged", ]

class MSImageSegmentationPredictor(Predictor):    def init(self, ckpt: ModelCheckpoint):        from modelscope.pipelines import pipeline        from modelscope.utils.constant import Tasks
        cfg = ckpt.get_cfg()
        self.classes = cfg.params.additional_args["classes"]        self.segmentor = pipeline(Tasks.image_segmentation, model=cfg.path)
    def _predict_numpy(        self,        data: t.Union[np.ndarray, t.Dict[str, np.ndarray]],        dtype: t.Optional[np.dtype] = None,    ) -> t.Dict[str, np.ndarray]:        predict_outputs = self.predict_udf(data["image"])
        return pd.DataFrame({"image": predict_outputs})
    def predict_udf(self, data: np.ndarray):        data = list(data)
        h, w, _ = data[0].shape        segments = self.segmentor(data)
        predict_outputs = []        for seg in segments:            predict_output = np.zeros((h, w), dtype=np.uint8)            for mask, label in zip(seg["masks"], seg["labels"]):                if label in self.classes:                    label_idx = COCO_PANOPTIC_CLASSES.index(label)                    predict_output[mask == 1] = label_idx            predict_outputs.append(predict_output)
        return predict_outputs

from aiearth.predict.checkpoint import ModelCheckpointfrom aiearth.predict.processors import Chw2HwcProcessor, SqueezeNdimProcessorfrom aiearth.predict.pipelines import GeoSegmentationPredictPipeline
from aiearth.predict.logging import root_logger as logger
classes = ["road"]ckpt = ModelCheckpoint.from_model_scope(    "damo/cv_swinL_panoptic-segmentation_cocopan",    classes=classes,)
pipe = GeoSegmentationPredictPipeline(    model_checkpoint=ckpt,    predictor_cls=MSImageSegmentationPredictor,    pre_processors=[Chw2HwcProcessor(["image"])],    post_processors=[SqueezeNdimProcessor(["image"])],)
from aiearth.predict.adapter.toolbox import parse_input_args, InputArgsinput_args: InputArgs = parse_input_args()
logger.info("run pipeline")# 对云平台工具箱模式的参数进行适配pipe(    uri=input_args.src,    out_dir=input_args.result_dir,    temp_dir=input_args.temp_dir,    out_filename=input_args.out_filename,)创建code_snippets目录，将模型pipeline代码保存到code_snippets目录下面，并命名为pipe.py。

5. 工具箱部署

通过模型版本客户端将模型pipeline代码部署到云平台工具箱模式。from aiearth.predict.deploy.schema import (    JobConfig,    JobOutType,    ToolboxRunConfig,)
job_config = JobConfig(    entrypoint="python pipe.py",    runtime_env={"working_dir": "./code_snippets"},)run_config = ToolboxRunConfig(out_type=JobOutType.VECTOR)
deploy = version.deploy_as_toolbox(job_config, run_config)
6. 查看部署状态

查询模型部署的状态，部署状态类型有：

waiting：等待部署deploying：部署中finished：部署完成failed：部署失败undeploying：等待取消部署undeployed：部署已取消

deploy_status = deploy.get_deploy_status()print(deploy_status)
7. 提交解译任务

当模型pipeline部署完成之后，可通过云平台的工具箱模式从页面提交AI解译任务。



参考资料

1、ModelScope模型库-Mask2Former-SwinL全景分割：https://modelscope.cn/models/damo/cv_swinL_panoptic-segmentation_cocopan/summary
2、通用图像分割任务: 使用 Mask2Former 和 OneFormer：https://huggingface.co/blog/zh/mask2former

关于AI Earth地球科学云平台

AI Earth地球科学云平台基于达摩院-视觉技术实验室在深度学习、计算机视觉、地理空间分析等方向上的技术积累，结合阿里云强大算力支撑，提供多源观测数据的云计算分析服务，用数据感知地球世界，让AI助力科学研究。





AI Earth官网



AI Earth钉钉群

关注我们 了解更多动态

往

期

推

荐

AI Earth支持用户自定义函数（User Defined Function）接入

快来创建你的第一个AI Earth在线应用！

“智联时空，有融乃大”——AIE-SEG V2.0发布及时空智能技术应用论坛成功举办！