30354831214

Committed 28 Jul 2026 11:26AM UTC coverage: 64.319%. First build

Build # 30354831214

Build Type

Pull #897

github

Committed by

web-flow

Commit Message

Merge b61c802ae into 81e93d5b0

Pull Request Pull Request #897: adds utils to conveniently read and analyze rtl traces

Coverage Stats

226 of 255 new or added lines in 2 files covered. (88.63%)

43415 of 67499 relevant lines covered (64.32%)

726.13 hits per line

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

88.54

/python/docc/benchmarks/rtl.py

"""Read, validate and query daisy RTL instrumentation traces.

The RTL library (``rtl/src/instrumentation.cpp``) emits a Chrome-trace subset
extended with docc-specific metadata. This module loads such a trace, validates
it against the bundled JSON schema (``daisy_trace.schema.json``) and exposes a
typed, queryable :class:`Trace` object.

Two trace shapes exist, selected at capture time by ``__DAISY_INSTRUMENTATION_MODE``:

* per-invocation (``cat == "region,daisy"``) — one event per region entry/exit,
  with ``metrics`` a flat ``{counter: int}`` map.
* aggregated (``cat == "aggregated_region,daisy"``) — one event per region with
  Welford statistics per counter plus a ``runtime`` stat block.

Example::

    from docc.benchmarks import Trace

    trace = Trace.load("daisy_trace.json")
    for region in trace.regions:
        print(region.name, region.runtime_mean_us)

    hot = trace.filter(target_type="CUDA")
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from functools import cached_property
from pathlib import Path
from typing import Any, Dict, Iterator, List, Optional, Union

__all__ = [
    "Trace",
    "TraceRegion",
    "LoopInfo",
    "SourcePos",
    "SourceRange",
    "MetricStat",
    "TraceValidationError",
    "validate",
    "PER_INVOCATION_CAT",
    "AGGREGATED_CAT",
    "STATIC_PREFIX",
]

PER_INVOCATION_CAT = "region,daisy"
AGGREGATED_CAT = "aggregated_region,daisy"
STATIC_PREFIX = "static:::"
_RUNTIME_KEY = "runtime"

_SCHEMA_FILENAME = "daisy_trace.schema.json"


def _find_schema() -> Path:
    """Locate the daisy-trace JSON schema.

    In an installed/built wheel the schema is shipped as package data next to this
    module (CMake installs ``rtl/schema/daisy_trace.schema.json`` into
    ``docc/benchmarks``). When running from a source checkout that install has not
    happened, so fall back to the canonical copy under ``rtl/schema``.
    """
    bundled = Path(__file__).with_name(_SCHEMA_FILENAME)
    if bundled.is_file():
        return bundled

    # rtl.py -> docc/python/docc/benchmarks/rtl.py ; schema -> docc/rtl/schema/...
    repo_schema = (
        Path(__file__).resolve().parents[3] / "rtl" / "schema" / _SCHEMA_FILENAME
    )
    if repo_schema.is_file():
        return repo_schema

    raise FileNotFoundError(
        f"Could not locate {_SCHEMA_FILENAME}; expected it next to {bundled} "
        f"(installed wheel) or at {repo_schema} (source checkout)."
    )


def load_schema() -> dict:
    """Return the bundled daisy-trace JSON schema as a dict."""
    with open(_find_schema(), "r") as f:
        return json.load(f)


class TraceValidationError(ValueError):
    """Raised when a trace does not conform to the daisy-trace schema."""

    def __init__(self, errors: List[str]):
        self.errors = errors
        super().__init__("Trace failed schema validation:\n" + "\n".join(errors))


def validate(instance: Any, schema: Optional[dict] = None) -> List[str]:
    """Validate a parsed trace object against the schema.

    Returns a list of human-readable error strings (empty when valid). Requires
    the optional ``jsonschema`` dependency.
    """
    try:
        from jsonschema import Draft202012Validator
    except ImportError as exc:  # pragma: no cover - exercised only without dep
        raise ImportError(
            "Validating daisy traces requires the 'jsonschema' package. "
            "Install it with `pip install jsonschema`."
        ) from exc

    if schema is None:
        schema = load_schema()

    validator = Draft202012Validator(schema)
    errors: List[str] = []
    for error in sorted(
        validator.iter_errors(instance), key=lambda e: list(e.absolute_path)
    ):
        location = "/".join(str(p) for p in error.absolute_path) or "<root>"
        errors.append(f"{location}: {error.message}")
    return errors


@dataclass(frozen=True)
class SourcePos:
    line: int
    col: int

    @classmethod
    def from_dict(cls, d: dict) -> "SourcePos":
        return cls(line=d["line"], col=d["col"])


@dataclass(frozen=True)
class SourceRange:
    file: str
    begin: SourcePos
    end: SourcePos

    @classmethod
    def from_dict(cls, d: dict) -> "SourceRange":
        return cls(
            file=d["file"],
            begin=SourcePos.from_dict(d["from"]),
            end=SourcePos.from_dict(d["to"]),
        )


@dataclass(frozen=True)
class LoopInfo:
    loopnest_index: int
    num_loops: int
    num_maps: int
    num_fors: int
    num_whiles: int
    max_depth: int
    is_perfectly_nested: bool
    is_perfectly_parallel: bool
    is_elementwise: bool
    has_side_effects: bool

    @classmethod
    def from_dict(cls, d: dict) -> "LoopInfo":
        return cls(
            loopnest_index=d["loopnest_index"],
            num_loops=d["num_loops"],
            num_maps=d["num_maps"],
            num_fors=d["num_fors"],
            num_whiles=d["num_whiles"],
            max_depth=d["max_depth"],
            is_perfectly_nested=d["is_perfectly_nested"],
            is_perfectly_parallel=d["is_perfectly_parallel"],
            is_elementwise=d["is_elementwise"],
            has_side_effects=d["has_side_effects"],
        )


@dataclass(frozen=True)
class MetricStat:
    """Aggregated statistics for a single counter (aggregated traces).

    ``mean``/``variance``/``min``/``max`` may be ``None`` for provided
    (``static:::``) metrics whose value was non-finite at capture time.
    """

    mean: Optional[float]
    variance: Optional[float]
    count: int
    min: Optional[float]
    max: Optional[float]

    @classmethod
    def from_dict(cls, d: dict) -> "MetricStat":
        return cls(
            mean=d["mean"],
            variance=d["variance"],
            count=d["count"],
            min=d["min"],
            max=d["max"],
        )


class TraceRegion:
    """A single region event, wrapping the raw trace dict with typed accessors."""

    def __init__(self, raw: dict):
        self._raw = raw

    # -- Raw / identity ----------------------------------------------------
    @property
    def raw(self) -> dict:
        return self._raw

    @property
    def category(self) -> str:
        return self._raw["cat"]

    @property
    def is_aggregated(self) -> bool:
        return self.category == AGGREGATED_CAT

    @property
    def name(self) -> str:
        return self._raw["name"]

    @property
    def pid(self) -> int:
        return self._raw["pid"]

    @property
    def tid(self) -> int:
        return self._raw["tid"]

    @property
    def ts_us(self) -> Optional[float]:
        return self._raw.get("ts")

    @property
    def dur_us(self) -> Optional[float]:
        return self._raw.get("dur")

    # -- Source metadata ---------------------------------------------------
    @property
    def _args(self) -> dict:
        return self._raw["args"]

    @property
    def function(self) -> str:
        return self._args["function"]

    @property
    def module(self) -> str:
        return self._args["module"]

    @property
    def target_type(self) -> str:
        return self._args.get("target_type", "")

    @cached_property
    def source_ranges(self) -> List[SourceRange]:
        return [SourceRange.from_dict(r) for r in self._args.get("source_ranges", [])]

    # -- docc metadata -----------------------------------------------------
    @property
    def _docc(self) -> dict:
        return self._args["docc"]

    @property
    def sdfg_name(self) -> str:
        return self._docc.get("sdfg_name", "")

    @property
    def sdfg_file(self) -> str:
        return self._docc.get("sdfg_file", "")

    @property
    def arg_capture_path(self) -> str:
        return self._docc.get("arg_capture_path", "")

    @property
    def features_file(self) -> str:
        return self._docc.get("features_file", "")

    @property
    def opt_report_file(self) -> str:
        return self._docc.get("opt_report_file", "")

    @property
    def has_element(self) -> bool:
        """True when SDFG element metadata (id/type/loop_info) is present."""
        return "element_id" in self._docc

    @property
    def element_id(self) -> Optional[int]:
        return self._docc.get("element_id")

    @property
    def element_type(self) -> Optional[str]:
        return self._docc.get("element_type")

    @property
    def loopnest_index(self) -> Optional[int]:
        return self._docc.get("loopnest_index")

    @cached_property
    def loop_info(self) -> Optional[LoopInfo]:
        li = self._docc.get("loop_info")
        return LoopInfo.from_dict(li) if li is not None else None

    # -- Metrics -----------------------------------------------------------
    @property
    def _metrics(self) -> dict:
        return self._args.get("metrics", {})

    @cached_property
    def counters(self) -> Dict[str, int]:
        """PAPI counter values for a per-invocation region (empty if aggregated)."""
        if self.is_aggregated:
            return {}
        return dict(self._metrics)

    @cached_property
    def counter_stats(self) -> Dict[str, MetricStat]:
        """Aggregated PAPI counter statistics (empty for per-invocation regions).

        Excludes the reserved ``runtime`` block and ``static:::`` provided metrics.
        """
        if not self.is_aggregated:
            return {}
        out: Dict[str, MetricStat] = {}
        for key, val in self._metrics.items():
            if key == _RUNTIME_KEY or key.startswith(STATIC_PREFIX):
                continue
            out[key] = MetricStat.from_dict(val)
        return out

    @cached_property
    def static_metrics(self) -> Dict[str, MetricStat]:
        """User-provided (``static:::``) metric statistics, keyed without the prefix."""
        if not self.is_aggregated:
            return {}
        out: Dict[str, MetricStat] = {}
        for key, val in self._metrics.items():
            if key.startswith(STATIC_PREFIX):
                out[key[len(STATIC_PREFIX) :]] = MetricStat.from_dict(val)
        return out

    @cached_property
    def runtime(self) -> Optional[MetricStat]:
        """Aggregated runtime statistics in microseconds (aggregated traces only)."""
        rt = self._metrics.get(_RUNTIME_KEY) if self.is_aggregated else None
        return MetricStat.from_dict(rt) if rt is not None else None

    @property
    def runtime_mean_us(self) -> Optional[float]:
        """Mean runtime in microseconds.

        For aggregated traces this is ``runtime.mean``; for per-invocation traces
        it is this event's own ``dur``.
        """
        if self.is_aggregated:
            return self.runtime.mean if self.runtime is not None else None
        return self.dur_us

    def metric(self, name: str) -> Optional[Union[int, float]]:
        """Return a representative value for ``name``.

        For per-invocation regions this is the raw counter value; for aggregated
        regions it is the counter/static metric mean. Returns ``None`` if absent.
        """
        if not self.is_aggregated:
            return self.counters.get(name)
        if name in self.counter_stats:
            return self.counter_stats[name].mean
        if name in self.static_metrics:
            return self.static_metrics[name].mean
        return None

    def __repr__(self) -> str:
        return (
            f"TraceRegion(name={self.name!r}, module={self.module!r}, "
            f"target={self.target_type!r}, aggregated={self.is_aggregated})"
        )


class Trace:
    """A parsed, validated daisy trace: a collection of :class:`TraceRegion`."""

    def __init__(self, raw: dict):
        self._raw = raw
        self._regions = [TraceRegion(e) for e in raw.get("traceEvents", [])]

    # -- Construction ------------------------------------------------------
    @classmethod
    def from_dict(cls, data: dict, validate_schema: bool = True) -> "Trace":
        """Build a :class:`Trace` from an already-parsed dict.

        Raises :class:`TraceValidationError` if ``validate_schema`` is set and the
        data does not conform to the schema.
        """
        if validate_schema:
            errors = validate(data)
            if errors:
                raise TraceValidationError(errors)
        return cls(data)

    @classmethod
    def load(cls, path: Union[str, Path], validate_schema: bool = True) -> "Trace":
        """Read a trace JSON file, optionally validate it, and return a :class:`Trace`."""
        with open(path, "r") as f:
            data = json.load(f)
        return cls.from_dict(data, validate_schema=validate_schema)

    # -- Container protocol ------------------------------------------------
    @property
    def regions(self) -> List[TraceRegion]:
        return self._regions

    @property
    def raw(self) -> dict:
        return self._raw

    def __len__(self) -> int:
        return len(self._regions)

    def __iter__(self) -> Iterator[TraceRegion]:
        return iter(self._regions)

    def __getitem__(self, index: int) -> TraceRegion:
        return self._regions[index]

    @property
    def is_aggregated(self) -> bool:
        """True if the trace was captured in aggregate mode.

        Determined from the first region; a valid trace never mixes modes.
        """
        return bool(self._regions) and self._regions[0].is_aggregated

    # -- Queries -----------------------------------------------------------
    def filter(
        self,
        *,
        module: Optional[str] = None,
        function: Optional[str] = None,
        target_type: Optional[str] = None,
        element_type: Optional[str] = None,
        sdfg_name: Optional[str] = None,
    ) -> List[TraceRegion]:
        """Return regions matching all provided (non-``None``) criteria."""
        out = []
        for r in self._regions:
            if module is not None and r.module != module:
                continue
            if function is not None and r.function != function:
                continue
            if target_type is not None and r.target_type != target_type:
                continue
            if element_type is not None and r.element_type != element_type:
                continue
            if sdfg_name is not None and r.sdfg_name != sdfg_name:
                continue
            out.append(r)
        return out

    def by_element_id(self, element_id: int) -> List[TraceRegion]:
        """Return regions carrying the given SDFG ``element_id``."""
        return [r for r in self._regions if r.element_id == element_id]

    def hottest(self, n: Optional[int] = None) -> List[TraceRegion]:
        """Regions sorted by mean runtime (descending); ``n`` limits the count."""
        ranked = sorted(
            (r for r in self._regions if r.runtime_mean_us is not None),
            key=lambda r: r.runtime_mean_us,
            reverse=True,
        )
        return ranked[:n] if n is not None else ranked

    def total_runtime_us(self) -> float:
        """Sum of mean runtimes across all regions (microseconds)."""
        return sum((r.runtime_mean_us or 0.0) for r in self._regions)

1	"""Read, validate and query daisy RTL instrumentation traces.
2
3	The RTL library (``rtl/src/instrumentation.cpp``) emits a Chrome-trace subset
4	extended with docc-specific metadata. This module loads such a trace, validates
5	it against the bundled JSON schema (``daisy_trace.schema.json``) and exposes a
6	typed, queryable :class:`Trace` object.
7
8	Two trace shapes exist, selected at capture time by ``__DAISY_INSTRUMENTATION_MODE``:
9
10	* per-invocation (``cat == "region,daisy"``) — one event per region entry/exit,
11	with ``metrics`` a flat ``{counter: int}`` map.
12	* aggregated (``cat == "aggregated_region,daisy"``) — one event per region with
13	Welford statistics per counter plus a ``runtime`` stat block.
14
15	Example::
16
17	from docc.benchmarks import Trace
18
19	trace = Trace.load("daisy_trace.json")
20	for region in trace.regions:
21	print(region.name, region.runtime_mean_us)
22
23	hot = trace.filter(target_type="CUDA")
24	"""
25
26	from __future__ import annotations	4✔
27
28	import json	4✔
29	from dataclasses import dataclass	4✔
30	from functools import cached_property	4✔
31	from pathlib import Path	4✔
32	from typing import Any, Dict, Iterator, List, Optional, Union	4✔
33
34	__all__ = [	4✔
35	"Trace",
36	"TraceRegion",
37	"LoopInfo",
38	"SourcePos",
39	"SourceRange",
40	"MetricStat",
41	"TraceValidationError",
42	"validate",
43	"PER_INVOCATION_CAT",
44	"AGGREGATED_CAT",
45	"STATIC_PREFIX",
46	]
47
48	PER_INVOCATION_CAT = "region,daisy"	4✔
49	AGGREGATED_CAT = "aggregated_region,daisy"	4✔
50	STATIC_PREFIX = "static:::"	4✔
51	_RUNTIME_KEY = "runtime"	4✔
52
53	_SCHEMA_FILENAME = "daisy_trace.schema.json"	4✔
54
55
56	def _find_schema() -> Path:	4✔
57	"""Locate the daisy-trace JSON schema.
58
59	In an installed/built wheel the schema is shipped as package data next to this
60	module (CMake installs ``rtl/schema/daisy_trace.schema.json`` into
61	``docc/benchmarks``). When running from a source checkout that install has not
62	happened, so fall back to the canonical copy under ``rtl/schema``.
63	"""
64	bundled = Path(__file__).with_name(_SCHEMA_FILENAME)	4✔
65	if bundled.is_file():	4✔
NEW 66	return bundled	×
67
68	# rtl.py -> docc/python/docc/benchmarks/rtl.py ; schema -> docc/rtl/schema/...
69	repo_schema = (	4✔
70	Path(__file__).resolve().parents[3] / "rtl" / "schema" / _SCHEMA_FILENAME
71	)
72	if repo_schema.is_file():	4✔
73	return repo_schema	4✔
74
NEW 75	raise FileNotFoundError(	×
76	f"Could not locate {_SCHEMA_FILENAME}; expected it next to {bundled} "
77	f"(installed wheel) or at {repo_schema} (source checkout)."
78	)
79
80
81	def load_schema() -> dict:	4✔
82	"""Return the bundled daisy-trace JSON schema as a dict."""
83	with open(_find_schema(), "r") as f:	4✔
84	return json.load(f)	4✔
85
86
87	class TraceValidationError(ValueError):	4✔
88	"""Raised when a trace does not conform to the daisy-trace schema."""
89
90	def __init__(self, errors: List[str]):	4✔
91	self.errors = errors	4✔
92	super().__init__("Trace failed schema validation:\n" + "\n".join(errors))	4✔
93
94
95	def validate(instance: Any, schema: Optional[dict] = None) -> List[str]:	4✔
96	"""Validate a parsed trace object against the schema.
97
98	Returns a list of human-readable error strings (empty when valid). Requires
99	the optional ``jsonschema`` dependency.
100	"""
101	try:	4✔
102	from jsonschema import Draft202012Validator	4✔
103	except ImportError as exc: # pragma: no cover - exercised only without dep
104	raise ImportError(
105	"Validating daisy traces requires the 'jsonschema' package. "
106	"Install it with `pip install jsonschema`."
107	) from exc
108
109	if schema is None:	4✔
110	schema = load_schema()	4✔
111
112	validator = Draft202012Validator(schema)	4✔
113	errors: List[str] = []	4✔
114	for error in sorted(	4✔
115	validator.iter_errors(instance), key=lambda e: list(e.absolute_path)
116	):
117	location = "/".join(str(p) for p in error.absolute_path) or "<root>"	4✔
118	errors.append(f"{location}: {error.message}")	4✔
119	return errors	4✔
120
121
122	@dataclass(frozen=True)	4✔
123	class SourcePos:	4✔
124	line: int	4✔
125	col: int	4✔
126
127	@classmethod	4✔
128	def from_dict(cls, d: dict) -> "SourcePos":	4✔
129	return cls(line=d["line"], col=d["col"])	4✔
130
131
132	@dataclass(frozen=True)	4✔
133	class SourceRange:	4✔
134	file: str	4✔
135	begin: SourcePos	4✔
136	end: SourcePos	4✔
137
138	@classmethod	4✔
139	def from_dict(cls, d: dict) -> "SourceRange":	4✔
140	return cls(	4✔
141	file=d["file"],
142	begin=SourcePos.from_dict(d["from"]),
143	end=SourcePos.from_dict(d["to"]),
144	)
145
146
147	@dataclass(frozen=True)	4✔
148	class LoopInfo:	4✔
149	loopnest_index: int	4✔
150	num_loops: int	4✔
151	num_maps: int	4✔
152	num_fors: int	4✔
153	num_whiles: int	4✔
154	max_depth: int	4✔
155	is_perfectly_nested: bool	4✔
156	is_perfectly_parallel: bool	4✔
157	is_elementwise: bool	4✔
158	has_side_effects: bool	4✔
159
160	@classmethod	4✔
161	def from_dict(cls, d: dict) -> "LoopInfo":	4✔
162	return cls(	4✔
163	loopnest_index=d["loopnest_index"],
164	num_loops=d["num_loops"],
165	num_maps=d["num_maps"],
166	num_fors=d["num_fors"],
167	num_whiles=d["num_whiles"],
168	max_depth=d["max_depth"],
169	is_perfectly_nested=d["is_perfectly_nested"],
170	is_perfectly_parallel=d["is_perfectly_parallel"],
171	is_elementwise=d["is_elementwise"],
172	has_side_effects=d["has_side_effects"],
173	)
174
175
176	@dataclass(frozen=True)	4✔
177	class MetricStat:	4✔
178	"""Aggregated statistics for a single counter (aggregated traces).
179
180	``mean``/``variance``/``min``/``max`` may be ``None`` for provided
181	(``static:::``) metrics whose value was non-finite at capture time.
182	"""
183
184	mean: Optional[float]	4✔
185	variance: Optional[float]	4✔
186	count: int	4✔
187	min: Optional[float]	4✔
188	max: Optional[float]	4✔
189
190	@classmethod	4✔
191	def from_dict(cls, d: dict) -> "MetricStat":	4✔
192	return cls(	4✔
193	mean=d["mean"],
194	variance=d["variance"],
195	count=d["count"],
196	min=d["min"],
197	max=d["max"],
198	)
199
200
201	class TraceRegion:	4✔
202	"""A single region event, wrapping the raw trace dict with typed accessors."""
203
204	def __init__(self, raw: dict):	4✔
205	self._raw = raw	4✔
206
207	# -- Raw / identity ----------------------------------------------------
208	@property	4✔
209	def raw(self) -> dict:	4✔
NEW 210	return self._raw	×
211
212	@property	4✔
213	def category(self) -> str:	4✔
214	return self._raw["cat"]	4✔
215
216	@property	4✔
217	def is_aggregated(self) -> bool:	4✔
218	return self.category == AGGREGATED_CAT	4✔
219
220	@property	4✔
221	def name(self) -> str:	4✔
222	return self._raw["name"]	4✔
223
224	@property	4✔
225	def pid(self) -> int:	4✔
NEW 226	return self._raw["pid"]	×
227
228	@property	4✔
229	def tid(self) -> int:	4✔
NEW 230	return self._raw["tid"]	×
231
232	@property	4✔
233	def ts_us(self) -> Optional[float]:	4✔
NEW 234	return self._raw.get("ts")	×
235
236	@property	4✔
237	def dur_us(self) -> Optional[float]:	4✔
238	return self._raw.get("dur")	4✔
239
240	# -- Source metadata ---------------------------------------------------
241	@property	4✔
242	def _args(self) -> dict:	4✔
243	return self._raw["args"]	4✔
244
245	@property	4✔
246	def function(self) -> str:	4✔
NEW 247	return self._args["function"]	×
248
249	@property	4✔
250	def module(self) -> str:	4✔
NEW 251	return self._args["module"]	×
252
253	@property	4✔
254	def target_type(self) -> str:	4✔
255	return self._args.get("target_type", "")	4✔
256
257	@cached_property	4✔
258	def source_ranges(self) -> List[SourceRange]:	4✔
259	return [SourceRange.from_dict(r) for r in self._args.get("source_ranges", [])]	4✔
260
261	# -- docc metadata -----------------------------------------------------
262	@property	4✔
263	def _docc(self) -> dict:	4✔
264	return self._args["docc"]	4✔
265
266	@property	4✔
267	def sdfg_name(self) -> str:	4✔
NEW 268	return self._docc.get("sdfg_name", "")	×
269
270	@property	4✔
271	def sdfg_file(self) -> str:	4✔
NEW 272	return self._docc.get("sdfg_file", "")	×
273
274	@property	4✔
275	def arg_capture_path(self) -> str:	4✔
NEW 276	return self._docc.get("arg_capture_path", "")	×
277
278	@property	4✔
279	def features_file(self) -> str:	4✔
NEW 280	return self._docc.get("features_file", "")	×
281
282	@property	4✔
283	def opt_report_file(self) -> str:	4✔
NEW 284	return self._docc.get("opt_report_file", "")	×
285
286	@property	4✔
287	def has_element(self) -> bool:	4✔
288	"""True when SDFG element metadata (id/type/loop_info) is present."""
289	return "element_id" in self._docc	4✔
290
291	@property	4✔
292	def element_id(self) -> Optional[int]:	4✔
293	return self._docc.get("element_id")	4✔
294
295	@property	4✔
296	def element_type(self) -> Optional[str]:	4✔
297	return self._docc.get("element_type")	4✔
298
299	@property	4✔
300	def loopnest_index(self) -> Optional[int]:	4✔
NEW 301	return self._docc.get("loopnest_index")	×
302
303	@cached_property	4✔
304	def loop_info(self) -> Optional[LoopInfo]:	4✔
305	li = self._docc.get("loop_info")	4✔
306	return LoopInfo.from_dict(li) if li is not None else None	4✔
307
308	# -- Metrics -----------------------------------------------------------
309	@property	4✔
310	def _metrics(self) -> dict:	4✔
311	return self._args.get("metrics", {})	4✔
312
313	@cached_property	4✔
314	def counters(self) -> Dict[str, int]:	4✔
315	"""PAPI counter values for a per-invocation region (empty if aggregated)."""
316	if self.is_aggregated:	4✔
NEW 317	return {}	×
318	return dict(self._metrics)	4✔
319
320	@cached_property	4✔
321	def counter_stats(self) -> Dict[str, MetricStat]:	4✔
322	"""Aggregated PAPI counter statistics (empty for per-invocation regions).
323
324	Excludes the reserved ``runtime`` block and ``static:::`` provided metrics.
325	"""
326	if not self.is_aggregated:	4✔
327	return {}	4✔
328	out: Dict[str, MetricStat] = {}	4✔
329	for key, val in self._metrics.items():	4✔
330	if key == _RUNTIME_KEY or key.startswith(STATIC_PREFIX):	4✔
331	continue	4✔
332	out[key] = MetricStat.from_dict(val)	4✔
333	return out	4✔
334
335	@cached_property	4✔
336	def static_metrics(self) -> Dict[str, MetricStat]:	4✔
337	"""User-provided (``static:::``) metric statistics, keyed without the prefix."""
338	if not self.is_aggregated:	4✔
NEW 339	return {}	×
340	out: Dict[str, MetricStat] = {}	4✔
341	for key, val in self._metrics.items():	4✔
342	if key.startswith(STATIC_PREFIX):	4✔
343	out[key[len(STATIC_PREFIX) :]] = MetricStat.from_dict(val)	4✔
344	return out	4✔
345
346	@cached_property	4✔
347	def runtime(self) -> Optional[MetricStat]:	4✔
348	"""Aggregated runtime statistics in microseconds (aggregated traces only)."""
349	rt = self._metrics.get(_RUNTIME_KEY) if self.is_aggregated else None	4✔
350	return MetricStat.from_dict(rt) if rt is not None else None	4✔
351
352	@property	4✔
353	def runtime_mean_us(self) -> Optional[float]:	4✔
354	"""Mean runtime in microseconds.
355
356	For aggregated traces this is ``runtime.mean``; for per-invocation traces
357	it is this event's own ``dur``.
358	"""
359	if self.is_aggregated:	4✔
360	return self.runtime.mean if self.runtime is not None else None	4✔
361	return self.dur_us	4✔
362
363	def metric(self, name: str) -> Optional[Union[int, float]]:	4✔
364	"""Return a representative value for ``name``.
365
366	For per-invocation regions this is the raw counter value; for aggregated
367	regions it is the counter/static metric mean. Returns ``None`` if absent.
368	"""
369	if not self.is_aggregated:	4✔
370	return self.counters.get(name)	4✔
371	if name in self.counter_stats:	4✔
NEW 372	return self.counter_stats[name].mean	×
373	if name in self.static_metrics:	4✔
374	return self.static_metrics[name].mean	4✔
NEW 375	return None	×
376
377	def __repr__(self) -> str:	4✔
NEW 378	return (	×
379	f"TraceRegion(name={self.name!r}, module={self.module!r}, "
380	f"target={self.target_type!r}, aggregated={self.is_aggregated})"
381	)
382
383
384	class Trace:	4✔
385	"""A parsed, validated daisy trace: a collection of :class:`TraceRegion`."""
386
387	def __init__(self, raw: dict):	4✔
388	self._raw = raw	4✔
389	self._regions = [TraceRegion(e) for e in raw.get("traceEvents", [])]	4✔
390
391	# -- Construction ------------------------------------------------------
392	@classmethod	4✔
393	def from_dict(cls, data: dict, validate_schema: bool = True) -> "Trace":	4✔
394	"""Build a :class:`Trace` from an already-parsed dict.
395
396	Raises :class:`TraceValidationError` if ``validate_schema`` is set and the
397	data does not conform to the schema.
398	"""
399	if validate_schema:	4✔
400	errors = validate(data)	4✔
401	if errors:	4✔
402	raise TraceValidationError(errors)	4✔
403	return cls(data)	4✔
404
405	@classmethod	4✔
406	def load(cls, path: Union[str, Path], validate_schema: bool = True) -> "Trace":	4✔
407	"""Read a trace JSON file, optionally validate it, and return a :class:`Trace`."""
NEW 408	with open(path, "r") as f:	×
NEW 409	data = json.load(f)	×
NEW 410	return cls.from_dict(data, validate_schema=validate_schema)	×
411
412	# -- Container protocol ------------------------------------------------
413	@property	4✔
414	def regions(self) -> List[TraceRegion]:	4✔
NEW 415	return self._regions	×
416
417	@property	4✔
418	def raw(self) -> dict:	4✔
NEW 419	return self._raw	×
420
421	def __len__(self) -> int:	4✔
422	return len(self._regions)	4✔
423
424	def __iter__(self) -> Iterator[TraceRegion]:	4✔
NEW 425	return iter(self._regions)	×
426
427	def __getitem__(self, index: int) -> TraceRegion:	4✔
428	return self._regions[index]	4✔
429
430	@property	4✔
431	def is_aggregated(self) -> bool:	4✔
432	"""True if the trace was captured in aggregate mode.
433
434	Determined from the first region; a valid trace never mixes modes.
435	"""
436	return bool(self._regions) and self._regions[0].is_aggregated	4✔
437
438	# -- Queries -----------------------------------------------------------
439	def filter(	4✔
440	self,
441	*,
442	module: Optional[str] = None,
443	function: Optional[str] = None,
444	target_type: Optional[str] = None,
445	element_type: Optional[str] = None,
446	sdfg_name: Optional[str] = None,
447	) -> List[TraceRegion]:
448	"""Return regions matching all provided (non-``None``) criteria."""
449	out = []	4✔
450	for r in self._regions:	4✔
451	if module is not None and r.module != module:	4✔
NEW 452	continue	×
453	if function is not None and r.function != function:	4✔
NEW 454	continue	×
455	if target_type is not None and r.target_type != target_type:	4✔
456	continue	4✔
457	if element_type is not None and r.element_type != element_type:	4✔
NEW 458	continue	×
459	if sdfg_name is not None and r.sdfg_name != sdfg_name:	4✔
NEW 460	continue	×
461	out.append(r)	4✔
462	return out	4✔
463
464	def by_element_id(self, element_id: int) -> List[TraceRegion]:	4✔
465	"""Return regions carrying the given SDFG ``element_id``."""
466	return [r for r in self._regions if r.element_id == element_id]	4✔
467
468	def hottest(self, n: Optional[int] = None) -> List[TraceRegion]:	4✔
469	"""Regions sorted by mean runtime (descending); ``n`` limits the count."""
470	ranked = sorted(	4✔
471	(r for r in self._regions if r.runtime_mean_us is not None),
472	key=lambda r: r.runtime_mean_us,
473	reverse=True,
474	)
475	return ranked[:n] if n is not None else ranked	4✔
476
477	def total_runtime_us(self) -> float:	4✔
478	"""Sum of mean runtimes across all regions (microseconds)."""
479	return sum((r.runtime_mean_us or 0.0) for r in self._regions)	4✔

daisytuner / docc / 30354831214

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