데이터 표현

런타임에 블록이 받는 Python 데이터 유형: Batch, WorkflowImageData, VideoMetadata.

많은 프레임워크는 개발자가 다룰 표준 데이터 타입을 강제하며, Workflows 생태계도 예외는 아닙니다. 비록 종류 Workflow에서의 종류는 전달되는 데이터의 고수준 추상화를 나타내므로, 다음에 제공될 구체적인 데이터 타입을 이해하는 것이 중요합니다. WorkflowBlock.run(...) 메서드입니다.

바로 이것이 여기서 배우게 될 내용입니다.

배치

Workflow 블록이 배치 처리를 선언하면 Batch라는 특수한 컨테이너 타입을 사용합니다. 배치 지향 매개변수는 모두 Batch[X]로 감싸지며, 여기서 X는 데이터 타입입니다:

from inference.core.workflows.execution_engine.entities.base import Batch
from inference.core.workflows.prototypes.block import BlockResult

# Workflow 블록의 run 메서드
def run(self, x: Batch[int], y: Batch[float]) -> BlockResult:
   pass

이 배치 타입은 Python 리스트와 유사하게 작동하지만, 한 가지 중요한 차이점이 있습니다. 즉, 그것은 읽기 전용. 요소를 수정할 수 없고, 요소를 추가하거나 제거할 수도 없습니다. 하지만 몇 가지 유용한 작업을 사용할 수 있습니다:

요소 반복

from inference.core.workflows.execution_engine.entities.base import Batch

def iterate(batch: Batch[int]) -> None:
    for element in batch:
        print(element)

여러 배치 묶기

배치 정렬은 걱정하지 마세요

Execution Engine은 run 메서드에 제공되는 배치가 동일한 크기가 되도록 보장하여, 반복 중 배치 크기 불일치로 인한 요소 손실을 방지합니다.

from inference.core.workflows.execution_engine.entities.base import Batch

def zip_batches(batch_1: Batch[int], batch_2: Batch[float]) -> None:
    for element_1, element_2 in zip(batch_1, batch_2):
        print(element_1, element_2)

배치 요소 인덱스 가져오기

각 튜플이 잠재적으로 중첩된 배치 구조에서 배치 요소의 위치를 나타내는 튜플 목록을 반환합니다.

from inference.core.workflows.execution_engine.entities.base import Batch

def discover_indices(batch: Batch[int]) -> None:
    for index in batch.indices:
        print(index)  # 예: 1차원 배치는 (0,), 2차원 중첩 배치는 (1, 3) 등

요소와 인덱스를 함께 가져오며 반복하기

from inference.core.workflows.execution_engine.entities.base import Batch

def iterate_with_indices(batch: Batch[int]) -> None:
    for index, element in batch.iter_with_indices():
        print(index, element)

의 추가 메서드 배치 컨테이너

Batch 인터페이스에는 다음과 같은 다른 메서드도 있습니다: remove_by_indices(...) 또는 broadcast(...)이 있지만, 이들은 Workflow 블록 내부에서 사용하도록 նախատեսված된 것은 아닙니다. 이러한 메서드는 주로 블록에 데이터를 제공할 때 Execution Engine에서 사용합니다.

WorkflowImageData

WorkflowImageData 이미지와 해당 메타데이터를 함께 캡슐화하는 데이터클래스로, Workflow 블록 내에서 이미지 표현을 조작하는 데 유용한 메서드를 제공합니다.

일부 사용자는 다음을 기대할 수 있습니다: np.ndarray Execution Engine에서 직접 제공되기를 기대할 수 있습니다, 다음의 경우에 이미지 종류가 선언될 때. 이는 편리하고 직관적인 접근일 수 있지만, 다음과 같은 한계를 도입합니다:

• 메타데이터 부족: 오직 하나의 np.ndarray, 데이터 계보나 원본 파일 내 이미지의 위치(예: 잘린 이미지를 다룰 때)와 같은 메타데이터를 첨부할 방법이 없습니다.

• 여러 표현을 캐시할 수 없음: 여러 블록이 이미지를 직렬화하여 HTTP로 전송해야 할 경우, WorkflowImageData는 base64 인코딩 버전과 같은 서로 다른 이미지 표현을 캐시할 수 있어 효율성을 높입니다.

비디오 메타데이터

Execution Enginge 이후 v1.2.0, 우리는 다음을 추가했습니다: video_metadata 에 WorkflowImageData. 이 객체는 비디오 처리의 컨텍스트를 담도록 되어 있으며, 비디오 처리 블록에서만 관련이 있습니다. 다른 블록은 출력 이미지를 생성하지 않는다면 그 존재를 무시해도 됩니다(다음 섹션에서 다룹니다).

다루는 것은 WorkflowImageData 인터페이스를 이해하면 꽤 간단합니다. 주요 메서드와 속성은 다음과 같습니다:

from inference.core.workflows.execution_engine.entities.base import WorkflowImageData

def operate_on_image(workflow_image: WorkflowImageData) -> None:
    # 이미지를 나타내는 np.ndarray를 가져오기.
    numpy_image = workflow_image.numpy_image  
    
    # API 전송에 이상적인, base64로 인코딩된 JPEG 형식의 이미지를 가져오기.
    base64_image: str = workflow_image.base64_image  
    
    # 이미지를 inference 모델과 호환되는 형식으로 변환하기.
    inference_format = workflow_image.to_inference_format()  
    
    # 부모 이미지와 관련된 메타데이터에 접근합니다
    parent_metadata = workflow_image.parent_metadata
    print(parent_metadata.parent_id)  # 부모 식별자
    origin_coordinates = parent_metadata.origin_coordinates  # 좌표가 있는 선택적 객체
    print(
        origin_coordinates.left_top_x, origin_coordinates.left_top_y, 
        origin_coordinates.origin_width, origin_coordinates.origin_height,
    )
    
    # 또는 루트 메타데이터에도 동일하게 적용됩니다(이미지의 가장 오래된 선조 - Workflow 입력 이미지)
    root_metadata = workflow_image.workflow_root_ancestor_metadata
    
    # `VideoMetadata` 객체를 가져오기 - 아래 사용 가이드 섹션을 참조하세요
    # `workflow_image`에 `VideoMetadata`가 제공되지 않으면 - 기본 메타데이터 객체가 
    # 속성에 접근할 때 생성됩니다
    video_metadata = workflow_image.video_metadata 

아래에서 이미지를 변환하면서 메타데이터를 보존하는 방법을 보여주는 예제를 확인할 수 있습니다

import numpy as np

from inference.core.workflows.execution_engine.entities.base import WorkflowImageData

def transform_image(image: WorkflowImageData) -> WorkflowImageData:
    transformed_image = some_transformation(image.numpy_image)
    # `WorkflowImageData`는 보존된 메타데이터를 유지한 채
    # 새 객체를 반환하는 도우미 메서드를 제공합니다. 메타데이터 보존
    # 은 출력 이미지가 다음과 관련해 호환될 때만 사용해야 합니다
    # 데이터 계보(이미지의 선행-후행 관계).
    # 공통 선행 이미지가 없는 이미지의 자르기와 병합에서는 계보가 보존되지 않습니다
    # - 아래에서 구현 팁을 확인할 수 있습니다.
    return WorkflowImageData.copy_and_replace(
        origin_image_data=image,
        numpy_image=transformed_image,
    )

def some_transformation(image: np.ndarray) -> np.ndarray:
    ...

이미지 자르기

블록이 차원을 늘리고 다음과 같은 출력 이미지 종류를 제공할 때, 보통 이는 이미지를 자른다는 의미입니다. 이 경우 입력 이미지는 video_metadata 제거되어야 합니다(일반적으로 이를 유지하는 것은 의미가 없으며, 기반이 되는 비디오 처리 블록은 동적으로 생성된 블록에서는 올바르게 작동하지 않기 때문입니다).

아래에서 이 작업의 구현 초안을 확인할 수 있습니다:

from typing import List, Tuple

from dataclasses import replace
from inference.core.workflows.execution_engine.entities.base import WorkflowImageData

def crop_images(
    image: WorkflowImageData, 
    crops: List[Tuple[str, int, int, int, int]],
) -> List[WorkflowImageData]:
    crops = []
    original_image = image.numpy_image
    for crop_id, x_min, y_min, x_max, y_max in crops:
        cropped_image = original_image[y_min:y_max, x_min:x_max]
        if not cropped_image.size:
            # 빈 잘라내기 결과는 버립니다
            continue
        result_crop = WorkflowImageData.create_crop(
            origin_image_data=image, 
            crop_identifier=crop_id,
            cropped_image=cropped_image,
            offset_x=x_min,
            offset_y=y_min,
        )
        crops.append(result_crop)
    return crops

경우에 따라 다음을 보존하고 싶을 수 있습니다: video_metadata. 이러한 상황의 예는 블록이 고정 좌표를 기반으로 잘라낸 결과를 생성하는 경우입니다(예: 개별 트래커를 적용할 여러 고정 관심 영역이 있는 단일 비디오 영상). 이때 결과 잘라내기들은 별도의 카메라에서 생성된 것처럼 비디오의 컨텍스트에서 처리되기를 원하게 됩니다. 다음의 동작을 조정하려면 create_crop(...) 메서드에 단순히 다음을 추가하세요: preserve_video_metadata=True:

def crop_images(
    image: WorkflowImageData, 
    crops: List[Tuple[str, int, int, int, int]],
) -> List[WorkflowImageData]:
    # [...]
    result_crop = WorkflowImageData.create_crop(
        origin_image_data=image, 
        crop_identifier=crop_id,
        cropped_image=cropped_image,
        offset_x=x_min,
        offset_y=y_min,
        preserve_video_metadata=True
    )
    # [...]

공통 선행 이미지가 없는 이미지 병합

공통 parent_metadata 병합하려는 여러 이미지에 대해 가리킬 수 없다면, Workflow에 "새로운" 이미지가 나타났음을 표시해야 합니다. 이를 위해 간단히 다음과 같이 하세요:

from typing import List, Tuple

from dataclasses import replace
from inference.core.workflows.execution_engine.entities.base import \\
    WorkflowImageData, ImageParentMetadata

def merge_images(image_1: WorkflowImageData, image_2: WorkflowImageData) -> WorkflowImageData:
    merged_image = some_mergin_operation(
        image_1=image_1.numpy_image,
        image_2=image_2.numpy_image
    )
    new_parent_metadata = ImageParentMetadata(
        # 이는 ID를 생성하는 여러 옵션 중 하나일 뿐이지만, 타당한 방법입니다
        parent_id=f"{image_1.parent_metadata.parent_id} + {image_2.parent_metadata.parent_id}"
    )
    return WorkflowImageData(
        parent_metadata=new_parent_metadata,
        numpy_image=merged_imagem
    )

VideoMetadata

사용 중단

video_metadata 종류 는 사용이 중단되었습니다. 새 블록에서는 해당 종류를 사용하지 않는 것을 권장합니다. VideoMetadata 데이터 표현은 다음의 멤버가 되었습니다: WorkflowImageData Execution Engine에서 v1.2.0 (inference 릴리스 v0.23.0)

VideoMetadata 는 비디오 프레임과 비디오 소스에 대한 다음 메타데이터를 제공하는 데이터클래스입니다:

from inference.core.workflows.execution_engine.entities.base import VideoMetadata

def inspect_vide_metadata(video_metadata: VideoMetadata) -> None:
    # 비디오를 위한 식별 문자열. 불투명한 값으로 취급해야 합니다.
    print(video_metadata.video_identifier)
    
    # 프레임의 순차 번호
    print(video_metadata.frame_number)
    
    # 비디오 프레임의 타임스탬프입니다. 비디오를 처리할 때는 
    # `fps`와 `frame_number`에 의존하는 것이 좋습니다. 실제 경과 시간은 
    # 비디오 파일의 경과 시간과 일치하지 않기 때문입니다
    print(video_metadata.frame_timestamp)
    
    # 이 필드는 FPS 값을 나타냅니다(가능한 경우에 한해 가져올 수 있음)(선택 사항)
    print(video_metadata.fps)
    
    # 이 필드는 라이브 스트림의 측정된 FPS를 나타냅니다(선택 사항)
    print(video_metadata.measured_fps)
    
    # 이 필드는 프레임이 비디오 파일에서 왔는지 스트림에서 왔는지를 알려주는 플래그입니다.
    # 판별할 수 없는 경우 - None
    print(video_metadata.comes_from_video_file)