14097016956

Committed 27 Mar 2025 01:41AM UTC coverage: 96.23% (+15.7%) from 80.557%

Build # 14097016956

Build Type

Pull #92

github

Specific Base 5758a3

Committed by

dearlordylord

Commit Message

feat(api): client-to-provider dry

Pull Request Pull Request #92: feat: Managed Service Kebabs

Run Details

53 of 55 new or added lines in 11 files covered. (96.36%)

44 existing lines in 10 files now uncovered.

2578 of 2679 relevant lines covered (96.23%)

3.85 hits per line

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

93.75

/scope3ai/api/tracer.py

from typing import List, Optional
from uuid import uuid4

from .typesgen import ImpactResponse, ModeledRow


# TODO Tracer is not BaseTracer?
class Tracer:
    """
    Tracer is responsible for tracking and aggregating environmental impact metrics
    from AI model interactions. It supports nested tracing, async operations, and
    provides detailed impact breakdowns for energy, emissions and water usage.

    Args:
        name (str, optional): Name identifier for the tracer. Defaults to None.
        keep_traces (bool, optional): Whether to keep trace history after completion.
            Defaults to False.
        client_id (str, optional): Client identifier for categorizing traces.
            Overrides global `SCOPE3AI_CLIENT_ID` setting. Defaults to None.
        project_id (str, optional): Project identifier for categorizing traces.
            Overrides global `SCOPE3AI_PROJECT_ID` setting. Defaults to None.
        application_id (str, optional): Application identifier for categorizing traces.
            Overrides global `SCOPE3AI_APPLICATION_ID` setting. Defaults to None.
        session_id (str, optional): Session identifier for tracking user sessions.
            Only available at tracer level. Defaults to None.
        trace_id (str, optional): Unique identifier for the trace.
            Auto-generated if not provided. Defaults to None.
    """

    def __init__(
        self,
        name: str = None,
        keep_traces: bool = False,
        client_id: Optional[str] = None,
        project_id: Optional[str] = None,
        application_id: Optional[str] = None,
        session_id: Optional[str] = None,
        trace_id: Optional[str] = None,
    ) -> None:
        from scope3ai.lib import Scope3AI

        self.trace_id = trace_id or uuid4().hex
        self.scope3ai = Scope3AI.get_instance()
        self.name = name
        self.keep_traces = keep_traces
        self.children: List[Tracer] = []
        self.rows: List[ModeledRow] = []
        self.traces = []  # type: List[Scope3AIContext]

        self.client_id = client_id
        self.project_id = project_id
        self.application_id = application_id
        self.session_id = session_id

    def impact(self, timeout: Optional[int] = None) -> ImpactResponse:
        """
        Return an aggregated impact response for the current tracer and its children.

        As the impact is computed asynchronously, this method will wait for the
        impact response to be available before returning it.
        """
        for trace in self.traces:
            trace.wait_impact(timeout)
        return self._impact()

    async def aimpact(self, timeout: Optional[int] = None) -> ImpactResponse:
        """
        Async version of Tracer::impact.
        """
        for trace in self.traces:
            await trace.await_impact(timeout)
        return self._impact()

    def _impact(self) -> ImpactResponse:
        """
        Return an aggregated impact response for the current tracer and its children.
        """
        all_rows = self.get_all_rows()
        return ImpactResponse(
            rows=all_rows,
            total_energy_wh=sum([row.total_impact.usage_energy_wh for row in all_rows]),
            total_gco2e=sum(
                [row.total_impact.usage_emissions_gco2e for row in all_rows]
            ),
            total_mlh2o=sum([row.total_impact.usage_water_ml for row in all_rows]),
            has_errors=any([row.error is not None for row in all_rows]),
        )

    def add_impact(self, impact: ModeledRow) -> None:
        self.rows.append(impact)

    def get_all_rows(self) -> List[ModeledRow]:
        all_rows = self.rows[:]
        for child in self.children:
            all_rows.extend(child.get_all_rows())
        return all_rows

    def _link_parent(self, parent: Optional["Tracer"]) -> None:
        if parent and (self not in parent.children):
            parent.children.append(self)

    def _unlink_parent(self, parent: Optional["Tracer"]) -> None:
        pass

    def _link_trace(self, trace) -> None:
        if trace not in self.traces:
            self.traces.append(trace)

    def _unlink_trace(self, trace) -> None:
        if self.keep_traces:
            return
        if trace in self.traces:
            self.traces.remove(trace)

1	from typing import List, Optional	4✔
2	from uuid import uuid4	4✔
3
4	from .typesgen import ImpactResponse, ModeledRow	4✔
5
6
7	# TODO Tracer is not BaseTracer?
8	class Tracer:	4✔
9	"""
10	Tracer is responsible for tracking and aggregating environmental impact metrics
11	from AI model interactions. It supports nested tracing, async operations, and
12	provides detailed impact breakdowns for energy, emissions and water usage.
13
14	Args:
15	name (str, optional): Name identifier for the tracer. Defaults to None.
16	keep_traces (bool, optional): Whether to keep trace history after completion.
17	Defaults to False.
18	client_id (str, optional): Client identifier for categorizing traces.
19	Overrides global `SCOPE3AI_CLIENT_ID` setting. Defaults to None.
20	project_id (str, optional): Project identifier for categorizing traces.
21	Overrides global `SCOPE3AI_PROJECT_ID` setting. Defaults to None.
22	application_id (str, optional): Application identifier for categorizing traces.
23	Overrides global `SCOPE3AI_APPLICATION_ID` setting. Defaults to None.
24	session_id (str, optional): Session identifier for tracking user sessions.
25	Only available at tracer level. Defaults to None.
26	trace_id (str, optional): Unique identifier for the trace.
27	Auto-generated if not provided. Defaults to None.
28	"""
29
30	def __init__(	4✔
31	self,
32	name: str = None,
33	keep_traces: bool = False,
34	client_id: Optional[str] = None,
35	project_id: Optional[str] = None,
36	application_id: Optional[str] = None,
37	session_id: Optional[str] = None,
38	trace_id: Optional[str] = None,
39	) -> None:
40	from scope3ai.lib import Scope3AI	4✔
41
42	self.trace_id = trace_id or uuid4().hex	4✔
43	self.scope3ai = Scope3AI.get_instance()	4✔
44	self.name = name	4✔
45	self.keep_traces = keep_traces	4✔
46	self.children: List[Tracer] = []	4✔
47	self.rows: List[ModeledRow] = []	4✔
48	self.traces = [] # type: List[Scope3AIContext]	4✔
49
50	self.client_id = client_id	4✔
51	self.project_id = project_id	4✔
52	self.application_id = application_id	4✔
53	self.session_id = session_id	4✔
54
55	def impact(self, timeout: Optional[int] = None) -> ImpactResponse:	4✔
56	"""
57	Return an aggregated impact response for the current tracer and its children.
58
59	As the impact is computed asynchronously, this method will wait for the
60	impact response to be available before returning it.
61	"""
62	for trace in self.traces:	4✔
63	trace.wait_impact(timeout)	4✔
64	return self._impact()	4✔
65
66	async def aimpact(self, timeout: Optional[int] = None) -> ImpactResponse:	4✔
67	"""
68	Async version of Tracer::impact.
69	"""
UNCOV 70	for trace in self.traces:	×
UNCOV 71	await trace.await_impact(timeout)	×
UNCOV 72	return self._impact()	×
73
74	def _impact(self) -> ImpactResponse:	4✔
75	"""
76	Return an aggregated impact response for the current tracer and its children.
77	"""
78	all_rows = self.get_all_rows()	4✔
79	return ImpactResponse(	4✔
80	rows=all_rows,
81	total_energy_wh=sum([row.total_impact.usage_energy_wh for row in all_rows]),
82	total_gco2e=sum(
83	[row.total_impact.usage_emissions_gco2e for row in all_rows]
84	),
85	total_mlh2o=sum([row.total_impact.usage_water_ml for row in all_rows]),
86	has_errors=any([row.error is not None for row in all_rows]),
87	)
88
89	def add_impact(self, impact: ModeledRow) -> None:	4✔
90	self.rows.append(impact)	4✔
91
92	def get_all_rows(self) -> List[ModeledRow]:	4✔
93	all_rows = self.rows[:]	4✔
94	for child in self.children:	4✔
95	all_rows.extend(child.get_all_rows())	4✔
96	return all_rows	4✔
97
98	def _link_parent(self, parent: Optional["Tracer"]) -> None:	4✔
99	if parent and (self not in parent.children):	4✔
100	parent.children.append(self)	4✔
101
102	def _unlink_parent(self, parent: Optional["Tracer"]) -> None:	4✔
103	pass	4✔
104
105	def _link_trace(self, trace) -> None:	4✔
106	if trace not in self.traces:	4✔
107	self.traces.append(trace)	4✔
108
109	def _unlink_trace(self, trace) -> None:	4✔
110	if self.keep_traces:	4✔
111	return	4✔
112	if trace in self.traces:	4✔
113	self.traces.remove(trace)	4✔

scope3data / scope3ai-py / 14097016956

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