12680309189

Committed 08 Jan 2025 10:40PM UTC coverage: 94.312% (+13.8%) from 80.557%

Build # 12680309189

Build Type

Pull #57

github

Specific Base 5d8466

Committed by

tito

Commit Message

feat(openai): add support for speech translation

Pull Request Pull Request #57: feat(openai): add support for speech translation

Run Details

67 of 73 new or added lines in 4 files covered. (91.78%)

36 existing lines in 7 files now uncovered.

1741 of 1846 relevant lines covered (94.31%)

3.77 hits per line

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

94.34

/scope3ai/tracers/openai/translation.py

import logging
import time
from typing import Any, Callable, Optional, Union

import tiktoken
from openai.resources.audio.translations import AsyncTranslations, Translations
from openai.types.audio.translation import Translation as _Translation
from openai.types.audio.translation_verbose import (
    TranslationVerbose as _TranslationVerbose,
)

from scope3ai.api.types import ImpactRow, Model, Scope3AIContext, Task
from scope3ai.constants import PROVIDERS
from scope3ai.lib import Scope3AI

from .utils import _get_file_audio_duration

PROVIDER = PROVIDERS.OPENAI.value

logger = logging.getLogger(__name__)


class AnnotatedStr(str):
    scope3ai: Optional[Scope3AIContext] = None


class Translation(_Translation):
    scope3ai: Optional[Scope3AIContext] = None


class TranslationVerbose(_TranslationVerbose):
    scope3ai: Optional[Scope3AIContext] = None


def _openai_translation_wrapper(
    response: Any, request_latency: float, kwargs: dict
) -> Union[Translation, TranslationVerbose, AnnotatedStr]:
    model = kwargs["model"]
    encoder = tiktoken.get_encoding("cl100k_base")

    if isinstance(response, (_Translation, _TranslationVerbose)):
        output_tokens = len(encoder.encode(response.text))
    elif isinstance(response, str):
        output_tokens = len(encoder.encode(response))
    else:
        output_tokens = None

    options = {}
    duration = _get_file_audio_duration(kwargs["file"])
    if duration is not None:
        options["input_audio_seconds"] = int(duration)

    scope3_row = ImpactRow(
        model=Model(id=model),
        provider=PROVIDER,
        output_tokens=output_tokens,
        request_duration_ms=request_latency,
        task=Task.translation,
        **options,
    )
    scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)

    if isinstance(response, _Translation):
        result = Translation.model_construct(**response.model_dump())
    elif isinstance(response, _TranslationVerbose):
        result = TranslationVerbose.model_construct(**response.model_dump())
    elif isinstance(response, str):
        result = AnnotatedStr(str)
    else:
        logger.error(f"Unexpected response type: {type(response)}")
        return response
    result.scope3ai = scope3_ctx
    return result


def openai_translation_wrapper(
    wrapped: Callable, instance: Translations, args: Any, kwargs: Any
) -> Union[Translation, TranslationVerbose, AnnotatedStr]:
    timer_start = time.perf_counter()
    response = wrapped(*args, **kwargs)
    request_latency = (time.perf_counter() - timer_start) * 1000
    return _openai_translation_wrapper(response, request_latency, kwargs)


async def openai_async_translation_wrapper(
    wrapped: Callable, instance: AsyncTranslations, args: Any, kwargs: Any
) -> Union[Translation, TranslationVerbose, AnnotatedStr]:
    timer_start = time.perf_counter()
    response = await wrapped(*args, **kwargs)
    request_latency = (time.perf_counter() - timer_start) * 1000
    return _openai_translation_wrapper(response, request_latency, kwargs)

1	import logging	4✔
2	import time	4✔
3	from typing import Any, Callable, Optional, Union	4✔
4
5	import tiktoken	4✔
6	from openai.resources.audio.translations import AsyncTranslations, Translations	4✔
7	from openai.types.audio.translation import Translation as _Translation	4✔
8	from openai.types.audio.translation_verbose import (	4✔
9	TranslationVerbose as _TranslationVerbose,
10	)
11
12	from scope3ai.api.types import ImpactRow, Model, Scope3AIContext, Task	4✔
13	from scope3ai.constants import PROVIDERS	4✔
14	from scope3ai.lib import Scope3AI	4✔
15
16	from .utils import _get_file_audio_duration	4✔
17
18	PROVIDER = PROVIDERS.OPENAI.value	4✔
19
20	logger = logging.getLogger(__name__)	4✔
21
22
23	class AnnotatedStr(str):	4✔
24	scope3ai: Optional[Scope3AIContext] = None	4✔
25
26
27	class Translation(_Translation):	4✔
28	scope3ai: Optional[Scope3AIContext] = None	4✔
29
30
31	class TranslationVerbose(_TranslationVerbose):	4✔
32	scope3ai: Optional[Scope3AIContext] = None	4✔
33
34
35	def _openai_translation_wrapper(	4✔
36	response: Any, request_latency: float, kwargs: dict
37	) -> Union[Translation, TranslationVerbose, AnnotatedStr]:
38	model = kwargs["model"]	4✔
39	encoder = tiktoken.get_encoding("cl100k_base")	4✔
40
41	if isinstance(response, (_Translation, _TranslationVerbose)):	4✔
42	output_tokens = len(encoder.encode(response.text))	4✔
43	elif isinstance(response, str):	4✔
44	output_tokens = len(encoder.encode(response))	4✔
45	else:
NEW 46	output_tokens = None	×
47
48	options = {}	4✔
49	duration = _get_file_audio_duration(kwargs["file"])	4✔
50	if duration is not None:	4✔
51	options["input_audio_seconds"] = int(duration)	4✔
52
53	scope3_row = ImpactRow(	4✔
54	model=Model(id=model),
55	provider=PROVIDER,
56	output_tokens=output_tokens,
57	request_duration_ms=request_latency,
58	task=Task.translation,
59	**options,
60	)
61	scope3_ctx = Scope3AI.get_instance().submit_impact(scope3_row)	4✔
62
63	if isinstance(response, _Translation):	4✔
64	result = Translation.model_construct(**response.model_dump())	4✔
65	elif isinstance(response, _TranslationVerbose):	4✔
66	result = TranslationVerbose.model_construct(**response.model_dump())	4✔
67	elif isinstance(response, str):	4✔
68	result = AnnotatedStr(str)	4✔
69	else:
NEW 70	logger.error(f"Unexpected response type: {type(response)}")	×
NEW 71	return response	×
72	result.scope3ai = scope3_ctx	4✔
73	return result	4✔
74
75
76	def openai_translation_wrapper(	4✔
77	wrapped: Callable, instance: Translations, args: Any, kwargs: Any
78	) -> Union[Translation, TranslationVerbose, AnnotatedStr]:
79	timer_start = time.perf_counter()	4✔
80	response = wrapped(args, *kwargs)	4✔
81	request_latency = (time.perf_counter() - timer_start) * 1000	4✔
82	return _openai_translation_wrapper(response, request_latency, kwargs)	4✔
83
84
85	async def openai_async_translation_wrapper(	4✔
86	wrapped: Callable, instance: AsyncTranslations, args: Any, kwargs: Any
87	) -> Union[Translation, TranslationVerbose, AnnotatedStr]:
88	timer_start = time.perf_counter()	4✔
89	response = await wrapped(args, *kwargs)	4✔
90	request_latency = (time.perf_counter() - timer_start) * 1000	4✔
91	return _openai_translation_wrapper(response, request_latency, kwargs)	4✔

scope3data / scope3ai-py / 12680309189

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