Make profiling sampling a stub package

Author: JelleZijlstra

stdlib/profiling/sampling.pyi

@@ -0,0 +1,17 @@
+from .collector import Collector as Collector
+from .gecko_collector import GeckoCollector as GeckoCollector
+from .heatmap_collector import HeatmapCollector as HeatmapCollector
+from .jsonl_collector import JsonlCollector as JsonlCollector
+from .pstats_collector import PstatsCollector as PstatsCollector
+from .stack_collector import CollapsedStackCollector as CollapsedStackCollector
+from .string_table import StringTable as StringTable
+
+__all__ = (
+    "Collector",
+    "PstatsCollector",
+    "CollapsedStackCollector",
+    "HeatmapCollector",
+    "GeckoCollector",
+    "JsonlCollector",
+    "StringTable",
+)

stdlib/profiling/sampling/__init__.pyi

@@ -0,0 +1,58 @@
+from _typeshed import StrOrBytesPath
+from abc import ABC, abstractmethod
+from collections.abc import Sequence
+from typing import Protocol, TypeAlias
+
+_Location: TypeAlias = int | tuple[int, int, int, int] | _LocationInfo | None
+_Frame: TypeAlias = _FrameInfo | tuple[str, _Location, str, int | None]
+_Timestamps: TypeAlias = Sequence[int] | None
+_StackFrames: TypeAlias = Sequence[_InterpreterInfo] | Sequence[_AwaitedInfo]
+
+class _LocationInfo(Protocol):
+    lineno: int
+    end_lineno: int
+    col_offset: int
+    end_col_offset: int
+    def __getitem__(self, index: int, /) -> int: ...
+
+class _FrameInfo(Protocol):
+    filename: str
+    location: _Location
+    funcname: str
+    opcode: int | None
+    def __getitem__(self, index: int, /) -> object: ...
+
+class _ThreadInfo(Protocol):
+    thread_id: int
+    status: int
+    frame_info: Sequence[_Frame]
+
+class _InterpreterInfo(Protocol):
+    interpreter_id: int
+    threads: Sequence[_ThreadInfo]
+
+class _CoroInfo(Protocol):
+    call_stack: Sequence[_Frame]
+    task_name: int | str
+
+class _TaskInfo(Protocol):
+    task_id: int
+    task_name: str
+    coroutine_stack: Sequence[_CoroInfo]
+    awaited_by: Sequence[_CoroInfo]
+
+class _AwaitedInfo(Protocol):
+    thread_id: int
+    awaited_by: Sequence[_TaskInfo]
+
+def normalize_location(location: _Location) -> tuple[int, int, int, int]: ...
+def extract_lineno(location: _Location) -> int: ...
+def filter_internal_frames(frames: Sequence[_Frame]) -> list[_Frame]: ...
+def iter_async_frames(awaited_info_list: Sequence[_AwaitedInfo]) -> object: ...
+
+class Collector(ABC):
+    @abstractmethod
+    def collect(self, stack_frames: _StackFrames, timestamps_us: _Timestamps = None) -> None: ...
+    def collect_failed_sample(self) -> None: ...
+    @abstractmethod
+    def export(self, filename: StrOrBytesPath) -> None: ...

stdlib/profiling/sampling/collector.pyi

@@ -0,0 +1,8 @@
+from _typeshed import StrOrBytesPath
+
+from .collector import Collector, _StackFrames, _Timestamps
+
+class GeckoCollector(Collector):
+    def __init__(self, sample_interval_usec: int, *, skip_idle: bool = False, opcodes: bool = False) -> None: ...
+    def collect(self, stack_frames: _StackFrames, timestamps_us: _Timestamps = None) -> None: ...
+    def export(self, filename: StrOrBytesPath) -> None: ...

stdlib/profiling/sampling/gecko_collector.pyi

@@ -0,0 +1,20 @@
+from _typeshed import StrOrBytesPath
+from collections.abc import Sequence
+
+from .collector import Collector, _Frame, _StackFrames, _Timestamps
+
+class HeatmapCollector(Collector):
+    FILE_INDEX_FORMAT: str
+    def __init__(self, sample_interval_usec: int, *, skip_idle: bool = False) -> None: ...
+    def collect(self, stack_frames: _StackFrames, timestamps_us: _Timestamps = None) -> None: ...
+    def export(self, output_path: StrOrBytesPath) -> None: ...
+    def process_frames(self, frames: Sequence[_Frame], thread_id: int, weight: int = 1) -> None: ...
+    def set_stats(
+        self,
+        sample_interval_usec: int,
+        duration_sec: float,
+        sample_rate: float,
+        error_rate: float | None = None,
+        missed_samples: float | None = None,
+        **kwargs: object,
+    ) -> None: ...

stdlib/profiling/sampling/heatmap_collector.pyi

@@ -0,0 +1,11 @@
+from _typeshed import StrOrBytesPath
+from collections.abc import Sequence
+
+from .collector import _Frame, _StackFrames, _Timestamps
+from .stack_collector import StackTraceCollector
+
+class JsonlCollector(StackTraceCollector):
+    def __init__(self, sample_interval_usec: int, *, skip_idle: bool = False, mode: int | None = None) -> None: ...
+    def collect(self, stack_frames: _StackFrames, timestamps_us: _Timestamps = None) -> None: ...
+    def export(self, filename: StrOrBytesPath) -> None: ...
+    def process_frames(self, frames: Sequence[_Frame], _thread_id: int, weight: int = 1) -> None: ...

stdlib/profiling/sampling/jsonl_collector.pyi

@@ -0,0 +1,12 @@
+from _typeshed import StrOrBytesPath
+
+from .collector import Collector, _StackFrames, _Timestamps
+
+class PstatsCollector(Collector):
+    def __init__(self, sample_interval_usec: int, *, skip_idle: bool = False) -> None: ...
+    def collect(self, stack_frames: _StackFrames, timestamps_us: _Timestamps = None) -> None: ...
+    def export(self, filename: StrOrBytesPath) -> None: ...
+    def create_stats(self) -> None: ...
+    def print_stats(
+        self, sort: int = -1, limit: int | None = None, show_summary: bool = True, mode: int | None = None
+    ) -> None: ...

stdlib/profiling/sampling/pstats_collector.pyi

@@ -0,0 +1,33 @@
+from _typeshed import StrOrBytesPath
+from abc import ABCMeta
+from collections.abc import Sequence
+
+from .collector import Collector, _Frame, _StackFrames, _Timestamps
+
+class StackTraceCollector(Collector, metaclass=ABCMeta):
+    def __init__(self, sample_interval_usec: int, *, skip_idle: bool = False) -> None: ...
+    def collect(self, stack_frames: _StackFrames, timestamps_us: _Timestamps = None) -> None: ...
+    def process_frames(self, frames: Sequence[_Frame], thread_id: int, weight: int = 1) -> None: ...
+
+class CollapsedStackCollector(StackTraceCollector):
+    def __init__(self, sample_interval_usec: int, *, skip_idle: bool = False) -> None: ...
+    def process_frames(self, frames: Sequence[_Frame], thread_id: int, weight: int = 1) -> None: ...
+    def export(self, filename: StrOrBytesPath) -> None: ...
+
+class FlamegraphCollector(StackTraceCollector):
+    def __init__(self, sample_interval_usec: int, *, skip_idle: bool = False) -> None: ...
+    def collect(self, stack_frames: _StackFrames, timestamps_us: _Timestamps = None) -> None: ...
+    def set_stats(
+        self,
+        sample_interval_usec: int,
+        duration_sec: float,
+        sample_rate: float,
+        error_rate: float | None = None,
+        missed_samples: float | None = None,
+        mode: int | None = None,
+    ) -> None: ...
+    def export(self, filename: StrOrBytesPath) -> None: ...
+    def process_frames(self, frames: Sequence[_Frame], thread_id: int, weight: int = 1) -> None: ...
+
+class DiffFlamegraphCollector(FlamegraphCollector):
+    def __init__(self, sample_interval_usec: int, *, baseline_binary_path: StrOrBytesPath, skip_idle: bool = False) -> None: ...

stdlib/profiling/sampling/stack_collector.pyi

@@ -0,0 +1,5 @@
+class StringTable:
+    def intern(self, string: object) -> int: ...
+    def get_string(self, index: int) -> str: ...
+    def get_strings(self) -> list[str]: ...
+    def __len__(self) -> int: ...