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