/
/
/
1"""
2Synchronous data collectors and formatting helpers for the Profiler provider.
3
4All functions in this module are plain synchronous code so the provider can run
5the (potentially) expensive ones in an executor thread. None of the collected
6data contains user-specific information: only code identifiers (function names,
7module paths, line numbers), counters and byte/time metrics are captured so the
8resulting report is safe to share publicly.
9"""
10
11from __future__ import annotations
12
13import collections
14import gc
15import json
16import logging
17import os
18import time
19import tracemalloc
20from pathlib import Path
21from typing import TYPE_CHECKING, Any
22
23import psutil
24import yappi
25
26from music_assistant.helpers.datetime import utc
27
28if TYPE_CHECKING:
29 import asyncio
30
31# fields of a single flight-recorder entry (also the stats.csv column order)
32RECORDER_FIELDS = (
33 "ts_unix",
34 "rss_mb",
35 "cpu_pct",
36 "loop_lag_avg_ms",
37 "loop_lag_max_ms",
38 "asyncio_tasks",
39 "tracked_tasks",
40 "tracked_timers",
41 "event_subscribers",
42 "ws_clients",
43 "ffmpeg_processes",
44 "events_per_s",
45 "log_errors_per_s",
46)
47
48MAX_CSV_SIZE = 4 * 1024 * 1024
49
50
51class LogErrorCounter(logging.Handler):
52 """
53 Log handler that counts warnings/errors without storing any message content.
54
55 Only the source code location, log level and exception type are recorded, so
56 the aggregated counts are safe to include in a shareable report.
57 """
58
59 max_unique_keys = 200
60
61 def __init__(self) -> None:
62 """Initialize the counter for WARNING level records and above."""
63 super().__init__(level=logging.WARNING)
64 self.total = 0
65 self.dropped = 0
66 self._counts: dict[tuple[str, str, str], int] = {}
67 self._last_seen: dict[tuple[str, str, str], int] = {}
68
69 def emit(self, record: logging.LogRecord) -> None:
70 """Count the log record by source code location, level and exception type."""
71 exc_type = "-"
72 if record.exc_info and record.exc_info[0]:
73 exc_type = record.exc_info[0].__name__
74 # group by code location: logger names may embed user-set names or device ids
75 source = f"{sanitize_code_path(record.pathname)}:{record.lineno}"
76 key = (source, record.levelname, exc_type)
77 self.total += 1
78 if key not in self._counts and len(self._counts) >= self.max_unique_keys:
79 self.dropped += 1
80 return
81 self._counts[key] = self._counts.get(key, 0) + 1
82 self._last_seen[key] = int(time.time())
83
84 def summarize(self, top_n: int = 50) -> dict[str, Any]:
85 """Return the aggregated warning/error counts, largest first."""
86 # take a snapshot under the handler lock as emit() may run in other threads
87 self.acquire()
88 try:
89 top_counts = sorted(self._counts.items(), key=lambda kv: kv[1], reverse=True)[:top_n]
90 last_seen = dict(self._last_seen)
91 finally:
92 self.release()
93 top = [
94 {
95 "source": key[0],
96 "level": key[1],
97 "exception": key[2],
98 "count": count,
99 "last_seen_ts_unix": last_seen.get(key, 0),
100 }
101 for key, count in top_counts
102 ]
103 return {"total_since_load": self.total, "unique_keys_dropped": self.dropped, "top": top}
104
105
106def sanitize_code_path(path: str) -> str:
107 """Reduce a code file path to a package-relative form without user-specific parts."""
108 path = path.replace("\\", "/")
109 for marker in ("/site-packages/", "/music_assistant/", "/lib/"):
110 if (idx := path.rfind(marker)) >= 0:
111 return path[idx + (1 if marker == "/music_assistant/" else len(marker)) :]
112 return path.rsplit("/", 1)[-1]
113
114
115def extract_cpu_profile(duration_s: float, out_dir: str, top_n: int = 40) -> dict[str, Any]:
116 """
117 Convert the stats of a just-stopped yappi run into a result dict and a .pstats file.
118
119 Must be called after ``yappi.stop()``; the collected stats are cleared when done.
120
121 :param duration_s: The actual (wall clock) duration the profile window ran for.
122 :param out_dir: Directory to store the .pstats file for offline analysis.
123 :param top_n: Maximum number of top functions to include in the result.
124 """
125 stats = yappi.get_func_stats()
126 captured_at = utc()
127 pstats_file: str | None = None
128 top_functions: list[dict[str, Any]] = []
129 if not stats.empty():
130 pstats_file = f"cpu_profile_{captured_at.strftime('%Y%m%d_%H%M%S')}.pstats"
131 stats.save(os.path.join(out_dir, pstats_file), type="pstat")
132 stats.sort("ttot", "desc")
133 for stat in stats:
134 if len(top_functions) >= top_n:
135 break
136 top_functions.append(
137 {
138 "name": stat.name,
139 "location": f"{sanitize_code_path(stat.module)}:{stat.lineno}",
140 "ncall": stat.ncall,
141 "tsub_s": round(stat.tsub, 4),
142 "ttot_s": round(stat.ttot, 4),
143 "tavg_ms": round(stat.tavg * 1000, 3),
144 }
145 )
146 yappi.clear_stats()
147 _prune_files(out_dir, "cpu_profile_", keep=10)
148 return {
149 "captured_at": captured_at.isoformat(timespec="seconds"),
150 "duration_s": round(duration_s, 1),
151 "clock_type": "cpu",
152 "top_functions": top_functions,
153 "pstats_file": pstats_file,
154 }
155
156
157def collect_memory_stats(proc: psutil.Process) -> dict[str, Any]:
158 """Collect process-level and garbage-collector memory statistics."""
159 mem = proc.memory_info()
160 try:
161 open_fds: int | None = proc.num_fds()
162 except AttributeError, psutil.Error:
163 open_fds = None
164 return {
165 "rss_mb": round(mem.rss / 1024**2, 1),
166 "vms_mb": round(mem.vms / 1024**2, 1),
167 "num_threads": proc.num_threads(),
168 "open_fds": open_fds,
169 "gc_enabled": gc.isenabled(),
170 "gc_counts": list(gc.get_count()),
171 "gc_thresholds": list(gc.get_threshold()),
172 }
173
174
175def collect_object_census(top_n: int = 30) -> list[dict[str, Any]]:
176 """
177 Count all live Python objects by type, largest counts first.
178
179 This walks the entire object heap and can stall the process for multiple
180 seconds on large installations; run it in an executor and only on request.
181 """
182 counter: collections.Counter[str] = collections.Counter(
183 type(obj).__name__ for obj in gc.get_objects()
184 )
185 return [{"type": name, "count": count} for name, count in counter.most_common(top_n)]
186
187
188def collect_tracemalloc_stats(
189 previous: tracemalloc.Snapshot | None, top_n: int = 30
190) -> tuple[dict[str, Any], tracemalloc.Snapshot]:
191 """
192 Summarize current tracemalloc data and the growth since the previous snapshot.
193
194 :param previous: Snapshot from the previous report to diff against (or None).
195 :param top_n: Maximum number of allocation sites to include per list.
196 :return: The summary dict plus the new snapshot to diff against next time.
197 """
198 snapshot = tracemalloc.take_snapshot().filter_traces(
199 (tracemalloc.Filter(False, tracemalloc.__file__),)
200 )
201 traced_current, traced_peak = tracemalloc.get_traced_memory()
202 top_sites = []
203 for stat in snapshot.statistics("lineno")[:top_n]:
204 frame = stat.traceback[0]
205 top_sites.append(
206 {
207 "location": f"{sanitize_code_path(frame.filename)}:{frame.lineno}",
208 "size_kb": round(stat.size / 1024, 1),
209 "count": stat.count,
210 }
211 )
212 growth = []
213 if previous is not None:
214 # only report allocation sites that actually grew (compare_to sorts by absolute
215 # difference, so large frees would otherwise dominate the list)
216 grown = [diff for diff in snapshot.compare_to(previous, "lineno") if diff.size_diff > 0]
217 for diff in grown[:top_n]:
218 frame = diff.traceback[0]
219 growth.append(
220 {
221 "location": f"{sanitize_code_path(frame.filename)}:{frame.lineno}",
222 "size_diff_kb": round(diff.size_diff / 1024, 1),
223 "count_diff": diff.count_diff,
224 }
225 )
226 summary = {
227 "traced_current_mb": round(traced_current / 1024**2, 1),
228 "traced_peak_mb": round(traced_peak / 1024**2, 1),
229 "top_allocation_sites": top_sites,
230 "growth_since_previous_report": growth,
231 }
232 return summary, snapshot
233
234
235def collect_task_stats(tasks: set[asyncio.Task[Any]], top_n: int = 50) -> dict[str, Any]:
236 """
237 Summarize all asyncio tasks by the code location where they are suspended.
238
239 Only coroutine names and file:line locations are included - never task
240 names or arguments, as those may contain user-specific data.
241 """
242 locations: collections.Counter[str] = collections.Counter()
243 for task in tasks:
244 coro = task.get_coro()
245 name = getattr(coro, "__qualname__", None) or type(coro).__name__
246 frame = getattr(coro, "cr_frame", None)
247 if frame is not None:
248 name = f"{name} @ {sanitize_code_path(frame.f_code.co_filename)}:{frame.f_lineno}"
249 locations[name] += 1
250 return {
251 "total": len(tasks),
252 "top_by_location": [
253 {"coroutine": name, "count": count} for name, count in locations.most_common(top_n)
254 ],
255 }
256
257
258def finalize_recorder_entry(
259 proc: psutil.Process, entry: dict[str, Any], csv_path: str
260) -> dict[str, Any]:
261 """Add process-level metrics to a flight-recorder entry and append it to the stats CSV."""
262 mem = proc.memory_info()
263 entry["rss_mb"] = round(mem.rss / 1024**2, 1)
264 # cpu percent is measured over the interval since the previous sample
265 entry["cpu_pct"] = round(proc.cpu_percent(), 1)
266 try:
267 entry["ffmpeg_processes"] = sum(
268 1 for child in proc.children() if "ffmpeg" in _process_name(child)
269 )
270 except psutil.Error:
271 entry["ffmpeg_processes"] = None
272 _append_csv_row(csv_path, entry)
273 return entry
274
275
276def persist_report(out_dir: str, report: dict[str, Any]) -> str:
277 """Write the report to disk as report.json + report.md and return the markdown version."""
278 markdown = render_markdown(report)
279 with open(os.path.join(out_dir, "report.json"), "w", encoding="utf-8") as _file:
280 json.dump(report, _file, indent=2, default=str)
281 with open(os.path.join(out_dir, "report.md"), "w", encoding="utf-8") as _file:
282 _file.write(markdown)
283 return markdown
284
285
286def render_markdown(report: dict[str, Any]) -> str:
287 """Render the report dict as human-readable markdown for pasting into a chat or issue."""
288 lines: list[str] = ["# Music Assistant profiler report", ""]
289 lines.extend(
290 f"- {key}: {value}" for key, value in report.items() if not isinstance(value, dict | list)
291 )
292 for key, value in report.items():
293 if isinstance(value, dict):
294 lines.extend(("", f"## {key}", *_render_section(value, depth=3)))
295 elif isinstance(value, list):
296 lines.extend(("", f"## {key}", *_md_table(value)))
297 return "\n".join(lines) + "\n"
298
299
300def _render_section(mapping: dict[str, Any], depth: int) -> list[str]:
301 """Render a report section as markdown lines (scalars first, then nested content)."""
302 lines: list[str] = []
303 nested: list[tuple[str, Any]] = []
304 for key, value in mapping.items():
305 if isinstance(value, dict) or (value and isinstance(value, list)):
306 nested.append((key, value))
307 else:
308 lines.append(f"- {key}: {value}")
309 for key, value in nested:
310 lines.extend(("", f"{'#' * min(depth, 6)} {key}"))
311 if isinstance(value, dict):
312 lines.extend(_render_section(value, depth + 1))
313 elif value and isinstance(value[0], dict):
314 lines.extend(_md_table(value))
315 else:
316 lines.extend(f"- {item}" for item in value)
317 return lines
318
319
320def _md_table(rows: list[dict[str, Any]]) -> list[str]:
321 """Render a list of uniform dicts as a markdown table."""
322 if not rows:
323 return ["(none)"]
324 columns = list(rows[0])
325 lines = ["| " + " | ".join(columns) + " |", "|" + "---|" * len(columns)]
326 lines.extend(
327 "| " + " | ".join(str(row.get(column, "")) for column in columns) + " |" for row in rows
328 )
329 return lines
330
331
332def _append_csv_row(csv_path: str, entry: dict[str, Any]) -> None:
333 """Append a recorder entry to the stats CSV, restarting the file when it grows too large."""
334 path = Path(csv_path)
335 write_header = True
336 if path.is_file():
337 if path.stat().st_size > MAX_CSV_SIZE:
338 path.unlink()
339 else:
340 write_header = False
341 with path.open("a", encoding="utf-8") as _file:
342 if write_header:
343 _file.write(",".join(RECORDER_FIELDS) + "\n")
344 _file.write(",".join(str(entry.get(field, "")) for field in RECORDER_FIELDS) + "\n")
345
346
347def _process_name(proc: psutil.Process) -> str:
348 """Return the process name, or an empty string if the process is already gone."""
349 try:
350 return str(proc.name())
351 except psutil.Error:
352 return ""
353
354
355def _prune_files(out_dir: str, prefix: str, keep: int) -> None:
356 """Remove the oldest files with the given prefix, keeping only the newest ones."""
357 files = sorted(path for path in Path(out_dir).iterdir() if path.name.startswith(prefix))
358 for path in files[:-keep]:
359 path.unlink()
360