import ctypes
import gc
import sys
import weakref
from abc import ABC, abstractmethod
from contextlib import suppress
from typing import (Any, Generic, Iterator, Optional, Tuple, Type, TypeVar,
Union)
from _pointers import add_ref, remove_ref
from typing_extensions import final
from ._utils import deref, force_set_attr, move_to_mem
from .exceptions import DereferenceError, FreedMemoryError, NullPointerError
from .util import NULL, Nullable, handle
__all__ = (
"BasePointer",
"BaseObjectPointer",
"BasicPointer",
"BaseCPointer",
"BaseAllocatedPointer",
"Dereferencable",
"IterDereferencable",
)
T = TypeVar("T")
A = TypeVar("A", bound="BasicPointer")
class BasicPointer(ABC):
"""Base class representing a pointer with no operations."""
@property
@abstractmethod
def address(self) -> Optional[int]:
"""Address that the pointer is looking at."""
...
@abstractmethod
def __repr__(self) -> str:
...
@final
def __str__(self) -> str:
return f"{type(self).__name__}({hex(self.address or 0)})"
@abstractmethod
def _cleanup(self) -> None:
...
@final
def __eq__(self, data: object) -> bool:
if not isinstance(data, BasePointer):
return False
return data.address == self.address
@final
def ensure(self) -> int:
"""Ensure that the pointer is not null.
Raises:
NullPointerError: Address of pointer is `None`
Returns:
Address of the pointer.
Example:
```py
ptr = to_ptr(NULL)
address = ptr.ensure() # NullPointerError
ptr >>= 1
address = ptr.ensure() # works just fine
```"""
if not self.address:
raise NullPointerError("pointer is NULL")
return self.address
class Movable(ABC, Generic[T, A]):
@abstractmethod
def move(
self,
data: Union[A, T],
*,
unsafe: bool = False,
) -> None:
"""Move/copy a value into the memory at the pointers address."""
...
def __ilshift__(self, data: Union[A, T]):
self.move(data)
return self
def __ixor__(self, data: Union[A, T]):
self.move(data, unsafe=True)
return self
class Dereferencable(ABC, Generic[T]):
"""Abstract class for an object that may be dereferenced."""
@abstractmethod
def dereference(self) -> T:
"""Dereference the pointer.
Returns:
Value at the pointers address."""
...
@final
def __invert__(self) -> T:
return self.dereference()
class IterDereferencable(Dereferencable[T], Generic[T]):
"""
Abstract class for an object that may be dereferenced via * (`__iter__`)
"""
def __iter__(self) -> Iterator[T]:
return iter({self.dereference()})
class BasePointer(
Dereferencable[T],
Movable[T, "BasePointer[T]"],
BasicPointer,
ABC,
Generic[T],
):
"""Base class representing a pointer."""
@abstractmethod
def __repr__(self) -> str:
...
@abstractmethod
def _cleanup(self) -> None:
...
class Sized(BasicPointer, ABC):
"""Base class for a pointer that has a size attribute."""
@property
@abstractmethod
def size(self) -> int:
"""Size of the target value."""
...
@handle
@final
def make_ct_pointer(self) -> "ctypes._PointerLike":
"""Convert the address to a ctypes pointer.
Returns:
The created ctypes pointer.
"""
return ctypes.cast(
self.ensure(),
ctypes.POINTER(ctypes.c_char * self.size),
)
@abstractmethod
def _make_stream_and_ptr(
self,
size: int,
address: int,
) -> Tuple["ctypes._PointerLike", bytes]:
...
class BaseObjectPointer(
IterDereferencable[T],
BasePointer[T],
ABC,
):
"""Abstract class for a pointer to a Python object."""
def __init__(
self,
address: Optional[int],
increment_ref: bool = False,
) -> None:
"""
Args:
address: Address of the underlying value.
increment_ref: Should the reference count on the target object get incremented.
""" # noqa
self._address: Optional[int] = address
if increment_ref and address:
add_ref(~self)
self._origin_size = sys.getsizeof(~self if address else None)
weakref.finalize(self, self._cleanup)
@handle
def set_attr(self, key: str, value: Any) -> None:
"""Force setting an attribute on the object the pointer is looking at.""" # noqa
v: Any = ~self # mypy gets angry if this isnt any
if not isinstance(~self, type):
v = type(v)
force_set_attr(v, key, value)
@handle
def assign(
self,
target: Nullable[Union["BaseObjectPointer[T]", T]],
) -> None:
"""Point to a new address.
Args:
target: New pointer or value to look at.
"""
if target is NULL:
self._address = None
return
new: BasePointer[T] = self._get_ptr(target) # type: ignore
if not isinstance(new, BaseObjectPointer):
raise ValueError(
"can only point to object pointer",
)
with suppress(NullPointerError):
remove_ref(~self)
self._address = new.address
add_ref(~self)
@property
def address(self) -> Optional[int]:
return self._address
@handle
def dereference(self) -> T:
return deref(self.ensure())
def __irshift__(
self,
value: Nullable[Union["BaseObjectPointer[T]", T]],
):
self.assign(value)
return self
@classmethod
@abstractmethod
def make_from(cls, obj: T) -> "BaseObjectPointer[T]":
"""Create a new instance of the pointer.
Args:
obj: Object to create pointer to.
Returns:
Created pointer.
Example:
```py
ptr = Pointer.make_from(1)
```"""
...
@classmethod
def _get_ptr(cls, obj: Union[T, "BasePointer[T]"]) -> "BasePointer[T]":
return (
obj
if isinstance(
obj,
BasePointer,
)
else cls.make_from(obj)
)
def _cleanup(self) -> None:
if self.address:
remove_ref(~self)
class BaseCPointer(
Movable[T, "BaseCPointer[T]"],
IterDereferencable[T],
Sized,
ABC,
):
def __init__(self, address: int, size: int):
self._address = address
self._size = size
weakref.finalize(self, self._cleanup)
@property
def address(self) -> Optional[int]:
return self._address
def _make_stream_and_ptr(
self,
size: int,
address: int,
) -> Tuple["ctypes._PointerLike", bytes]:
bytes_a = (ctypes.c_ubyte * size).from_address(address)
return self.make_ct_pointer(), bytes(bytes_a)
@handle
def move(
self,
data: Union["BaseCPointer[T]", T],
*,
unsafe: bool = False,
) -> None:
"""Move data to the target address."""
if not isinstance(data, BaseCPointer):
raise ValueError(
f'"{type(data).__name__}" object is not a valid C pointer',
)
ptr, byte_stream = self._make_stream_and_ptr(
data.size,
data.ensure(),
)
move_to_mem(ptr, byte_stream, unsafe=unsafe, target="C data")
def __ilshift__(self, data: Union["BaseCPointer[T]", T]):
self.move(data)
return self
def __ixor__(self, data: Union["BaseCPointer[T]", T]):
self.move(data, unsafe=True)
return self
@handle
def make_ct_pointer(self):
"""Turn the pointer into a ctypes pointer."""
return ctypes.cast(
self.ensure(),
ctypes.POINTER(ctypes.c_char * self.size),
)
@abstractmethod
def _as_parameter_(self) -> "ctypes._CData":
"""Convert the data into something that ctypes understands."""
...
@abstractmethod
def _cleanup(self) -> None:
...
class BaseAllocatedPointer(BasePointer[T], Sized, ABC):
@property
@abstractmethod
def address(self) -> Optional[int]:
...
@address.setter
def address(self, value: int) -> None:
...
@property
@abstractmethod
def freed(self) -> bool:
...
@freed.setter
def freed(self, value: bool) -> None:
self._freed = value
@property
def assigned(self) -> bool:
"""Whether the allocated memory has been assigned a value."""
return self._assigned
@assigned.setter
def assigned(self, value: bool) -> None:
self._assigned = value
@handle
def move(
self,
data: Union[BasePointer[T], T],
unsafe: bool = False,
) -> None:
add_ref(data)
self.ensure_valid()
from .object_pointer import to_ptr
data_ptr = data if isinstance(data, BasePointer) else to_ptr(data)
if (sys.version_info.minor >= 11) and (gc.is_tracked(~data_ptr)):
remove_ref(data)
raise RuntimeError("allocation on tracked types is not supported on 3.11+")
ptr, byte_stream = self._make_stream_and_ptr(
sys.getsizeof(~data_ptr),
data_ptr.ensure(),
)
move_to_mem(ptr, byte_stream, unsafe=unsafe)
self.assigned = True
remove_ref(data)
@handle
def dereference(self) -> T:
if self.freed:
raise FreedMemoryError(
"cannot dereference memory that has been freed",
)
if not self.assigned:
raise DereferenceError(
"cannot dereference allocated memory that has no value",
)
return deref(self.ensure())
@abstractmethod
def __add__(self, amount: int) -> "BaseAllocatedPointer[Any]":
...
@abstractmethod
def __sub__(self, amount: int) -> "BaseAllocatedPointer[Any]":
...
def _cleanup(self) -> None:
pass
def _make_stream_and_ptr(
self,
size: int,
address: int,
) -> Tuple["ctypes._PointerLike", bytes]:
if self.freed:
raise FreedMemoryError("memory has been freed")
bytes_a = (ctypes.c_ubyte * size).from_address(address) # fmt: off
return self.make_ct_pointer(), bytes(bytes_a)
@abstractmethod
def free(self) -> None:
"""Free the memory."""
...
def ensure_valid(self) -> None:
"""Ensure the memory has not been freed."""
if self.freed:
raise FreedMemoryError(
f"{self} has been freed",
)
@property
def size(self) -> int:
return self._size
@size.setter
def size(self, value: int) -> None:
self._size = value