Python ctypes 模块，c_int16() 实例源码

def button(self, y: ctypes.c_int16, action : Action):
        mouse = pygame.mouse.get_pos()
        click = pygame.mouse.get_pressed()

        if 40 < mouse[0] < self.res[0] - 40 and y < mouse[1] < y + 24:
            if click[0] == 0:
                self.MAIN.blit(_button2, (40, y))
            elif click[0] == 1:
                self.MAIN.blit(_button3, (40, y))

                if action.histoireCiblee != None:
                    Histoire.histoire = action.histoireCiblee
                if Histoire.page.son != None:
                    st.stopsound(Histoire.page.son[0])
                Histoire.page = Histoire.chargerpage(action.cible)

        else:
            self.MAIN.blit(_button1, (40, y))

        self._write(action.desc[0], action.desc[1], 20, self.res[0] // 2, y + 13, center = True)

def read_sphere_wav(file_name):
    wav_file = open(file_name, 'rb')
    raw_header = wav_file.read(1024).decode('utf-8')
    raw_data = wav_file.read()
    sample_count = len(raw_data) // 2

    wav_data = np.zeros(shape=[sample_count], dtype=np.int32)

    for i in range(sample_count):
        wav_data[i] = ctypes.c_int16(ord(raw_data[2 * i + 1]) << 8).value + ctypes.c_int16(ord(raw_data[2 * i])).value

    header_list = raw_header.split("\n")
    sphere_header = {}
    for s in header_list:
        if len(s) > 0 and s != "end_head":
            tmp = s.split(" ")
            if len(tmp) < 3 and len(tmp) > 0:
                sphere_header['Name'] = tmp[0]
            elif len(tmp[0]) > 0:
                sphere_header[tmp[0]] = tmp[2]

    return wav_data, sphere_header

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 read_var_short(self):
        ret = 0
        size = 0
        while size < 16:
            byte = self.read_byte()
            has_next = (byte & 0x80) == 0x80
            if size > 0:
                ret = ret + ((byte & 0x7F) << size)
            else:
                ret = ret + (byte & 0x7F)
            size = size + 7
            if not has_next:
                if ret > 32767:
                    ret = ret - 65536
                return ctypes.c_int16(ret).value
        raise RuntimeError("Too much data")

def _write(self, text: str, textColor: pygame.Color, size : ctypes.c_int16, x: ctypes.c_int16, y: ctypes.c_int16, center: bool = False):
        visual = pygame.font.SysFont("Candara", size)
        surface = visual.render(text, True, textColor)

        if not center:
            rectangle = surface.get_rect(x = x, y = y)
        else:
            rectangle = surface.get_rect(centerx = x, centery = y)

        self.MAIN.blit(surface, rectangle)

def read_sphere_wav(file_name):
    wav_file = open(file_name, 'rb')
    raw_header = wav_file.read(1024).decode('utf-8')
    raw_data = wav_file.read()
    sample_count = len(raw_data) // 2

    wav_data = np.zeros(shape=[sample_count], dtype=np.int32)

    for i in range(sample_count):
        wav_data[i] = ctypes.c_int16(ord(raw_data[2 * i + 1]) << 8).value + ctypes.c_int16(ord(raw_data[2 * i])).value

    header_list = raw_header.split("\n")
    sphere_header = {}
    for s in header_list:
        if len(s) > 0 and s != "end_head":
            tmp = s.split(" ")
            if 0 < len(tmp) < 3:
                sphere_header['Name'] = tmp[0]
            elif len(tmp[0]) > 0:
                sphere_header[tmp[0]] = tmp[2]

    return wav_data, sphere_header


# axis=0 for col vectors
# axis=1 for row vectors

def test_intfield_with_variable_bit_lenth(self):
        """
        This test verifies that setting an integer value of variable size is correctly exported to the to_bytes 
        function.  This also tests the ability to set a value for the packet upon instantiation.  
        """
        class int_packet_with_varied_sized_int_fields(models.Packet):
            int_field = models.IntField()
            int_field_signed = models.IntField(signed=True)
            int_field_4_bits = models.IntField(bit_len=4)
            int_field_12_bits = models.IntField(bit_len=12)

        pkt = int_packet_with_varied_sized_int_fields(
            int_field = 0xbeef,
            int_field_signed = 0xdead,
            int_field_4_bits = 0xa,
            int_field_12_bits = 0xbc
        )

        class c_pkt_struct(ctypes.Structure):
            _fields_ = (
                ('int_field', ctypes.c_uint16),
                ('int_field_signed', ctypes.c_int16),
                ('int_field_4_bits', ctypes.c_uint16, 4),
                ('int_field_12_bits', ctypes.c_uint16, 12),
            )

        c_pkt = c_pkt_struct()
        c_pkt.int_field = 0xbeef
        c_pkt.int_field_signed = 0xdead
        c_pkt.int_field_4_bits = 0xa
        c_pkt.int_field_12_bits = 0xbc

        b_str = ctypes.string_at(ctypes.addressof(c_pkt), ctypes.sizeof(c_pkt))

        self.assertEquals(b_str, pkt.to_bytes())

def validate_i16(val, p, key=None):
    if isinstance(val, int) and ctypes.c_int16(val).value == val:
        return
    raise_validation_error(ErrInvalidI16, val, p, key=key)

def _branch_always(self, i):
        """Emulate the BRA and BRL instructions.

        Args:
            i: The instruction.

        Returns:
            The target of the branch.
        """
        if i.address_mode == Mode.RELATIVE:
            self.pc += c_int8(i.operand).value
        elif i.address_mode == Mode.RELATIVE_LONG:
            self.pc += c_int16(i.operand).value
        return self.pc

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 decode(decoder, data, length, frame_size, decode_fec, channels=2):
    """Decode an Opus frame

    Unlike the `opus_decode` function , this function takes an additional
    parameter `channels`, which indicates the number of channels in the frame
    """

    pcm_size = frame_size * channels * ctypes.sizeof(ctypes.c_int16)
    pcm = (ctypes.c_int16 * pcm_size)()
    pcm_pointer = ctypes.cast(pcm, opuslib.api.c_int16_pointer)

    # Converting from a boolean to int
    decode_fec = int(bool(decode_fec))

    try:
        result = _decode(
            decoder, data, length, pcm_pointer, frame_size, decode_fec)
    except ctypes.ArgumentError:
        result = _decode(
            decoder, bytes(data.encode('latin-1')), length, pcm_pointer,
            frame_size, decode_fec)
    if result < 0:
        raise opuslib.exceptions.OpusError(result)

    return array.array('h', pcm[:result * channels]).tostring()

def setup(self):

        # counter "ValueError: number of bits invalid for bit field"
        monkeypatch_pyclibrary_ctypes_struct()

        # register header- and library paths
        # https://pyclibrary.readthedocs.org/en/latest/get_started/configuration.html#specifying-headers-and-libraries-locations
        # TODO: this probably acts on a global basis; think about it
        if self.include_path:
            add_header_locations([self.include_path])
        if self.library_path:
            add_library_locations([self.library_path])

        # define extra types suitable for embedded use
        types = {
            'uint8_t': c_uint8,
            'uint16_t': c_uint16,
            'uint32_t': c_uint32,
            'int8_t': c_int8,
            'int16_t': c_int16,
            'int32_t': c_int32,
            }

        # TODO: this probably acts on a global basis; think about it
        if not (CParser._init or CLibrary._init):
            auto_init(extra_types=types)

def rawread(self):
        aty = ctypes.c_int16 * self.samplesPerPacket
        xi = aty()
        xq = aty()
        firstSampleNum = ctypes.c_uint(0)
        grChanged = ctypes.c_uint(0)
        rfChanged = ctypes.c_uint(0)
        fsChanged = ctypes.c_uint(0)
        err = self.lib.mir_sdr_ReadPacket(ctypes.byref(xi),
                                          ctypes.byref(xq),
                                          ctypes.byref(firstSampleNum),
                                          ctypes.byref(grChanged),
                                          ctypes.byref(rfChanged),
                                          ctypes.byref(fsChanged))
        if err != 0:
            sys.stderr.write("sdrplay: mir_sdrReadPacket failed: %d\n" % (err))
            sys.exit(1)

        # I don't know if these are needed.
        if grChanged.value != 0:
            self.lib.mir_sdr_ResetUpdateFlags(1, 0, 0)
        if rfChanged.value != 0:
            self.lib.mir_sdr_ResetUpdateFlags(0, 1, 0)
        if fsChanged.value != 0:
            self.lib.mir_sdr_ResetUpdateFlags(0, 0, 1)

        # ii = numpy.fromstring(xi, dtype=numpy.int16)
        # qq = numpy.fromstring(xq, dtype=numpy.int16)

        ii = [ xi[i] for i in range(0, self.samplesPerPacket) ]
        ii = numpy.array(ii).astype(numpy.float64)

        qq = [ xq[i] for i in range(0, self.samplesPerPacket) ]
        qq = numpy.array(qq).astype(numpy.float64)

        iq = ii + 1j*qq

        # theory: firstSampleNum is only 16 bits, upper bits are junk.
        num = firstSampleNum.value
        num = num & 0xffff
        if self.expect != None and num != self.expect:
            print("%d vs %d -- gap %d" % (self.expect, num, num - self.expect))

        self.expect = (num + self.samplesPerPacket) & 0xffff

        return iq

    # blocking read.
    # returns [ samples, end_time ]