12777439528

Committed 14 Jan 2025 10:21PM UTC coverage: 95.153% (+14.6%) from 80.557%

Build # 12777439528

Build Type

Pull #62

github

Specific Base 3071c5

Committed by

web-flow

Commit Message

Merge 20548d47d into 42c868843

Pull Request Pull Request #62: feat: Hugging face - speech to text async fix

Run Details

14 of 14 new or added lines in 1 file covered. (100.0%)

33 existing lines in 9 files now uncovered.

2022 of 2125 relevant lines covered (95.15%)

3.8 hits per line

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

98.04

/scope3ai/tracers/huggingface/speech_to_text.py

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

from aiohttp import ClientResponse
from huggingface_hub import (
    AutomaticSpeechRecognitionOutput as _AutomaticSpeechRecognitionOutput,
    AsyncInferenceClient,
)
from huggingface_hub import InferenceClient  # 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_SPEECH_TO_TEXT_TASK = "automatic-speech-recognition"


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


def _hugging_face_automatic_recognition_wrapper(
    timer_start: Any,
    model: Any,
    response: Any,
    http_response: Union[ClientResponse, Response],
    args: Any,
    kwargs: Any,
) -> AutomaticSpeechRecognitionOutput:
    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=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,
    )

    scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)
    result = AutomaticSpeechRecognitionOutput(**asdict(response))
    result.scope3ai = scope3_ctx
    return result


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 len(http_responses) > 0:
            http_response = http_responses[-1]
    model = kwargs.get("model") or instance.get_recommended_model(
        HUGGING_FACE_SPEECH_TO_TEXT_TASK
    )
    return _hugging_face_automatic_recognition_wrapper(
        timer_start, model, response, http_response, args, kwargs
    )


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 len(http_responses) > 0:
            http_response = http_responses[-1]
    model = kwargs.get("model") or instance.get_recommended_model(
        HUGGING_FACE_SPEECH_TO_TEXT_TASK
    )
    return _hugging_face_automatic_recognition_wrapper(
        timer_start, model, response, http_response, args, kwargs
    )

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

scope3data / scope3ai-py / 12777439528

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