我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用ctypes.py_object()。
def test_get_raw(self): from ctypes import pythonapi, c_void_p, py_object try: Bytes_FromString = pythonapi.PyBytes_FromString except: Bytes_FromString = pythonapi.PyString_FromString Bytes_FromString.restype = c_void_p Bytes_FromString.argtypes = [py_object] mixer.init() try: samples = as_bytes('abcdefgh') # keep byte size a multiple of 4 snd = mixer.Sound(buffer=samples) raw = snd.get_raw() self.assertTrue(isinstance(raw, bytes_)) self.assertNotEqual(snd._samples_address, Bytes_FromString(samples)) self.assertEqual(raw, samples) finally: mixer.quit()
def extra_ivars(cls, base): c_type = c_typeobj(cls) c_base = c_typeobj(base) t_size = c_type.tp_basicsize b_size = c_base.tp_basicsize sizeof_pyobj = ctypes.sizeof(ctypes.py_object) if c_type.tp_itemsize or c_base.tp_itemsize: return t_size != b_size or c_type.tp_itemsize != c_base.tp_itemsize if is_heap_type(cls): if c_type.tp_weaklistoffset and c_base.tp_weaklistoffset == 0: if c_type.tp_weaklistoffset + sizeof_pyobj == t_size: t_size -= sizeof_pyobj if c_type.tp_dictoffset and c_base.tp_dictoffset == 0: if c_type.tp_dictoffset + sizeof_pyobj == t_size: t_size -= sizeof_pyobj return t_size != b_size
def set_window_user_pointer(window, pointer): ''' Sets the user pointer of the specified window. You may pass a normal python object into this function and it will be wrapped automatically. The object will be kept in existence until the pointer is set to something else or until the window is destroyed. Wrapper for: void glfwSetWindowUserPointer(GLFWwindow* window, void* pointer); ''' data = (False, pointer) if not isinstance(pointer, ctypes.c_void_p): data = (True, pointer) # Create a void pointer for the python object pointer = ctypes.cast(ctypes.pointer(ctypes.py_object(pointer)), ctypes.c_void_p) window_addr = ctypes.cast(ctypes.pointer(window), ctypes.POINTER(ctypes.c_long)).contents.value _window_user_data_repository[window_addr] = data _glfw.glfwSetWindowUserPointer(window, pointer)
def pre_work( self, task, ): self.update_current_task( task=task, ) interval = self.worker_config['timeouts']['soft_timeout'] if interval == 0: interval = None self.current_timers[threading.get_ident()] = threading.Timer( interval=interval, function=ctypes.pythonapi.PyThreadState_SetAsyncExc, args=( ctypes.c_long(threading.get_ident()), ctypes.py_object(worker.WorkerSoftTimedout), ) ) self.current_timers[threading.get_ident()].start()
def qwidget_winid_to_hwnd(winid): """ Convert the winid for a QWidget to an HWND. :param int winid: The QWidget winid to convert. :returns: Window handle """ # Setup arguments and return types. ctypes.pythonapi.PyCObject_AsVoidPtr.restype = ctypes.c_void_p ctypes.pythonapi.PyCObject_AsVoidPtr.argtypes = [ctypes.py_object] # Convert PyCObject to a void pointer. hwnd = ctypes.pythonapi.PyCObject_AsVoidPtr(winid) return hwnd
def raise_exc(self, excobj): """Raise exception processing. """ if not self.isAlive(): return tid = next((k for k, v in threading._active.items() if v is self), None) # pylint: disable=protected-access if tid is None: return with self._thr_lock: res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), ctypes.py_object(excobj)) if res == 0: raise ValueError("nonexistent thread id") elif res > 1: # if it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), 0) raise SystemError("PyThreadState_SetAsyncExc failed") # the thread was alive when we entered the loop, but was not found # in the dict, hence it must have been already terminated. should we raise # an exception here? silently ignore?
def terminate_thread(thread): """Terminates a python thread from another thread. :param thread: a threading.Thread instance """ if not thread.isAlive(): return exc = ctypes.py_object(SystemExit) res = ctypes.pythonapi.PyThreadState_SetAsyncExc( ctypes.c_long(thread.ident), exc) if res == 0: raise ValueError("nonexistent thread id") elif res > 1: # """if it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect""" ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None) raise SystemError("PyThreadState_SetAsyncExc failed")
def media_open_cb(opaque, datap, sizep): """LibVLC callback used to point the player to the video buffer upon opening the media. opaque: pointer to our media object. datap: the libVLC video buffer. sizep: length of the media stream (or sys.maxsize if unknown). """ datap.contents.value = opaque sizep.contents.value = sys.maxsize container = ctypes.cast(opaque, ctypes.POINTER( ctypes.py_object)).contents.value return container.open()
def test_SDL_GetSetWindowData(self): # TODO: fix this window = video.SDL_CreateWindow(b"Test", 10, 10, 10, 10, 0) self.assertIsInstance(window.contents, video.SDL_Window) values = {b"text": py_object("Teststring"), b"object": py_object(self), b"list": py_object([1, 2, 3, 4]), b"tuple": py_object(("a", 1, self)) } for k, v in values.items(): retval = video.SDL_GetWindowData(window, k) self.assertFalse(retval) video.SDL_SetWindowData(window, k, v) retval = video.SDL_GetWindowData(window, k) self.assertEqual(retval.contents.value, v.value) video.SDL_DestroyWindow(window)
def terminate_thread(thread): """Terminate a python thread from another thread. :param thread: a threading.Thread instance """ if not thread.isAlive(): return exc = ctypes.py_object(SystemExit) res = ctypes.pythonapi.PyThreadState_SetAsyncExc( ctypes.c_long(thread.ident), exc) if res == 0: raise ValueError("nonexistent thread id") elif res > 1: # If it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None) raise SystemError("PyThreadState_SetAsyncExc failed")
def patchable_builtin(klass): # It's important to create variables here, we want those objects alive # within this whole scope. name = klass.__name__ target = klass.__dict__ # Hardcore introspection to find the `PyProxyDict` object that contains the # precious `dict` attribute. proxy_dict = SlotsProxy.from_address(id(target)) namespace = {} # This is the way I found to `cast` this `proxy_dict.dict` into a python # object, cause the `from_address()` function returns the `py_object` # version ctypes.pythonapi.PyDict_SetItem( ctypes.py_object(namespace), ctypes.py_object(name), proxy_dict.dict, ) return namespace[name]
def pack_l2s(lnum, sep='', case='lower'): import ctypes PyLong_AsByteArray = ctypes.pythonapi._PyLong_AsByteArray PyLong_AsByteArray.argtypes = [ctypes.py_object, ctypes.c_char_p, ctypes.c_size_t, ctypes.c_int, ctypes.c_int] a = ctypes.create_string_buffer(lnum.bit_length()//8 + 1) PyLong_AsByteArray(lnum, a, len(a), 0, 1) hexbytes = ["{:02x}".format(ord(b)) for b in a.raw] while hexbytes[0] == '00': hexbytes.pop(0) if case == 'upper': return sep.join(hexbytes).upper() return sep.join(hexbytes)
def terminate_thread(thread): """Terminate a python thread from another thread. :param thread: a threading.Thread instance """ if not thread.isAlive(): return # print('valet watcher thread: notifier thread is alive... - kill it...') exc = ctypes.py_object(SystemExit) res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), exc) if res == 0: raise ValueError("nonexistent thread id") elif res > 1: # If it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None) raise SystemError("PyThreadState_SetAsyncExc failed") # print('valet watcher thread exits')
def _async_raise(tid, exctype): '''Raises an exception in the threads with id tid''' if not inspect.isclass(exctype): raise TypeError("Only types can be raised (not instances)") try: res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), ctypes.py_object(exctype)) except AttributeError: # To catch: undefined symbol: PyThreadState_SetAsyncExc return if res == 0: raise ValueError("invalid thread id") elif res != 1: # "if it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect" ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), 0) raise SystemError("PyThreadState_SetAsyncExc failed")
def kill_thread(thread=None, name=None, tid=0, exctype=SystemExit): """raises the exception, performs cleanup if needed""" if name: thread = search_thread(name=name, part=True) if thread and isinstance(thread, threading.Thread): if not thread.is_alive(): return '??????' tid = thread.ident if not tid: return '?????' if not isinstance(tid, ctypes.c_longlong): tid = ctypes.c_longlong(tid) if not inspect.isclass(exctype): raise TypeError("Only types can be raised (not instances)") res = ctypes.pythonapi.PyThreadState_SetAsyncExc( tid, ctypes.py_object(exctype)) if res == 0: raise ValueError("invalid thread id") elif res != 1: # """if it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect""" ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, 0) raise SystemError("PyThreadState_SetAsyncExc failed")
def _async_raise(tid, excobj): res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(excobj)) if res == 0: raise ValueError("nonexistent thread id") elif res > 1: # """if it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect""" ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, 0) raise SystemError("PyThreadState_SetAsyncExc failed")
def _unwrap_window_id(window_id): try: return int(window_id) except: ctypes.pythonapi.PyCapsule_GetPointer.restype = ctypes.c_void_p ctypes.pythonapi.PyCapsule_GetPointer.argtypes = [ctypes.py_object] return int(ctypes.pythonapi.PyCapsule_GetPointer(window_id, None))
def _async_raise(tid, exctype): """raises the exception, performs cleanup if needed""" if not inspect.isclass(exctype): raise TypeError("Only types can be raised (not instances)") if not issubclass(exctype, BaseException): raise ValueError("Only sub classes of BaseException can be raised") # PyThreadState_SetAsyncExc requires GIL to be held gil_state = ctypes.pythonapi.PyGILState_Ensure() try: res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), ctypes.py_object(exctype)) if res == 0: # The thread is likely dead already. raise InvalidThreadIdError(tid) elif res != 1: # If more than one threads are affected (WTF?), we're in trouble, and # we try our best to revert the effect, although this may not work. ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), None) raise AsyncRaiseError("PyThreadState_SetAsyncExc failed", res) finally: ctypes.pythonapi.PyGILState_Release(gil_state)
def _pythonapi_geterrno(): """ Read errno using Python C API: raise an exception with PyErr_SetFromErrno and then read error code 'errno'. This function may raise an RuntimeError. """ try: pythonapi.PyErr_SetFromErrno(py_object(OSError)) except OSError, err: return err.errno else: raise RuntimeError("get_errno() is unable to get error code")
def _import_cast(self): """Use ctypes to access the C function. Raise any sort of error: we just support this where ctypes works as expected. """ import ctypes lib = ctypes.pydll.LoadLibrary(psycopg2._psycopg.__file__) cast = lib.typecast_BINARY_cast cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object] cast.restype = ctypes.py_object return cast
def locals_to_fast(self, frame): try: ltf = ctypes.pythonapi.PyFrame_LocalsToFast ltf.argtypes = [ctypes.py_object, ctypes.c_int] ltf(frame, 1) except: pass
def _async_raise(tid, exctype): """raises the exception, performs cleanup if needed""" if not inspect.isclass(exctype): raise TypeError("Only types can be raised (not instances)") res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), ctypes.py_object(exctype)) if res == 0: raise ValueError("invalid thread id") elif res != 1: # """if it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect""" ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), None) raise SystemError("PyThreadState_SetAsyncExc failed")
def struct_to_py_object(d): if not isinstance(d, CDataType): raise TypeError("Must be a ctypes._CData instance") if isinstance(d, ctypes._Pointer): cdata = d.contents else: cdata = d cdata_obj = CDataObject.from_address(id(cdata)) cdata_obj.b_needsfree = 0 ns = {} pythonapi.PyDict_SetItem(py_object(ns), py_object(None), ctypes.pointer(cdata)) return ns[None]
def _async_raise(tid, exctype): """raises the exception, performs cleanup if needed""" tid = ctypes.c_long(tid) if not inspect.isclass(exctype): exctype = type(exctype) res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype)) if res == 0: raise ValueError("invalid thread id") elif res != 1: # """if it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect""" ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None) raise SystemError("PyThreadState_SetAsyncExc failed")
def __init__(self, size): assert size > 0, "Array size must be > 0" self._size = size # Create the array structure using the ctypes module PyArrayType = ctypes.py_object * size self._elements = PyArrayType() # Initialize each element. self.clear(None) # Returns the size of the array
def __call__(self, p_struct, *args): py_object_addr = getattr(p_struct.contents, self._py_object_field) py_object = ctypes.cast(py_object_addr, ctypes.POINTER(ctypes.py_object)).contents.value return self._fn(py_object, *args)
def __init__(self): self.struct = self.CStructure() self.this = ctypes.pointer(self.struct) if self._pyobject_: # save pointer to python object setattr(self.struct, self._pyobject_, ctypes.cast(ctypes.pointer(ctypes.py_object(self)), ctypes.c_void_p)) self.override_vtable()
def _async_raise(self, tid, exctype): """raises the exception, performs cleanup if needed""" tid = ctypes.c_long(tid) if not inspect.isclass(exctype): exctype = type(exctype) res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype)) if res == 0: log.error("invalid thread id") elif res != 1: """if it returns a number greater than one, you're in trouble, and you should call it again with exc=NULL to revert the effect""" ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None) log.error("PyThreadState_SetAsyncExc failed")
def _import_cast(self): """Use ctypes to access the C function. Raise any sort of error: we just support this where ctypes works as expected. """ import ctypes lib = ctypes.cdll.LoadLibrary(psycopg2._psycopg.__file__) cast = lib.typecast_BINARY_cast cast.argtypes = [ctypes.c_char_p, ctypes.c_size_t, ctypes.py_object] cast.restype = ctypes.py_object return cast
def stop(cls, task_name): if task_name not in cls._tasks: raise Exception("task {} is not exists".format(task_name)) tid = cls._tasks[task_name] exc_type = SystemExit tid = ctypes.c_long(tid) res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exc_type)) if res == 0: raise ValueError("invalid thread id?the task may be already stop") elif res != 1: # """if it returns a number greater than one, you're in trouble, # and you should call it again with exc=NULL to revert the effect""" ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None) raise SystemError("PyThreadState_SetAsyncExc failed") del cls._tasks[task_name]
def dd_DDMatrix2Poly(matrixptr): error = ctypes.c_int() polyhedraptr = libcdd.dd_DDMatrix2Poly(matrixptr, ctypes.byref(error)) # Return None on error. # The error values are enums, so they aren't exposed. if error.value != DD_NO_ERRORS: # Dump out the errors to stderr libcdd.dd_WriteErrorMessages( ctypes.pythonapi.PyFile_AsFile(ctypes.py_object(sys.stdout)), error) dd_FreePolyhedra(polyhedraptr) return None return polyhedraptr
def make_save_locals_impl(): """ Factory for the 'save_locals_impl' method. This may seem like a complicated pattern but it is essential that the method is created at module load time. Inner imports after module load time would cause an occasional debugger deadlock due to the importer lock and debugger lock being taken in different order in different threads. """ try: if '__pypy__' in sys.builtin_module_names: import __pypy__ # @UnresolvedImport pypy_locals_to_fast = __pypy__.locals_to_fast except: pass else: if '__pypy__' in sys.builtin_module_names: def save_locals_pypy_impl(frame): pypy_locals_to_fast(frame) return save_locals_pypy_impl try: import ctypes locals_to_fast = ctypes.pythonapi.PyFrame_LocalsToFast except: pass else: def save_locals_ctypes_impl(frame): locals_to_fast(ctypes.py_object(frame), ctypes.c_int(0)) return save_locals_ctypes_impl return None
def let_variable(self, frame_id, var_name, expression_value): """ Let a frame's var with a value by building then eval a let expression with breakoints disabled. """ breakpoints_backup = IKBreakpoint.backup_breakpoints_state() IKBreakpoint.disable_all_breakpoints() let_expression = "%s=%s" % (var_name, expression_value,) eval_frame = ctypes.cast(frame_id, ctypes.py_object).value global_vars = eval_frame.f_globals local_vars = eval_frame.f_locals try: exec(let_expression, global_vars, local_vars) error_message="" except Exception as e: t, result = sys.exc_info()[:2] if isinstance(t, str): result_type = t else: result_type = str(t.__name__) error_message = "%s: %s" % (result_type, result,) IKBreakpoint.restore_breakpoints_state(breakpoints_backup) _logger.e_debug("let_variable(%s) => %s", let_expression, error_message or 'succeed') return error_message
def __init__(self, vlc_instance): # Cast this container to a c pointer to use in the callbacks self._opaque = ctypes.cast(ctypes.pointer( ctypes.py_object(self)), ctypes.c_void_p) # Create the vlc callbacks, these will in turn call the methods defined # in this container self.media = vlc_instance.media_new_callbacks( cb.CALLBACKS["read"], cb.CALLBACKS["open"], cb.CALLBACKS["seek"], cb.CALLBACKS["close"], self._opaque )
