13416015178

import time
from dataclasses import asdict, dataclass
from typing import Any, Callable, Optional, Tuple, 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.lib import Scope3AI
from scope3ai.response_interceptor.aiohttp_interceptor import aiohttp_response_capture
from scope3ai.response_interceptor.requests_interceptor import requests_response_capture

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,
) -> Tuple[AutomaticSpeechRecognitionOutput, ImpactRow]:
    compute_audio_length = None
    compute_time = 0
    if http_response:
        compute_audio_length = http_response.headers.get("x-compute-audio-length")
        compute_time = http_response.headers.get("x-compute-time")
    if not compute_time:
        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=float(compute_audio_length),
        request_duration_ms=float(compute_time) * 1000,
    )
    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, Tuple, 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.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	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_get_impact_row(	4✔
30	timer_start: Any,
31	model: Any,
32	response: Any,
33	http_response: Optional[Union[ClientResponse, Response]],
34	args: Any,
35	kwargs: Any,
36	) -> Tuple[AutomaticSpeechRecognitionOutput, ImpactRow]:
37	compute_audio_length = None	4✔
38	compute_time = 0	4✔
39	if http_response:	4✔
40	compute_audio_length = http_response.headers.get("x-compute-audio-length")	4✔
41	compute_time = http_response.headers.get("x-compute-time")	4✔
42	if not compute_time:	4✔
43	compute_time = time.perf_counter() - timer_start	×
44	if not compute_audio_length:	4✔
45	compute_audio_length = 0	4✔
46	scope3_row = ImpactRow(	4✔
47	model_id=model,
48	task=Task.text_to_speech,
49	input_audio_seconds=float(compute_audio_length),
50	request_duration_ms=float(compute_time) * 1000,
51	)
52	result = AutomaticSpeechRecognitionOutput(**asdict(response))	4✔
53	return result, scope3_row	4✔
54
55
56	def huggingface_automatic_recognition_output_wrapper(	4✔
57	wrapped: Callable, instance: InferenceClient, args: Any, kwargs: Any
58	) -> AutomaticSpeechRecognitionOutput:
59	timer_start = time.perf_counter()	4✔
60	http_response: Response \| None = None	4✔
61	with requests_response_capture() as responses:	4✔
62	response = wrapped(args, *kwargs)	4✔
63	http_responses = responses.get()	4✔
64	if http_responses:	4✔
65	http_response = http_responses[-1]	4✔
66	model = kwargs.get("model") or instance.get_recommended_model(	4✔
67	HUGGING_FACE_SPEECH_TO_TEXT_TASK
68	)
69	result, impact_row = _hugging_face_automatic_recognition_get_impact_row(	4✔
70	timer_start, model, response, http_response, args, kwargs
71	)
72	scope3_ctx = Scope3AI.get_instance().submit_impact(impact_row)	4✔
73	result.scope3ai = scope3_ctx	4✔
74	return result	4✔
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 http_responses:	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	result, impact_row = _hugging_face_automatic_recognition_get_impact_row(	4✔
91	timer_start, model, response, http_response, args, kwargs
92	)
93	scope3_ctx = await Scope3AI.get_instance().asubmit_impact(impact_row)	4✔
94	result.scope3ai = scope3_ctx	4✔
95	return result	4✔

scope3data / scope3ai-py / 13416015178

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