12753874046

Committed 13 Jan 2025 06:40PM UTC coverage: 95.076% (+14.5%) from 80.557%

Build # 12753874046

Build Type

Pull #61

github

Specific Base 3a8d3f

Committed by

kevdevg

Commit Message

fix: vision pillow read bytes

Pull Request Pull Request #61: feat(Hugging face): Vision methods - image classification / image segmentation / object detection

Run Details

179 of 189 new or added lines in 5 files covered. (94.71%)

34 existing lines in 9 files now uncovered.

2008 of 2112 relevant lines covered (95.08%)

3.8 hits per line

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

95.0

/scope3ai/tracers/huggingface/vision/image_classification.py

import io
import time
from dataclasses import dataclass
from typing import Any, Callable, Optional, Union, List
from PIL import Image
from aiohttp import ClientResponse
from huggingface_hub import (
    InferenceClient,
    AsyncInferenceClient,
    ImageClassificationOutputElement,
)  # type: ignore[import-untyped]
from requests import Response

from scope3ai.api.types import Scope3AIContext, Model, ImpactRow
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_CLASSIFICATION_TASK = "image-classification"


@dataclass
class ImageClassificationOutput:
    elements: List[ImageClassificationOutputElement] = None
    scope3ai: Optional[Scope3AIContext] = None


def _hugging_face_image_classification_wrapper(
    timer_start: Any,
    model: Any,
    response: Any,
    http_response: Union[ClientResponse, Response],
    args: Any,
    kwargs: Any,
) -> ImageClassificationOutput:
    input_tokens = 0
    if http_response:
        compute_time = http_response.headers.get("x-compute-time")
    else:
        compute_time = time.perf_counter() - timer_start
    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 = [
            ("{width}x{height}".format(width=input_width, height=input_height))
        ]
    except Exception:
        pass
    scope3_row = ImpactRow(
        model=Model(id=model),
        input_tokens=input_tokens,
        task=Task.image_classification,
        output_images=[],  # No images to output in classification
        request_duration_ms=float(compute_time) * 1000,
        managed_service_id=PROVIDER,
        input_images=input_images,
    )

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


def huggingface_image_classification_wrapper(
    wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
) -> ImageClassificationOutput:
    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 len(http_responses) > 0:
            http_response = http_responses[-1]
    model = kwargs.get("model") or instance.get_recommended_model(
        HUGGING_FACE_IMAGE_CLASSIFICATION_TASK
    )
    return _hugging_face_image_classification_wrapper(
        timer_start, model, response, http_response, args, kwargs
    )


async def huggingface_image_classification_wrapper_async(
    wrapped: Callable, instance: AsyncInferenceClient, args: Any, kwargs: Any
) -> ImageClassificationOutput:
    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 len(http_responses) > 0:
            http_response = http_responses[-1]
    model = kwargs.get("model") or instance.get_recommended_model(
        HUGGING_FACE_IMAGE_CLASSIFICATION_TASK
    )
    return _hugging_face_image_classification_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, List	4✔
5	from PIL import Image	4✔
6	from aiohttp import ClientResponse	4✔
7	from huggingface_hub import (	4✔
8	InferenceClient,
9	AsyncInferenceClient,
10	ImageClassificationOutputElement,
11	) # type: ignore[import-untyped]
12	from requests import Response	4✔
13
14	from scope3ai.api.types import Scope3AIContext, Model, ImpactRow	4✔
15	from scope3ai.api.typesgen import Task	4✔
16	from scope3ai.constants import PROVIDERS	4✔
17	from scope3ai.lib import Scope3AI	4✔
18	from scope3ai.response_interceptor.aiohttp_interceptor import aiohttp_response_capture	4✔
19	from scope3ai.response_interceptor.requests_interceptor import requests_response_capture	4✔
20
21	PROVIDER = PROVIDERS.HUGGINGFACE_HUB.value	4✔
22	HUGGING_FACE_IMAGE_CLASSIFICATION_TASK = "image-classification"	4✔
23
24
25	@dataclass	4✔
26	class ImageClassificationOutput:	4✔
27	elements: List[ImageClassificationOutputElement] = None	4✔
28	scope3ai: Optional[Scope3AIContext] = None	4✔
29
30
31	def _hugging_face_image_classification_wrapper(	4✔
32	timer_start: Any,
33	model: Any,
34	response: Any,
35	http_response: Union[ClientResponse, Response],
36	args: Any,
37	kwargs: Any,
38	) -> ImageClassificationOutput:
39	input_tokens = 0	4✔
40	if http_response:	4✔
41	compute_time = http_response.headers.get("x-compute-time")	4✔
42	else:
NEW 43	compute_time = time.perf_counter() - timer_start	×
44	try:	4✔
45	image_param = args[0] if len(args) > 0 else kwargs["image"]	4✔
46	if type(image_param) is str:	4✔
47	input_image = Image.open(args[0] if len(args) > 0 else kwargs["image"])	4✔
48	else:
49	input_image = Image.open(io.BytesIO(image_param))	4✔
50	input_width, input_height = input_image.size	4✔
51	input_images = [	4✔
52	("{width}x{height}".format(width=input_width, height=input_height))
53	]
NEW 54	except Exception:	×
NEW 55	pass	×
56	scope3_row = ImpactRow(	4✔
57	model=Model(id=model),
58	input_tokens=input_tokens,
59	task=Task.image_classification,
60	output_images=[], # No images to output in classification
61	request_duration_ms=float(compute_time) * 1000,
62	managed_service_id=PROVIDER,
63	input_images=input_images,
64	)
65
66	scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)	4✔
67	result = ImageClassificationOutput()	4✔
68	result.elements = response	4✔
69	result.scope3ai = scope3_ctx	4✔
70	return result	4✔
71
72
73	def huggingface_image_classification_wrapper(	4✔
74	wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
75	) -> ImageClassificationOutput:
76	timer_start = time.perf_counter()	4✔
77	http_response: Response \| None = None	4✔
78	with requests_response_capture() as responses:	4✔
79	response = wrapped(args, *kwargs)	4✔
80	http_responses = responses.get()	4✔
81	if len(http_responses) > 0:	4✔
82	http_response = http_responses[-1]	4✔
83	model = kwargs.get("model") or instance.get_recommended_model(	4✔
84	HUGGING_FACE_IMAGE_CLASSIFICATION_TASK
85	)
86	return _hugging_face_image_classification_wrapper(	4✔
87	timer_start, model, response, http_response, args, kwargs
88	)
89
90
91	async def huggingface_image_classification_wrapper_async(	4✔
92	wrapped: Callable, instance: AsyncInferenceClient, args: Any, kwargs: Any
93	) -> ImageClassificationOutput:
94	timer_start = time.perf_counter()	4✔
95	http_response: ClientResponse \| None = None	4✔
96	with aiohttp_response_capture() as responses:	4✔
97	response = await wrapped(args, *kwargs)	4✔
98	http_responses = responses.get()	4✔
99	if len(http_responses) > 0:	4✔
100	http_response = http_responses[-1]	4✔
101	model = kwargs.get("model") or instance.get_recommended_model(	4✔
102	HUGGING_FACE_IMAGE_CLASSIFICATION_TASK
103	)
104	return _hugging_face_image_classification_wrapper(	4✔
105	timer_start, model, response, http_response, args, kwargs
106	)

scope3data / scope3ai-py / 12753874046

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