Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SDK]Fix dict subscribing in py38 #836

Merged
merged 1 commit into from
Oct 20, 2023
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 8 additions & 9 deletions src/promptflow/promptflow/_utils/multimedia_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,14 @@
import imghdr
import os
import re
import requests
import uuid

from functools import partial
from pathlib import Path
from typing import Any, Callable
from typing import Any, Callable, Dict
from urllib.parse import urlparse

import requests

from promptflow.contracts._errors import InvalidImageInput
from promptflow.contracts.flow import FlowInputDefinition
from promptflow.contracts.multimedia import Image, PFBytes
Expand Down Expand Up @@ -146,12 +146,10 @@ def create_image(value: any, base_dir: Path = None):
def save_image_to_file(image: Image, file_name: str, folder_path: Path, relative_path: Path = None):
ext = get_extension_from_mime_type(image._mime_type)
file_name = f"{file_name}.{ext}" if ext else file_name
image_reference = {
f"data:{image._mime_type};path": str(relative_path / file_name) if relative_path else file_name
}
image_reference = {f"data:{image._mime_type};path": str(relative_path / file_name) if relative_path else file_name}
path = folder_path / relative_path if relative_path else folder_path
os.makedirs(path, exist_ok=True)
with open(os.path.join(path, file_name), 'wb') as file:
with open(os.path.join(path, file_name), "wb") as file:
file.write(image)
return image_reference

Expand All @@ -163,6 +161,7 @@ def pfbytes_file_reference_encoder(obj):
file_name = str(uuid.uuid4())
return save_image_to_file(obj, file_name, folder_path, relative_path)
raise TypeError(f"Not supported to dump type '{type(obj).__name__}'.")

return pfbytes_file_reference_encoder


Expand All @@ -185,7 +184,7 @@ def convert_multimedia_data_to_base64(value: Any):


# TODO: Move this function to a more general place and integrate serialization to this function.
def recursive_process(value: Any, process_funcs: dict[type, Callable] = None) -> dict:
def recursive_process(value: Any, process_funcs: Dict[type, Callable] = None) -> dict:
if process_funcs:
for cls, f in process_funcs.items():
if isinstance(value, cls):
Expand All @@ -197,7 +196,7 @@ def recursive_process(value: Any, process_funcs: dict[type, Callable] = None) ->
return value


def load_multimedia_data(inputs: dict[str, FlowInputDefinition], line_inputs: dict, base_dir: Path):
def load_multimedia_data(inputs: Dict[str, FlowInputDefinition], line_inputs: dict, base_dir: Path):
updated_inputs = dict(line_inputs or {})
for key, value in inputs.items():
if value.type == ValueType.IMAGE:
Expand Down
Loading