メディア

Copied!
1
2
3
4
5
6
7
from functions.api import function, Media
# Media 型は foundry_sdk_runtime からもインポートできます
# from foundry_sdk_runtime.media import Media

@function
def echo_media(media: Media) -> Media:
    return media

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
from ontology_sdk import FoundryClient
from foundry_sdk_runtime.media import Media
from functions.api import function

@function
def upload_media(body: str, media_set_filename: str) -> Media:
    client = FoundryClient()
    media: Media = client.ontology.media.upload_media(
        body=body.encode("utf8"),
        filename=media_set_filename,
    )
    return media

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
from ontology_sdk import FoundryClient
from foundry_sdk_runtime.media import Media
from functions.api import function

@function
async def upload_media(body: str, media_set_filename: str) -> Media:
    client = FoundryClient()
    media_coroutine = client.ontology.media.async_upload_media(
        body=body.encode("utf8"),
        filename=media_set_filename,
    )
    # media_coroutine は await 可能です。
    return await media_coroutine

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# Python オントロジー SDK 2.198 以降を使用していることを確認してください

from ontology_sdk import FoundryClient
from ontology_sdk.ontology.objects import Aircraft
from functions.api import function, OntologyEdit
from foundry_sdk_runtime.media import Media

@function(edits=[Aircraft])
def upload_text_to_new_plane() -> list[OntologyEdit]:
    client = FoundryClient()
    edits = client.ontology.edits()
    media: Media = client.ontology.media.upload_media(
        body="Hello, world".encode("utf8"),
        filename="/planes/aircraft.txt",
    )
    edits.objects.Aircraft.create(
        pk = "primary_key",
        my_media_property=media,
        # ...
    )
    return edits.get_edits()

Copied!
1
2
3
4
5
6
get_media_content(self) -> BytesIO: ...

from io import BytesIO

# https://docs.python.org/3/library/io.html#io.BytesIO
raw_data: BytesIO = my_aircraft.my_media_property.get_media_content()

Copied!
1
2
3
4
5
6
7
from foundry_sdk_runtime.media import MediaMetadata

# 使用例：
media_metadata: MediaMetadata = my_aircraft.my_media_property.get_media_metadata()
path = media_metadata.path
size_bytes = media_metadata.size_bytes
media_type = media_metadata.media_type

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
get_media_full_metadata(self) -> MediaFullMetadata: ...

# ほかのバリアントには AudioMediaItemMetadata、VideoMediaItemMetadata、
# SpreadsheetMediaItemMetadata、Model3dMediaItemMetadata、DicomMediaItemMetadata、
# EmailMediaItemMetadata、UntypedMediaItemMetadata があります。スキーマの全体を参照してください。
# https://github.com/palantir/foundry-platform-python/blob/develop/docs/v2/MediaSets/models/MediaItemMetadata.md

from foundry_sdk.v2.media_sets.models import (
    DocumentMediaItemMetadata,
    ImageryMediaItemMetadata,
)
from foundry_sdk_runtime.media import MediaFullMetadata

full_metadata: MediaFullMetadata = my_aircraft.my_media_property.get_media_full_metadata()
item = full_metadata.item_metadata

if isinstance(item, DocumentMediaItemMetadata):
    page_count = item.pages
    title = item.title
elif isinstance(item, ImageryMediaItemMetadata):
    dimensions = item.dimensions
    bands = item.bands

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
from foundry_sdk.v2.media_sets.models import (
    ImageTransformation,
    JpgFormat,
    RotateImageOperation,
)
from foundry_sdk_runtime.errors import (
    MediaTransformationFailedError,
    MediaTransformationTimeoutError,
)
from foundry_sdk_runtime.media import Media
from functions.api import function
from ontology_sdk import FoundryClient

@function(beta=True)
def image_transform(document: Media) -> Media:
    client = FoundryClient()
    transformation = ImageTransformation(
        encoding=JpgFormat(),
        operations=[RotateImageOperation(angle="DEGREE_180")],
    )
    try:
        transformed_bytes: bytes = client.ontology.media.transform_and_wait(
            media_reference=document.get_media_reference(),
            transformation=transformation,
            poll_interval_seconds=3.0,
            poll_timeout_seconds=30.0,
        )
    except MediaTransformationFailedError:
        # 変換ジョブが FAILED ステータスを報告しました。
        raise
    except MediaTransformationTimeoutError:
        # ジョブが完了する前に poll_timeout_seconds が経過しました。
        raise
    # 関数が Media を返すように、変換後のバイト列を再アップロードします。
    return client.ontology.media.upload_media(body=transformed_bytes, filename="rotated.jpg")

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
from foundry_sdk.v2.media_sets.models import (
    DocumentMediaItemMetadata,
    DocumentToTextTransformation,
    OcrHocrOutputFormat,
    OcrLanguageWrapper,
    OcrOnPageOperation,
    OcrParameters,
)
from foundry_sdk_runtime.media import Media
from functions.api import function
from ontology_sdk import FoundryClient

@function(beta=True)
def ocr_pdf_pages(document: Media) -> list[bytes]:
    """PDF のすべてのページで OCR を実行し、ページごとに hOCR のバイト列を返します。

    各 hOCR ドキュメントには、検出された単語、行、
    段落の `bbox` 属性が含まれます。任意の HTML パーサーでパースすると、1回の処理でテキストと
    境界ボックスの両方を復元できます。
    """
    client = FoundryClient()
    metadata = document.get_media_full_metadata().item_metadata
    if not isinstance(metadata, DocumentMediaItemMetadata) or metadata.pages is None:
        raise ValueError("Expected a PDF document with a known page count")

    media_reference = document.get_media_reference()
    page_results: list[bytes] = []

    for page_number in range(metadata.pages):
        transformation = DocumentToTextTransformation(
            operation=OcrOnPageOperation(
                page_number=page_number,
                parameters=OcrParameters(
                    output_format=OcrHocrOutputFormat(),
                    languages=[OcrLanguageWrapper(language="ENG")],
                ),
            ),
        )
        hocr_bytes: bytes = client.ontology.media.transform_and_wait(
            media_reference=media_reference,
            transformation=transformation,
            poll_timeout_seconds=120.0,
        )
        page_results.append(hocr_bytes)

    return page_results

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 foundry_sdk.v2.media_sets.models import (
    DocumentToImageTransformation,
    PngFormat,
    RenderPageOperation,
)
from foundry_sdk_runtime.media import Media
from functions.api import function
from ontology_sdk import FoundryClient

@function(beta=True)
def render_first_page_as_png(document: Media) -> Media:
    """PDF のページ0を幅 1200px の PNG としてレンダリングし、Media として返します。"""
    client = FoundryClient()
    transformation = DocumentToImageTransformation(
        encoding=PngFormat(),
        operation=RenderPageOperation(page_number=0, width=1200),
    )
    rendered_png: bytes = client.ontology.media.transform_and_wait(
        media_reference=document.get_media_reference(),
        transformation=transformation,
    )
    # 関数が Media を返すように、レンダリングしたページを再アップロードします。
    return client.ontology.media.upload_media(body=rendered_png, filename="page.png")

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
from foundry_sdk.v2.media_sets.models import (
    DocumentToDocumentTransformation,
    PdfFormat,
    SlicePdfRangeOperation,
)
from foundry_sdk_runtime.media import Media
from functions.api import function
from ontology_sdk import FoundryClient

@function(beta=True)
def slice_first_ten_pages(document: Media) -> Media:
    """入力 PDF のページ0～9を含む新しい PDF を Media として返します。"""
    client = FoundryClient()
    transformation = DocumentToDocumentTransformation(
        encoding=PdfFormat(),
        operation=SlicePdfRangeOperation(
            start_page_inclusive=0,
            end_page_exclusive=10,
            strictly_enforce_end_page=False,  # 10ページ未満のドキュメントを許容します
        ),
    )
    sliced_pdf: bytes = client.ontology.media.transform_and_wait(
        media_reference=document.get_media_reference(),
        transformation=transformation,
    )
    # 関数が Media を返すように、切り出した PDF を再アップロードします。
    return client.ontology.media.upload_media(body=sliced_pdf, filename="slice.pdf")

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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
from foundry_sdk.v2.media_sets.models import (
    AnnotateImageOperation,
    Annotation,
    BoundingBox,
    BoundingBoxGeometry,
    DocumentMediaItemMetadata,
    DocumentToImageTransformation,
    DocumentToTextTransformation,
    ImageTransformation,
    OcrHocrOutputFormat,
    OcrLanguageWrapper,
    OcrOnPageOperation,
    OcrParameters,
    PngFormat,
    RenderPageOperation,
)
from foundry_sdk_runtime.media import Media
from functions.api import function
from ontology_sdk import FoundryClient

@function(beta=True)
def annotate_pdf_with_ocr_boxes(document: Media) -> list[Media]:
    """PDF のすべてのページをレンダリングし、各ページで OCR を実行してテキストの境界ボックスを見つけ、
    レンダリングした画像に描画し、ページごとに注釈付きの Media を1個返します。"""
    client = FoundryClient()
    media_reference = document.get_media_reference()

    # 完全なメタデータを使ってページ数を調べます。
    metadata = document.get_media_full_metadata().item_metadata
    if not isinstance(metadata, DocumentMediaItemMetadata) or metadata.pages is None:
        raise ValueError("Expected a PDF document with a known page count")

    annotated_pages: list[Media] = []
    for page_number in range(metadata.pages):
        # 1. ページを PNG としてレンダリングします。
        rendered_png: bytes = client.ontology.media.transform_and_wait(
            media_reference=media_reference,
            transformation=DocumentToImageTransformation(
                encoding=PngFormat(),
                operation=RenderPageOperation(page_number=page_number, width=1200),
            ),
        )

        # 2. 同じページで hOCR モードの OCR を実行し、単語単位の境界ボックスを取得します。
        hocr_bytes: bytes = client.ontology.media.transform_and_wait(
            media_reference=media_reference,
            transformation=DocumentToTextTransformation(
                operation=OcrOnPageOperation(
                    page_number=page_number,
                    parameters=OcrParameters(
                        output_format=OcrHocrOutputFormat(),
                        languages=[OcrLanguageWrapper(language="ENG")],
                    ),
                ),
            ),
            poll_timeout_seconds=120.0,
        )

        # 3. hOCR をパースし、画像のピクセル単位で境界ボックスを取得します。
        # parse_hocr_bounding_boxes ヘルパーはここでは省略しています。例の下の注記を参照してください。
        boxes: list[tuple[str, BoundingBox]] = parse_hocr_bounding_boxes(hocr_bytes)

        # 4. レンダリングした PNG を一時的なメディアアイテムとして再アップロードします。
        rendered_media = client.ontology.media.upload_media(
            body=rendered_png, filename=f"page-{page_number}.png"
        )

        # 5. Media の変換を使って、レンダリングしたページに注釈を付けます。
        annotated_bytes: bytes = client.ontology.media.transform_and_wait(
            media_reference=rendered_media.get_media_reference(),
            transformation=ImageTransformation(
                encoding=PngFormat(),
                operations=[
                    AnnotateImageOperation(
                        annotations=[
                            Annotation(
                                geometry=BoundingBoxGeometry(bounding_box=box),
                                label=label,
                            )
                            for label, box in boxes
                        ],
                    ),
                ],
            ),
        )

        # 6. 関数が Media を返すように、注釈を付けたページを再アップロードします。
        annotated_pages.append(
            client.ontology.media.upload_media(
                body=annotated_bytes, filename=f"page-{page_number}-annotated.png"
            )
        )

    return annotated_pages

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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import asyncio

from foundry_sdk.v2.media_sets.models import (
    AnnotateImageOperation,
    Annotation,
    BoundingBox,
    BoundingBoxGeometry,
    DocumentMediaItemMetadata,
    DocumentToImageTransformation,
    DocumentToTextTransformation,
    ImageTransformation,
    OcrHocrOutputFormat,
    OcrLanguageWrapper,
    OcrOnPageOperation,
    OcrParameters,
    PngFormat,
    RenderPageOperation,
)
from foundry_sdk_runtime.media import Media
from functions.api import function
from ontology_sdk import FoundryClient

@function(beta=True)
async def annotate_pdf_with_ocr_boxes(document: Media) -> list[Media]:
    """PDF のすべてのページをレンダリングし、各ページで OCR を実行して注釈を付け、ページごとに Media を1個返します。"""
    client = FoundryClient()
    media_reference = document.get_media_reference()

    metadata = document.get_media_full_metadata().item_metadata
    if not isinstance(metadata, DocumentMediaItemMetadata) or metadata.pages is None:
        raise ValueError("Expected a PDF document with a known page count")

    async def annotate_page(page_number: int) -> Media:
        # ページの PNG としてのレンダリングと、同じページの OCR を同時に実行します。
        # どちらの変換も同じソースドキュメントを読み取り、互いに独立しているため、
        # asyncio.gather により、順番にではなく並列でポーリングできます。
        rendered_png, hocr_bytes = await asyncio.gather(
            client.ontology.media.async_transform_and_wait(
                media_reference=media_reference,
                transformation=DocumentToImageTransformation(
                    encoding=PngFormat(),
                    operation=RenderPageOperation(page_number=page_number, width=1200),
                ),
            ),
            client.ontology.media.async_transform_and_wait(
                media_reference=media_reference,
                transformation=DocumentToTextTransformation(
                    operation=OcrOnPageOperation(
                        page_number=page_number,
                        parameters=OcrParameters(
                            output_format=OcrHocrOutputFormat(),
                            languages=[OcrLanguageWrapper(language="ENG")],
                        ),
                    ),
                ),
                poll_timeout_seconds=120.0,
            ),
        )

        # hOCR のパースによる境界ボックスの取得（同期版の例を参照）と、
        # レンダリングした PNG の一時的なメディアアイテムとしての再アップロードを同時に実行します。
        boxes, rendered_media = await asyncio.gather(
            async_parse_hocr_bounding_boxes(hocr_bytes),
            client.ontology.media.async_upload_media(
                body=rendered_png,
                filename=f"page-{page_number}.png",
            ),
        )

        # Media の変換を使って、レンダリングしたページに注釈を付けます。
        annotated_bytes: bytes = await client.ontology.media.async_transform_and_wait(
            media_reference=rendered_media.get_media_reference(),
            transformation=ImageTransformation(
                encoding=PngFormat(),
                operations=[
                    AnnotateImageOperation(
                        annotations=[
                            Annotation(
                                geometry=BoundingBoxGeometry(bounding_box=box),
                                label=label,
                            )
                            for label, box in boxes
                        ],
                    ),
                ],
            ),
        )

        # 関数が Media を返すように、注釈を付けたページを再アップロードします。
        return await client.ontology.media.async_upload_media(
            body=annotated_bytes,
            filename=f"page-{page_number}-annotated.png",
        )

    # すべてのページを同時に処理します。
    return list(await asyncio.gather(*(annotate_page(p) for p in range(metadata.pages))))