Skills › Scientific Agent Skills › autoskill › cluster.py
cluster.py
54 lines · skills/autoskill/scripts/cluster.py at 5f2bc03
1from collections import defaultdict
4def segment_sessions(events, idle_gap_seconds, min_session_seconds):
5 if not events:
6 return []
7 events = sorted(events, key=lambda e: e["ts"])
8 groups = [[events[0]]]
9 for prev, curr in zip(events, events[1:]):
10 if curr["ts"] - prev["ts"] > idle_gap_seconds:
11 groups.append([curr])
12 else:
13 groups[-1].append(curr)
15 sessions = []
16 for group in groups:
17 duration = group[-1]["ts"] - group[0]["ts"]
18 if duration < min_session_seconds:
19 continue
20 apps, seen = [], set()
21 for evt in group:
22 if evt["app"] not in seen:
23 seen.add(evt["app"])
24 apps.append(evt["app"])
25 sessions.append({
26 "start_ts": group[0]["ts"],
27 "end_ts": group[-1]["ts"],
28 "duration_seconds": duration,
29 "apps": apps,
30 "window_titles": [e["window_title"] for e in group if e.get("window_title")],
31 })
32 return sessions
35def cluster_sessions(sessions, min_cluster_size):
36 buckets = defaultdict(list)
37 for s in sessions:
38 buckets[tuple(s["apps"])].append(s)
40 clusters = []
41 for apps, members in buckets.items():
42 if len(members) < min_cluster_size:
43 continue
44 example_titles = []
45 for m in members:
46 if m["window_titles"]:
47 example_titles.append(m["window_titles"][0])
48 clusters.append({
49 "apps": list(apps),
50 "session_count": len(members),
51 "total_duration_seconds": sum(m["duration_seconds"] for m in members),
52 "example_titles": example_titles,
53 })
54 return clusters