Reference

class NULL:
    """Unique object representing a NULL address.

    May be used with object pointers or passed to bindings.
    """

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

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 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 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 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 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 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 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 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]

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 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 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

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()

class AllocatedPointer(IterDereferencable[T], BaseAllocatedPointer[T]):
    """Pointer to allocated memory."""

    def __init__(
        self,
        address: int,
        size: int,
        assigned: bool = False,
        parent: AllocatedPointer[T] | None = None,
    ) -> 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
        self._parent: AllocatedPointer[T] | None = parent

    def _indexed(self, amount: int) -> AllocatedPointer[T]:
        return AllocatedPointer(
            self.ensure() + amount,
            self.size - amount,
            self.assigned,
            parent=self._parent,
        )

    def _get_parent(self) -> AllocatedPointer[T]:
        parent = self

        while parent._parent:
            parent = parent._parent

        return parent

    @property
    def freed(self) -> bool:
        return self._get_parent()._freed

    @freed.setter
    def freed(self, value: bool) -> None:
        self._get_parent()._freed = value

    @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) -> AllocatedPointer[T]:
        return self._indexed(amount)

    def __sub__(self, amount: int) -> AllocatedPointer[T]:
        return self._indexed(amount)

    @handle
    def free(self) -> None:
        self.ensure_valid()
        c_free(self.make_ct_pointer())
        self.freed = True

    @handle
    def __getitem__(self, index: int) -> AllocatedPointer[T]:
        if not isinstance(index, int):
            raise ValueError(
                f"memory indices must be int, not {type(index).__name__}",
            )

        return self._indexed(index)

    @handle
    def __setitem__(self, index: int, value: T) -> None:
        if not isinstance(index, int):
            raise ValueError(
                f"memory indices must be int, not {type(index).__name__}",
            )

        ptr = self._indexed(index)
        ptr <<= value