python类calcsize()的实例源码

def decrypt_file(key, in_filename, out_filename=None, chunksize=24*1024):

    # Split .crypt extension to restore file format
    if not out_filename:
        out_filename = os.path.splitext(in_filename)[0]

    with open(in_filename, 'rb') as infile:
        origsize = struct.unpack('<Q', infile.read(struct.calcsize('Q')))[0]
        iv = infile.read(16)
        decryptor = AES.new(key, AES.MODE_CBC, iv)

        with open(out_filename, 'wb') as outfile:
            while True:
                chunk = infile.read(chunksize)
                if len(chunk) == 0:
                    break
                outfile.write(decryptor.decrypt(chunk))

        # Truncate file to original size
            outfile.truncate(origsize)

def run(self):
        print("VEDIO server starts...")
        self.sock.bind(self.ADDR)
        self.sock.listen(1)
        conn, addr = self.sock.accept()
        print("remote VEDIO client success connected...")
        data = "".encode("utf-8")
        payload_size = struct.calcsize("L")
        cv2.namedWindow('Remote', cv2.WINDOW_NORMAL)
        while True:
            while len(data) < payload_size:
                data += conn.recv(81920)
            packed_size = data[:payload_size]
            data = data[payload_size:]
            msg_size = struct.unpack("L", packed_size)[0]
            while len(data) < msg_size:
                data += conn.recv(81920)
            zframe_data = data[:msg_size]
            data = data[msg_size:]
            frame_data = zlib.decompress(zframe_data)
            frame = pickle.loads(frame_data)
            cv2.imshow('Remote', frame)
            if cv2.waitKey(1) & 0xFF == 27:
                break

def _getOffsets(self):
        """
        Calculate offsets to VDMX_Group records.
        For each ratRange return a list of offset values from the beginning of
        the VDMX table to a VDMX_Group.
        """
        lenHeader = sstruct.calcsize(VDMX_HeaderFmt)
        lenRatRange = sstruct.calcsize(VDMX_RatRangeFmt)
        lenOffset = struct.calcsize('>H')
        lenGroupHeader = sstruct.calcsize(VDMX_GroupFmt)
        lenVTable = sstruct.calcsize(VDMX_vTableFmt)
        # offset to the first group
        pos = lenHeader + self.numRatios*lenRatRange + self.numRatios*lenOffset
        groupOffsets = []
        for group in self.groups:
            groupOffsets.append(pos)
            lenGroup = lenGroupHeader + len(group) * lenVTable
            pos += lenGroup  # offset to next group
        offsets = []
        for ratio in self.ratRanges:
            groupIndex = ratio['groupIndex']
            offsets.append(groupOffsets[groupIndex])
        return offsets

def decompile(self, data, ttFont):
        sstruct.unpack2(Silf_hdr_format, data, self)
        if self.version >= 5.0:
            (data, self.scheme) = grUtils.decompress(data)
            sstruct.unpack2(Silf_hdr_format_3, data, self)
            base = sstruct.calcsize(Silf_hdr_format_3)
        elif self.version < 3.0:
            self.numSilf = struct.unpack('>H', data[4:6])
            self.scheme = 0
            self.compilerVersion = 0
            base = 8
        else:
            self.scheme = 0
            sstruct.unpack2(Silf_hdr_format_3, data, self)
            base = sstruct.calcsize(Silf_hdr_format_3)

        silfoffsets = struct.unpack_from(('>%dL' % self.numSilf), data[base:])
        for offset in silfoffsets:
            s = Silf()
            self.silfs.append(s)
            s.decompile(data[offset:], ttFont, self.version)

def read_f(self, offset=None):
        if offset is not None:
            self.fid.seek(offset)
        d = {}
        for key, fmt in self.description:
            buf = self.fid.read(struct.calcsize(fmt))
            if len(buf) != struct.calcsize(fmt):
                return None
            val = list(struct.unpack(fmt, buf))
            for i, ival in enumerate(val):
                if hasattr(ival, 'replace'):
                    ival = ival.replace(unicode.encode('\x03'), unicode.encode(''))
                    ival = ival.replace(unicode.encode('\x00'), unicode.encode(''))
                    val[i] = ival.decode("utf-8")
            if len(val) == 1:
                val = val[0]
            d[key] = val
        return d

def read_f(self, offset =None):
        if offset is not None :
            self.fid.seek(offset)
        d = { }
        for key, fmt in self.description :
            fmt = '<' + fmt # insures use of standard sizes
            buf = self.fid.read(struct.calcsize(fmt))
            if len(buf) != struct.calcsize(fmt) : return None
            val = list(struct.unpack(fmt , buf))
            for i, ival in enumerate(val):
                if hasattr(ival, 'split'):
                    val[i] = ival.split('\x00', 1)[0]
            if len(val) == 1:
                val = val[0]
            d[key] = val
        return d

def read_f(self, offset=None):
        if offset is not None:
            self.fid.seek(offset)
        d = {}
        for key, fmt in self.description:
            buf = self.fid.read(struct.calcsize(fmt))
            if len(buf) != struct.calcsize(fmt):
                return None
            val = list(struct.unpack(fmt, buf))
            for i, ival in enumerate(val):
                if hasattr(ival, 'replace'):
                    ival = ival.replace(str.encode('\x03'), str.encode(''))
                    ival = ival.replace(str.encode('\x00'), str.encode(''))
                    val[i] = ival.decode("utf-8")
            if len(val) == 1:
                val = val[0]
            d[key] = val
        return d

def read_f(self, offset =None):
        if offset is not None :
            self.fid.seek(offset)
        d = { }
        for key, fmt in self.description :
            fmt = '<' + fmt # insures use of standard sizes
            buf = self.fid.read(struct.calcsize(fmt))
            if len(buf) != struct.calcsize(fmt) : return None
            val = list(struct.unpack(fmt , buf))
            for i, ival in enumerate(val):
                if hasattr(ival, 'split'):
                    val[i] = ival.split('\x00', 1)[0]
            if len(val) == 1:
                val = val[0]
            d[key] = val
        return d

def _RegisterWndClass(self):
        className = "PythonSplash"
        global g_registeredClass
        if not g_registeredClass:
            message_map = {}
            wc = win32gui.WNDCLASS()
            wc.SetDialogProc() # Make it a dialog class.
            self.hinst = wc.hInstance = win32api.GetModuleHandle(None)
            wc.lpszClassName = className
            wc.style = 0
            wc.hCursor = win32gui.LoadCursor( 0, win32con.IDC_ARROW )
            wc.hbrBackground = win32con.COLOR_WINDOW + 1
            wc.lpfnWndProc = message_map # could also specify a wndproc.
            wc.cbWndExtra = win32con.DLGWINDOWEXTRA + struct.calcsize("Pi")
            classAtom = win32gui.RegisterClass(wc)
            g_registeredClass = 1
        return className

def _extract_images(filename):
        """???????????????????

        :param filename: ?????
        :return: 4??numpy??[index, y, x, depth]? ???np.float32
        """
        images = []
        print('Extracting {}'.format(filename))
        with gzip.GzipFile(fileobj=open(filename, 'rb')) as f:
            buf = f.read()
            index = 0
            magic, num_images, rows, cols = struct.unpack_from('>IIII', buf, index)
            if magic != 2051:
                raise ValueError('Invalid magic number {} in MNIST image file: {}'.format(magic, filename))
            index += struct.calcsize('>IIII')
            for i in range(num_images):
                img = struct.unpack_from('>784B', buf, index)
                index += struct.calcsize('>784B')
                img = np.array(img, dtype=np.float32)
                # ????[0,255]???[0,1]
                img = np.multiply(img, 1.0 / 255.0)
                img = img.reshape(rows, cols, 1)
                images.append(img)
        return np.array(images, dtype=np.float32)

def _extract_labels(filename, num_classes=10):
        """???????????????

        :param filename: ?????
        :param num_classes: ??one-hot??????????10?
        :return: 2??numpy??[index, num_classes]? ???np.float32
        """
        labels = []
        print('Extracting {}'.format(filename))
        with gzip.GzipFile(fileobj=open(filename, 'rb')) as f:
            buf = f.read()
            index = 0
            magic, num_labels = struct.unpack_from('>II', buf, index)
            if magic != 2049:
                raise ValueError('Invalid magic number {} in MNIST label file: {}'.format(magic, filename))
            index += struct.calcsize('>II')
            for i in range(num_labels):
                label = struct.unpack_from('>B', buf, index)
                index += struct.calcsize('>B')
                label_one_hot = np.zeros(num_classes, dtype=np.float32)
                label_one_hot[label[0]] = 1
                labels.append(label_one_hot)
        return np.array(labels, dtype=np.float32)

def read_image(filename):
    f = open(filename, 'rb')
    index = 0
    buf = f.read()
    f.close()

    magic, images, rows, columns = struct.unpack_from('>IIII' , buf , index)
    index += struct.calcsize('>IIII')
    for i in xrange(images):
        #for i in xrange(2000):
        image = Image.new('L', (columns, rows))

        for x in xrange(rows):
            for y in xrange(columns):
                image.putpixel((y, x), int(struct.unpack_from('>B', buf, index)[0]))
                index += struct.calcsize('>B')


        print 'save ' + str(i) + 'image'
        image.save('./test/' + str(i) + '.png')

def read_label(filename, saveFilename):
    f = open(filename, 'rb')
    index = 0
    buf = f.read()
    f.close()

    magic, labels = struct.unpack_from('>II' , buf , index)
    index += struct.calcsize('>II')

    labelArr = [0] * labels
    #labelArr = [0] * 2000

    for x in xrange(labels):
        #for x in xrange(2000):
        labelArr[x] = int(struct.unpack_from('>B', buf, index)[0])
        index += struct.calcsize('>B')

    save = open(saveFilename, 'w')

    save.write(','.join(map(lambda x: str(x), labelArr)))
    save.write('\n')

    save.close()
    print 'save labels success'

def unpack(self, buf):
        if len(buf) < self.pyr_len:
            raise StorageError("Buffer too short")
        self._magic, essidlen = struct.unpack(self.pyr_head, \
                                              buf[:self.pyr_len])
        if self._magic == 'PYR2':
            self._delimiter = '\n'
        elif self._magic == 'PYRT':
            self._delimiter = '\00'
        else:
            raise StorageError("Not a PYRT- or PYR2-buffer.")
        headfmt = "<%ssi16s" % (essidlen, )
        headsize = struct.calcsize(headfmt)
        header = buf[self.pyr_len:self.pyr_len + headsize]
        if len(header) != headsize:
            raise StorageError("Invalid header size")
        header = struct.unpack(headfmt, header)
        self.essid, self._numElems, self._digest = header
        pmkoffset = self.pyr_len + headsize
        pwoffset = pmkoffset + self._numElems * 32
        self._pwbuffer = buf[pwoffset:]
        self._pmkbuffer = buf[pmkoffset:pwoffset]
        if len(self._pmkbuffer) % 32 != 0:
            raise StorageError("pmkbuffer seems truncated")

def valid_ranges(*types):
    # given a sequence of numeric types, collect their _type_
    # attribute, which is a single format character compatible with
    # the struct module, use the struct module to calculate the
    # minimum and maximum value allowed for this format.
    # Returns a list of (min, max) values.
    result = []
    for t in types:
        fmt = t._type_
        size = struct.calcsize(fmt)
        a = struct.unpack(fmt, ("\x00"*32)[:size])[0]
        b = struct.unpack(fmt, ("\xFF"*32)[:size])[0]
        c = struct.unpack(fmt, ("\x7F"+"\x00"*32)[:size])[0]
        d = struct.unpack(fmt, ("\x80"+"\xFF"*32)[:size])[0]
        result.append((min(a, b, c, d), max(a, b, c, d)))
    return result

def _types_bitmap(cls, data, error):
        bits = []
        o = 0
        while o < len(data):
            fmt = "!BB"
            fmtsz = struct.calcsize(fmt)
            dat = data[o:o+fmtsz]
            if len(dat) != fmtsz:
                e = ("_types_bitmap", o, 'offset out of range: data size = %d' % len(data))
                error.append(e)
                return None
            block, bytes = struct.unpack(fmt, dat)
            o += fmtsz
            for i in range(bytes):
                b = struct.unpack("!B", data[o+i:o+i+1])[0]
                for j in range(8):
                        if b & (1 << (7-j)):
                                bits.append((block*32+i)*8+j)
            o += bytes
        return bits

def _do_query(cls, buf, offset, error):
        qry = {}
        res = cls._do_name(buf, offset, 0, error)
        if res is None:
            e = ("_do_query", offset, "_do_name failed")
            error.append(e)
            return None
        offset, name = res
        qry['Qname'] = name

        fmt = "!HH"
        reqlen = struct.calcsize(fmt)
        strng = buf[offset:offset + reqlen]
        if len(strng) != reqlen:
            e = ("_do_query", offset, 'offset out of range: buf size = %d' % len(buf))
            error.append(e)
            return None
        res = struct.unpack(fmt, strng)
        qry['Qtype'] = cls._type_to_text(res[0])
        qry['Qclass'] = cls._class_to_text(res[1])

        return offset + reqlen, qry

def calc_padding(fmt, align):
    """Calculate how many padding bytes needed for ``fmt`` to be aligned to
    ``align``.

    Args:
        fmt (str): :mod:`struct` format.
        align (int): alignment (2, 4, 8, etc.)

    Returns:
        str: padding format (e.g., various number of 'x').

    >>> calc_padding('b', 2)
    'x'

    >>> calc_padding('b', 3)
    'xx'
    """
    remain = struct.calcsize(fmt) % align
    if remain == 0:
        return ""
    return 'x' * (align - remain)

def _read_extras(self):
        dbfile = self.dbfile

        # Read the extras
        HashReader._read_extras(self)

        # Set up for reading the index array
        indextype = self.extras["indextype"]
        self.indexbase = dbfile.tell()
        self.indexlen = self.extras["indexlen"]
        self.indexsize = struct.calcsize(indextype)
        # Set up the function to read values from the index array
        if indextype == "B":
            self._get_pos = dbfile.get_byte
        elif indextype == "H":
            self._get_pos = dbfile.get_ushort
        elif indextype == "i":
            self._get_pos = dbfile.get_int
        elif indextype == "I":
            self._get_pos = dbfile.get_uint
        elif indextype == "q":
            self._get_pos = dbfile.get_long
        else:
            raise Exception("Unknown index type %r" % indextype)

def doVersion(checkForArgs=True):
  forceCheck = simple = False
  if checkForArgs:
    while Cmd.ArgumentsRemaining():
      myarg = getArgument()
      if myarg == u'check':
        forceCheck = True
      elif myarg == u'simple':
        simple = True
      else:
        unknownArgumentExit()
  if simple:
    writeStdout(__version__)
    return
  import struct
  version_data = u'GAM {0} - {1}\n{2}\nPython {3}.{4}.{5} {6}-bit {7}\ngoogle-api-python-client {8}\noauth2client {9}\n{10} {11}\nPath: {12}\n'
  writeStdout(version_data.format(__version__, GAM_URL, __author__, sys.version_info[0],
                                  sys.version_info[1], sys.version_info[2], struct.calcsize(u'P')*8,
                                  sys.version_info[3], googleapiclient.__version__, oauth2client.__version__, platform.platform(),
                                  platform.machine(), GM.Globals[GM.GAM_PATH]))
  if forceCheck:
    doGAMCheckForUpdates(forceCheck=True)

# gam help

def receive(channel):
    """ Receive a message from a channel """

    size = struct.calcsize("L")
    size = channel.recv(size)
    try:
        size = socket.ntohl(struct.unpack("L", size)[0])
    except struct.error as e:
        return ''

    buf = ""

    while len(buf) < size:
        buf = channel.recv(size - len(buf))

    return pickle.loads(buf)[0]

def UnpackNMITEMACTIVATE(lparam):
    format = _nmhdr_fmt + _nmhdr_align_padding
    if is64bit:
        # the struct module doesn't handle this correctly as some of the items
        # are actually structs in structs, which get individually aligned.
        format = format + "iiiiiiixxxxP"
    else:
        format = format + "iiiiiiiP"
    buf = win32gui.PyMakeBuffer(struct.calcsize(format), lparam)
    return _MakeResult("NMITEMACTIVATE hwndFrom idFrom code iItem iSubItem uNewState uOldState uChanged actionx actiony lParam",
                       struct.unpack(format, buf))

# MENUITEMINFO struct
# http://msdn.microsoft.com/library/default.asp?url=/library/en-us/winui/WinUI/WindowsUserInterface/Resources/Menus/MenuReference/MenuStructures/MENUITEMINFO.asp
# We use the struct module to pack and unpack strings as MENUITEMINFO
# structures.  We also have special handling for the 'fMask' item in that
# structure to avoid the caller needing to explicitly check validity
# (None is used if the mask excludes/should exclude the value)

def PackMENUINFO(dwStyle = None, cyMax = None,
                 hbrBack = None, dwContextHelpID = None, dwMenuData = None,
                 fMask = 0):
    if dwStyle is None: dwStyle = 0
    else: fMask |= win32con.MIM_STYLE
    if cyMax is None: cyMax = 0
    else: fMask |= win32con.MIM_MAXHEIGHT
    if hbrBack is None: hbrBack = 0
    else: fMask |= win32con.MIM_BACKGROUND
    if dwContextHelpID is None: dwContextHelpID = 0
    else: fMask |= win32con.MIM_HELPID
    if dwMenuData is None: dwMenuData = 0
    else: fMask |= win32con.MIM_MENUDATA
    # Create the struct.
    item = struct.pack(
                _menuinfo_fmt,
                struct.calcsize(_menuinfo_fmt), # cbSize
                fMask,
                dwStyle,
                cyMax,
                hbrBack,
                dwContextHelpID,
                dwMenuData)
    return array.array("b", item)

def _StructPackDecoder(wire_type, format):
  """Return a constructor for a decoder for a fixed-width field.

  Args:
      wire_type:  The field's wire type.
      format:  The format string to pass to struct.unpack().
  """

  value_size = struct.calcsize(format)
  local_unpack = struct.unpack

  # Reusing _SimpleDecoder is slightly slower than copying a bunch of code, but
  # not enough to make a significant difference.

  # Note that we expect someone up-stack to catch struct.error and convert
  # it to _DecodeError -- this way we don't have to set up exception-
  # handling blocks every time we parse one value.

  def InnerDecode(buffer, pos):
    new_pos = pos + value_size
    result = local_unpack(format, buffer[pos:new_pos])[0]
    return (result, new_pos)
  return _SimpleDecoder(wire_type, InnerDecode)

def _process_scrape(self, payload, trans):
        info_struct = '!LLL'
        info_size = struct.calcsize(info_struct)
        info_count = len(payload) / info_size
        hashes = trans['sent_hashes']
        response = {}
        for info_offset in xrange(info_count):
            off = info_size * info_offset
            info = payload[off:off + info_size]
            seeders, completed, leechers = struct.unpack(info_struct, info)
            response[hashes[info_offset]] = {
                'seeders': seeders,
                'completed': completed,
                'leechers': leechers,
            }
        return response

def send_one_ping(my_socket, dest_addr, ID):
    """
    Send one ping to the given >dest_addr<.
    """
    dest_addr  =  socket.gethostbyname(dest_addr)

    # Header is type (8), code (8), checksum (16), id (16), sequence (16)
    my_checksum = 0

    # Make a dummy heder with a 0 checksum.
    header = struct.pack("bbHHh", ICMP_ECHO_REQUEST, 0, my_checksum, ID, 1)
    bytesInDouble = struct.calcsize("d")
    data = (192 - bytesInDouble) * "Q"
    data = struct.pack("d", default_timer()) + data

    # Calculate the checksum on the data and the dummy header.
    my_checksum = checksum(header + data)

    # Now that we have the right checksum, we put that in. It's just easier
    # to make up a new header than to stuff it into the dummy.
    header = struct.pack(
        "bbHHh", ICMP_ECHO_REQUEST, 0, socket.htons(my_checksum), ID, 1
    )
    packet = header + data
    my_socket.sendto(packet, (dest_addr, 1)) # Don't know about the 1

def get_platform():
    """ get_platform()

    Get a string that specifies the platform more specific than
    sys.platform does. The result can be: linux32, linux64, win32,
    win64, osx32, osx64. Other platforms may be added in the future.
    """
    # Get platform
    if sys.platform.startswith('linux'):
        plat = 'linux%i'
    elif sys.platform.startswith('win'):
        plat = 'win%i'
    elif sys.platform.startswith('darwin'):
        plat = 'osx%i'
    else:  # pragma: no cover
        return None

    return plat % (struct.calcsize('P') * 8)  # 32 or 64 bits

def decryptStreamChunkOld(self,response, wfile, chunksize=24*1024, startOffset=0):
            if ENCRYPTION_ENABLE == 0:
                return
    #    with open(in_filename, 'rb') as infile:
            origsize = struct.unpack('<Q', response.read(struct.calcsize('Q')))[0]
            decryptor = AES.new(self.key, AES.MODE_ECB)

            count = 0
            while True:
                chunk = response.read(chunksize)
                count = count + 1
                if len(chunk) == 0:
                    break
                responseChunk = decryptor.decrypt(chunk)
                if count == 1 and startOffset !=0:
                    wfile.write(responseChunk[startOffset:])
                elif (len(chunk)) < (len(responseChunk.strip())):
                    wfile.write(responseChunk.strip())
                else:
                    wfile.write(responseChunk)

def decryptCalculatePadding(self,response, chunksize=24*1024):
            if ENCRYPTION_ENABLE == 0:
                return
    #    with open(in_filename, 'rb') as infile:
            origsize = struct.unpack('<Q', response.read(struct.calcsize('Q')))[0]
            decryptor = AES.new(self.key, AES.MODE_ECB)

            count = 0
            while True:
                chunk = response.read(chunksize)
                count = count + 1
                if len(chunk) == 0:
                    break

                responseChunk = decryptor.decrypt(chunk)
                return int(len(chunk) - len(responseChunk.strip()))

def run(self):
        print("VEDIO server starts...")
        self.sock.bind(self.ADDR)
        self.sock.listen(1)
        conn, addr = self.sock.accept()
        print("remote VEDIO client success connected...")
        data = "".encode("utf-8")
        payload_size = struct.calcsize("L")
        cv2.namedWindow('Remote', cv2.WINDOW_NORMAL)
        while True:
            while len(data) < payload_size:
                data += conn.recv(81920)
            packed_size = data[:payload_size]
            data = data[payload_size:]
            msg_size = struct.unpack("L", packed_size)[0]
            while len(data) < msg_size:
                data += conn.recv(81920)
            zframe_data = data[:msg_size]
            data = data[msg_size:]
            frame_data = zlib.decompress(zframe_data)
            frame = pickle.loads(frame_data)
            cv2.imshow('Remote', frame)
            if cv2.waitKey(1) & 0xFF == 27:
                break