python类c_bool()的实例源码

def window__get_thread_window_handles(thread_process_id):
    """

    :param thread_process_id: thread process ID, as from window__get_thread_process_id
    :return: hwnd for all top-level windows with this thread process id.
    """
    ret = []

    def callback(hwnd, lparam):
        ret.append(hwnd)
        return True

    enum_win_proc = WINFUNCTYPE(c_bool, POINTER(c_int), POINTER(c_int))
    windll.user32.EnumThreadWindows(wintypes.DWORD(thread_process_id), enum_win_proc(callback), None)

    return ret

def add_export(self,export_range,export_dir):
        logger.debug("Adding new video export process.")
        should_terminate = mp.Value(c_bool,False)
        frames_to_export  = mp.Value(c_int,0)
        current_frame = mp.Value(c_int,0)

        rec_dir = self.g_pool.rec_dir
        user_dir = self.g_pool.user_dir
        start_frame= export_range.start
        end_frame= export_range.stop+1 #end_frame is exclusive
        frames_to_export.value = end_frame-start_frame

        # Here we make clones of every plugin that supports it.
        # So it runs in the current config when we lauch the exporter.
        plugins = self.g_pool.plugins.get_initializers()

        out_file_path=verify_out_file_path(self.rec_name,export_dir)
        process = Export_Process(target=export, args=(should_terminate,frames_to_export,current_frame, rec_dir,user_dir,self.g_pool.min_data_confidence,start_frame,end_frame,plugins,out_file_path))
        self.new_export = process

def AwaObjectLinkArray_IsValid(self, array, index):
        self._lib.AwaObjectLinkArray_IsValid.restype = c_bool
        self._lib.AwaObjectLinkArray_IsValid.argtypes = [c_void_p, c_ulong]
        return self._lib.AwaObjectLinkArray_IsValid(array, index)

# * @}

def AwaOpaqueArray_IsValid(self, array, index):
        self._lib.AwaOpaqueArray_IsValid.restype = c_bool
        self._lib.AwaOpaqueArray_IsValid.argtypes = [c_void_p, c_ulong]
        return self._lib.AwaOpaqueArray_IsValid(array, index)

def AwaTimeArray_IsValid(self, array, index):
        self._lib.AwaTimeArray_IsValid.restype = c_bool
        self._lib.AwaTimeArray_IsValid.argtypes = [c_void_p, c_ulong]
        return self._lib.AwaTimeArray_IsValid(array, index)

def AwaCStringArrayIterator_Next(self, iterator):
        self._lib.AwaCStringArrayIterator_Next.restype = c_bool
        self._lib.AwaCStringArrayIterator_Next.argtypes = [c_void_p]
        return self._lib.AwaCStringArrayIterator_Next(iterator)

def AwaIntegerArrayIterator_Next(self, iterator):
        self._lib.AwaIntegerArrayIterator_Next.restype = c_bool
        self._lib.AwaIntegerArrayIterator_Next.argtypes = [c_void_p]
        return self._lib.AwaIntegerArrayIterator_Next(iterator)

def AwaBooleanArrayIterator_GetValue(self, iterator):
        self._lib.AwaBooleanArrayIterator_GetValue.restype = c_bool
        self._lib.AwaBooleanArrayIterator_GetValue.argtypes = [c_void_p]
        return self._lib.AwaBooleanArrayIterator_GetValue(iterator)

def make_array(shape=(1,), dtype=np.float32, shared=False, fill_val=None):  
    np_type_to_ctype = {np.float32: ctypes.c_float,
                        np.float64: ctypes.c_double,
                        np.bool: ctypes.c_bool,
                        np.uint8: ctypes.c_ubyte,
                        np.uint64: ctypes.c_ulonglong}

    if not shared:
        np_arr = np.empty(shape, dtype=dtype)
    else:
        numel = np.prod(shape)
        arr_ctypes = sharedctypes.RawArray(np_type_to_ctype[dtype], numel)
        np_arr = np.frombuffer(arr_ctypes, dtype=dtype, count=numel)
        np_arr.shape = shape

    if not fill_val is None:
        np_arr[...] = fill_val

    return np_arr

def set_translation(handle, pos):
    current = get_win_info(handle)

    err = ctypes.windll.user32.MoveWindow(
        handle,
        ctypes.c_int(pos[0]),
        ctypes.c_int(pos[1]),
        ctypes.c_int(current.width),
        ctypes.c_int(current.height),
        ctypes.c_bool(False),
    )

    # Returns 0 on failure
    # https://msdn.microsoft.com/en-us/library/ms633534(VS.85).aspx
    if err == ctypes.c_bool(0):
        raise ctypes.WinError()

def callback_friend_typing(self, callback, user_data):
        """
        Set the callback for the `friend_typing` event. Pass NULL to unset.

        This event is triggered when a friend starts or stops typing.

        :param callback: Python function. Should take pointer (c_void_p) to Tox object,
        The friend number (c_uint32) of the friend who started or stopped typing,
        The result of calling tox_friend_get_typing (c_bool) on the passed friend_number,
        pointer (c_void_p) to user_data
        :param user_data: pointer (c_void_p) to user data
        """
        c_callback = CFUNCTYPE(None, c_void_p, c_uint32, c_bool, c_void_p)
        self.friend_typing_cb = c_callback(callback)
        Tox.libtoxcore.tox_callback_friend_typing(self._tox_pointer, self.friend_typing_cb, c_void_p(user_data))

    # -----------------------------------------------------------------------------------------------------------------
    # Sending private messages
    # -----------------------------------------------------------------------------------------------------------------

def self_set_typing(self, friend_number, typing):
        """
        Set the client's typing status for a friend.

        The client is responsible for turning it on or off.

        :param friend_number: The friend to which the client is typing a message.
        :param typing: The typing status. True means the client is typing.
        :return: True on success.
        """
        tox_err_set_typing = c_int()
        result = Tox.libtoxcore.tox_self_set_typing(self._tox_pointer, c_uint32(friend_number),
                                                    c_bool(typing), byref(tox_err_set_typing))
        tox_err_set_typing = tox_err_set_typing.value
        if tox_err_set_typing == TOX_ERR_SET_TYPING['OK']:
            return bool(result)
        elif tox_err_set_typing == TOX_ERR_SET_TYPING['FRIEND_NOT_FOUND']:
            raise ArgumentError('The friend number did not designate a valid friend.')

def callback_call(self, callback, user_data):
        """
        Set the callback for the `call` event. Pass None to unset.

        :param callback: The function for the call callback.

        Should take pointer (c_void_p) to ToxAV object,
        The friend number (c_uint32) from which the call is incoming.
        True (c_bool) if friend is sending audio.
        True (c_bool) if friend is sending video.
        pointer (c_void_p) to user_data
        :param user_data: pointer (c_void_p) to user data
        """
        c_callback = CFUNCTYPE(None, c_void_p, c_uint32, c_bool, c_bool, c_void_p)
        self.call_cb = c_callback(callback)
        ToxAV.libtoxav.toxav_callback_call(self._toxav_pointer, self.call_cb, user_data)

def __SetDLLReturnTypes(self):
        self.nnotesdll.NotesInitExtended.restype = ctypes.c_uint16
        self.nnotesdll.NotesTerm.restype = ctypes.c_uint16
        self.nnotesdll.NSFDbOpen.restype = ctypes.c_uint16
        self.nnotesdll.NSFDbClose.restype = ctypes.c_uint16
        self.nnotesdll.NSFNoteOpenExt.restype = ctypes.c_uint16
        self.nnotesdll.NSFNoteOpenByUNID.restype = ctypes.c_uint16
        self.nnotesdll.NSFNoteClose.restype = ctypes.c_uint16
        self.nnotesdll.NSFNoteCopy.restype = ctypes.c_uint16
        self.nnotesdll.NSFNoteGetInfo.restype = None
        self.nnotesdll.NSFNoteIsSignedOrSealed.restype = ctypes.c_bool
        self.nnotesdll.NSFNoteDecrypt.restype = ctypes.c_uint16
        self.nnotesdll.NSFItemDelete.restype = ctypes.c_uint16
        self.nnotesdll.NSFNoteHasMIMEPart.restype = ctypes.c_bool
        self.nnotesdll.NSFNoteHasMIME.restype = ctypes.c_bool
        self.nnotesdll.NSFNoteHasComposite.restype = ctypes.c_bool
        self.nnotesdll.MMCreateConvControls.restype = ctypes.c_uint16
        self.nnotesdll.MMDestroyConvControls.restype = ctypes.c_uint16
        self.nnotesdll.MMSetMessageContentEncoding.restype = None
        self.nnotesdll.MIMEConvertCDParts.restype = ctypes.c_uint16
        self.nnotesdll.MIMEConvertMIMEPartCC.restype = ctypes.c_uint16
        self.nnotesdll.NSFNoteUpdate.restype = ctypes.c_uint16

def on_main_thread(func):
    if not callable(func):
        raise TypeError('expected a callable')
    def new_func(*args, **kwargs):
        if NSThread.isMainThread(restype=c_bool, argtypes=[]):
            return func(*args, **kwargs)
        dispatcher = OMMainThreadDispatcher.new()
        dispatcher.func = func
        dispatcher.args = args
        dispatcher.kwargs = kwargs
        dispatcher.retval = None
        dispatcher.performSelectorOnMainThread_withObject_waitUntilDone_(sel('invoke'), None, True)
        retval = dispatcher.retval
        dispatcher.release()
        return retval
    return new_func

def _load_data_imp(_cmd, _block_ptr):
    global _dragged_item_path
    handler = runtime.ObjCBlockPointer(_block_ptr,
                                       argtypes=[ctypes.c_void_p, ctypes.c_bool, ctypes.c_void_p])

    if _dragged_item_path:
        NSURL = ObjCClass('NSURL')
        url = NSURL.fileURLWithPath_isDirectory_(ns(_dragged_item_path), os.path.isdir(_dragged_item_path))

        _dragged_item_path = None
    else:
        url = None

    if not url:
        error = NSError.errorWithDomain_code_userInfo_('com.robertvojta.blackmamba', 1, None)
        handler(None, None, error)
    else:
        handler(url.ptr, False, None)

def callback_friend_typing(self, callback, user_data):
        """
        Set the callback for the `friend_typing` event. Pass NULL to unset.

        This event is triggered when a friend starts or stops typing.

        :param callback: Python function. Should take pointer (c_void_p) to Tox object,
        The friend number (c_uint32) of the friend who started or stopped typing,
        The result of calling tox_friend_get_typing (c_bool) on the passed friend_number,
        pointer (c_void_p) to user_data
        :param user_data: pointer (c_void_p) to user data
        """
        c_callback = CFUNCTYPE(None, c_void_p, c_uint32, c_bool, c_void_p)
        self.friend_typing_cb = c_callback(callback)
        Tox.libtoxcore.tox_callback_friend_typing(self._tox_pointer, self.friend_typing_cb, c_void_p(user_data))

    # -----------------------------------------------------------------------------------------------------------------
    # Sending private messages
    # -----------------------------------------------------------------------------------------------------------------

def self_set_typing(self, friend_number, typing):
        """
        Set the client's typing status for a friend.

        The client is responsible for turning it on or off.

        :param friend_number: The friend to which the client is typing a message.
        :param typing: The typing status. True means the client is typing.
        :return: True on success.
        """
        tox_err_set_typing = c_int()
        result = Tox.libtoxcore.tox_self_set_typing(self._tox_pointer, c_uint32(friend_number),
                                                    c_bool(typing), byref(tox_err_set_typing))
        tox_err_set_typing = tox_err_set_typing.value
        if tox_err_set_typing == TOX_ERR_SET_TYPING['OK']:
            return bool(result)
        elif tox_err_set_typing == TOX_ERR_SET_TYPING['FRIEND_NOT_FOUND']:
            raise ArgumentError('The friend number did not designate a valid friend.')

def list_alttab_windows(cls):
        """
        Return the list of the windows handles that are currently guessed to be
        eligible to the Alt+Tab panel.
        Raises a OSError exception on error.
        """
        # LPARAM is defined as LONG_PTR (signed type)
        if ctypes.sizeof(ctypes.c_long) == ctypes.sizeof(ctypes.c_void_p):
            LPARAM = ctypes.c_long
        elif ctypes.sizeof(ctypes.c_longlong) == ctypes.sizeof(ctypes.c_void_p):
            LPARAM = ctypes.c_longlong
        EnumWindowsProc = ctypes.WINFUNCTYPE(
                                ctypes.c_bool, ctypes.c_void_p, LPARAM)

        def _enum_proc(hwnd, lparam):
            try:
                if cls.is_alttab_window(hwnd):
                    handles.append(hwnd)
            except OSError:
                pass
            return True

        handles = []
        ctypes.windll.user32.EnumWindows(EnumWindowsProc(_enum_proc), 0)
        return handles

def __init__(self, array):
        CTypeArray.__init__(self, c_bool, array)

def window__enum_window_handles(callback):

    # noinspection PyUnusedLocal
    def callback_wrapper(hwnd, lparam):
        return callback(hwnd)

    enum_win_proc = WINFUNCTYPE(c_bool, c_int, POINTER(c_int))
    windll.user32.EnumWindows(enum_win_proc(callback_wrapper), None)

def window__get_child_window_handles(hwnd_parent):
    ret = []

    def callback(hwnd, lparam):
        ret.append(hwnd)
        return True

    enum_win_proc = WINFUNCTYPE(c_bool, POINTER(c_int), POINTER(c_int))
    windll.user32.EnumChildWindows(hwnd_parent, enum_win_proc(callback), None)

    return ret

def monitor__find_monitors():
    ret = []

    def callback(h_monitor, hdc_monitor, lprc_monitor, lParam):
        # hMonitor: display monitor handle
        # hdcMonitor: device context handle; color attributes + clipping region
        # lprcMonitor: RECT structure w/ device-context coordinates.
        ret.append({
            'monitor_handle': h_monitor,
            'device_context_handle': hdc_monitor,
            'left': lprc_monitor.contents.left,
            'right': lprc_monitor.contents.right,
            'top': lprc_monitor.contents.top,
            'bottom': lprc_monitor.contents.bottom,
            'width': lprc_monitor.contents.right - lprc_monitor.contents.left,
            'height': lprc_monitor.contents.bottom - lprc_monitor.contents.top,
            'primary': len(ret) == 0,
            'index': len(ret),
        })
        return True

    enum_monitor_proc = WINFUNCTYPE(c_bool, POINTER(c_int), POINTER(c_int), POINTER(wintypes.RECT), POINTER(c_int))
    if EnumDisplayMonitors(None, None, enum_monitor_proc(callback), None) == 0:
        raise Exception("Could not find monitors")

    # for i in ret:
    #     info = POINTER(wintypes.MONITORINFOEX())
    #     if GetMonitorInfo(i['monitor_handle'], info) != 0:
    #         i['primary'] = info.contents.dwFlags == MONITORINFOF_PRIMARY
    #         i['name'] = str(info.contents.szDevice)

    return ret

def libvlc_dialog_post_login(p_id, psz_username, psz_password, b_store):
    '''Post a login answer
    After this call, p_id won't be valid anymore
    See libvlc_dialog_cbs.pf_display_login.
    @param p_id: id of the dialog.
    @param psz_username: valid and non empty string.
    @param psz_password: valid string (can be empty).
    @param b_store: if true, store the credentials.
    @return: 0 on success, or -1 on error.
    @version: LibVLC 3.0.0 and later.
    '''
    f = _Cfunctions.get('libvlc_dialog_post_login', None) or \
        _Cfunction('libvlc_dialog_post_login', ((1,), (1,), (1,), (1,),), None,
                    ctypes.c_int, ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_bool)
    return f(p_id, psz_username, psz_password, b_store)

def add_exports(self):
        outfiles = set()
        for d in self.new_exports:
            logger.debug("Adding new export.")
            should_terminate = Value(c_bool, False)
            frames_to_export = Value(c_int, 0)
            current_frame = Value(c_int, 0)
            start_frame = None
            end_frame = None
            export_dir = d
            user_dir = self.g_pool.user_dir

            # we need to know the timestamps of our exports.
            try:  # 0.4
                frames_to_export.value = len(np.load(os.path.join(export_dir, 'world_timestamps.npy')))
            except:  # <0.4
                frames_to_export.value = len(np.load(os.path.join(export_dir, 'timestamps.npy')))

            # Here we make clones of every plugin that supports it.
            # So it runs in the current config when we lauch the exporter.
            plugins = self.g_pool.plugins.get_initializers()

            # make a unique name created from rec_session and dir name
            rec_session, rec_dir = export_dir.rsplit(os.path.sep, 2)[1:]
            out_name = rec_session+"_"+rec_dir+".mp4"
            out_file_path = os.path.join(self.destination_dir, out_name)
            if out_file_path in outfiles:
                logger.error("This export setting would try to save {} at least twice please rename dirs to prevent this. Skipping File".format(out_file_path))
            else:
                outfiles.add(out_file_path)
                logger.info("Exporting to: {}".format(out_file_path))

                process = Export_Process(target=export, args=(should_terminate, frames_to_export, current_frame,
                                                              export_dir, user_dir, self.g_pool.min_data_confidence,
                                                              start_frame, end_frame, plugins, out_file_path))
                self.exports.append(process)

def init_marker_cacher(self):
        from marker_detector_cacher import fill_cache
        visited_list = [False if x == False else True for x in self.cache]
        video_file_path =  self.g_pool.capture.source_path
        timestamps = self.g_pool.capture.timestamps
        self.cache_queue = mp.Queue()
        self.cacher_seek_idx = mp.Value('i',0)
        self.cacher_run = mp.Value(c_bool,True)
        self.cacher = mp.Process(target=fill_cache, args=(visited_list,video_file_path,timestamps,self.cache_queue,self.cacher_seek_idx,self.cacher_run,self.min_marker_perimeter_cacher))
        self.cacher.start()

def is_process_64_from_handle(hProcess):
    """ Take a process handle. return True if process is 64 bits, and False otherwise. """
    iswow64 = c_bool(False)
    if IsWow64Process is None:
        return False
    if not IsWow64Process(hProcess, byref(iswow64)):
        raise WinError()
    return not iswow64.value

def __init__(self):
        self.lock = multiprocessing.Lock()
        self.readers_condition = multiprocessing.Condition(self.lock)
        self.writer_condition = multiprocessing.Condition(self.lock)
        self.readers = multiprocessing.RawValue(ctypes.c_uint, 0)
        self.writer = multiprocessing.RawValue(ctypes.c_bool, False)

def is_process_64_from_handle(hProcess):
    """ Take a process handle. return True if process is 64 bits, and False otherwise. """
    iswow64 = c_bool(False)
    if IsWow64Process is None:
        return False
    if not IsWow64Process(hProcess, byref(iswow64)):
        raise WinError()
    return not iswow64.value

def __init__(self,
                 mitmproxy_options: mitmproxy_extensions.options.MitmproxyExtendedOptions) -> None:
        self.mitmproxy_options = mitmproxy_options
        # setup proxy server from options
        proxy_config = mitmproxy.proxy.config.ProxyConfig(mitmproxy_options)
        self._proxy_server = mitmproxy.proxy.server.ProxyServer(proxy_config)

        self._results_queue = multiprocessing.Queue()
        self._producer_push_event = multiprocessing.Event()  # type: ignore
        self._has_active_flows_state = multiprocessing.Value(ctypes.c_bool, False)

        self._proxy_proc = None  # type: t.Optional[ProxyProc]