| from __future__ import annotations |
|
|
| import os |
| import time |
| from pathlib import Path |
| from typing import Mapping |
|
|
| from .classifier import classify_all |
| from .dryrun import build_dry_run |
| from .models import DashboardSnapshot, ProcessSample |
| from .procfs import scan_processes |
| from .risk import assess_groups |
| from .tree import build_groups, build_nodes |
|
|
|
|
| def build_snapshot( |
| proc_root: str | Path = "/proc", |
| *, |
| previous: Mapping[str, ProcessSample] | None = None, |
| exclude_self: bool = True, |
| ) -> DashboardSnapshot: |
| exclude_pids = {os.getpid()} if exclude_self else set() |
| processes = scan_processes( |
| proc_root, |
| previous=previous, |
| current_uid_only=True, |
| exclude_pids=exclude_pids, |
| ) |
| classifications = classify_all(processes) |
| nodes = build_nodes(processes, classifications) |
| groups = build_groups(nodes) |
| stats, warnings = assess_groups(groups) |
| dry_run = build_dry_run(groups) |
| return DashboardSnapshot( |
| sample_ts=time.time(), |
| processes=processes, |
| classifications=classifications, |
| nodes=nodes, |
| groups=groups, |
| dry_run=dry_run, |
| stats=stats, |
| warnings=warnings, |
| ) |
|
|
|
|
| def samples_by_key(snapshot: DashboardSnapshot) -> dict[str, ProcessSample]: |
| return {sample.process_key: sample for sample in snapshot.processes} |
|
|
|
|
