Skip to main content
Version: 2.0.0

3. Custom Model 만들기

NuFiStreamer nufi element에서 이용할 수 있는 커스텀 model을 만들어봅니다. Gstreamer에 대한 이해가 필요하다면 Gstreamer docs를 참고해주세요.

Model 설명​

  • Model은 nufi element의 model property로 설정되어 이미지를 처리하기 위한 객체입니다.
    • nufi element에 대한 세부정보는 nufi element 문서를 참고하세요.
  • 기본적으로 NuFiStreamer에서 제공하는 Model들은 다음과 같습니다.
  • 제공하는 Model 이외에도 커스텀하게 만든 image processing Model도 nufi element에서 이용할 수 있습니다.
  • 커스텀 Model 구현 시 nufi.models 하위의 BaseModel 추상 클래스를 구현해야 합니다.
    • BaseModel은 다음과 같은 메소드를 필수적으로 구현해야 합니다.
      • def process(self, image: np.ndarray, pipeline_id: Optional[int] = None) -> List[NufiResult]
        • 이미지를 처리하는 Model의 핵심 메소드입니다.
        • Input
          • image: 처리하고자 하는 타겟 이미지 데이터
          • pipeline_id: 해당 Model을 이용하는 nufi element가 속한 pipeline id (optional)
        • Output
          • List[NufiResult]

NufiResult 설명​

  • NufiResult는 nufi.core.postprocess 하위에 존재합니다.
  • NufiResult가 가지는 property는 다음과 같습니다.
    • data (Any): 이미지에 그려지지 않는 메타데이터를 포함하는 데이터 (예: image frame seq, fall detection 쓰러짐 여부 등)
      • data는 어떠한 데이터든 올 수 있지만 Json encode, decode가 가능한 데이터여야만 합니다.
      • 구현한 Class를 경우 nufi.core.postprocess 하위에 SerializableBase를 상속하는 Class여야만 합니다.
    • drawables (List[Drawable]): 실제로 이미지에 metadata를 그리는 객체인 Drawable list
    • is_draw (bool): 해당 NufiResult가 가진 drawables 출력 여부

Drawable 설명​

  • Drawable은 nufi.core.postprocess 하위에 존재합니다.
  • 기본적으로 NuFiStreamer에서는 다음과 같은 Drawable 구현체를 제공합니다.
    • Box: 사각형을 그리기 위한 Drawable 구현체입니다.
    • Line: 선을 그리기 위한 Drawable 구현체입니다.
    • Point: 점을 그리기 위한 Drawable 구현체입니다.
  • 커스텀 Model을 구현할 때 process 메소드가 제공되는 Drawable 구현체들을 가지는 NufiResult를 반환해도 됩니다.
  • 이미지에 그려야 하는 모양이 제공되는 Drawable 구현체로 그릴 수 없다면 커스텀하게 만듭니다.
  • 커스텀 구현 시 nufi.core.postprocess 하위의 Drawable 추상 클래스를 구현해야 합니다.
    • Drawable은 다음과 같은 메소드를 필수적으로 구현해야 합니다.
      • def draw(self, image: np.ndarray)
        • 이미지에 metadata를 그리는 메소드입니다.
        • Input
          • image: 처리하고자 하는 타겟 이미지 데이터

구현 예시​

Drawable 예제 - 제공된 Drawable 구현체를 이용하는 경우​

  • MidRectanglePrinter 모델은 NuFiStreamer에서 제공하는 Drawable 구현체인 Box를 이용합니다.
  • 해당 모델을 이용하는 pipeline 예시는 Triangle 모델 파이프라인 예시를 참고하세요.
from nufi.models import BaseModel
from nufi.core.postprocess import Box, NufiResult


class MidRectanglePrinter(BaseModel):
    def __init__(self, id):
        self.id = id
        print(f"Initialize {self.id} rectangle printer")

    def process(
        self, image: ndarray, pipeline_id: Optional[int] = None
    ) -> List[NufiResult]:
        box_points = self._calculate_center_box(image)
        box = Box(
            "rectangle",
            box_points,
            (255, 0, 0),
            is_draw=True,
        )
        return [NufiResult(drawables=[box])]

    def clean(self, pipeline_id: Optional[int] = None):
        return super().clean(pipeline_id)

    def _calculate_center_box(self, image: ndarray) -> List[int]:
        height, width = image.shape[:2]

        box_width = width // 4
        box_height = height // 4

        x1 = (width - box_width) // 2
        y1 = (height - box_height) // 2

        x2 = x1 + box_width
        y2 = y1 + box_height

        return [x1, y1, x2, y2]

Drawable 예제 - 커스텀하게 Drawable 구현체를 만드는 경우​

  • 커스텀 Drawable 구현체인 Triangle을 만듭니다.
    • Triangle은 draw 메소드에서 삼각형을 이미지에 그리도록 설계했습니다.
  • 이를 이용해서 MidTrianglePrinter Model에서 NufiResult를 반환합니다.
  • 해당 모델을 이용하는 pipeline 예시는 Triangle 모델 파이프라인 예시를 참고하세요.
from nufi.models import BaseModel
from nufi.core.postprocess import NufiResult, Drawable


class Triangle(Drawable):
    def __init__(
        self,
        xyxyxy: List[tuple],
        color=(0, 0, 0),
        thickness=2,
        is_draw=True,
    ):
        self.xyxyxy = xyxyxy
        self.color = color
        self.thickness = thickness
        self.is_draw = is_draw

    def draw(self, image: ndarray):
        if self.is_draw:
            triangle_pts = np.array(self.xyxyxy, np.int32)
            cv2.polylines(
                image,
                [triangle_pts],
                isClosed=True,
                color=self.color,
                thickness=self.thickness,
            )


class MidTrianglePrinter(BaseModel):
    def __init__(self, id):
        self.id = id
        print(f"Initialize {self.id} rectangle printer")

    def process(
        self, image: ndarray, pipeline_id: Optional[int] = None
    ) -> List[NufiResult]:
        points = self.calculate_points(image)
        triangle = Triangle(
            xyxyxy=points,
            color=(255, 0, 0),
            thickness=3,
            is_draw=True,
        )
        return [NufiResult(drawables=[triangle])]

    def clean(self, pipeline_id: Optional[int] = None):
        return super().clean(pipeline_id)

    def calculate_points(self, image: ndarray) -> List[int]:
        height, width = image.shape[:2]

        tri_width = width // 3
        tri_height = height // 3

        x1 = (width - tri_width) // 2
        y1 = height - tri_height

        x2 = x1 + tri_width
        y2 = y1

        x3 = (x1 + x2) // 2
        y3 = y2 - tri_height

        return [(x1, y1), (x2, y2), (x3, y3)]

NufiResult 예제 - data property 설정하는 방법​

Class​

  • Class를 NufiResult data property에 넣기 위해선 nufi.core.postprocess 하위의 SerializableBase를 상속해야합니다.
    • SerializableBase에 의해서 Json encode, decode가 가능해집니다.
import numpy as np
from nufi.models import BaseModel, SerializableBase
from nufi.core.postprocess import Box, NufiResult


class ImageFrameInfo(SerializableBase):
    def __init__(self, width, height):
        self.width = width
        self.height = height

    def __repr__(self) -> str:
        return f"ImageFrameInfo(width={self.width}, height={self.height})"

class MidRectanglePrinter(BaseModel):
    def __init__(self, id):
        self.id = id
        print(f"Initialize {self.id} rectangle printer")

    def process(
        self, image: np.ndarray, pipeline_id: Optional[int] = None
    ) -> List[NufiResult]:
        box_points = self._calculate_center_box(image)
        height, width = image.shape[:2]
        data = ImageFrameInfo(width=width, height=height)
        box = Box(
            "rectangle",
            box_points,
            (255, 0, 0),
            is_draw=True,
        )

        return [NufiResult(data=data, drawables=[box])]

    def clean(self, pipeline_id: Optional[int] = None):
        return super().clean(pipeline_id)

    def _calculate_center_box(self, image: ndarray) -> List[int]:
        height, width = image.shape[:2]

        box_width = width // 4
        box_height = height // 4

        x1 = (width - box_width) // 2
        y1 = (height - box_height) // 2

        x2 = x1 + box_width
        y2 = y1 + box_height

        return [x1, y1, x2, y2]

Dictionary​

  • image width, height 정보를 data에 dictionary로 변환하여 입력합니다.
import numpy as np
from nufi.models import BaseModel, SerializableBase
from nufi.core.postprocess import Box, NufiResult


class MidRectanglePrinter(BaseModel):
    def __init__(self, id: int):
        self.id = id
        print(f"Initialize {self.id} rectangle printer")

    def process(
        self, image: np.ndarray, pipeline_id: Optional[int] = None
    ) -> List[NufiResult]:
        box_points = self._calculate_center_box(image)
        height, width = image.shape[:2]
        data = {"width": width, "height": height} # dictionary NufiResult에 입력
        box = Box(
            "rectangle",
            box_points,
            (255, 0, 0),
            is_draw=True,
        )

        return [NufiResult(data=data, drawables=[box])]

    def clean(self, pipeline_id: Optional[int] = None):
        return super().clean(pipeline_id)

    def _calculate_center_box(self, image: ndarray) -> List[int]:
        height, width = image.shape[:2]

        box_width = width // 4
        box_height = height // 4

        x1 = (width - box_width) // 2
        y1 = (height - box_height) // 2

        x2 = x1 + box_width
        y2 = y1 + box_height

        return [x1, y1, x2, y2]

Primitive Type​

  • MidRectanglePrinter id 정보를 data에 primitive type으로 입력합니다.
import numpy as np
from nufi.models import BaseModel, SerializableBase
from nufi.core.postprocess import Box, NufiResult


class MidRectanglePrinter(BaseModel):
    def __init__(self, id):
        self.id = id
        print(f"Initialize {self.id} rectangle printer")

    def process(
        self, image: np.ndarray, pipeline_id: Optional[int] = None
    ) -> List[NufiResult]:
        box_points = self._calculate_center_box(image)
        data = self.id # primitive type NufiResult에 입력
        box = Box(
            "rectangle",
            box_points,
            (255, 0, 0),
            is_draw=True,
        )

        return [NufiResult(data=data, drawables=[box])]

    def clean(self, pipeline_id: Optional[int] = None):
        return super().clean(pipeline_id)

    def _calculate_center_box(self, image: ndarray) -> List[int]:
        height, width = image.shape[:2]

        box_width = width // 4
        box_height = height // 4

        x1 = (width - box_width) // 2
        y1 = (height - box_height) // 2

        x2 = x1 + box_width
        y2 = y1 + box_height

        return [x1, y1, x2, y2]