_remote_debugging.pyi

from collections.abc import Callable, Iterable
from os import PathLike
import sys
from typing import Any, Final, Generic, TypeAlias, final, Self, TypeVar

_T_co = TypeVar("_T_co", covariant=True)

# from _typeshed
# This is an internal CPython type that is like, but subtly different from, a NamedTuple
# Subclasses of this type are found in multiple modules.
# In typeshed, `structseq` is only ever used as a mixin in combination with a fixed-length `Tuple`
# See discussion at #6546 & #6560
# `structseq` classes are unsubclassable, so are all decorated with `@final`.
class structseq(Generic[_T_co]):
    n_fields: Final[int]
    n_unnamed_fields: Final[int]
    n_sequence_fields: Final[int]
    # The first parameter will generally only take an iterable of a specific length.
    # E.g. `os.uname_result` takes any iterable of length exactly 5.
    #
    # The second parameter will accept a dict of any kind without raising an exception,
    # but only has any meaning if you supply it a dict where the keys are strings.
    # https://github.com/python/typeshed/pull/6560#discussion_r767149830
    def __new__(cls, sequence: Iterable[_T_co], dict: dict[str, Any] = ...) -> Self: ...
    if sys.version_info >= (3, 13):
        def __replace__(self, **kwargs: Any) -> Self: ...

StrOrBytesPath: TypeAlias = str | bytes | PathLike[str] | PathLike[bytes]
_Location: TypeAlias = tuple[int, int, int, int] | LocationInfo | None
_Frame: TypeAlias = tuple[str, _Location, str, int | None] | FrameInfo
_Stats: TypeAlias = dict[str, int | float]

PROCESS_VM_READV_SUPPORTED: Final[int]
THREAD_STATUS_GIL_REQUESTED: Final[int]
THREAD_STATUS_HAS_EXCEPTION: Final[int]
THREAD_STATUS_HAS_GIL: Final[int]
THREAD_STATUS_MAIN_THREAD: Final[int]
THREAD_STATUS_ON_CPU: Final[int]
THREAD_STATUS_UNKNOWN: Final[int]

@final
class LocationInfo(structseq[int], tuple[int, int, int, int]):
    __match_args__: Final = ("lineno", "end_lineno", "col_offset", "end_col_offset")
    @property
    def lineno(self) -> int: ...
    @property
    def end_lineno(self) -> int: ...
    @property
    def col_offset(self) -> int: ...
    @property
    def end_col_offset(self) -> int: ...

@final
class FrameInfo(structseq[object], tuple[str, _Location, str, int | None]):
    __match_args__: Final = ("filename", "location", "funcname", "opcode")
    @property
    def filename(self) -> str: ...
    @property
    def location(self) -> _Location: ...
    @property
    def funcname(self) -> str: ...
    @property
    def opcode(self) -> int | None: ...

@final
class CoroInfo(structseq[object], tuple[list[_Frame], int | str]):
    __match_args__: Final = ("call_stack", "task_name")
    @property
    def call_stack(self) -> list[_Frame]: ...
    @property
    def task_name(self) -> int | str: ...

@final
class TaskInfo(structseq[object], tuple[int, str, list[CoroInfo], list[CoroInfo]]):
    __match_args__: Final = ("task_id", "task_name", "coroutine_stack", "awaited_by")
    @property
    def task_id(self) -> int: ...
    @property
    def task_name(self) -> str: ...
    @property
    def coroutine_stack(self) -> list[CoroInfo]: ...
    @property
    def awaited_by(self) -> list[CoroInfo]: ...

@final
class ThreadInfo(structseq[object], tuple[int, int, list[_Frame]]):
    __match_args__: Final = ("thread_id", "status", "frame_info")
    @property
    def thread_id(self) -> int: ...
    @property
    def status(self) -> int: ...
    @property
    def frame_info(self) -> list[_Frame]: ...

@final
class InterpreterInfo(structseq[object], tuple[int, list[ThreadInfo]]):
    __match_args__: Final = ("interpreter_id", "threads")
    @property
    def interpreter_id(self) -> int: ...
    @property
    def threads(self) -> list[ThreadInfo]: ...

@final
class AwaitedInfo(structseq[object], tuple[int, list[TaskInfo]]):
    __match_args__: Final = ("thread_id", "awaited_by")
    @property
    def thread_id(self) -> int: ...
    @property
    def awaited_by(self) -> list[TaskInfo]: ...

@final
class GCStatsInfo(structseq[object], tuple[int, int, int, int, int, int, int, int, int, float]):
    __match_args__: Final = (
        "gen",
        "iid",
        "ts_start",
        "ts_stop",
        "collections",
        "collected",
        "uncollectable",
        "candidates",
        "heap_size",
        "duration",
    )
    @property
    def gen(self) -> int: ...
    @property
    def iid(self) -> int: ...
    @property
    def ts_start(self) -> int: ...
    @property
    def ts_stop(self) -> int: ...
    @property
    def collections(self) -> int: ...
    @property
    def collected(self) -> int: ...
    @property
    def uncollectable(self) -> int: ...
    @property
    def candidates(self) -> int: ...
    @property
    def heap_size(self) -> int: ...
    @property
    def duration(self) -> float: ...

@final
class RemoteUnwinder:
    def __init__(
        self,
        pid: int,
        *,
        all_threads: bool = False,
        only_active_thread: bool = False,
        mode: int = 0,
        debug: bool = False,
        skip_non_matching_threads: bool = True,
        native: bool = False,
        gc: bool = False,
        opcodes: bool = False,
        cache_frames: bool = False,
        stats: bool = False,
    ) -> None: ...
    def get_stack_trace(self) -> list[InterpreterInfo]: ...
    def get_all_awaited_by(self) -> list[AwaitedInfo]: ...
    def get_async_stack_trace(self) -> list[AwaitedInfo]: ...
    def get_stats(self) -> _Stats: ...
    def pause_threads(self) -> bool: ...
    def resume_threads(self) -> bool: ...

@final
class GCMonitor:
    def __init__(self, pid: int, *, debug: bool = False) -> None: ...
    def get_gc_stats(self, all_interpreters: bool = False) -> list[GCStatsInfo]: ...

@final
class BinaryWriter:
    def __init__(
        self, filename: StrOrBytesPath, sample_interval_us: int, start_time_us: int, *, compression: int = 0
    ) -> None: ...
    @property
    def total_samples(self) -> int: ...
    def write_sample(self, stack_frames: list[InterpreterInfo], timestamp_us: int) -> None: ...
    def finalize(self) -> None: ...
    def close(self) -> None: ...
    def __enter__(self) -> Self: ...
    def __exit__(self, exc_type: object = None, exc_val: object = None, exc_tb: object = None) -> bool: ...
    def get_stats(self) -> _Stats: ...

@final
class BinaryReader:
    def __init__(self, filename: StrOrBytesPath) -> None: ...
    @property
    def sample_count(self) -> int: ...
    @property
    def sample_interval_us(self) -> int: ...
    def replay(self, collector: object, progress_callback: Callable[[int, int], object] | None = None) -> int: ...
    def get_info(self) -> dict[str, object]: ...
    def get_stats(self) -> _Stats: ...
    def close(self) -> None: ...
    def __enter__(self) -> Self: ...
    def __exit__(self, exc_type: object = None, exc_val: object = None, exc_tb: object = None) -> bool: ...

def zstd_available() -> bool: ...
def get_child_pids(pid: int, *, recursive: bool = True) -> list[int]: ...
def is_python_process(pid: int) -> bool: ...
def get_gc_stats(pid: int, *, all_interpreters: bool = False) -> list[GCStatsInfo]: ...