Fix ruff lint errors in unified profiling script

Refactor main() into parse_args(), profile_command(), and
report_results() to fix C901 (complexity) and PLR0915 (too many
statements). Also fix S306 (mktemp → NamedTemporaryFile), PLW1510
(explicit check=False), and add noqa for intentional os.path usage
(PTH112) and subprocess with CLI args (S603).
This commit is contained in:
Kevin Turcios 2026-04-15 04:07:11 -05:00
parent c54b7d2125
commit 5988f54605

View file

@ -28,12 +28,12 @@ import tempfile
import time
import tracemalloc
BASELINE_PATH = "/tmp/deep_baseline_total"
def main():
args = sys.argv[1:]
# Parse: <source_root> -- <command...>
if "--" not in args or len(args) < 3:
def parse_args(args):
"""Parse <source_root> -- <command...> from argv."""
if "--" not in args or len(args) < 3: # noqa: PLR2004
print(
"Usage: python deep_profile.py <source_root> -- <command> [args...]",
file=sys.stderr,
@ -48,35 +48,46 @@ def main():
source_root = os.path.abspath(args[sep - 1])
cmd = args[sep + 1 :]
if not os.path.isdir(source_root):
print(f"Error: source root '{source_root}' is not a directory", file=sys.stderr)
if not os.path.isdir(source_root): # noqa: PTH112
print(
f"Error: source root '{source_root}' is not a directory",
file=sys.stderr,
)
sys.exit(1)
# --- GC tracking ---
return source_root, cmd
def profile_command(cmd):
"""Run cmd under cProfile + tracemalloc + GC tracking."""
gc_times = []
def gc_callback(phase, info):
def gc_callback(phase, _info):
if phase == "start":
gc_callback._start = time.perf_counter()
elif phase == "stop":
gc_times.append(time.perf_counter() - gc_callback._start)
gc.callbacks.append(gc_callback)
# --- Profile the command ---
tracemalloc.start()
profiler = cProfile.Profile()
profiler.enable()
result = subprocess.run(cmd)
result = subprocess.run(cmd, check=False) # noqa: S603
profiler.disable()
mem_snapshot = tracemalloc.take_snapshot()
prof_path = tempfile.mktemp(suffix=".prof", prefix="deep_cpu_")
with tempfile.NamedTemporaryFile(
suffix=".prof", prefix="deep_cpu_", delete=False
) as prof_file:
prof_path = prof_file.name
profiler.dump_stats(prof_path)
# --- Report ---
return result, mem_snapshot, gc_times, prof_path
def report_results(source_root, mem_snapshot, gc_times, prof_path):
"""Print unified profile report."""
print("\n" + "=" * 60)
print("UNIFIED PROFILE RESULTS")
print("=" * 60)
@ -93,33 +104,39 @@ def main():
# CPU top functions (project-only)
print(f"\n=== CPU: Top project functions (source root: {source_root}) ===")
p = pstats.Stats(prof_path)
stats = p.stats
project_funcs = []
for (file, line, name), (cc, nc, tt, ct, callers) in stats.items():
for (file, line, name), (cc, nc, tt, ct, callers) in p.stats.items():
if not os.path.abspath(file).startswith(source_root):
continue
project_funcs.append((ct, tt, nc, name, file, line))
project_funcs.sort(reverse=True)
total = project_funcs[0][0] if project_funcs else 1
# Save baseline total for delta comparisons
baseline_path = "/tmp/deep_baseline_total"
if not os.path.exists(baseline_path):
with open(baseline_path, "w") as f:
# Baseline delta tracking
if not os.path.exists(BASELINE_PATH):
with open(BASELINE_PATH, "w") as f:
f.write(str(total))
print(f" (baseline recorded: {total:.3f}s)")
else:
with open(baseline_path) as f:
with open(BASELINE_PATH) as f:
baseline = float(f.read().strip())
delta = (total - baseline) / baseline * 100
print(f" (baseline: {baseline:.3f}s, current: {total:.3f}s, delta: {delta:+.1f}%)")
print(
f" (baseline: {baseline:.3f}s, current: {total:.3f}s, delta: {delta:+.1f}%)"
)
for ct, tt, nc, name, file, line in project_funcs[:15]:
pct = ct / total * 100
relpath = os.path.relpath(file)
print(f" {name:30s} {pct:5.1f}% cumtime {tt:.3f}s self {nc:>6d} calls {relpath}:{line}")
print(
f" {name:30s} {pct:5.1f}% cumtime {tt:.3f}s self {nc:>6d} calls {relpath}:{line}"
)
# Exit with the profiled command's exit code
def main():
source_root, cmd = parse_args(sys.argv[1:])
result, mem_snapshot, gc_times, prof_path = profile_command(cmd)
report_results(source_root, mem_snapshot, gc_times, prof_path)
sys.exit(result.returncode)