12840122239

Committed 18 Jan 2025 02:37AM UTC coverage: 95.824% (+15.3%) from 80.557%

Build # 12840122239

Build Type

Pull #68

github

Specific Base 308311

Committed by

tito

Commit Message

fix(huggingface): fixing aiohttp not working with VCR if passing filename

Pull Request Pull Request #68: fix(huggingface): fixing aiohttp not working with VCR if passing filename

Run Details

3 of 3 new or added lines in 2 files covered. (100.0%)

45 existing lines in 9 files now uncovered.

2111 of 2203 relevant lines covered (95.82%)

3.83 hits per line

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

98.18

/scope3ai/tracers/huggingface/speech_to_text.py

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

from aiohttp import ClientResponse
from huggingface_hub import (
    AsyncInferenceClient,
    InferenceClient,  # type: ignore[import-untyped]
)
from huggingface_hub import (
    AutomaticSpeechRecognitionOutput as _AutomaticSpeechRecognitionOutput,
)
from requests import Response

from scope3ai.api.types import ImpactRow, Scope3AIContext
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_SPEECH_TO_TEXT_TASK = "automatic-speech-recognition"


@dataclass
class AutomaticSpeechRecognitionOutput(_AutomaticSpeechRecognitionOutput):
    scope3ai: Optional[Scope3AIContext] = None


def _hugging_face_automatic_recognition_get_impact_row(
    timer_start: Any,
    model: Any,
    response: Any,
    http_response: Optional[Union[ClientResponse, Response]],
    args: Any,
    kwargs: Any,
) -> (AutomaticSpeechRecognitionOutput, ImpactRow):
    compute_audio_length = None
    if http_response:
        compute_audio_length = http_response.headers.get("x-compute-audio-length")
        compute_time = http_response.headers.get("x-compute-time")
    else:
        compute_time = time.perf_counter() - timer_start
    if not compute_audio_length:
        compute_audio_length = 0
    scope3_row = ImpactRow(
        model_id=model,
        task=Task.text_to_speech,
        input_audio_seconds=int(float(compute_audio_length)),
        request_duration_ms=float(compute_time) * 1000,
        managed_service_id=PROVIDER,
    )
    result = AutomaticSpeechRecognitionOutput(**asdict(response))
    return result, scope3_row


def huggingface_automatic_recognition_output_wrapper(
    wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
) -> AutomaticSpeechRecognitionOutput:
    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_SPEECH_TO_TEXT_TASK
    )
    result, impact_row = _hugging_face_automatic_recognition_get_impact_row(
        timer_start, model, response, http_response, args, kwargs
    )
    scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)
    result.scope3ai = scope3_ctx
    return result


async def huggingface_automatic_recognition_output_wrapper_async(
    wrapped: Callable, instance: AsyncInferenceClient, args: Any, kwargs: Any
) -> AutomaticSpeechRecognitionOutput:
    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_SPEECH_TO_TEXT_TASK
    )
    result, impact_row = _hugging_face_automatic_recognition_get_impact_row(
        timer_start, model, response, http_response, args, kwargs
    )
    scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)
    result.scope3ai = scope3_ctx
    return result

1	import time	4✔
2	from dataclasses import asdict, dataclass	4✔
3	from typing import Any, Callable, Optional, Union	4✔
4
5	from aiohttp import ClientResponse	4✔
6	from huggingface_hub import (	4✔
7	AsyncInferenceClient,
8	InferenceClient, # type: ignore[import-untyped]
9	)
10	from huggingface_hub import (	4✔
11	AutomaticSpeechRecognitionOutput as _AutomaticSpeechRecognitionOutput,
12	)
13	from requests import Response	4✔
14
15	from scope3ai.api.types import ImpactRow, Scope3AIContext	4✔
16	from scope3ai.api.typesgen import Task	4✔
17	from scope3ai.constants import PROVIDERS	4✔
18	from scope3ai.lib import Scope3AI	4✔
19	from scope3ai.response_interceptor.aiohttp_interceptor import aiohttp_response_capture	4✔
20	from scope3ai.response_interceptor.requests_interceptor import requests_response_capture	4✔
21
22	PROVIDER = PROVIDERS.HUGGINGFACE_HUB.value	4✔
23	HUGGING_FACE_SPEECH_TO_TEXT_TASK = "automatic-speech-recognition"	4✔
24
25
26	@dataclass	4✔
27	class AutomaticSpeechRecognitionOutput(_AutomaticSpeechRecognitionOutput):	4✔
28	scope3ai: Optional[Scope3AIContext] = None	4✔
29
30
31	def _hugging_face_automatic_recognition_get_impact_row(	4✔
32	timer_start: Any,
33	model: Any,
34	response: Any,
35	http_response: Optional[Union[ClientResponse, Response]],
36	args: Any,
37	kwargs: Any,
38	) -> (AutomaticSpeechRecognitionOutput, ImpactRow):
39	compute_audio_length = None	4✔
40	if http_response:	4✔
41	compute_audio_length = http_response.headers.get("x-compute-audio-length")	4✔
42	compute_time = http_response.headers.get("x-compute-time")	4✔
43	else:
UNCOV 44	compute_time = time.perf_counter() - timer_start	×
45	if not compute_audio_length:	4✔
46	compute_audio_length = 0	4✔
47	scope3_row = ImpactRow(	4✔
48	model_id=model,
49	task=Task.text_to_speech,
50	input_audio_seconds=int(float(compute_audio_length)),
51	request_duration_ms=float(compute_time) * 1000,
52	managed_service_id=PROVIDER,
53	)
54	result = AutomaticSpeechRecognitionOutput(**asdict(response))	4✔
55	return result, scope3_row	4✔
56
57
58	def huggingface_automatic_recognition_output_wrapper(	4✔
59	wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
60	) -> AutomaticSpeechRecognitionOutput:
61	timer_start = time.perf_counter()	4✔
62	http_response: Response \| None = None	4✔
63	with requests_response_capture() as responses:	4✔
64	response = wrapped(args, *kwargs)	4✔
65	http_responses = responses.get()	4✔
66	if http_responses:	4✔
67	http_response = http_responses[-1]	4✔
68	model = kwargs.get("model") or instance.get_recommended_model(	4✔
69	HUGGING_FACE_SPEECH_TO_TEXT_TASK
70	)
71	result, impact_row = _hugging_face_automatic_recognition_get_impact_row(	4✔
72	timer_start, model, response, http_response, args, kwargs
73	)
74	scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)	4✔
75	result.scope3ai = scope3_ctx	4✔
76	return result	4✔
77
78
79	async def huggingface_automatic_recognition_output_wrapper_async(	4✔
80	wrapped: Callable, instance: AsyncInferenceClient, args: Any, kwargs: Any
81	) -> AutomaticSpeechRecognitionOutput:
82	timer_start = time.perf_counter()	4✔
83	http_response: ClientResponse \| None = None	4✔
84	with aiohttp_response_capture() as responses:	4✔
85	response = await wrapped(args, *kwargs)	4✔
86	http_responses = responses.get()	4✔
87	if http_responses:	4✔
88	http_response = http_responses[-1]	4✔
89	model = kwargs.get("model") or instance.get_recommended_model(	4✔
90	HUGGING_FACE_SPEECH_TO_TEXT_TASK
91	)
92	result, impact_row = _hugging_face_automatic_recognition_get_impact_row(	4✔
93	timer_start, model, response, http_response, args, kwargs
94	)
95	scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)	4✔
96	result.scope3ai = scope3_ctx	4✔
97	return result	4✔

scope3data / scope3ai-py / 12840122239

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