[ Web Proxy ]
URL:
Viewing: https://raw.githubusercontent.com/ZeroIntensity/pointers.py/master/src/pointers/c_pointer.py [Back]  [Original]

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

Web Proxy Viewer  |  New URL  |  Original Page