Reference
NULL
Unique object representing a NULL address.
May be used with object pointers or passed to bindings.
Source code in src/pointers/util.py
class NULL:
"""Unique object representing a NULL address.
May be used with object pointers or passed to bindings.
"""
handle(func)
Handle segment violation errors when called.
src/pointers/util.py
def handle(func: Callable[P, T]) -> Callable[P, T]:
"""Handle segment violation errors when called."""
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
try:
faulthandler.disable()
call = _handle(func, args, kwargs)
with suppress(UnsupportedOperation):
faulthandler.enable()
return call
except (RuntimeError, OSError) as e:
msg = str(e)
if not any(
{
msg.startswith("segment violation"),
msg.startswith("exception: access violation"),
}
):
raise
with suppress(UnsupportedOperation):
faulthandler.enable()
raise SegmentViolation(msg) from None
return wrapper
raw_type(ct)
Set a raw ctypes type for a struct.
src/pointers/util.py
def raw_type(ct: Type["ctypes._CData"]) -> Any:
"""Set a raw ctypes type for a struct."""
return RawType(ct)
struct_cast(ptr)
Cast a Struct
or StructPointer
to a Python object.
src/pointers/util.py
@handle
def struct_cast(ptr: Union["Struct", "StructPointer"]) -> Any:
"""Cast a `Struct` or `StructPointer` to a Python object."""
return ctypes.cast(ptr.get_existing_address(), ctypes.py_object).value
BaseAllocatedPointer
Bases: BasePointer[T]
, Sized
, ABC
Source code in src/pointers/base_pointers.py
class BaseAllocatedPointer(BasePointer[T], Sized, ABC):
@property
@abstractmethod
def address(self) -> Optional[int]:
...
@address.setter
def address(self, value: int) -> None:
...
@property
def freed(self) -> bool:
"""Whether the allocated memory has been freed."""
return self._freed
@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)
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
assigned: bool
property
writable
Whether the allocated memory has been assigned a value.
freed: bool
property
writable
Whether the allocated memory has been freed.
ensure_valid()
Ensure the memory has not been freed.
src/pointers/base_pointers.py
def ensure_valid(self) -> None:
"""Ensure the memory has not been freed."""
if self.freed:
raise FreedMemoryError(
f"{self} has been freed",
)
free()
abstractmethod
Free the memory.
src/pointers/base_pointers.py
@abstractmethod
def free(self) -> None:
"""Free the memory."""
...
BaseCPointer
Bases: IterDereferencable[T]
, Movable[T, 'BaseCPointer[T]']
, BasicPointer
, Sized
, ABC
Source code in src/pointers/base_pointers.py
class BaseCPointer(
IterDereferencable[T],
Movable[T, "BaseCPointer[T]"],
BasicPointer,
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):
return ctypes.cast(
self.ensure(),
ctypes.POINTER(ctypes.c_char * self.size),
)
@abstractmethod
def _as_parameter_(self) -> "ctypes._CData":
"""Convert the pointer to a ctypes pointer."""
...
@abstractmethod
def _cleanup(self) -> None:
...
move(data, *, unsafe=False)
Move data to the target address.
src/pointers/base_pointers.py
@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")
BaseObjectPointer
Bases: Typed[Type[T]]
, IterDereferencable[T]
, BasePointer[T]
, ABC
Abstract class for a pointer to a Python object.
Source code in src/pointers/base_pointers.py
class BaseObjectPointer(
Typed[Type[T]],
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)
@property
def type(self) -> Type[T]:
warnings.warn(
"BaseObjectPointer.type is deprecated, please use type(~ptr) instead", # noqa
DeprecationWarning,
)
return type(~self)
@handle
def set_attr(self, key: str, value: Any) -> None:
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)
__init__(address, increment_ref=False)
Parameters: |
|
---|
src/pointers/base_pointers.py
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)
assign(target)
Point to a new address.
Parameters: |
|
---|
src/pointers/base_pointers.py
@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)
make_from(obj)
abstractmethod
classmethod
Create a new instance of the pointer.
Parameters: |
|
---|
Returns: |
|
---|
Example
ptr = Pointer.make_from(1)
src/pointers/base_pointers.py
@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)
```"""
...
BasePointer
Bases: Dereferencable[T]
, Movable[T, 'BasePointer[T]']
, BasicPointer
, ABC
, Generic[T]
Base class representing a pointer.
Source code in src/pointers/base_pointers.py
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:
...
BasicPointer
Bases: ABC
Base class representing a pointer with no operations.
Source code in src/pointers/base_pointers.py
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
address: Optional[int]
abstractmethod
property
Address that the pointer is looking at.
ensure()
Ensure that the pointer is not null.
Raises: |
|
---|
Returns: |
|
---|
Example
ptr = to_ptr(NULL)
address = ptr.ensure() # NullPointerError
ptr >>= 1
address = ptr.ensure() # works just fine
src/pointers/base_pointers.py
@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
Dereferencable
Bases: ABC
, Generic[T]
Abstract class for an object that may be dereferenced.
Source code in src/pointers/base_pointers.py
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:
"""Dereference the pointer."""
return self.dereference()
__invert__()
Dereference the pointer.
src/pointers/base_pointers.py
@final
def __invert__(self) -> T:
"""Dereference the pointer."""
return self.dereference()
dereference()
abstractmethod
Dereference the pointer.
Returns: |
|
---|
src/pointers/base_pointers.py
@abstractmethod
def dereference(self) -> T:
"""Dereference the pointer.
Returns:
Value at the pointers address."""
...
IterDereferencable
Bases: Dereferencable[T]
, Generic[T]
Abstract class for an object that may be dereferenced via * (__iter__
)
Source code in src/pointers/base_pointers.py
class IterDereferencable(Dereferencable[T], Generic[T]):
"""
Abstract class for an object that may be dereferenced via * (`__iter__`)
"""
def __iter__(self) -> Iterator[T]:
"""Dereference the pointer."""
return iter({self.dereference()})
__iter__()
Dereference the pointer.
src/pointers/base_pointers.py
def __iter__(self) -> Iterator[T]:
"""Dereference the pointer."""
return iter({self.dereference()})
Sized
Bases: ABC
Base class for a pointer that has a size attribute.
Source code in src/pointers/base_pointers.py
class Sized(ABC):
"""Base class for a pointer that has a size attribute."""
@abstractmethod
def ensure(self) -> int:
...
@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]:
...
size: int
abstractmethod
property
Size of the target value.
make_ct_pointer()
Convert the address to a ctypes pointer.
Returns: |
|
---|
src/pointers/base_pointers.py
@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),
)
Typed
Bases: ABC
, Generic[T]
Base class for a pointer that has a type attribute.
Source code in src/pointers/base_pointers.py
class Typed(ABC, Generic[T]):
"""Base class for a pointer that has a type attribute."""
@property
@abstractmethod
def type(self) -> T:
"""Type of the value at the address."""
...
type: T
abstractmethod
property
Type of the value at the address.
CArrayPointer
Bases: _CDeref[List[T]]
, Typed[Type[T]]
, BaseCPointer[List[T]]
Class representing a pointer to a C array.
Source code in src/pointers/c_pointer.py
class CArrayPointer(_CDeref[List[T]], Typed[Type[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
def type(self) -> Type[T]: # type: ignore
return self._type
@property # type: ignore
@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]
dereference()
Dereference the pointer.
src/pointers/c_pointer.py
@handle
def dereference(self) -> List[T]:
"""Dereference the pointer."""
array = self._as_parameter_
return [array[i] for i in range(self._length)] # type: ignore
TypedCPointer
Bases: _CDeref[T]
, Typed[T]
, BaseCPointer[T]
Class representing a pointer with a known type.
Source code in src/pointers/c_pointer.py
class TypedCPointer(_CDeref[T], Typed[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):
ctype = get_mapped(self.type)
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})"
__iter__()
Dereference the pointer.
src/pointers/c_pointer.py
def __iter__(self) -> Iterator[T]:
"""Dereference the pointer."""
return iter({self.dereference()})
dereference()
Dereference the pointer.
src/pointers/c_pointer.py
@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
VoidPointer
Bases: BaseCPointer[Any]
Class representing a void pointer to a C object.
Source code in src/pointers/c_pointer.py
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
cast(ptr, data_type)
Cast a void pointer to a typed pointer.
src/pointers/c_pointer.py
@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,
)
to_c_ptr(data)
Convert a python type to a pointer to a C type.
src/pointers/c_pointer.py
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)
to_struct_ptr(struct)
Convert a struct to a pointer.
src/pointers/c_pointer.py
def to_struct_ptr(struct: A) -> "StructPointer[A]":
"""Convert a struct to a pointer."""
from .structure import StructPointer
return StructPointer(id(struct))
to_voidp(ptr)
Cast a typed pointer to a void pointer.
src/pointers/c_pointer.py
def to_voidp(ptr: TypedCPointer[Any]) -> VoidPointer:
"""Cast a typed pointer to a void pointer."""
return VoidPointer(ptr.ensure(), ptr.size)
decay(func)
Automatically convert values to pointers when called.
Example
@decay
def my_function(a: str, b: Pointer[str]):
print(a, *c)
my_function('a', 'b')
src/pointers/decay.py
def decay(func: Callable[P, T]) -> Callable[..., T]:
"""Automatically convert values to pointers when called.
Example:
```py
@decay
def my_function(a: str, b: Pointer[str]):
print(a, *c)
my_function('a', 'b')
```
"""
@wraps(func)
def inner(*args: P.args, **kwargs: P.kwargs) -> T:
actual = _decay_params(func, args, kwargs)
return func(**actual) # type: ignore
return inner
decay_annotated(func)
Example
@decay_annotated
def my_function(a: str, b: Annotated[str, Pointer]):
print(a, *c)
my_function('a', 'b')
src/pointers/decay.py
def decay_annotated(func: Callable[P, T]) -> Callable[P, T]:
"""
Example:
```py
@decay_annotated
def my_function(a: str, b: Annotated[str, Pointer]):
print(a, *c)
my_function('a', 'b')
```
"""
@wraps(func)
def wrapped(*args: P.args, **kwargs: P.kwargs):
hints, actual = _make_func_params(func, args, kwargs)
for param, hint in hints.items():
if get_origin(hint) is not Annotated:
continue
hint_arg = get_args(hint)[1]
if (hint_arg is Pointer) or (get_origin(hint_arg) is Pointer):
actual[param] = to_ptr(actual[param])
return func(**actual) # type: ignore
return wrapped
decay_wrapped(_)
Example
def my_function_wrapper(a: str, b: str, c: str) -> None:
...
@decay_wrapped(my_function_wrapper)
def my_function(a: str, b: str, c: Pointer[str]):
print(a, b, *c)
print(a, b, ~c)
my_function('a', 'b', 'c')
src/pointers/decay.py
def decay_wrapped(_: Callable[P, T]) -> Callable[..., Callable[P, T]]:
"""
Example:
```py
def my_function_wrapper(a: str, b: str, c: str) -> None:
...
@decay_wrapped(my_function_wrapper)
def my_function(a: str, b: str, c: Pointer[str]):
print(a, b, *c)
print(a, b, ~c)
my_function('a', 'b', 'c')
```
"""
def decorator(func: Callable[..., T]) -> Callable[P, T]:
@wraps(func)
def wrapped(*args: P.args, **kwargs: P.kwargs):
actual = _decay_params(func, args, kwargs)
return func(**actual)
return wrapped
return decorator # type: ignore
Struct
Abstract class representing a struct.
Source code in src/pointers/structure.py
class Struct:
"""Abstract class representing a struct."""
_hints: Dict[str, Any]
_void_p: List[str]
_internal_struct: Type[ctypes.Structure]
@classmethod
def _convert_tc_ptr(cls, typ: Any, name: str):
if typ is TypedCPointer:
raise TypeError(
"cannot instantiate: TypedCPointer has no type argument",
)
if getattr(typ, "__origin__", None) is TypedCPointer:
setattr(
cls,
name,
RawType(
ctypes.POINTER(get_mapped(typ.__args__[0])),
),
)
return typ
@classmethod
def _get_type(
cls,
ct: Type["ctypes._CData"],
name: str,
) -> Type["ctypes._CData"]:
attr = getattr(cls, name, None)
if isinstance(attr, RawType):
return attr.tp
if ct is ctypes.c_void_p:
cls._void_p.append(name)
return ct
def __init__(self, *args: Any, do_sync: bool = True):
class_typ: Type[Struct] = type(self)
if class_typ is Struct:
raise Exception(
"cannot instantiate Struct directly",
)
self._existing_address: Optional[int] = None
self._struct = self._internal_struct(
*[
i if not isinstance(i, BaseCPointer) else i._as_parameter_
for i in args # fmt: off
]
)
if do_sync:
self._sync()
def __init_subclass__(cls):
hints = cls.__annotations__
cls._void_p = []
cls._hints = {
k: cls._convert_tc_ptr(v, k)
for k, v in hints.items()
if k not in {"_hints", "_void_p", "_internal_struct"}
}
class _InternalStruct(ctypes.Structure):
_fields_ = [
(
name,
cls._get_type(get_mapped(typ), name),
)
for name, typ in cls._hints.items() # fmt: off
]
cls._internal_struct = _InternalStruct
@property
def _as_parameter_(self) -> ctypes.Structure:
return self._struct
@classmethod
def from_existing(cls, struct: ctypes.Structure) -> "Struct":
"""Build a new struct from an existing ctypes structure.
Args:
struct: Existing `ctypes.Structure` object
Returns:
Created struct object.
"""
instance = cls(do_sync=False)
instance._struct = struct # type: ignore
# mypy is getting angry here for whatever reason
instance._sync()
instance._existing_address = ctypes.addressof(struct)
return instance
@handle
def __getattribute__(self, name: str):
attr = super().__getattribute__(name)
with suppress(AttributeError):
hints = super().__getattribute__("_hints")
if (name in hints) and (type(attr)) is bytes:
attr = attempt_decode(attr)
if isinstance(attr, ctypes._Pointer): # type: ignore
value = attr.contents
ct = type(value)
if ct is ctypes.c_void_p:
add_ref(ct)
return VoidPointer(
ctypes.addressof(value),
ctypes.sizeof(value),
)
py_type = get_py(ct)
return TypedCPointer(
ctypes.addressof(value),
py_type,
ctypes.sizeof(value),
False,
)
if name in super().__getattribute__("_void_p"):
ct = ctypes.c_void_p(attr) # type: ignore
return VoidPointer(attr, ctypes.sizeof(ct)) # type: ignore
return attr
def __setattr__(self, name: str, value: Any):
if hasattr(self, "_struct"):
self._struct.__setattr__(name, value)
super().__setattr__(name, value)
def _sync(self) -> None:
for name in self._hints:
setattr(self, name, getattr(self._struct, name))
def __repr__(self) -> str:
return f"<struct {type(self).__name__} at {hex(ctypes.addressof(self._struct))}>" # noqa
@property
def struct(self) -> ctypes.Structure:
"""Raw internal Structure object."""
return self._struct
def get_existing_address(self) -> int:
if not self._existing_address:
raise ValueError("instance has not been created from a C struct")
return self._existing_address
struct: ctypes.Structure
property
Raw internal Structure object.
from_existing(struct)
classmethod
Build a new struct from an existing ctypes structure.
Parameters: |
|
---|
Returns: |
|
---|
src/pointers/structure.py
@classmethod
def from_existing(cls, struct: ctypes.Structure) -> "Struct":
"""Build a new struct from an existing ctypes structure.
Args:
struct: Existing `ctypes.Structure` object
Returns:
Created struct object.
"""
instance = cls(do_sync=False)
instance._struct = struct # type: ignore
# mypy is getting angry here for whatever reason
instance._sync()
instance._existing_address = ctypes.addressof(struct)
return instance
StructPointer
Bases: Pointer[T]
Class representing a pointer to a struct.
Source code in src/pointers/structure.py
class StructPointer(Pointer[T]):
"""Class representing a pointer to a struct."""
def __init__(
self,
address: int,
existing: Optional["Struct"] = None,
):
self._existing = existing
super().__init__(address, True)
@property # type: ignore
@handle
def _as_parameter_(
self,
) -> Union[int, "ctypes._PointerLike"]:
existing = self._existing
if existing:
return ctypes.pointer(existing.struct)
return self.ensure()
def __repr__(self) -> str:
return f"StructPointer(address={self.address}, existing={self._existing!r})" # noqa
def __rich__(self) -> str:
return f"<[bold blue]pointer[/] to struct at {str(self)}>"
def get_existing_address(self) -> int:
return (~self).get_existing_address()
AllocatedPointer
Bases: IterDereferencable[T]
, BaseAllocatedPointer[T]
Pointer to allocated memory.
Source code in src/pointers/malloc.py
class AllocatedPointer(IterDereferencable[T], BaseAllocatedPointer[T]):
"""Pointer to allocated memory."""
def __init__(
self,
address: int,
size: int,
assigned: bool = False,
) -> None:
"""
Args:
address: Address of the allocated memory.
size: Size of the allocated memory.
assigned: Whether an object is currently inside the memory.
"""
self._address = address
self._size = size
self._freed = False
self._assigned = assigned
@property
def address(self) -> Optional[int]:
return self._address
@address.setter
def address(self, value: int) -> None:
self._address = value
def __repr__(self) -> str:
return f"AllocatedPointer(address={self.address}, size={self.size})"
def __add__(self, amount: int):
return AllocatedPointer(
self.ensure() + amount,
self.size,
self.assigned,
)
def __sub__(self, amount: int):
return AllocatedPointer(
self.ensure() - amount,
self.size,
self.assigned,
)
@handle
def free(self) -> None:
self.ensure_valid()
c_free(self.make_ct_pointer())
self.freed = True
__init__(address, size, assigned=False)
Parameters: |
|
---|
src/pointers/malloc.py
def __init__(
self,
address: int,
size: int,
assigned: bool = False,
) -> None:
"""
Args:
address: Address of the allocated memory.
size: Size of the allocated memory.
assigned: Whether an object is currently inside the memory.
"""
self._address = address
self._size = size
self._freed = False
self._assigned = assigned
free(target)
Equivalent to target.free()
Parameters: |
|
---|
Example
ptr = malloc(1)
free(ptr) # is the same as `ptr.free()`
src/pointers/malloc.py
def free(target: BaseAllocatedPointer):
"""Equivalent to `target.free()`
Args:
target: Pointer to free.
Example:
```py
ptr = malloc(1)
free(ptr) # is the same as `ptr.free()`
```"""
target.free()
malloc(size)
Allocate memory of a given size.
Parameters: |
|
---|
Returns: |
|
---|
Raises: |
|
---|
Example
ptr = malloc(1)
src/pointers/malloc.py
def malloc(size: int) -> AllocatedPointer[Any]:
"""Allocate memory of a given size.
Args:
size: Allocation size.
Returns:
Pointer to allocated memory.
Raises:
AllocationError: Raised when allocation fails, presumably due to no memory.
Example:
```py
ptr = malloc(1)
```
""" # noqa
mem = c_malloc(size)
if not mem:
raise AllocationError("failed to allocate memory")
return AllocatedPointer(mem, size)
realloc(target, size)
Resize a memory block created by malloc.
Parameters: |
|
---|
Returns: |
|
---|
Raises: |
|
---|
Example
ptr = malloc(1)
realloc(ptr, 2)
src/pointers/malloc.py
@handle
def realloc(target: A, size: int) -> A:
"""Resize a memory block created by malloc.
Args:
target: Pointer to reallocate.
size: New allocation size.
Returns:
Original object.
Raises:
InvalidSizeError: Object inside allocation is larger than attempted reallocation.
AllocationError: Raised when allocation fails, presumably due to no memory.
Example:
```py
ptr = malloc(1)
realloc(ptr, 2)
```
""" # noqa
if type(target) is StackAllocatedPointer:
raise TypeError("pointers to items on the stack may not be resized")
tsize: int = sys.getsizeof(~target)
if target.assigned and (tsize > size):
raise InvalidSizeError(
f"object inside memory is of size {tsize}, so memory cannot be set to size {size}", # noqa
)
addr = c_realloc(target.address, size)
if not addr:
raise AllocationError("failed to resize memory")
target.size = size
target.address = addr
return target
AllocatedArrayPointer
Bases: BaseAllocatedPointer[T]
Pointer to an allocated array.
Source code in src/pointers/calloc.py
class AllocatedArrayPointer(BaseAllocatedPointer[T]):
"""Pointer to an allocated array."""
def __init__(
self,
address: int,
chunks: int,
chunk_size: int,
current_index: int,
chunk_store: Optional[Dict[int, "AllocatedArrayPointer[T]"]] = None,
freed: bool = False,
origin_address: Optional[int] = None,
) -> None:
self._origin_address = origin_address or address
self._address = address
self._size = chunk_size
self._current_index = current_index
self._chunks = chunks
self._chunk_store = chunk_store or {0: self}
self._assigned = True
self._tracked = False
self._freed = freed
if chunk_store:
self._chunk_store[self.current_index] = self
@property # type: ignore
def address(self) -> Optional[int]:
return self._address
@property
def current_index(self) -> int:
"""Current chunk index."""
return self._current_index
@property
def chunks(self) -> int:
"""Number of allocated chunks."""
return self._chunks
def _get_chunk_at(self, index: int) -> "AllocatedArrayPointer[T]":
if index > self.chunks:
raise IndexError(
f"index is {index}, while allocation is {self.chunks}",
)
if index < 0: # for handling __sub__
raise IndexError("index is below zero")
if index not in self._chunk_store:
self._chunk_store[index] = AllocatedArrayPointer(
self._origin_address + (index * self.size),
self.chunks,
self.size,
index,
self._chunk_store,
self._freed,
self._origin_address,
)
return self._chunk_store[index]
def __add__(self, amount: int) -> "AllocatedArrayPointer[T]":
self.ensure_valid()
return self._get_chunk_at(self._current_index + amount)
def __sub__(self, amount: int) -> "AllocatedArrayPointer[T]":
return self.__add__(-amount)
def __repr__(self) -> str:
return f"AllocatedArrayPointer(address={self.address}, current_index={self.current_index})" # noqa
def __iter__(self) -> Iterator["AllocatedArrayPointer[T]"]:
for i in range(self.current_index, self.chunks):
yield self + i
def __getitem__(self, index: int) -> "AllocatedArrayPointer[T]":
return self._get_chunk_at(index)
def __setitem__(self, index: int, value: T) -> None:
chunk = self._get_chunk_at(index)
chunk <<= value
@handle
def free(self) -> None:
first = self[0]
first.ensure_valid()
for i in range(self._chunks): # using __iter__ breaks here
chunk = self._get_chunk_at(i)
chunk.freed = True
c_free(first.make_ct_pointer())
chunks: int
property
Number of allocated chunks.
current_index: int
property
Current chunk index.
calloc(num, size)
Allocate a number of blocks with a given size.
src/pointers/calloc.py
def calloc(num: int, size: int) -> AllocatedArrayPointer:
"""Allocate a number of blocks with a given size."""
address: int = c_calloc(num, size)
if not address:
raise AllocationError("failed to allocate memory")
return AllocatedArrayPointer(address, num, size, 0)
AllocationError
Bases: Exception
Raised when a memory allocation fails.
Source code in src/pointers/exceptions.py
class AllocationError(Exception):
"""Raised when a memory allocation fails."""
DereferenceError
Bases: Exception
Raised when dereferencing an address fails.
Source code in src/pointers/exceptions.py
class DereferenceError(Exception):
"""Raised when dereferencing an address fails.""" # noqa
FreedMemoryError
Bases: Exception
Raised when trying to perform an operation on freed memory.
Source code in src/pointers/exceptions.py
class FreedMemoryError(Exception):
"""Raised when trying to perform an operation on freed memory."""
InvalidBindingParameter
Bases: Exception
Raised when an invalid type is passed to a binding.
Source code in src/pointers/exceptions.py
class InvalidBindingParameter(Exception):
"""Raised when an invalid type is passed to a binding."""
InvalidSizeError
Bases: Exception
Raised when trying to move an object of the wrong size to an allocation.
Source code in src/pointers/exceptions.py
class InvalidSizeError(Exception):
"""Raised when trying to move an object of the wrong size to an allocation.""" # noqa
InvalidVersionError
Bases: Exception
Python version is not high enough.
Source code in src/pointers/exceptions.py
class InvalidVersionError(Exception):
"""Python version is not high enough."""
NullPointerError
Bases: Exception
Raised when a pointer is null.
Source code in src/pointers/exceptions.py
class NullPointerError(Exception):
"""Raised when a pointer is null."""
SegmentViolation
Bases: Exception
SIGSEGV was sent to Python.
Source code in src/pointers/exceptions.py
class SegmentViolation(Exception):
"""SIGSEGV was sent to Python."""
VariableLifetimeError
Bases: Exception
Variable is no longer available.
Source code in src/pointers/exceptions.py
class VariableLifetimeError(Exception):
"""Variable is no longer available."""
attempt_decode(data)
Attempt to decode a string of bytes.
src/pointers/_utils.py
def attempt_decode(data: bytes) -> Union[str, bytes]:
"""Attempt to decode a string of bytes."""
try:
return data.decode()
except UnicodeDecodeError:
return data
deref(address)
Get the value at the target address.
src/pointers/_utils.py
def deref(address: int) -> Any:
"""Get the value at the target address."""
return ctypes.cast(address, ctypes.py_object).value
force_set_attr(typ, key, value)
Force setting an attribute on the target type.
src/pointers/_utils.py
def force_set_attr(typ: Type[Any], key: str, value: Any) -> None:
"""Force setting an attribute on the target type."""
if not isinstance(typ, type):
raise ValueError(
f"{typ} does not derive from type (did you pass an instance and not a class)?", # noqa
)
_force_set_attr(typ, key, value)
get_mapped(typ)
Get the C mapped value of the given type.
src/pointers/_utils.py
def get_mapped(typ: Any) -> "Type[ctypes._CData]":
"""Get the C mapped value of the given type."""
from .c_pointer import VoidPointer
if getattr(typ, "__origin__", None) is Callable:
args = list(typ.__args__)
res = args.pop(-1)
return ctypes.CFUNCTYPE(get_mapped(res), *map(get_mapped, args))
if type(typ) in {FunctionType, MethodType}:
hints = typ.__annotations__.copy()
try:
res = hints.pop("return")
except KeyError as e:
raise TypeError(
"return type annotation is required to convert to a C function" # noqa
) from e
args = hints.values()
return ctypes.CFUNCTYPE(
get_mapped(res) if res else None,
*map(get_mapped, args),
)
# VoidPointer needs to be passed here to stop circular imports
return {**_C_TYPES, VoidPointer: ctypes.c_void_p}.get( # type: ignore
typ,
) or ctypes.py_object
get_py(data)
Map the specified C type to a Python type.
src/pointers/_utils.py
def get_py(
data: Type["ctypes._CData"],
) -> Type[Any]:
"""Map the specified C type to a Python type."""
from .base_pointers import BaseCPointer
if data.__name__.startswith("LP_"):
return BaseCPointer
try:
return _PY_TYPES[data]
except KeyError as e:
raise ValueError(
f"{data} is not a valid ctypes type",
) from e
is_mappable(typ)
Whether the specified type is mappable to C.
src/pointers/_utils.py
def is_mappable(typ: Any) -> bool:
"""Whether the specified type is mappable to C."""
try:
get_mapped(typ)
return True
except ValueError:
return False
make_py(data)
Convert the target C value to a Python object.
src/pointers/_utils.py
def make_py(data: "ctypes._CData"):
"""Convert the target C value to a Python object."""
typ = get_py(type(data))
res = typ(data)
if typ is bytes:
res = attempt_decode(res)
return res
map_type(data)
Map the specified data to a C type.
src/pointers/_utils.py
def map_type(data: Any) -> "ctypes._CData":
"""Map the specified data to a C type."""
typ = get_mapped(type(data))
return typ(data) # type: ignore
move_to_mem(ptr, stream, *, unsafe=False, target='memory allocation')
Move data to a C pointer.
src/pointers/_utils.py
def move_to_mem(
ptr: "ctypes._PointerLike",
stream: bytes,
*,
unsafe: bool = False,
target: str = "memory allocation",
):
"""Move data to a C pointer."""
slen = len(stream)
plen = len(ptr.contents) # type: ignore
if (slen > plen) and (not unsafe):
raise InvalidSizeError(
f"object is of size {slen}, while {target} is {plen}",
)
ctypes.memmove(ptr, stream, slen)