データ表現

ブロックが実行時に受け取るPythonデータ型: Batch、WorkflowImageData、VideoMetadata。

多くのフレームワークは、開発者が扱う標準データ型を強制しており、Workflows エコシステムも例外ではありません。 種類 Workflow における kind は、通過するデータの高レベルな抽象化を表しますが、提供される具体的なデータ型を理解することが重要です WorkflowBlock.run(...) Workflow ブロックを作成する際のメソッドです。

そして、まさにここでそれを学ぶことができます。

Batch

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

この Batch 型は Python の list と同様に機能しますが、1 つ重要な違いがあります。それは 読み取り専用です。要素を変更したり、要素を追加・削除したりすることはできません。ただし、いくつか便利な操作が利用できます:

要素の反復

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

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

複数のバッチを zip する

バッチの整列について心配する必要はありません

Execution Engine は、run メソッドに渡されるバッチが同じサイズになるよう保証するため、反復中にバッチサイズの不一致によって要素が失われることはありません。

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

def zip_batches(batch_1: Batch[int], batch_2: Batch[int]) -> 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 コンテナ

Batch インターフェースには、次のような他のメソッドもあります。 remove_by_indices(...) または broadcast(...)、これらは Workflow ブロック内で使用することを意図したものではありません。これらのメソッドは主に Execution Engine がブロックへデータを提供する際に使用します。

WorkflowImageData

WorkflowImageData は、画像とそのメタデータをまとめて格納する dataclass で、Workflow ブロック内で画像表現を操作するための便利なメソッドを提供します。

一部のユーザーは、 np.ndarray が、Execution Engine から直接提供されることを期待するかもしれません。 image kind が宣言されているとき。これは便利で分かりやすい方法ですが、次のような制約があります:

• メタデータが不足すること: 単に np.ndarrayだけでは、データ系譜や元ファイル内での画像の位置（たとえば切り抜き画像を扱う場合）といったメタデータを付与する方法がありません。

• 複数の表現をキャッシュできないこと: 複数のブロックが画像をシリアライズして HTTP 経由で送信する必要がある場合、WorkflowImageData は base64 エンコード版など異なる画像表現をキャッシュでき、効率が向上します。

動画メタデータ

Execution Engine では 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:
    ...

画像の切り抜き

ブロックが次元を増やし、次を持つ出力を返す場合 image kind - 通常それは画像の切り抜きを意味します。そのような場合、入力画像 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そのような状況の例は、ブロックが固定座標に基づいて切り抜きを生成する場合です（たとえば、複数の固定 Region of Interest に個別トラッカーを適用する単一動画フッテージなど）。その場合、結果の切り抜きを、あたかも別々のカメラで生成されたかのように、動画の文脈で処理したいはずです。次のように create_crop(...) の動作を調整するには 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 を作成する方法の 1 つにすぎませんが、妥当な方法です
        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 種類 は非推奨です。新しいブロックではこの kind を使用しないことを推奨します。 VideoMetadata データ表現は のメンバーとなりました WorkflowImageData Execution Engine の v1.2.0 (inference リリース v0.23.0)

VideoMetadata は、動画フレームと動画ソースに関する次のメタデータを提供する dataclass です:

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)