Python トランスフォームでメディアセットを使用する

Python トランスフォームでは、メディアセットを使用して、PDF テキスト抽出、光学式文字認識（OCR）、画像のタイル化、メタデータのパースなどを実行できます。以下のセクションでは、Python リポジトリでメディアセットを設定する方法と、Python トランスフォームでメディアセットの読み取りと書き込みを行う方法を説明します。

Code Repositories でメディアセットのバッチパイプラインを作成する手順を説明した例をご覧いただけます。

メディアセットを使用したインクリメンタル（差分処理）トランスフォームの作成についても学習できます。

transforms-media ライブラリをリポジトリにインポートする

メディアセット固有のデコレーターを使用するには、まず transforms-media ライブラリをリポジトリにインポートする必要があります。Code Repositories インターフェースの左側にあるライブラリファイルドロワーに移動します。transforms-media を検索し、ライブラリをインストールします。

コードリポジトリに transforms-media への依存関係を追加する。

メディアセットを扱う場合は、@transform デコレーターを使用する必要があります。メディアセットの入力と出力は、transforms.mediasets.MediaSetInput と transforms.mediasets.MediaSetOutput の指定を使用して渡すことができます。ビルド時には、入力の指定が transforms.mediasets.MediaSetInputParam オブジェクトに、出力の指定が transforms.mediasets.MediaSetOutputParam オブジェクトに解決されます。これらの MediaSetInputParam および MediaSetOutputParam オブジェクトを使用すると、計算関数内でメディアセットにアクセスできます。任意の数のメディアセットの入力や出力を、他の有効なトランスフォームの入力や出力（表形式のデータセットなど）と組み合わせて使用できます。たとえば、次のように指定します。

Copied!
1
2
3
4
5
6
7
8
9
10
from transforms.api import transform
from transforms.mediasets import MediaSetInput, MediaSetOutput


@transform.spark.using(
    images=MediaSetInput('/examples/images'),
    output_images=MediaSetOutput('/examples/output_images')
)
def translate_images(images, output_images):
    ...

メディアセットから読み取る

個々のメディアアイテムには、ファイルパスまたは RID でアクセスできます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
from transforms.api import transform
from transforms.mediasets import MediaSetInput, MediaSetOutput


@transform.spark.using(
    images=MediaSetInput('/examples/images'),
    output_images=MediaSetOutput('/examples/output_images')
)
def translate_images(images, output_images):
    image1 = images.get_media_item_by_path("image1")
    image2 = images.get_media_item("ri.mio.main.media-item.123")
    ...

ただし、多くの場合、メディアセット内のすべての項目を加工する必要があります。そのためには、まず listing メソッドを使用して項目をデータフレームに取り込む必要があります。以下の例では、入力メディアセット内のすべての項目を一覧化し、得られたデータフレームを表形式の出力に書き込みます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from transforms.api import transform, Output
from transforms.mediasets import MediaSetInput


@transform.spark.using(
    images=MediaSetInput('/examples/images'),
    listing_output=Output('/examples/listed_images')
)
def translate_images(ctx, images, listing_output):
    media_items_listing = images.list_media_items_by_path_with_media_reference(ctx)

    # media_items_listing に対して通常の PySpark のデータ加工を実行できます

    column_typeclasses = {'mediaReference': [{'kind': 'reference', 'name': 'media_reference'}]}
    listing_output.write_dataframe(media_items_listing, column_typeclasses=column_typeclasses)

メディアセット内の特定のパスに複数の項目がある場合、最新の項目のみが一覧に含まれます。一覧には以下のスキーマが適用されます。

+--------------------------+-----------+-------------------+
|        mediaItemRid      |    path   |  mediaReference   |
+--------------------------+-----------+-------------------+
| ri.mio.main.media-item.1 | item1.jpg |  {{reference1}}   |
| ri.mio.main.media-item.2 | item2.jpg |  {{reference2}}   |
| ri.mio.main.media-item.3 | item3.jpg |  {{reference3}}   |
+--------------------------+-----------+-------------------+

上記の例では、一覧の先頭 3 行のみを示しています。

mediaReference 列の型クラスを設定すると、その列を Media Reference として読み取ることができます。

get_media_item()、get_media_item_by_path() などの呼び出しは、Python のファイルライクなストリームオブジェクトを返します。io.open() ↗ が受け付けるすべてのオプションにも対応しています。項目はストリームとして読み取られるため、ランダムアクセスには対応していません。

項目全体をダウンロードせずに、個々のメディアアイテムのメタデータを返すこともできます。メタデータには、画像の寸法や音声の長さなどの情報が含まれます。利用可能なメタデータの完全なリファレンスについては、以下の付録を参照してください。以下の例では、各画像のメタデータを格納する列をメディアアイテムの一覧に追加します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from transforms.api import transform, Output
from transforms.mediasets import MediaSetInput
from pyspark.sql import functions as F
from pyspark.sql.types import StringType
from conjure_python_client import ConjureEncoder


@transform.spark.using(
    images=MediaSetInput('/examples/images'),
    listing_output_with_metadata=Output('/examples/listed_images_with_metadata')
)
def translate_images(ctx, images, listing_output_with_metadata):

    def get_metadata(media_item_rid):
        metadata = images.get_media_item_metadata(media_item_rid)
        return ConjureEncoder().default(metadata)

    metadata_udf = F.udf(get_metadata, StringType())

    media_items_listing = images.list_media_items_by_path_with_media_reference(ctx)
    listing_with_metadata = media_items_listing.withColumn('metadata', metadata_udf(F.col('mediaItemRid')))
    column_typeclasses = {'mediaReference': [{'kind': 'reference', 'name': 'media_reference'}]}
    listing_output_with_metadata.write_dataframe(listing_with_metadata, column_typeclasses=column_typeclasses)

メディアセットには、すぐに使用できる組み込みの変換がいくつか用意されています。API および対応している変換の一覧については、以下の付録を参照してください。これらの変換を呼び出した場合も、Python のファイルライクなストリームオブジェクトが返されます。組み込みの変換を使用するには、メディアセットの入力で適切なメソッドを呼び出します。たとえば、次のように呼び出します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@transform.spark.using(
    images=MediaSetInput('/examples/images'),
    image_text_output=Output('/examples/listed_images_with_text')
)
def translate_images(ctx, images, image_text_output):

    def get_ocr_on_image(media_item_rid):
        return images.transform_image_to_text_ocr_output_text(media_item_rid).read().decode('utf-8')

    ocr_on_image_udf = F.udf(get_ocr_on_image, StringType())

    media_items_listing = images.list_media_items_by_path_with_media_reference(ctx)
    listing_with_ocr = media_items_listing.withColumn('text', ocr_on_image_udf(F.col('mediaItemRid')))
    column_typeclasses = {'mediaReference': [{'kind': 'reference', 'name': 'media_reference'}]}
    image_text_output.write_dataframe(listing_with_ocr, column_typeclasses=column_typeclasses)

メディアセットを作成する

メディアセット作成ダイアログを使用してプロジェクトやフォルダー内にメディアセットを作成するだけでなく、Code Repositories 内で直接メディアセットを作成し、Python トランスフォームの出力として使用することもできます。

まず、MediaSetOutput の指定で、新しいメディアセットの出力先と名前を選択します。

定義したパスを選択します。この時点では、パスに赤い下線が表示されているはずです。ホバー時に、メディアセットが存在しないことを示すエラーメッセージが表示されます。

行の左側にある電球アイコンから、メディアセットを作成を選択します。

ダイアログの手順に従って目的のメディアセットスキーマを選択し、新しいメディアセットで必要となるその他の設定を完了します。

作成するを選択すると、MediaSetOutput の指定に、入力した詳細情報が反映されます。これらの注釈フィールドは、メディアセットの作成方法を定義します。

新しいメディアセットは、Python トランスフォームが初めてビルドされた後に作成されます。それ以降は、注釈フィールドを編集しないでください。

メディアセットに書き込む

MediaSetOutput の指定を使用すると、メディアセットをトランスフォームの出力として使用できます。

項目をアップロードするには、出力メディアセットで put_media_item() エンドポイントを呼び出します。このエンドポイントは、任意のファイルライクオブジェクトと、出力メディアセット内で項目を識別するために使用されるパスを受け付けます。以下に基本的な例を示します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
from transforms.api import transform
from transforms.mediasets import MediaSetInput, MediaSetOutput


@transform.spark.using(
    images=MediaSetInput('/examples/images'),
    output_images=MediaSetOutput('/examples/output_images')
)
def upload_images(images, output_images):
    with images.get_media_item_by_path("image1.jpg") as input_image:
        output_images.put_media_item(input_image, "copied_image1.jpg")

あるメディアセットから別のメディアセットに項目をコピーする場合は、出力で fast_copy_media_item() メソッドを使用できます。この方法は、メディアアイテムをダウンロードして再アップロードするよりも高速で効率的です。

Copied!
1
2
3
4
5
6
7
@transform.spark.using(
    images=MediaSetInput('/examples/images'),
    output_images=MediaSetOutput('/examples/output_images')
)
def upload_images(images, output_images):
    origin_media_item_rid = images.get_media_item_rid_by_path("image1.jpg").item
    output_images.fast_copy_media_item(images, origin_media_item_rid, "fast_copied_image1.jpg")

ユーザー定義関数（UDF）内でメディアセットに項目をアップロードすると、並列度を高めることができます。以下の例では、組み込みの PDF から JPEG への変換を使用して、入力メディアセット内の PDF を JPEG に変換し、それらの JPEG を新しい出力メディアセットにアップロードします。続いて、アップロードした JPEG の Media Reference を含む表形式のデータセットを書き出します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from transforms.api import transform, Output
from transforms.mediasets import MediaSetInput, MediaSetOutput
from pyspark.sql import functions as F
from pyspark.sql.types import StringType


@transform.spark.using(
    pdfs=MediaSetInput('/examples/PDFs'),
    output_images=MediaSetOutput('/examples/JPEGs'),
    output_references=Output('/examples/JPEG listing')
)
def upload_images(ctx, pdfs, output_images, output_references):

    def upload_jpg(media_item_rid, path):
        with pdfs.transform_document_to_jpg(media_item_rid, 0) as jpeg:
            response = output_images.put_media_item(jpeg, path)
        return response.media_item_rid

    upload_udf = F.udf(upload_jpg, StringType())

    listed_pdfs = pdfs.list_media_items_by_path(ctx)
    media_reference_template = output_images.media_reference_template()
    uploaded_jpegs = listed_pdfs\
        .withColumn('uploaded_media_item_rid', upload_udf(F.col('mediaItemRid'), F.col('path')))\
        .select('path', 'uploaded_media_item_rid')\
        .withColumn("mediaReference", F.format_string(media_reference_template,
                                                      'uploaded_media_item_rid'))

    column_typeclasses = {'mediaReference': [{'kind': 'reference', 'name': 'media_reference'}]}
    output_references.write_dataframe(uploaded_jpegs, column_typeclasses=column_typeclasses)

メディアセットの書き込みモード

メディアセットへの書き込みには、次の 2 種類の書き込みモードのいずれかを使用できます。

• modify：アップロードされた項目が、メディアセットのブランチ内の既存の項目に追加されます。
• replace：アップロードされた項目で、メディアセットのブランチ全体が置き換えられます。

デフォルトの書き込みモードは、メディアセットのトランザクションポリシーによって異なります。トランザクションありのメディアセットでは、デフォルトは replace です。トランザクションなしのメディアセットでは、modify 書き込みモードを使用します。トランザクションなしのメディアセットのブランチは空の状態にリセットできないため、このモードは変更できません。

メディアセットの出力の書き込みモードは、実行時に動的に変更できます。これは、出力全体を置き換えるかどうかを、パイプライン内のカスタム条件に基づいて判断する場合に役立ちます。

メディアセットの書き込みモードを変更するには、メディアセットの出力で .set_write_mode() メソッドを使用します。出力に項目がアップロードされるまでの任意の時点で、書き込みモードを変更できます。たとえば、次のように変更します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform, Input
from transforms.mediasets import MediaSetOutput

@transform.spark.using(
    input_PNGs=Input('/examples/input_PNGs'),
    output_PNGs=MediaSetOutput('/examples/output_PNGs'),
)
def upload_pngs(input_PNGs, output_PNGs):
    if should_replace(input_PNGs):
        output_PNGs.set_write_mode("replace")
    else:
        output_PNGs.set_write_mode("modify")

    output_PNGs.put_dataset_files(input_PNGs)

ファイルシステム（Catalog）データセットからアップロードする

put_dataset_files を使用する

Python メディアセット SDK には、Palantir ファイルシステム（Catalog と呼ばれます）の従来のデータセットからメディアセットにファイルをアップロードするためのツールが組み込まれています。たとえば、次のように使用します。

Copied!
1
2
3
4
5
6
7
8
9
10
from transforms.api import transform, Input
from transforms.mediasets import MediaSetOutput


@transform.spark.using(
    pdfs_dataset=Input('/examples/PDFs'),
    pdfs_media_set=MediaSetOutput('/examples/PDFs media set')
)
def upload_to_media_set(pdfs_dataset, pdfs_media_set):
    pdfs_media_set.put_dataset_files(pdfs_dataset, ignore_items_not_matching_schema=False)

このトランスフォームは、データセット内のすべての項目をメディアセットにアップロードします。メディアセットのスキーマに一致しない項目がある場合（たとえば、データセットに JPEG が含まれている場合）、ビルドに失敗します。ignore_items_not_matching_schema=True を設定すると、このような不一致は無視（確認）されます。

put_media_item を使用する

ファイルを個別にアップロードすることもできます。たとえば、次のようにアップロードします。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from transforms.api import transform, Input, Output, incremental
from transforms.mediasets import MediaSetInput, MediaSetOutput
import os


@transform.spark.using(
    output_media_set=MediaSetOutput("/path/media_set_output", should_snapshot=False),
    input_dataset=Input("/path/dataset_of_raw_files"),
)
def compute(input_dataset, output_media_set):
    all_files = list(input_dataset.filesystem().ls())
    for current_file in all_files:
        with input_dataset.filesystem().open(current_file.path, 'rb') as f:
            filename = os.path.basename(current_file.path)
            output_media_set.put_media_item(f, filename)

軽量トランスフォームへの対応

メディアセットは、軽量トランスフォームを使用して加工できます。API は標準の Python トランスフォームと同じですが、メディアアイテムを一覧化する際のデータフレーム操作には、pandas などの単一ノードのライブラリを使用します。以下の例は、入力メディアセット内のすべての項目を一覧化し、得られたデータフレームを表形式の出力に書き込む軽量トランスフォームを示しています。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
from transforms.api import transform, Output, lightweight, LightweightOutput
from transforms.mediasets import MediaSetInput, LightweightMediaSetInputParam

@transform.using(
    images=MediaSetInput('/examples/images'),
    listing_output=Output('/examples/listed_images')
)
def translate_images(images: LightweightMediaSetInputParam, listing_output: LightweightOutput):
    media_items_listing = images.list_media_items_by_path_with_media_reference().pandas()

    # media_items_listing に対して通常の pandas のデータ加工を実行できます

    listing_output.write_table(media_items_listing)

ドキュメントからレイアウトを考慮したコンテンツを抽出する

メディアセットを扱う場合は、トランスフォームを使用して、段落、ヘッダー、テーブルなどのドキュメントのコンテンツを、そのレイアウトに関する追加のメタデータとともに抽出できます。この抽出トランスフォームは、PDF と画像のどちらのメディアセットでも実行できます。

抽出した境界ボックスをビジョンモデルに渡すと、特に複雑なドキュメントでより良い結果が得られる場合があります。

PDF ドキュメントの場合、V2 レスポンスは、リクエストされた各ページのブロック配列を格納した pages 配列を持つオブジェクトです。各ブロックには、種類、信頼度、ID、境界ボックス、抽出されたテキスト、および 0 から始まるページ番号が含まれます。format を TEXT に設定した場合、ブロックのテキストは {"type": "text", "text": "Extracted content"} のようなタグ付きの値になります。

以下の Python トランスフォームは、OCR を使用して 1 ページずつ抽出し、結合した V2 レスポンスを JSON として text 列に保存します。各ドキュメントは 1 行に格納されます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import json

from transforms.api import transform, Output
from transforms.mediasets import MediaSetInput, MediaSetInputParam
from pyspark.sql.functions import udf


@transform.spark.using(
    output=Output('ri.foundry.main.dataset.0-1-2-3-4'),
    media_input=MediaSetInput('/examples/input_pdfs'),
)
def compute(media_input: MediaSetInputParam, output):

    def extract_all_text(media_item_rid):
        metadata = media_input.get_media_item_metadata(media_item_rid)
        pages = metadata.document.pages
        if pages is None:
            return ""

        extracted_pages = []
        for page in range(pages):
            response = media_input.transform_media_item(media_item_rid, str(page), {
                "type": "documentToText",
                "documentToText": {
                    "operation": {
                        "type": "extractLayoutAwareTextV2",
                        "extractLayoutAwareTextV2": {
                            "pageRange": {
                                "startPageInclusive": page,
                                "endPageExclusive": page + 1
                            },
                            "config": {
                                "mode": "SCAN",
                                "format": "TEXT",
                                "languages": [{"type": "language", "language": "ENG"}]
                            }
                        }
                    }
                }
            })
            extracted_pages.extend(response.json()["pages"])
        return json.dumps({"pages": extracted_pages})

    extract_text_udf = udf(extract_all_text)

    result = media_input.dataframe().withColumn("text", extract_text_udf("mediaItemRid"))

    column_typeclasses = {
        "mediaReference": [{"kind": "reference", "name": "media_reference"}]
    }

    output.write_dataframe(result, column_typeclasses=column_typeclasses)

この抽出トランスフォームを多数のドキュメントやページ数の多いドキュメントに対して実行する場合は、最適なパフォーマンスを得るために並列処理を使用することをお勧めします。

サイズの大きいドキュメントの場合は、pageRange を使用して、1 回のリクエストで複数のページをまとめて処理できます。ページ範囲の設定について詳しくはこちら。

リファレンス：組み込みの変換

利用可能なすべての変換メソッド、パラメーター、および例の完全なリファレンスについては、メディアセットのトランスフォーム API ドキュメントを参照してください。

PDF ドキュメントを JPEG 画像に変換する

ドキュメントのページを、指定した寸法の画像に変換します。

パラメーター：

• media_item_rid (Optional[str])：指定した場合、メディアセット全体ではなく、指定した項目に対して変換を実行します。デフォルトは None です。
• start_page (Optional[int])：変換を開始するページを、0 から始まるインデックスで指定します。デフォルトは 0（最初のページ）です。
• end_page (Optional[int])：変換を終了するページを、0 から始まるインデックスで指定します（このページは含まれません）。デフォルトは None（最後のページ）です。
• width (Optional[int])：出力画像の幅です。デフォルトは 1024 です。
• height (Optional[int])：出力画像の高さです。デフォルトは 1024 です。
• output_format (str)：PNG や JPEG など、出力画像の形式です。デフォルトは PNG です。

戻り値：

• ドキュメントから画像への変換を含む MediaSetInputTransform のインスタンスです。さらに変換を適用できます。

例：

Copied!
1
2
transform = pdf_input.transform().convert_document_to_images("ri.mio.main.media-item.1", 0, 5, 600, 900, output_format="jpg")
image_output.write(transform)

OCR を使用して PDF ドキュメントセットをテキストに変換する

このトランスフォームは、視覚言語モデルで抽出を行う AI ベースの OCR ではなく、従来の光学式文字認識（OCR）を使用して PDF ドキュメントからテキストを抽出します。視覚言語モデルを使用した PDF ドキュメントのコンテンツ抽出について詳しくはこちら。

パラメーター：

• languages (list[str])：OCR に使用する言語のリストです。デフォルトは英語です。有効なコードはすべて、Tesseract ドキュメント ↗ の languages に記載されています。
• scripts (Optional[list[str]])：OCR に使用する文字体系のリストです。デフォルトは None です。有効なコードはすべて、Tesseract ドキュメント ↗ の scripts に記載されています。
• media_item_rid (Optional[str])：指定した場合、メディアセット全体ではなく、指定した項目に対して変換を実行します。デフォルトは None です。
• start_page (Optional[int])：OCR を開始するページを、0 から始まるインデックスで指定します。PDF メディアセットにのみ適用されます。デフォルトは 0（最初のページ）です。
• end_page (Optional[int])：OCR を終了するページを、0 から始まるインデックスで指定します（このページは含まれません）。PDF メディアセットにのみ適用されます。デフォルトは None（最後のページ）です。
• return_structure (str)：item_per_row または page_per_row です。PDF メディアセット全体の変換にのみ適用されます。デフォルトは item_per_row です。
• suppress_errors (bool)：エラー処理の動作を指定します。True の場合、エラーが捕捉され、エラーメッセージが出力に返されます。False の場合、エラーは捕捉されず、ビルドに失敗します。デフォルトは True です。メディアセット全体の変換にのみ適用されます。

戻り値：

• DataFrame、文字列のリスト、または単一の文字列です。
  • str：単一の画像に対する変換。
  • list[str]：単一の PDF に対する変換。
  • DataFrame：メディアセット全体に対する変換。
    • PDF（item_per_row）の場合：列は media_item_rid、path、media_reference、extracted_text（list[str]）です。
    • PDF（page_per_row）または画像セットの場合：列は media_item_rid、path、media_reference、page_number、extracted_text（str）です。

例：

Copied!
1
2
df = media_set.transform().ocr()
dataset_output.write_dataframe(df)

音声をテキストに文字起こしする

発話を含む音声ファイルをテキストに文字起こしします。

パラメーター：

• media_item_rid (Optional[str])：指定した場合、メディアセット全体ではなく、指定した項目に対して変換を実行します。デフォルトは None です。
• language (Optional[str])：文字起こしに使用する言語です。デフォルトは None で、その場合は自動検出されます。有効な言語は、Whisper GitHub リポジトリ ↗ の LANGUAGES に記載されています。
• performance_mode (Literal["more_economical", "more_performant"])：文字起こしに使用するパフォーマンスモードです。デフォルトは more_economical です。
• output_format (Literal["text", "segments"])：出力の形式です。デフォルトは text です。
• add_timestamps (Optional[bool])：文字起こしにタイムスタンプを追加するかどうかを制御します。デフォルトは False です。output_format が text の場合にのみ適用されます。
• suppress_errors (bool)：エラー処理の動作を指定します。True の場合、エラーが捕捉され、エラーメッセージが出力に返されます。False の場合、エラーは捕捉されず、ビルドに失敗します。デフォルトは True です。メディアセット全体の変換にのみ適用されます。

戻り値：

• DataFrame または単一の文字列です。
  • str：単一の項目に対する変換に適用されます。
    • text：文字起こしされたテキスト。
    • segments：タイムスタンプ、セグメントの信頼度、その他の詳細を含む、文字起こしされたセグメントを格納する JSON オブジェクト。
  • DataFrame：列は media_item_rid、path、media_reference、transcription（str）です。メディアセット全体に対する変換に適用されます。

例：

Copied!
1
2
df = media_set.transform().transcribe("ri.mio.main.media-item.1", language="english")
dataset_output.write_dataframe(df)