pointers.py/src/pointers/c_pointer.py at master · ZeroIntensity/pointers.py · GitHub

Repository navigation

History

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

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

pointers.py/src/pointers/c_pointer.py at master · ZeroIntensity/pointers.py · GitHub

Repository navigation

Expand file tree

Breadcrumbs

c_pointer.py

Latest commit

History

Breadcrumbs

c_pointer.py

File metadata and controls