from __future__ import annotations
import ctypes
from abc import ABC, abstractmethod
from typing import (TYPE_CHECKING, Any, Callable, Iterator, List, Optional,
Type, TypeVar)
from _pointers import add_ref, remove_ref
from typing_extensions import ParamSpec
from ._utils import deref, get_mapped, map_type
from .base_pointers import BaseCPointer, IterDereferencable
from .util import handle
if TYPE_CHECKING:
from .structure import Struct, StructPointer
T = TypeVar("T")
A = TypeVar("A", bound="Struct")
P = ParamSpec("P")
__all__ = (
"TypedCPointer",
"VoidPointer",
"to_c_ptr",
"cast",
"to_voidp",
"array",
"to_struct_ptr",
"to_func_ptr",
)
class VoidPointer(BaseCPointer[Any]):
"""Class representing a void pointer to a C object."""
@property
def size(self) -> int:
return self._size
@property # type: ignore
@handle
def _as_parameter_(self) -> ctypes.c_void_p:
return ctypes.c_void_p(self.address)
@handle
def dereference(self) -> Optional[int]:
deref = ctypes.c_void_p.from_address(self.ensure())
return deref.value
def __repr__(self) -> str:
return f"VoidPointer(address={self.address}, size={self.size})"
def _cleanup(self) -> None:
pass
class _CDeref(IterDereferencable[T], ABC):
@abstractmethod
def address(self) -> Optional[int]:
...
@property
@abstractmethod
def decref(self) -> bool:
"""Whether the target objects reference count should be decremented when the pointer is garbage collected.""" # noqa
...
@handle
def _cleanup(self) -> None:
if self.address:
if (type(~self) is not str) and (self.decref):
remove_ref(~self)
class FunctionPointer(BaseCPointer[Callable[P, T]]): # type: ignore
def __init__(self, address: int) -> None:
cb = deref(address)
add_ref(cb)
self._decref = True
self._address = address
mapped: "ctypes._FuncPointer" = get_mapped(cb) # type: ignore
@ctypes.CFUNCTYPE(mapped._restype_, *mapped._argtypes_) # type: ignore
def transport(*args, **kwargs):
cb(*args, **kwargs)
add_ref(transport)
self._transport = transport
self._size = ctypes.sizeof(transport)
@handle
def _cleanup(self) -> None:
if self.address:
remove_ref(~self)
remove_ref(self._transport)
@property
def _as_parameter_(self) -> "ctypes._CData":
return self._transport
@property
def size(self) -> int:
return self._size
@property
def type(self) -> Type[Callable[P, T]]:
return type(~self)
@handle
def dereference(self) -> Callable[P, T]:
return deref(self.ensure())
@handle # type: ignore
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> T:
return (~self)(*args, **kwargs)
def __repr__(self) -> str:
return f"FunctionPointer(address={self.address})"
class TypedCPointer(_CDeref[T], BaseCPointer[T]):
"""Class representing a pointer with a known type."""
def __init__(
self,
address: int,
data_type: Type[T],
size: int,
void_p: bool = True,
decref: bool = True,
alt: bool = False,
) -> None:
self._void_p = void_p
super().__init__(address, size)
self._type = data_type
self._decref = decref
self.alt = alt
@property
def size(self) -> int:
return self._size
@property
def decref(self) -> bool:
return self._decref
@property
def type(self):
return self._type
@property
def address(self) -> Optional[int]:
return self._address
@property # type: ignore
@handle
def _as_parameter_(self) -> ctypes._CData:
ctype = get_mapped(self.type)
if (ctype is ctypes.c_char_p) and (self.alt):
deref = ctype(self.ensure())
return deref
else:
deref = ctype.from_address(self.ensure())
value = deref.value # type: ignore
if isinstance(value, (TypedCPointer, VoidPointer)):
return ctypes.pointer(value._as_parameter_) # type: ignore
return ctypes.pointer(deref)
@handle
def dereference(self) -> T:
"""Dereference the pointer."""
ctype = get_mapped(self.type)
if (ctype is ctypes.c_char_p) and (self.alt):
res = ctypes.c_char_p(self.ensure()).value
return res # type: ignore
ptr = (
ctype.from_address(self.ensure())
if not self._void_p
else ctypes.cast(
ctypes.c_void_p(self.address), ctypes.POINTER(ctype)
) # fmt: off
)
return ptr.value if not self._void_p else ptr.contents.value # type: ignore # noqa
def __iter__(self) -> Iterator[T]:
"""Dereference the pointer."""
return iter({self.dereference()})
def __repr__(self) -> str:
return f"TypedCPointer(address={self.address}, size={self.size})"
class CArrayPointer(_CDeref[List[T]], BaseCPointer[List[T]]):
"""Class representing a pointer to a C array."""
def __init__(
self,
address: int,
size: int,
length: int,
typ: Type[T],
):
self._length = length
self._type = typ
self._decref = False
super().__init__(address, size)
@property
def decref(self) -> bool:
return self._decref
@property
@handle
def _as_parameter_(self) -> "ctypes.Array[ctypes._CData]":
ctype = get_mapped(self._type)
deref = (ctype * self._length).from_address(self.ensure())
return deref
@handle
def dereference(self) -> List[T]:
"""Dereference the pointer."""
array = self._as_parameter_
return [array[i] for i in range(self._length)] # type: ignore
def __repr__(self) -> str:
return f"CArrayPointer(address={self.address}, size={self.size})"
def __getitem__(self, index: int) -> T:
array = ~self
return array[index]
@handle
def cast(ptr: VoidPointer, data_type: Type[T]) -> TypedCPointer[T]:
"""Cast a void pointer to a typed pointer."""
return TypedCPointer(
ptr.ensure(),
data_type,
ptr.size,
decref=False,
void_p=True,
alt=True,
)
def to_voidp(ptr: TypedCPointer[Any]) -> VoidPointer:
"""Cast a typed pointer to a void pointer."""
return VoidPointer(ptr.ensure(), ptr.size)
def to_c_ptr(data: T) -> TypedCPointer[T]:
"""Convert a python type to a pointer to a C type."""
ct = map_type(data)
add_ref(ct)
address = ctypes.addressof(ct)
typ = type(data)
return TypedCPointer(address, typ, ctypes.sizeof(ct), False)
def to_struct_ptr(struct: A) -> "StructPointer[A]":
"""Convert a struct to a pointer."""
from .structure import StructPointer
return StructPointer(id(struct))
@handle
def array(*seq: T) -> CArrayPointer[T]:
f_type = type(seq[0])
for i in seq:
if type(i) is not f_type: # dont use isinstance here
raise ValueError(
"all values in the array must be the same type",
)
length = len(seq)
ctype = get_mapped(f_type)
arr = (ctype * length)(*seq) # type: ignore
add_ref(arr)
return CArrayPointer(
ctypes.addressof(arr),
ctypes.sizeof(arr),
length,
f_type, # type: ignore
)
def to_func_ptr(fn: Callable[P, T]) -> FunctionPointer[P, T]:
return FunctionPointer(id(fn)) # type: ignore