1"""Time Profiler lane parser (schema `time-profile`).
2 
3Aggregates CPU samples by leaf symbol, keeps per-sample rows so that other
4lanes can correlate by timestamp window.
5"""
6from __future__ import annotations
7 
8from collections import defaultdict
9from pathlib import Path
10from typing import Any
11 
12from . import xctrace, xml_utils
13 
14PREFERRED_SCHEMAS = ("time-profile",)
15FALLBACK_SCHEMAS = ("time-sample",)  # no symbolication; used only if nothing else
16 
17 
18def analyze(
19    trace_path: Path,
20    toc_schemas: frozenset[str],
21    top_n: int = 10,
22    window: tuple[int, int] | None = None,
23    run: int = 1,
24) -> dict[str, Any]:
25    schema = _pick_schema(toc_schemas)
26    if schema is None:
27        return {
28            "lane": "time-profiler",
29            "available": False,
30            "notes": ["Time Profiler data not present in trace."],
31        }
32 
33    xml_bytes = xctrace.export_schema(trace_path, schema, run=run)
34    stream = xml_utils.RowStream(xml_bytes)
35 
36    samples: list[dict] = []
37    symbol_weight: dict[str, int] = defaultdict(int)
38    symbol_samples: dict[str, int] = defaultdict(int)
39    symbol_thread: dict[str, str] = {}
40    processes: set[str] = set()
41    total_weight = 0
42    min_time: int | None = None
43    max_time: int | None = None
44 
45    for row in stream:
46        time_el = row.get("time")
47        weight_el = row.get("weight")
48        thread_el = row.get("thread")
49        stack_el = row.get("stack")
50        if stack_el is None or time_el is None or thread_el is None:
51            continue
52 
53        sample_time_ns = xml_utils.int_text(stream.resolve(time_el))
54        if not xml_utils.in_window(sample_time_ns, window):
55            continue
56        weight_ns = xml_utils.int_text(stream.resolve(weight_el)) or 0
57        frames = xml_utils.extract_backtrace(stack_el, stream, max_frames=20)
58        if not frames:
59            continue
60 
61        thread = xml_utils.extract_thread(thread_el, stream)
62        process_name = (thread.get("process") or {}).get("name")
63        if process_name:
64            processes.add(process_name)
65 
66        leaf = xml_utils.top_symbol(frames)
67        symbol_weight[leaf] += weight_ns
68        symbol_samples[leaf] += 1
69        symbol_thread.setdefault(
70            leaf, "main" if thread["is_main"] else thread.get("name", "")
71        )
72        total_weight += weight_ns
73 
74        if sample_time_ns is not None:
75            min_time = sample_time_ns if min_time is None else min(min_time, sample_time_ns)
76            max_time = sample_time_ns if max_time is None else max(max_time, sample_time_ns)
77 
78            samples.append({
79                "time_ns": sample_time_ns,
80                "weight_ns": weight_ns,
81                "thread_name": thread["name"],
82                "is_main": thread["is_main"],
83                "process": process_name,
84                "leaf_symbol": leaf,
85                "frames": frames[:5],
86            })
87 
88    samples.sort(key=lambda s: s["time_ns"])
89 
90    top = sorted(
91        symbol_weight.items(), key=lambda kv: kv[1], reverse=True
92    )[:top_n]
93    top_offenders = [
94        {
95            "symbol": sym,
96            "weight_ns": w,
97            "weight_ms": round(w / 1_000_000, 2),
98            "samples": symbol_samples[sym],
99            "percent": round(100.0 * w / total_weight, 2) if total_weight else 0.0,
100            "thread": symbol_thread.get(sym, ""),
101        }
102        for sym, w in top
103    ]
104 
105    notes: list[str] = []
106    if schema in FALLBACK_SCHEMAS:
107        notes.append(
108            f"Using fallback schema `{schema}`; backtraces may be unsymbolicated."
109        )
110 
111    return {
112        "lane": "time-profiler",
113        "available": True,
114        "schema_used": schema,
115        "metrics": {
116            "total_samples": len(samples),
117            "total_weight_ns": total_weight,
118            "total_weight_ms": round(total_weight / 1_000_000, 2),
119            "window_start_ns": min_time,
120            "window_end_ns": max_time,
121            "processes": sorted(processes),
122        },
123        "top_offenders": top_offenders,
124        "notes": notes,
125        # Internal: retained for correlation. Stripped before JSON emission
126        # if --slim is requested by the orchestrator.
127        "_samples": samples,
128    }
129 
130 
131def _pick_schema(available: frozenset[str]) -> str | None:
132    for s in PREFERRED_SCHEMAS + FALLBACK_SCHEMAS:
133        if s in available:
134            return s
135    return None
136