1"""Animation hitches lane parser.
2 
3Xcode 26 schema `hitches` columns: start, duration (hitch time), process,
4is-system, swap-id, label, display, narrative-description. The
5narrative-description field carries Apple's own attribution (e.g.
6"Potentially expensive app update(s)") which is the highest-signal column.
7"""
8from __future__ import annotations
9 
10from collections import Counter
11from pathlib import Path
12from typing import Any
13 
14from . import xctrace, xml_utils
15 
16CANDIDATE_SCHEMAS = ("hitches", "animation-hitch", "hitch")
17 
18START_KEYS = ("start", "time", "sample-time")
19DURATION_KEYS = ("duration", "hitch-duration", "frame-duration")
20 
21 
22def analyze(
23    trace_path: Path,
24    toc_schemas: frozenset[str],
25    top_n: int = 10,
26    window: tuple[int, int] | None = None,
27    run: int = 1,
28) -> dict[str, Any]:
29    schema = _pick_schema(toc_schemas)
30    if schema is None:
31        return {
32            "lane": "hitches",
33            "available": False,
34            "notes": ["Animation hitches not present in trace."],
35        }
36 
37    xml_bytes = xctrace.export_schema(trace_path, schema, run=run)
38    stream = xml_utils.RowStream(xml_bytes)
39 
40    events: list[dict] = []
41    narrative_counts: Counter[str] = Counter()
42    system_count = 0
43 
44    for row in stream:
45        start_ns = _first_int(row, stream, START_KEYS)
46        duration_ns = _first_int(row, stream, DURATION_KEYS)
47        if start_ns is None or duration_ns is None:
48            continue
49        if not xml_utils.event_overlaps_window(start_ns, start_ns + duration_ns, window):
50            continue
51 
52        process_el = row.get("process")
53        process = (
54            xml_utils.extract_process(process_el, stream)
55            if process_el is not None else None
56        )
57 
58        narrative_el = row.get("narrative-description")
59        narrative = xml_utils.str_text(stream.resolve(narrative_el)) if narrative_el is not None else None
60        if narrative:
61            narrative_counts[narrative] += 1
62 
63        is_system_el = row.get("is-system")
64        is_system = _bool_text(stream.resolve(is_system_el)) if is_system_el is not None else None
65        if is_system:
66            system_count += 1
67 
68        events.append({
69            "start_ns": start_ns,
70            "end_ns": start_ns + duration_ns,
71            "duration_ns": duration_ns,
72            "hitch_duration_ns": duration_ns,  # Xcode 26 `duration` == hitch time
73            "frame_duration_ns": None,
74            "hitch_duration_ms": round(duration_ns / 1_000_000, 2),
75            "frame_duration_ms": None,
76            "start_ms": round(start_ns / 1_000_000, 2),
77            "process": (process or {}).get("name"),
78            "narrative": narrative,
79            "is_system": bool(is_system) if is_system is not None else None,
80        })
81 
82    events.sort(key=lambda e: e["duration_ns"], reverse=True)
83 
84    total_hitch_ms = sum(e["hitch_duration_ms"] for e in events)
85    worst = events[0] if events else None
86 
87    per_process: dict[str, int] = {}
88    for e in events:
89        key = e["process"] or "unknown"
90        per_process[key] = per_process.get(key, 0) + 1
91 
92    top_offenders = [
93        {
94            "start_ms": e["start_ms"],
95            "hitch_duration_ms": e["hitch_duration_ms"],
96            "frame_duration_ms": e["frame_duration_ms"],
97            "process": e["process"],
98            "narrative": e["narrative"],
99            "is_system": e["is_system"],
100        }
101        for e in events[:top_n]
102    ]
103 
104    return {
105        "lane": "hitches",
106        "available": True,
107        "schema_used": schema,
108        "metrics": {
109            "count": len(events),
110            "total_hitch_ms": round(total_hitch_ms, 2),
111            "worst_hitch_ms": worst["hitch_duration_ms"] if worst else 0,
112            "per_process": per_process,
113            "system_hitches": system_count,
114            "app_hitches": len(events) - system_count,
115            "narrative_breakdown": dict(narrative_counts.most_common()),
116        },
117        "top_offenders": top_offenders,
118        "notes": [],
119        "_events": events,
120    }
121 
122 
123def _pick_schema(available: frozenset[str]) -> str | None:
124    for s in CANDIDATE_SCHEMAS:
125        if s in available:
126            return s
127    return None
128 
129 
130def _first_int(row, stream, keys):
131    for key in keys:
132        el = row.get(key)
133        if el is None:
134            continue
135        val = xml_utils.int_text(stream.resolve(el))
136        if val is not None:
137            return val
138    return None
139 
140 
141def _bool_text(elem) -> bool | None:
142    txt = xml_utils.str_text(elem)
143    if txt is None:
144        return None
145    return txt.strip() in ("1", "true", "True", "YES", "Yes")
146