.config/coc/extensions/node_modules/coc-pyright/node_modules/pyright/dist/typeshed-fallback/stdlib/multiprocessing/sharedctypes.pyi

   1 import ctypes
   2 from collections.abc import Callable, Iterable, Sequence
   3 from ctypes import _CData, _SimpleCData, c_char
   4 from multiprocessing.context import BaseContext
   5 from multiprocessing.synchronize import _LockLike
   6 from typing import Any, Generic, Protocol, Type, TypeVar, overload
   7 from typing_extensions import Literal
   8
   9 _T = TypeVar("_T")
  10 _CT = TypeVar("_CT", bound=_CData)
  11
  12 @overload
  13 def RawValue(typecode_or_type: Type[_CT], *args: Any) -> _CT: ...
  14 @overload
  15 def RawValue(typecode_or_type: str, *args: Any) -> Any: ...
  16 @overload
  17 def RawArray(typecode_or_type: Type[_CT], size_or_initializer: int | Sequence[Any]) -> ctypes.Array[_CT]: ...
  18 @overload
  19 def RawArray(typecode_or_type: str, size_or_initializer: int | Sequence[Any]) -> Any: ...
  20 @overload
  21 def Value(typecode_or_type: Type[_CT], *args: Any, lock: Literal[False], ctx: BaseContext | None = ...) -> _CT: ...
  22 @overload
  23 def Value(
  24     typecode_or_type: Type[_CT], *args: Any, lock: Literal[True] | _LockLike, ctx: BaseContext | None = ...
  25 ) -> SynchronizedBase[_CT]: ...
  26 @overload
  27 def Value(
  28     typecode_or_type: str, *args: Any, lock: Literal[True] | _LockLike, ctx: BaseContext | None = ...
  29 ) -> SynchronizedBase[Any]: ...
  30 @overload
  31 def Value(
  32     typecode_or_type: str | Type[_CData], *args: Any, lock: bool | _LockLike = ..., ctx: BaseContext | None = ...
  33 ) -> Any: ...
  34 @overload
  35 def Array(
  36     typecode_or_type: Type[_CT], size_or_initializer: int | Sequence[Any], *, lock: Literal[False], ctx: BaseContext | None = ...
  37 ) -> _CT: ...
  38 @overload
  39 def Array(
  40     typecode_or_type: Type[_CT],
  41     size_or_initializer: int | Sequence[Any],
  42     *,
  43     lock: Literal[True] | _LockLike,
  44     ctx: BaseContext | None = ...,
  45 ) -> SynchronizedArray[_CT]: ...
  46 @overload
  47 def Array(
  48     typecode_or_type: str,
  49     size_or_initializer: int | Sequence[Any],
  50     *,
  51     lock: Literal[True] | _LockLike,
  52     ctx: BaseContext | None = ...,
  53 ) -> SynchronizedArray[Any]: ...
  54 @overload
  55 def Array(
  56     typecode_or_type: str | Type[_CData],
  57     size_or_initializer: int | Sequence[Any],
  58     *,
  59     lock: bool | _LockLike = ...,
  60     ctx: BaseContext | None = ...,
  61 ) -> Any: ...
  62 def copy(obj: _CT) -> _CT: ...
  63 @overload
  64 def synchronized(obj: _SimpleCData[_T], lock: _LockLike | None = ..., ctx: Any | None = ...) -> Synchronized[_T]: ...
  65 @overload
  66 def synchronized(obj: ctypes.Array[c_char], lock: _LockLike | None = ..., ctx: Any | None = ...) -> SynchronizedString: ...
  67 @overload
  68 def synchronized(obj: ctypes.Array[_CT], lock: _LockLike | None = ..., ctx: Any | None = ...) -> SynchronizedArray[_CT]: ...
  69 @overload
  70 def synchronized(obj: _CT, lock: _LockLike | None = ..., ctx: Any | None = ...) -> SynchronizedBase[_CT]: ...
  71
  72 class _AcquireFunc(Protocol):
  73     def __call__(self, block: bool = ..., timeout: float | None = ...) -> bool: ...
  74
  75 class SynchronizedBase(Generic[_CT]):
  76     acquire: _AcquireFunc
  77     release: Callable[[], None]
  78     def __init__(self, obj: Any, lock: _LockLike | None = ..., ctx: Any | None = ...) -> None: ...
  79     def __reduce__(self) -> tuple[Callable[..., Any], tuple[Any, _LockLike]]: ...
  80     def get_obj(self) -> _CT: ...
  81     def get_lock(self) -> _LockLike: ...
  82     def __enter__(self) -> bool: ...
  83     def __exit__(self, *args: Any) -> None: ...
  84
  85 class Synchronized(SynchronizedBase[_SimpleCData[_T]], Generic[_T]):
  86     value: _T
  87
  88 class SynchronizedArray(SynchronizedBase[ctypes.Array[_CT]], Generic[_CT]):
  89     def __len__(self) -> int: ...
  90     def __getitem__(self, i: int) -> _CT: ...
  91     def __setitem__(self, i: int, o: _CT) -> None: ...
  92     def __getslice__(self, start: int, stop: int) -> list[_CT]: ...
  93     def __setslice__(self, start: int, stop: int, values: Iterable[_CT]) -> None: ...
  94
  95 class SynchronizedString(SynchronizedArray[c_char]):
  96     value: bytes
  97     raw: bytes