> For the complete documentation index, see [llms.txt](https://docs.roboflow.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.roboflow.com/workflows/ja/gaido/developer-guide/data-representations.md).
# データ表現
多くのフレームワークは、開発者が扱う標準データ型を強制しており、Workflows エコシステムも例外ではありません。 [種類](/workflows/ja/gaido/developer-guide/kinds.md) Workflow における kind は、通過するデータの高レベルな抽象化を表しますが、提供される具体的なデータ型を理解することが重要です `WorkflowBlock.run(...)` Workflow ブロックを作成する際のメソッドです。
そして、まさにここでそれを学ぶことができます。
## `Batch`
Workflow ブロックがバッチ処理を宣言するとき、Batch と呼ばれる特別なコンテナ型を使用します。バッチ指向のすべてのパラメータは次のようにラップされます: `Batch[X]`ここで X はデータ型です:
```python
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 つ重要な違いがあります。それは **読み取り専用**です。要素を変更したり、要素を追加・削除したりすることはできません。ただし、いくつか便利な操作が利用できます:
### 要素の反復
```python
from inference.core.workflows.execution_engine.entities.base import Batch
def iterate(batch: Batch[int]) -> None:
for element in batch:
print(element)
```
### 複数のバッチを zip する
{% hint style="info" %}
**バッチの整列について心配する必要はありません**
Execution Engine は、run メソッドに渡されるバッチが同じサイズになるよう保証するため、反復中にバッチサイズの不一致によって要素が失われることはありません。
{% endhint %}
```python
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)
```
### バッチ要素のインデックスを取得する
これはタプルのリストを返します。各タプルは、入れ子になった可能性のあるバッチ構造内でのバッチ要素の位置を表します。
```python
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) など。
```
### 要素とそのインデックスを同時に取得しながら反復する
```python
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 エンコード版など異なる画像表現をキャッシュでき、効率が向上します。
{% hint style="info" %}
**動画メタデータ**
Execution Engine では `v1.2.0`で、追加されました `video_metadata` に `WorkflowImageData`このオブジェクトは動画処理のコンテキストを保持するためのもので、動画処理ブロックにのみ関連します。出力画像を作成しない場合、他のブロックはその存在を無視してもかまいません（次のセクションで説明します）。
{% endhint %}
を扱う `WorkflowImageData` は、そのインターフェースを理解すればかなり簡単です。主なメソッドとプロパティをいくつか示します:
```python
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
```
以下に、画像を変換しながらメタデータを保持する方法を示す例を示します
```python
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` は削除する必要があります（通常、それらを保持する意味がなく、基盤となる動画処理ブロックは動的に作成されたブロックに対して正しく動作しないためです）。
その操作の実装の下書きを以下に示します:
```python
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`:
```python
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 に「新しい」画像が現れることを示す必要があります。簡単には次のようにします:
```python
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`
{% hint style="warning" %}
**非推奨**
[`video_metadata` 種類](/workflows/ja/gaido/developer-guide/kinds/video-metadata.md) は非推奨です。新しいブロックではこの kind を使用しないことを推奨します。 `VideoMetadata` データ表現は のメンバーとなりました `WorkflowImageData` Execution Engine の `v1.2.0` (`inference` リリース `v0.23.0`)
{% endhint %}
`VideoMetadata` は、動画フレームと動画ソースに関する次のメタデータを提供する dataclass です:
```python
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)
```