add attribute recognition pipline

api_examples/pipelines/test_pedestrian_attribute_rec.py

@@ -0,0 +1,26 @@
+# copyright (c) 2024 PaddlePaddle Authors. All Rights Reserve.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from paddlex import create_pipeline
+
+pipeline = create_pipeline(pipeline="pedestrian_attribute_recognition")
+
+output = pipeline.predict(
+    "./test_samples/pedestrian_attribute_002.jpg", det_threshold=0.7, cls_threshold=0.7
+)
+
+for res in output:
+    res.print()  ## 打印预测的结构化输出
+    res.save_to_img("./output")  ## 保存结果可视化图像
+    res.save_to_json("./output/")  ## 保存预测的结构化输出

api_examples/pipelines/test_vehicle_attribute_rec.py

@@ -0,0 +1,26 @@
+# copyright (c) 2024 PaddlePaddle Authors. All Rights Reserve.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from paddlex import create_pipeline
+
+pipeline = create_pipeline(pipeline="vehicle_attribute_recognition")
+
+output = pipeline.predict(
+    "./test_samples/vehicle_attribute_002.jpg", det_threshold=0.7, cls_threshold=0.7
+)
+
+for res in output:
+    res.print()  ## 打印预测的结构化输出
+    res.save_to_img("./output")  ## 保存结果可视化图像
+    res.save_to_json("./output/")  ## 保存预测的结构化输出

paddlex/configs/pipelines/pedestrian_attribute_recognition.yaml

@@ -0,0 +1,15 @@
+pipeline_name: pedestrian_attribute_recognition
+
+SubModules:
+  Detection:
+    module_name: object_detection
+    model_name: PP-YOLOE-L_human
+    model_dir: null
+    batch_size: 1
+    threshold: 0.5 
+  Classification:
+    module_name: multilabel_classification
+    model_name: PP-LCNet_x1_0_pedestrian_attribute
+    model_dir: null
+    batch_size: 1
+    threshold: 0.5

paddlex/configs/pipelines/vehicle_attribute_recognition.yaml

@@ -0,0 +1,15 @@
+pipeline_name: vehicle_attribute_recognition
+
+SubModules:
+  Detection:
+    module_name: object_detection
+    model_name: PP-YOLOE-L_vehicle
+    model_dir: null
+    batch_size: 1  
+    threshold: 0.5  
+  Classification:
+    module_name: multilabel_classification
+    model_name: PP-LCNet_x1_0_vehicle_attribute
+    model_dir: null
+    batch_size: 1
+    threshold: 0.5

paddlex/inference/models_new/image_classification/predictor.py

@@ -114,7 +114,8 @@ class ClasPredictor(BasicPredictor):
         """
         batch_raw_imgs = self.preprocessors["Read"](imgs=batch_data)
         batch_imgs = self.preprocessors["Resize"](imgs=batch_raw_imgs)
-        batch_imgs = self.preprocessors["Crop"](imgs=batch_imgs)
+        if "Crop" in self.preprocessors:
+            batch_imgs = self.preprocessors["Crop"](imgs=batch_imgs)
         batch_imgs = self.preprocessors["Normalize"](imgs=batch_imgs)
         batch_imgs = self.preprocessors["ToCHW"](imgs=batch_imgs)
         x = self.preprocessors["ToBatch"](imgs=batch_imgs)

paddlex/inference/pipelines_new/__init__.py

@@ -31,6 +31,10 @@ from .anomaly_detection import AnomalyDetectionPipeline
 from .ts_forecasting import TSFcPipeline
 from .ts_anomaly_detection import TSAnomalyDetPipeline
 from .ts_classification import TSClsPipeline
+from .attribute_recognition import (
+    PedestrianAttributeRecPipeline,
+    VehicleAttributeRecPipeline,
+)
 
 
 def get_pipeline_path(pipeline_name: str) -> str:

paddlex/inference/pipelines_new/attribute_recognition/__init__.py

@@ -0,0 +1,15 @@
+# copyright (c) 2024 PaddlePaddle Authors. All Rights Reserve.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .pipeline import PedestrianAttributeRecPipeline, VehicleAttributeRecPipeline

paddlex/inference/pipelines_new/attribute_recognition/pipeline.py

@@ -0,0 +1,100 @@
+# copyright (c) 2024 PaddlePaddle Authors. All Rights Reserve.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import Any, Dict, Optional
+
+import pickle
+from pathlib import Path
+import numpy as np
+
+from ...utils.pp_option import PaddlePredictorOption
+from ...common.reader import ReadImage
+from ...common.batch_sampler import ImageBatchSampler
+from ..components import CropByBoxes
+from ..base import BasePipeline
+from .result import AttributeRecResult
+
+
+class AttributeRecPipeline(BasePipeline):
+    """Attribute Rec Pipeline"""
+
+    def __init__(
+        self,
+        config: Dict,
+        device: str = None,
+        pp_option: PaddlePredictorOption = None,
+        use_hpip: bool = False,
+        hpi_params: Optional[Dict[str, Any]] = None,
+    ):
+        super().__init__(
+            device=device, pp_option=pp_option, use_hpip=use_hpip, hpi_params=hpi_params
+        )
+
+        self.det_model = self.create_model(config["SubModules"]["Detection"])
+        self.cls_model = self.create_model(config["SubModules"]["Classification"])
+        self._crop_by_boxes = CropByBoxes()
+        self._img_reader = ReadImage(format="BGR")
+
+        self.det_threshold = config["SubModules"]["Detection"].get("threshold", 0.7)
+        self.cls_threshold = config["SubModules"]["Classification"].get(
+            "threshold", 0.7
+        )
+
+        self.batch_sampler = ImageBatchSampler(
+            batch_size=config["SubModules"]["Detection"]["batch_size"]
+        )
+        self.img_reader = ReadImage(format="BGR")
+
+    def predict(self, input, **kwargs):
+        det_threshold = kwargs.pop("det_threshold", self.det_threshold)
+        cls_threshold = kwargs.pop("cls_threshold", self.cls_threshold)
+        for img_id, batch_data in enumerate(self.batch_sampler(input)):
+            raw_imgs = self.img_reader(batch_data)
+            all_det_res = list(self.det_model(raw_imgs, threshold=det_threshold))
+            for input_data, raw_img, det_res in zip(batch_data, raw_imgs, all_det_res):
+                cls_res = self.get_cls_result(raw_img, det_res, cls_threshold)
+                yield self.get_final_result(input_data, raw_img, det_res, cls_res)
+
+    def get_cls_result(self, raw_img, det_res, cls_threshold):
+        subs_of_img = list(self._crop_by_boxes(raw_img, det_res["boxes"]))
+        img_list = [img["img"] for img in subs_of_img]
+        all_cls_res = list(self.cls_model(img_list, threshold=cls_threshold))
+        output = {"label": [], "score": []}
+        for res in all_cls_res:
+            output["label"].append(res["label_names"])
+            output["score"].append(res["scores"])
+        return output
+
+    def get_final_result(self, input_data, raw_img, det_res, rec_res):
+        single_img_res = {"input_path": input_data, "input_img": raw_img, "boxes": []}
+        for i, obj in enumerate(det_res["boxes"]):
+            rec_scores = rec_res["score"][i]
+            labels = rec_res["label"][i]
+            single_img_res["boxes"].append(
+                {
+                    "labels": labels,
+                    "rec_scores": rec_scores,
+                    "det_score": obj["score"],
+                    "coordinate": obj["coordinate"],
+                }
+            )
+        return AttributeRecResult(single_img_res)
+
+
+class PedestrianAttributeRecPipeline(AttributeRecPipeline):
+    entities = "pedestrian_attribute_recognition"
+
+
+class VehicleAttributeRecPipeline(AttributeRecPipeline):
+    entities = "vehicle_attribute_recognition"

paddlex/inference/pipelines_new/attribute_recognition/result.py

@@ -0,0 +1,90 @@
+# copyright (c) 2024 PaddlePaddle Authors. All Rights Reserve.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import cv2
+import numpy as np
+import PIL
+from PIL import Image, ImageDraw, ImageFont
+
+from ....utils.fonts import PINGFANG_FONT_FILE_PATH
+from ...utils.io import ImageReader
+from ...common.result import BaseCVResult
+from ...utils.color_map import get_colormap, font_colormap
+
+
+def draw_attribute_result(img, boxes):
+    """
+    Args:
+        img (PIL.Image.Image): PIL image
+        boxes (list): a list of dictionaries representing detection box information.
+    Returns:
+        img (PIL.Image.Image): visualized image
+    """
+    font_size = int((0.024 * int(img.width) + 2) * 0.7)
+    font = ImageFont.truetype(PINGFANG_FONT_FILE_PATH, font_size, encoding="utf-8")
+
+    draw_thickness = int(max(img.size) * 0.005)
+    draw = ImageDraw.Draw(img)
+    label2color = {}
+    catid2fontcolor = {}
+    color_list = get_colormap(rgb=True)
+
+    for i, dt in enumerate(boxes):
+        text_lines, bbox, score = dt["label"], dt["coordinate"], dt["score"]
+        if i not in label2color:
+            color_index = i % len(color_list)
+            label2color[i] = color_list[color_index]
+            catid2fontcolor[i] = font_colormap(color_index)
+        color = tuple(label2color[i]) + (255,)
+        font_color = tuple(catid2fontcolor[i])
+
+        xmin, ymin, xmax, ymax = bbox
+        # draw box
+        draw.line(
+            [(xmin, ymin), (xmin, ymax), (xmax, ymax), (xmax, ymin), (xmin, ymin)],
+            width=draw_thickness,
+            fill=color,
+        )
+        # draw label
+        current_y = ymin
+        for line in text_lines:
+            if tuple(map(int, PIL.__version__.split("."))) <= (10, 0, 0):
+                tw, th = draw.textsize(line, font=font)
+            else:
+                left, top, right, bottom = draw.textbbox((0, 0), line, font)
+                tw, th = right - left, bottom - top + 4
+
+            draw.text((5 + xmin + 1, current_y + 1), line, fill=(0, 0, 0), font=font)
+            draw.text((5 + xmin, current_y), line, fill=color, font=font)
+            current_y += th
+    return img
+
+
+class AttributeRecResult(BaseCVResult):
+
+    def _to_img(self):
+        """apply"""
+        img_reader = ImageReader(backend="pillow")
+        image = img_reader.read(self["input_path"])
+        boxes = [
+            {
+                "coordinate": box["coordinate"],
+                "label": box["labels"],
+                "score": box["det_score"],
+            }
+            for box in self["boxes"]
+        ]
+        image = draw_attribute_result(image, boxes)
+        return image