Function load_samples

scripts/analyze_runtime_memory_log.py:201–239

(paths: Iterable[Path])

def load_samples(paths: Iterable[Path]) -> list[Sample]:
    samples: list[Sample] = []
    for path in paths:
        try:
            lines = path.read_text().splitlines()
        except FileNotFoundError:
            continue
        for idx, line in enumerate(lines, start=1):
            line = line.strip()
            if not line:
                continue
            try:
                raw = json.loads(line)
            except json.JSONDecodeError:
                continue
            trigger = raw.get("trigger") or {}
            source = str(raw.get("source") or "")
            kind = infer_kind(raw, source)
            target = infer_target(raw, path)
            instance_id = infer_instance_id(raw, target)
            trigger_category, trigger_reason = infer_trigger(raw, kind, source, trigger)
            samples.append(
                Sample(
                    path=path,
                    line_no=idx,
                    raw=raw,
                    timestamp_ms=int(raw.get("timestamp_ms") or 0),
                    kind=kind,
                    target=target,
                    instance_id=instance_id,
                    source=source,
                    trigger_category=trigger_category,
                    trigger_reason=trigger_reason,
                    sessions=raw.get("sessions") if isinstance(raw.get("sessions"), dict) else None,
                    totals=raw.get("totals") if isinstance(raw.get("totals"), dict) else None,
                )
            )
    samples.sort(key=lambda sample: (sample.timestamp_ms, str(sample.path), sample.line_no))
    return samples


def infer_instance_id(raw: dict[str, Any], target: str) -> str: