12799820886

Committed 16 Jan 2025 12:51AM UTC coverage: 95.594% (+15.0%) from 80.557%

Build # 12799820886

Build Type

Pull #63

github

Specific Base 3fcc9d

Committed by

tito

Commit Message

fix(openai): fix input_images format

Pull Request Pull Request #63: feat(api): update with latest API yaml from aiapi

Run Details

196 of 202 new or added lines in 21 files covered. (97.03%)

33 existing lines in 7 files now uncovered.

2061 of 2156 relevant lines covered (95.59%)

3.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.12

/scope3ai/tracers/huggingface/image_to_image.py

import io
import time
from dataclasses import dataclass
from typing import Any, Callable, Optional, Union

import tiktoken
from aiohttp import ClientResponse
from huggingface_hub import (  # type: ignore[import-untyped]
    AsyncInferenceClient,
    InferenceClient,
)
from huggingface_hub import ImageToImageOutput as _ImageToImageOutput
from PIL import Image
from requests import Response

from scope3ai.api.types import ImpactRow, Scope3AIContext
from scope3ai.api.typesgen import Image as RootImage
from scope3ai.api.typesgen import Task
from scope3ai.constants import PROVIDERS
from scope3ai.lib import Scope3AI
from scope3ai.response_interceptor.aiohttp_interceptor import aiohttp_response_capture
from scope3ai.response_interceptor.requests_interceptor import requests_response_capture

PROVIDER = PROVIDERS.HUGGINGFACE_HUB.value
HUGGING_FACE_IMAGE_TO_IMAGE_TASK = "chat"


@dataclass
class ImageToImageOutput(_ImageToImageOutput):
    scope3ai: Optional[Scope3AIContext] = None


def _hugging_face_image_to_image_wrapper(
    timer_start: Any,
    model: Any,
    response: Any,
    http_response: Optional[Union[ClientResponse, Response]],
    args: Any,
    kwargs: Any,
) -> ImageToImageOutput:
    compute_time = time.perf_counter() - timer_start
    input_tokens = 0
    if http_response:
        compute_time = http_response.headers.get("x-compute-time") or compute_time
        input_tokens = http_response.headers.get("x-compute-characters")
    if not input_tokens:
        encoder = tiktoken.get_encoding("cl100k_base")
        prompt = args[1] if len(args) > 1 else kwargs.get("prompt", "")
        input_tokens = len(encoder.encode(prompt)) if prompt != "" else 0
    try:
        image_param = args[0] if len(args) > 0 else kwargs["image"]
        if type(image_param) is str:
            input_image = Image.open(args[0] if len(args) > 0 else kwargs["image"])
        else:
            input_image = Image.open(io.BytesIO(image_param))
        input_width, input_height = input_image.size
        input_images = [RootImage(root=f"{input_width}x{input_height}")]
    except Exception:
        input_images = []
        pass
    output_width, output_height = response.size
    scope3_row = ImpactRow(
        model_id=model,
        input_tokens=int(input_tokens),
        task=Task.image_generation,
        request_duration_ms=float(compute_time) * 1000,
        managed_service_id=PROVIDER,
        output_images=[RootImage(root=f"{output_width}x{output_height}")],
        input_images=input_images,
    )

    scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
    result = ImageToImageOutput(response)
    result.scope3ai = scope3_ctx
    return result


def huggingface_image_to_image_wrapper(
    wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
) -> ImageToImageOutput:
    timer_start = time.perf_counter()
    http_response: Response | None = None
    with requests_response_capture() as responses:
        response = wrapped(*args, **kwargs)
        http_responses = responses.get()
        if http_responses:
            http_response = http_responses[-1]
    model = kwargs.get("model") or instance.get_recommended_model(
        HUGGING_FACE_IMAGE_TO_IMAGE_TASK
    )
    return _hugging_face_image_to_image_wrapper(
        timer_start, model, response, http_response, args, kwargs
    )


async def huggingface_image_to_image_wrapper_async(
    wrapped: Callable, instance: AsyncInferenceClient, args: Any, kwargs: Any
) -> ImageToImageOutput:
    timer_start = time.perf_counter()
    http_response: ClientResponse | None = None
    with aiohttp_response_capture() as responses:
        response = await wrapped(*args, **kwargs)
        http_responses = responses.get()
        if http_responses:
            http_response = http_responses[-1]
    model = kwargs.get("model") or instance.get_recommended_model(
        HUGGING_FACE_IMAGE_TO_IMAGE_TASK
    )
    return _hugging_face_image_to_image_wrapper(
        timer_start, model, response, http_response, args, kwargs
    )

1	import io	4✔
2	import time	4✔
3	from dataclasses import dataclass	4✔
4	from typing import Any, Callable, Optional, Union	4✔
5
6	import tiktoken	4✔
7	from aiohttp import ClientResponse	4✔
8	from huggingface_hub import ( # type: ignore[import-untyped]	4✔
9	AsyncInferenceClient,
10	InferenceClient,
11	)
12	from huggingface_hub import ImageToImageOutput as _ImageToImageOutput	4✔
13	from PIL import Image	4✔
14	from requests import Response	4✔
15
16	from scope3ai.api.types import ImpactRow, Scope3AIContext	4✔
17	from scope3ai.api.typesgen import Image as RootImage	4✔
18	from scope3ai.api.typesgen import Task	4✔
19	from scope3ai.constants import PROVIDERS	4✔
20	from scope3ai.lib import Scope3AI	4✔
21	from scope3ai.response_interceptor.aiohttp_interceptor import aiohttp_response_capture	4✔
22	from scope3ai.response_interceptor.requests_interceptor import requests_response_capture	4✔
23
24	PROVIDER = PROVIDERS.HUGGINGFACE_HUB.value	4✔
25	HUGGING_FACE_IMAGE_TO_IMAGE_TASK = "chat"	4✔
26
27
28	@dataclass	4✔
29	class ImageToImageOutput(_ImageToImageOutput):	4✔
30	scope3ai: Optional[Scope3AIContext] = None	4✔
31
32
33	def _hugging_face_image_to_image_wrapper(	4✔
34	timer_start: Any,
35	model: Any,
36	response: Any,
37	http_response: Optional[Union[ClientResponse, Response]],
38	args: Any,
39	kwargs: Any,
40	) -> ImageToImageOutput:
41	compute_time = time.perf_counter() - timer_start	4✔
42	input_tokens = 0	4✔
43	if http_response:	4✔
44	compute_time = http_response.headers.get("x-compute-time") or compute_time	4✔
45	input_tokens = http_response.headers.get("x-compute-characters")	4✔
46	if not input_tokens:	4✔
47	encoder = tiktoken.get_encoding("cl100k_base")	4✔
48	prompt = args[1] if len(args) > 1 else kwargs.get("prompt", "")	4✔
49	input_tokens = len(encoder.encode(prompt)) if prompt != "" else 0	4✔
50	try:	4✔
51	image_param = args[0] if len(args) > 0 else kwargs["image"]	4✔
52	if type(image_param) is str:	4✔
53	input_image = Image.open(args[0] if len(args) > 0 else kwargs["image"])	4✔
54	else:
55	input_image = Image.open(io.BytesIO(image_param))	×
56	input_width, input_height = input_image.size	4✔
57	input_images = [RootImage(root=f"{input_width}x{input_height}")]	4✔
58	except Exception:	×
NEW 59	input_images = []	×
60	pass	×
61	output_width, output_height = response.size	4✔
62	scope3_row = ImpactRow(	4✔
63	model_id=model,
64	input_tokens=int(input_tokens),
65	task=Task.image_generation,
66	request_duration_ms=float(compute_time) * 1000,
67	managed_service_id=PROVIDER,
68	output_images=[RootImage(root=f"{output_width}x{output_height}")],
69	input_images=input_images,
70	)
71
72	scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)	4✔
73	result = ImageToImageOutput(response)	4✔
74	result.scope3ai = scope3_ctx	4✔
75	return result	4✔
76
77
78	def huggingface_image_to_image_wrapper(	4✔
79	wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
80	) -> ImageToImageOutput:
81	timer_start = time.perf_counter()	4✔
82	http_response: Response \| None = None	4✔
83	with requests_response_capture() as responses:	4✔
84	response = wrapped(args, *kwargs)	4✔
85	http_responses = responses.get()	4✔
86	if http_responses:	4✔
87	http_response = http_responses[-1]	4✔
88	model = kwargs.get("model") or instance.get_recommended_model(	4✔
89	HUGGING_FACE_IMAGE_TO_IMAGE_TASK
90	)
91	return _hugging_face_image_to_image_wrapper(	4✔
92	timer_start, model, response, http_response, args, kwargs
93	)
94
95
96	async def huggingface_image_to_image_wrapper_async(	4✔
97	wrapped: Callable, instance: AsyncInferenceClient, args: Any, kwargs: Any
98	) -> ImageToImageOutput:
99	timer_start = time.perf_counter()	4✔
100	http_response: ClientResponse \| None = None	4✔
101	with aiohttp_response_capture() as responses:	4✔
102	response = await wrapped(args, *kwargs)	4✔
103	http_responses = responses.get()	4✔
104	if http_responses:	4✔
105	http_response = http_responses[-1]	4✔
106	model = kwargs.get("model") or instance.get_recommended_model(	4✔
107	HUGGING_FACE_IMAGE_TO_IMAGE_TASK
108	)
109	return _hugging_face_image_to_image_wrapper(	4✔
110	timer_start, model, response, http_response, args, kwargs
111	)

scope3data / scope3ai-py / 12799820886

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous