我们从Python开源项目中，提取了以下48个代码示例，用于说明如何使用ctypes.c_uint64()。
def get(self): arg = DrmModeObjGetPropertiesC() arg.obj_id = self.obj_id arg.obj_type = self.obj_type fcntl.ioctl(self._drm.fd, DRM_IOCTL_MODE_OBJ_GETPROPERTIES, arg) prop_ids = (ctypes.c_uint32*arg.count_props)() arg.props_ptr = ctypes.cast(ctypes.pointer(prop_ids), ctypes.c_void_p).value prop_values = (ctypes.c_uint64*arg.count_props)() arg.prop_values_ptr = ctypes.cast(ctypes.pointer(prop_values), ctypes.c_void_p).value fcntl.ioctl(self._drm.fd, DRM_IOCTL_MODE_OBJ_GETPROPERTIES, arg) self._arg = arg vals = [v for i, v in zip(prop_ids, prop_values) if i == self.id] return vals[0]
def add(self, uuid, tstamp, values): """Add an event in TrailDB. uuid -- UUID of this event. tstamp -- Timestamp of this event (datetime or integer). values -- value of each field. """ if isinstance(tstamp, datetime): tstamp = int(time.mktime(tstamp.timetuple())) n = len(self.ofields) value_array = (c_char_p * n)(*values) value_lengths = (c_uint64 * n)(*[len(v) for v in values]) f = lib.tdb_cons_add(self._cons, uuid_raw(uuid), tstamp, value_array, value_lengths) if f: raise TrailDBError("Too many values: %s" % values[f])
def _return_ctype(self): """ Returns the associated ctype of a given datatype. """ _datatype_ctype = { DataType.Bool: ctypes.c_uint8, DataType.I8: ctypes.c_int8, DataType.U8: ctypes.c_uint8, DataType.I16: ctypes.c_int16, DataType.U16: ctypes.c_uint16, DataType.I32: ctypes.c_int32, DataType.U32: ctypes.c_uint32, DataType.I64: ctypes.c_int64, DataType.U64: ctypes.c_uint64, DataType.Sgl: ctypes.c_float, DataType.Dbl: ctypes.c_double, } return _datatype_ctype[self]
def callback_file_recv(self, callback, user_data): """ Set the callback for the `file_recv` event. Pass None to unset. This event is triggered when a file transfer request is received. :param callback: Python function. The client should acquire resources to be associated with the file transfer. Incoming file transfers start in the PAUSED state. After this callback returns, a transfer can be rejected by sending a TOX_FILE_CONTROL_CANCEL control command before any other control commands. It can be accepted by sending TOX_FILE_CONTROL_RESUME. Should take pointer (c_void_p) to Tox object, The friend number (c_uint32) of the friend who is sending the file transfer request. The friend-specific file number (c_uint32) the data received is associated with. The meaning of the file (c_uint32) to be sent. Size in bytes (c_uint64) of the file the client wants to send, UINT64_MAX if unknown or streaming. Name of the file (c_char_p). Does not need to be the actual name. This name will be sent along with the file send request. Size in bytes (c_size_t) of the filename. 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_uint32, c_uint32, c_uint64, c_char_p, c_size_t, c_void_p) self.file_recv_cb = c_callback(callback) self.libtoxcore.tox_callback_file_recv(self._tox_pointer, self.file_recv_cb, user_data)
def calcHash(buf): global _nhash #buf = b"\x61\x24\x7f\xbf\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00" buf = list(bytearray(buf)) #print buf num_bytes = len(buf) array_type = ctypes.c_ubyte * num_bytes data = _nhash.compute_hash(array_type(*buf), ctypes.c_uint32(num_bytes)); #print data return ctypes.c_uint64(data).value
def fetch(self): self.type_name = "enum" arg = DrmModeObjGetPropertyC() arg.prop_id = self.id fcntl.ioctl(self._drm.fd, DRM_IOCTL_MODE_GETPROPERTY, arg) if not (arg.count_enum_blobs and (arg.flags & DRM_MODE_PROP_ENUM)): raise ValueError("not an enum property") if arg.count_values != arg.count_enum_blobs: raise ValueError("count_values != count_enum_blobs") values = (ctypes.c_uint64*arg.count_values)() arg.values_ptr = ctypes.cast(ctypes.pointer(values), ctypes.c_void_p).value enum_blobs = (DrmModePropertyEnumC*arg.count_enum_blobs)() arg.enum_blob_ptr = ctypes.cast(ctypes.pointer(enum_blobs), ctypes.c_void_p).value fcntl.ioctl(self._drm.fd, DRM_IOCTL_MODE_GETPROPERTY, arg) self.enum = {} for i in range(arg.count_enum_blobs): self.enum[int(values[i])] = enum_blobs[i].name self._arg = arg self.name = arg.name self.flags = arg.flags self.immutable = True if (arg.flags & DRM_MODE_PROP_IMMUTABLE) else False
def __waitForMultipleObjects( self, all_handles ): num_handles = len( all_handles ) wait_handles_t = ctypes.c_uint64 * num_handles wait_handles = wait_handles_t( *all_handles ) return ctypes.windll.kernel32.WaitForMultipleObjects( num_handles, ctypes.byref( wait_handles ), 0, INFINITE )
def sendfile(fdout, fdin, offset, nbytes): if sys.platform == 'darwin': _sendfile.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_uint64, ctypes.POINTER(ctypes.c_uint64), ctypes.c_voidp, ctypes.c_int] _nbytes = ctypes.c_uint64(nbytes) result = _sendfile(fdin, fdout, offset, _nbytes, None, 0) if result == -1: e = ctypes.get_errno() if e == errno.EAGAIN and _nbytes.value is not None: return _nbytes.value raise OSError(e, os.strerror(e)) return _nbytes.value elif sys.platform in ('freebsd', 'dragonfly',): _sendfile.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_uint64, ctypes.c_uint64, ctypes.c_voidp, ctypes.POINTER(ctypes.c_uint64), ctypes.c_int] _sbytes = ctypes.c_uint64() result = _sendfile(fdin, fdout, offset, nbytes, None, _sbytes, 0) if result == -1: e = ctypes.get_errno() if e == errno.EAGAIN and _sbytes.value is not None: return _sbytes.value raise OSError(e, os.strerror(e)) return _sbytes.value else: _sendfile.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.POINTER(ctypes.c_uint64), ctypes.c_size_t] _offset = ctypes.c_uint64(offset) sent = _sendfile(fdout, fdin, _offset, nbytes) if sent == -1: e = ctypes.get_errno() raise OSError(e, os.strerror(e)) return sent
def open(self): nbyte = ctypes.c_uint64() #print '>ipcbuf_get_next_read' ptr = _dada.ipcbuf_get_next_read(self.buf, nbyte) nbyte = nbyte.value block_id = 0 return ptr, nbyte, block_id
def open(self): nbyte = ctypes.c_uint64() block_id = ctypes.c_uint64() #print '>ipcio_open_block_read' ptr = _dada.ipcio_open_block_read(self.io, nbyte, block_id) nbyte = nbyte.value block_id = block_id.value #print 'block_id =', block_id return ptr, nbyte, block_id
def open(self): nbyte = self.size_bytes() block_id = ctypes.c_uint64() #print '>ipcio_open_block_write' ptr = _dada.ipcio_open_block_write(self.io, block_id) block_id = block_id.value #print 'block_id =', block_id return ptr, nbyte, block_id
def __init__(self, path): """Open a TrailDB at path.""" self._db = db = lib.tdb_init() res = lib.tdb_open(self._db, path) if res != 0: raise TrailDBError("Could not open %s, error code %d" % (path, res)) self.num_trails = lib.tdb_num_trails(db) self.num_events = lib.tdb_num_events(db) self.num_fields = lib.tdb_num_fields(db) self.fields = [lib.tdb_get_field_name(db, i) for i in xrange(self.num_fields)] self._event_cls = namedtuple('event', self.fields, rename=True) self._uint64_ptr = pointer(c_uint64())
def validate_u64(val, p, key=None): if isinstance(val, (int, long)) and ctypes.c_uint64(val).value == val: return raise_validation_error(ErrInvalidU64, val, p, key=key)
def struct(cls, ea, **sid): """Return the structure_t at address ``ea`` as a dict of ctypes. If the structure ``sid`` is specified, then use that specific structure type. """ ea = interface.address.within(ea) if any(n in sid for n in ('sid','struc','structure','id')): res = sid['sid'] if 'sid' in sid else sid['struc'] if 'struc' in sid else sid['structure'] if 'structure' in sid else sid['id'] if 'id' in sid else None sid = res.id if isinstance(res, structure.structure_t) else res else: sid = type.structure.id(ea) st = structure.instance(sid, offset=ea) typelookup = { (int,-1) : ctypes.c_int8, (int,1) : ctypes.c_uint8, (int,-2) : ctypes.c_int16, (int,2) : ctypes.c_uint16, (int,-4) : ctypes.c_int32, (int,4) : ctypes.c_uint32, (int,-8) : ctypes.c_int64, (int,8) : ctypes.c_uint64, (float,4) : ctypes.c_float, (float,8) : ctypes.c_double, } res = {} for m in st.members: t, val = m.type, read(m.offset, m.size) or '' try: ct = typelookup[t] except KeyError: ty, sz = t if isinstance(t, __builtin__.tuple) else (m.type, 0) if isinstance(t, __builtin__.list): t = typelookup[tuple(ty)] ct = t*sz elif ty in (chr,str): ct = ctypes.c_char*sz else: ct = None finally: res[m.name] = val if any(_ is None for _ in (ct,val)) else ctypes.cast(ctypes.pointer(ctypes.c_buffer(val)),ctypes.POINTER(ct)).contents return res
def raise_an_exception(): """ A helper for NiFpgaStatusExceptionTest """ session = ctypes.c_int32(0x0000beef) fifo = ctypes.c_uint32(0x0000f1f0) data = ctypes.c_uint64(0x0000da7a) number_of_elements = ctypes.c_size_t(0x100) timeout_ms = ctypes.c_size_t(0x200) elements_remaining = ctypes.c_size_t(0x300) bogus_string_argument = ctypes.c_char_p(b"I am a string") exception = nifpga.FifoTimeoutError( function_name="Dummy Function Name", argument_names=["session", "fifo", "data", "number of elements", "timeout ms", "elements remaining", "a bogus string argument"], function_args=(session, fifo, data, number_of_elements, timeout_ms, elements_remaining, bogus_string_argument)) raise exception
def file_seek(self, friend_number, file_number, position): """ Sends a file seek control command to a friend for a given file transfer. This function can only be called to resume a file transfer right before TOX_FILE_CONTROL_RESUME is sent. :param friend_number: The friend number of the friend the file is being received from. :param file_number: The friend-specific identifier for the file transfer. :param position: The position that the file should be seeked to. :return: True on success. """ tox_err_file_seek = c_int() result = Tox.libtoxcore.tox_file_control(self._tox_pointer, c_uint32(friend_number), c_uint32(file_number), c_uint64(position), byref(tox_err_file_seek)) tox_err_file_seek = tox_err_file_seek.value if tox_err_file_seek == TOX_ERR_FILE_SEEK['OK']: return bool(result) elif tox_err_file_seek == TOX_ERR_FILE_SEEK['FRIEND_NOT_FOUND']: raise ArgumentError('The friend_number passed did not designate a valid friend.') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['FRIEND_NOT_CONNECTED']: raise ArgumentError('This client is currently not connected to the friend.') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['NOT_FOUND']: raise ArgumentError('No file transfer with the given file number was found for the given friend.') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['SEEK_DENIED']: raise IOError('File was not in a state where it could be seeked.') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['INVALID_POSITION']: raise ArgumentError('Seek position was invalid') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['SENDQ']: raise RuntimeError('Packet queue is full.')
def callback_file_chunk_request(self, callback, user_data): """ Set the callback for the `file_chunk_request` event. Pass None to unset. This event is triggered when Core is ready to send more file data. :param callback: Python function. If the length parameter is 0, the file transfer is finished, and the client's resources associated with the file number should be released. After a call with zero length, the file number can be reused for future file transfers. If the requested position is not equal to the client's idea of the current file or stream position, it will need to seek. In case of read-once streams, the client should keep the last read chunk so that a seek back can be supported. A seek-back only ever needs to read from the last requested chunk. This happens when a chunk was requested, but the send failed. A seek-back request can occur an arbitrary number of times for any given chunk. In response to receiving this callback, the client should call the function `tox_file_send_chunk` with the requested chunk. If the number of bytes sent through that function is zero, the file transfer is assumed complete. A client must send the full length of data requested with this callback. Should take pointer (c_void_p) to Tox object, The friend number (c_uint32) of the receiving friend for this file. The file transfer identifier (c_uint32) returned by tox_file_send. The file or stream position (c_uint64) from which to continue reading. The number of bytes (c_size_t) requested for the current chunk. 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_uint32, c_uint64, c_size_t, c_void_p) self.file_chunk_request_cb = c_callback(callback) self.libtoxcore.tox_callback_file_chunk_request(self._tox_pointer, self.file_chunk_request_cb, user_data) # ----------------------------------------------------------------------------------------------------------------- # File transmission: receiving # -----------------------------------------------------------------------------------------------------------------
def callback_file_recv_chunk(self, callback, user_data): """ Set the callback for the `file_recv_chunk` event. Pass NULL to unset. This event is first triggered when a file transfer request is received, and subsequently when a chunk of file data for an accepted request was received. :param callback: Python function. When length is 0, the transfer is finished and the client should release the resources it acquired for the transfer. After a call with length = 0, the file number can be reused for new file transfers. If position is equal to file_size (received in the file_receive callback) when the transfer finishes, the file was received completely. Otherwise, if file_size was UINT64_MAX, streaming ended successfully when length is 0. Should take pointer (c_void_p) to Tox object, The friend number (c_uint32) of the friend who is sending the file. The friend-specific file number (c_uint32) the data received is associated with. The file position (c_uint64) of the first byte in data. A byte array (c_char_p) containing the received chunk. The length (c_size_t) of the received chunk. 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_uint32, c_uint64, POINTER(c_uint8), c_size_t, c_void_p) self.file_recv_chunk_cb = c_callback(callback) self.libtoxcore.tox_callback_file_recv_chunk(self._tox_pointer, self.file_recv_chunk_cb, user_data) # ----------------------------------------------------------------------------------------------------------------- # Low-level custom packet sending and receiving # -----------------------------------------------------------------------------------------------------------------
def getindex(self): index_size = ctypes.c_uint64(0) index_data = ctypes.POINTER(ctypes.c_uint64)() check_call(_LIB.MXDataIterGetIndex(self.handle, ctypes.byref(index_data), ctypes.byref(index_size))) address = ctypes.addressof(index_data.contents) dbuffer = (ctypes.c_uint64* index_size.value).from_address(address) np_index = np.frombuffer(dbuffer, dtype=np.uint64) return np_index.copy()
def translate_type(t): table = { "float": 'c_float', "uint32_t": 'c_uint32', "uint64_t": 'c_uint64', "size_t": 'c_size_t', "float": 'c_float', 'int32_t': 'c_int32', 'int': 'c_int32', 'uint8_t': 'c_int8', "char": "c_char", "void": "None", "void*": "c_void_p", "const void*": 'c_void_p', "const char*": 'c_char_p', "const char* const*": 'POINTER(c_char_p)', "struct wl_display*": "POINTER(wl_display)", "struct wl_surface*": "POINTER(wl_surface)", "const ObjectTableEntryNVX* const*": "POINTER(POINTER(ObjectTableEntryNVX))", 'v': '' } if t in table.keys(): return table[t] if t.endswith("*"): if t.startswith("const"): ttype = t[6:len(t)-1] ttype = table[ttype] if ttype in table else ttype return "POINTER({})".format(ttype) else: ttype = t[:len(t)-1] ttype = table[ttype] if ttype in table else ttype return "POINTER({})".format(ttype) return t
def parse_handles_def(f): f.write("# Handles types\n") handles = re.findall("VK_DEFINE_HANDLE\(Vk(\w+)\)", src, re.S) for h in handles: f.write("{} = c_size_t\n".format(h)) handles_non_dispatchable = re.findall("VK_DEFINE_NON_DISPATCHABLE_HANDLE\(Vk(\w+)\)", src, re.S) for h in handles_non_dispatchable: f.write("{} = c_uint64\n".format(h))
def parse_allocation_callback(f): # Allocation callback must be defined before the structs, but there are no good way to differenciate them # from the function pointers. Hence why they are hardcoded here f.write(""" # Allocation callback FnAllocationFunction = FUNCTYPE(c_void_p, c_void_p, c_size_t, c_size_t, SystemAllocationScope) FnReallocationFunction = FUNCTYPE(c_void_p, c_void_p, c_size_t, c_size_t, SystemAllocationScope) FnFreeFunction = FUNCTYPE(None, c_void_p, c_void_p) FnInternalAllocationNotification = FUNCTYPE(None, c_void_p, c_size_t, InternalAllocationType, SystemAllocationScope) FnInternalFreeNotification = FUNCTYPE(None, c_void_p, c_size_t, InternalAllocationType, SystemAllocationScope) FnDebugReportCallbackEXT = FUNCTYPE(Bool32, DebugReportFlagsEXT, DebugReportObjectTypeEXT, c_uint64, c_size_t, c_uint32, c_char_p, c_char_p, c_void_p) """[1::])
def file_seek(self, friend_number, file_number, position): """ Sends a file seek control command to a friend for a given file transfer. This function can only be called to resume a file transfer right before TOX_FILE_CONTROL_RESUME is sent. :param friend_number: The friend number of the friend the file is being received from. :param file_number: The friend-specific identifier for the file transfer. :param position: The position that the file should be seeked to. :return: True on success. """ tox_err_file_seek = c_int() result = Tox.libtoxcore.tox_file_control(self._tox_pointer, c_uint32(friend_number), c_uint32(file_number), c_uint64(position), byref(tox_err_file_seek)) tox_err_file_seek = tox_err_file_seek.value if tox_err_file_seek == TOX_ERR_FILE_SEEK['OK']: return bool(result) elif tox_err_file_seek == TOX_ERR_FILE_SEEK['FRIEND_NOT_FOUND']: raise ArgumentError('The friend_number passed did not designate a valid friend.') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['FRIEND_NOT_CONNECTED']: raise RuntimeError('This client is currently not connected to the friend.') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['NOT_FOUND']: raise ArgumentError('No file transfer with the given file number was found for the given friend.') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['SEEK_DENIED']: raise ArgumentError('File was not in a state where it could be seeked.') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['INVALID_POSITION']: raise ArgumentError('Seek position was invalid') elif tox_err_file_seek == TOX_ERR_FILE_SEEK['SENDQ']: raise ArgumentError('Packet queue is full.')
def callback_file_recv_chunk(self, callback, user_data): """ Set the callback for the `file_recv_chunk` event. Pass NULL to unset. This event is first triggered when a file transfer request is received, and subsequently when a chunk of file data for an accepted request was received. :param callback: Python function. When length is 0, the transfer is finished and the client should release the resources it acquired for the transfer. After a call with length = 0, the file number can be reused for new file transfers. If position is equal to file_size (received in the file_receive callback) when the transfer finishes, the file was received completely. Otherwise, if file_size was UINT64_MAX, streaming ended successfully when length is 0. Should take pointer (c_void_p) to Tox object, The friend number (c_uint32) of the friend who is sending the file. The friend-specific file number (c_uint32) the data received is associated with. The file position (c_uint64) of the first byte in data. A byte array (c_char_p) containing the received chunk. The length (c_size_t) of the received chunk. 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_uint32, c_uint64, POINTER(c_uint8), c_size_t, c_void_p) self.file_recv_chunk_cb = c_callback(callback) self.libtoxcore.tox_callback_file_recv_chunk(self._tox_pointer, self.file_recv_chunk_cb, user_data) # ----------------------------------------------------------------------------------------------------------------- # Low-level network information # -----------------------------------------------------------------------------------------------------------------
def read_var_uh_long(self): _loc3_ = 0 _loc2_low = 0 _loc2_high = 0 _loc4_ = 0 while True: _loc3_ = self.read_unsigned_byte() if _loc4_ == 28: break if _loc3_ >= 128: _loc2_low = _loc2_low | (_loc3_ & 127) << _loc4_ _loc4_ = _loc4_ + 7 continue _loc2_low = _loc2_low | _loc3_ << _loc4_ return ctypes.c_uint64(_loc2_high * 4294967296 + _loc2_low).value if _loc3_ >= 128: _loc3_ = _loc3_ & 127 _loc2_low = _loc2_low | _loc3_ << _loc4_ _loc2_high = _lrshift32(_loc3_, 4) _loc4_ = 3 while True: _loc3_ = self.read_unsigned_byte() if _loc4_ < 32: if _loc3_ >= 128: _loc2_high = _loc2_high | (_loc3_ & 127) << _loc4_ else: break _loc4_ = _loc4_ + 7 _loc2_high = _loc2_high | (_loc3_ << _loc4_) return ctypes.c_uint64(_loc2_high * 4294967296 + _loc2_low).value _loc2_low = _loc2_low | (_loc3_ << _loc4_) _loc2_high = _lrshift32(_loc3_, 4) return ctypes.c_uint64(_loc2_high * 4294967296 + _loc2_low).value
def call_hash(self, buf): buf = list(bytearray(buf)) num_bytes = len(buf) array_type = ctypes.c_ubyte * num_bytes data = self._hash_lib.compute_hash(array_type(*buf), ctypes.c_uint32(num_bytes)) return ctypes.c_uint64(data).value
def __init__(self, drm, id_, type_): self._drm = drm self.id = id_ self.type = type_ self.props = [] arg = DrmModeObjGetPropertiesC() arg.obj_id = self.id arg.obj_type = self.type fcntl.ioctl(self._drm.fd, DRM_IOCTL_MODE_OBJ_GETPROPERTIES, arg) if arg.count_props == 0: #print("DrmProperties(%d, 0x%x): arg.count_props=%d" % (self.id, self.type, arg.count_props)) return prop_ids = (ctypes.c_uint32*arg.count_props)() arg.props_ptr = ctypes.cast(ctypes.pointer(prop_ids), ctypes.c_void_p).value prop_values = (ctypes.c_uint64*arg.count_props)() arg.prop_values_ptr = ctypes.cast(ctypes.pointer(prop_values), ctypes.c_void_p).value fcntl.ioctl(self._drm.fd, DRM_IOCTL_MODE_OBJ_GETPROPERTIES, arg) self._arg = arg for i in range(arg.count_props): propid = int(prop_ids[i]) propc = DrmModeObjGetPropertyC() propc.prop_id = propid fcntl.ioctl(self._drm.fd, DRM_IOCTL_MODE_GETPROPERTY, propc) if propc.count_enum_blobs: if propc.flags & DRM_MODE_PROP_ENUM: prop = DrmPropertyEnum(self._drm, propid, self.id, self.type) elif propc.flags & DRM_MODE_PROP_BITMASK: prop = DrmPropertyBitmask(self._drm, propid, self.id, self.type, propc) else: raise ValueError("count_enum_blobs: propc.flags=0x%x" % propc.flags) elif propc.flags & DRM_MODE_PROP_BLOB: prop = DrmPropertyBlob(self._drm, propid, self.id, self.type, propc) else: raise ValueError("not count_enum_blobs: propc.flags=0x%x" % propc.flags) self.props.append(prop)
def file_send_chunk(self, friend_number, file_number, position, data): """ Send a chunk of file data to a friend. This function is called in response to the `file_chunk_request` callback. The length parameter should be equal to the one received though the callback. If it is zero, the transfer is assumed complete. For files with known size, Core will know that the transfer is complete after the last byte has been received, so it is not necessary (though not harmful) to send a zero-length chunk to terminate. For streams, core will know that the transfer is finished if a chunk with length less than the length requested in the callback is sent. :param friend_number: The friend number of the receiving friend for this file. :param file_number: The file transfer identifier returned by tox_file_send. :param position: The file or stream position from which to continue reading. :param data: Chunk of file data :return: true on success. """ tox_err_file_send_chunk = c_int() result = self.libtoxcore.tox_file_send_chunk(self._tox_pointer, c_uint32(friend_number), c_uint32(file_number), c_uint64(position), c_char_p(data), c_size_t(len(data)), byref(tox_err_file_send_chunk)) tox_err_file_send_chunk = tox_err_file_send_chunk.value if tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['OK']: return bool(result) elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['NULL']: raise ArgumentError('The length parameter was non-zero, but data was NULL.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['FRIEND_NOT_FOUND']: ArgumentError('The friend_number passed did not designate a valid friend.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['FRIEND_NOT_CONNECTED']: raise ArgumentError('This client is currently not connected to the friend.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['NOT_FOUND']: raise ArgumentError('No file transfer with the given file number was found for the given friend.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['NOT_TRANSFERRING']: raise ArgumentError('File transfer was found but isn\'t in a transferring state: (paused, done, broken, ' 'etc...) (happens only when not called from the request chunk callback).') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['INVALID_LENGTH']: raise ArgumentError('Attempted to send more or less data than requested. The requested data size is ' 'adjusted according to maximum transmission unit and the expected end of the file. ' 'Trying to send less or more than requested will return this error.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['SENDQ']: raise RuntimeError('Packet queue is full.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['WRONG_POSITION']: raise ArgumentError('Position parameter was wrong.')
def file_send_chunk(self, friend_number, file_number, position, data): """ Send a chunk of file data to a friend. This function is called in response to the `file_chunk_request` callback. The length parameter should be equal to the one received though the callback. If it is zero, the transfer is assumed complete. For files with known size, Core will know that the transfer is complete after the last byte has been received, so it is not necessary (though not harmful) to send a zero-length chunk to terminate. For streams, core will know that the transfer is finished if a chunk with length less than the length requested in the callback is sent. :param friend_number: The friend number of the receiving friend for this file. :param file_number: The file transfer identifier returned by tox_file_send. :param position: The file or stream position from which to continue reading. :param data: Chunk of file data :return: true on success. """ tox_err_file_send_chunk = c_int() result = self.libtoxcore.tox_file_send_chunk(self._tox_pointer, c_uint32(friend_number), c_uint32(file_number), c_uint64(position), c_char_p(data), c_size_t(len(data)), byref(tox_err_file_send_chunk)) tox_err_file_send_chunk = tox_err_file_send_chunk.value if tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['OK']: return bool(result) elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['NULL']: raise ArgumentError('The length parameter was non-zero, but data was NULL.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['FRIEND_NOT_FOUND']: ArgumentError('The friend_number passed did not designate a valid friend.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['FRIEND_NOT_CONNECTED']: raise ArgumentError('This client is currently not connected to the friend.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['NOT_FOUND']: raise ArgumentError('No file transfer with the given file number was found for the given friend.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['NOT_TRANSFERRING']: raise ArgumentError('File transfer was found but isn\'t in a transferring state: (paused, done, broken, ' 'etc...) (happens only when not called from the request chunk callback).') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['INVALID_LENGTH']: raise ArgumentError('Attempted to send more or less data than requested. The requested data size is ' 'adjusted according to maximum transmission unit and the expected end of the file. ' 'Trying to send less or more than requested will return this error.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['SENDQ']: raise ArgumentError('Packet queue is full.') elif tox_err_file_send_chunk == TOX_ERR_FILE_SEND_CHUNK['WRONG_POSITION']: raise ArgumentError('Position parameter was wrong.')
