python implementation for Qt's QDataStream（看一下QDataStream的结构）

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import print_function
from __future__ import unicode_literals
from __future__ import absolute_import
from __future__ import division

"""
python implemention for QDataStream (Qt 4.0 - Qt 5.6)

"""

import sys
import struct
import io

Qt_1_0 = 1
Qt_2_0 = 2
Qt_2_1 = 3
Qt_3_0 = 4
Qt_3_1 = 5
Qt_3_3 = 6
Qt_4_0 = 7
Qt_4_1 = Qt_4_0
Qt_4_2 = 8
Qt_4_3 = 9
Qt_4_4 = 10
Qt_4_5 = 11
Qt_4_6 = 12
Qt_4_7 = Qt_4_6
Qt_4_8 = Qt_4_7
Qt_4_9 = Qt_4_8
Qt_5_0 = 13
Qt_5_1 = 14
Qt_5_2 = 15
Qt_5_3 = Qt_5_2
Qt_5_4 = 16
Qt_5_5 = Qt_5_4
Qt_5_6 = 17

SinglePrecision = 0
DoublePrecision = 1


#懒得支持 Qt 4 以下的格式了。

class Serializer:
    def __init__(self):
        self.buffer = io.BytesIO()
        self.byte_order = "big"
        self.version = Qt_4_6
        self.floatingPointPrecision = DoublePrecision

    def write_char(self, c):
        if len(c) != 1:
            raise Exception("write_char() only accept bytes of length 1.")
        self.buffer.write(c)

    def write_int8(self, i):
        self.buffer.write(struct.pack("b", i))

    def write_bool(self, b):
        self.write_int8(1 if b else 0)

    def write_uint8(self, i):
        self.buffer.write(struct.pack("B", i))

    def write_int16(self, i):
        pattern = ">h" if self.byte_order == "big" else "<h"
        self.buffer.write(struct.pack(pattern, i))

    def write_uint16(self, i):
        pattern = ">H" if self.byte_order == "big" else "<H"
        self.buffer.write(struct.pack(pattern, i))

    def write_int32(self, i):
        pattern = ">i" if self.byte_order == "big" else "<i"
        self.buffer.write(struct.pack(pattern, int(i)))

    def write_uint32(self, i):
        pattern = ">I" if self.byte_order == "big" else "<I"
        self.buffer.write(struct.pack(pattern, int(i)))

    def write_int64(self, i):
        pattern = ">q" if self.byte_order == "big" else "<q"
        self.buffer.write(struct.pack(pattern, int(i)))

    def write_uint64(self, i):
        pattern = ">Q" if self.byte_order == "big" else "<Q"
        self.buffer.write(struct.pack(pattern, int(i)))

    def write_float(self, f):
        if self.version >= Qt_4_6 and self.floatingPointPrecision == DoublePrecision:
            self.write_double(f)
        else:
            pattern = ">f" if self.byte_order == "big" else "<f"
            self.buffer.write(struct.pack(pattern, f))

    def write_double(self, d):
        if self.version >= Qt_4_6 and self.floatingPointPrecision == SinglePrecision:
            self.write_float(d)
        else:
            pattern = ">d" if self.byte_order == "big" else "<d"
            self.buffer.write(struct.pack(pattern, d))

    def write_bytes(self, bs):
        if not bs:
            self.write_uint32(0xffffffff)
            return
        self.write_uint32(len(bs))
        self.buffer.write(bs)

    def write_raw_bytes(self, bs):
        self.buffer.write(bs)

    def write_string(self, s):
        if not s:
            self.write_bytes(b"")
            return
        if self.byte_order == "big":
            self.write_bytes(s.encode("utf-16be"))
        else:
            self.write_bytes(s.encode("utf-16le"))

    def write_map(self, d, key_type_or_func, value_type_or_func):
        self.write_uint32(len(d))
        for k, v in d.items():
            if hasattr(key_type_or_func, "__call__"):
                key_type_or_func(self, k)
            else:
                self.write(key_type_or_func, k)
            if hasattr(value_type_or_func, "__call__"):
                value_type_or_func(self, v)
            else:
                self.write(value_type_or_func, v)

    def write_list(self, l, element_type_or_func):
        self.write_uint32(len(l))
        if hasattr(element_type_or_func, "__call__"):
            for e in l:
                element_type_or_func(self, e)
        else:
            for e in l:
                self.write(element_type_or_func, e)

    def write(self, vt, v):
        return {
            "int8": self.write_int8,
            "uint8": self.write_uint8,
            "int16": self.write_int16,
            "uint16": self.write_uint16,
            "int32": self.write_int32,
            "uint32": self.write_uint32,
            "int64": self.write_int64,
            "uint64": self.write_uint64,
            "float": self.write_float,
            "double": self.write_double,
            "str": self.write_string,
            "bytes": self.write_bytes,
            "bool": self.write_bool,
        }[vt](v)

    def get_value(self):
        return self.buffer.getvalue()


class Deserializer:
    def __init__(self, buffer):
        self.buffer = io.BytesIO(buffer)
        self.byte_order = "big"
        self.double = False
        self.version = Qt_4_6
        self.floatingPointPrecision = DoublePrecision

    def read_int8(self):
        i, = struct.unpack("b", self.buffer.read(1))
        return i

    def read_bool(self):
        return bool(self.read_int8())

    def read_uint8(self):
        i, = struct.unpack("B", self.buffer.read(1))
        return i

    def read_int16(self):
        pattern = ">h" if self.byte_order == "big" else "<h"
        i, = struct.unpack(pattern, self.buffer.read(2))
        return i

    def read_uint16(self):
        pattern = ">H" if self.byte_order == "big" else "<H"
        i, = struct.unpack(pattern, self.buffer.read(2))
        return i

    def read_int32(self):
        pattern = ">i" if self.byte_order == "big" else "<i"
        i, = struct.unpack(pattern, self.buffer.read(4))
        return i

    def read_uint32(self):
        pattern = ">I" if self.byte_order == "big" else "<I"
        i, = struct.unpack(pattern, self.buffer.read(4))
        return i

    def read_int64(self):
        pattern = ">q" if self.byte_order == "big" else "<q"
        i, = struct.unpack(pattern, self.buffer.read(8))
        return i

    def read_uint64(self):
        pattern = ">Q" if self.byte_order == "big" else "<Q"
        i, = struct.unpack(pattern, self.buffer.read(8))
        return i

    def read_float(self):
        if self.version >= Qt_4_6 and self.floatingPointPrecision == DoublePrecision:
            return self.read_double()
        else:
            pattern = ">f" if self.byte_order == "big" else "<f"
            f, = struct.unpack(pattern, self.buffer.read(4))
            return f

    def read_double(self):
        if self.version >= Qt_4_6 and self.floatingPointPrecision == SinglePrecision:
            return self.read_float()
        else:
            pattern = ">d" if self.byte_order == "big" else "<d"
            d, = struct.unpack(pattern, self.buffer.read(8))
            return d

    def read_bytes(self):
        length = self.read_uint32()
        if length == 0xffffffff:
            return b""
        return self.buffer.read(length)

    def read_raw_bytes(self, length):
        return self.buffer.read(length)

    def read_string(self):
        buf = self.read_bytes()
        if self.byte_order == "little":
            return buf.decode("utf-16le")
        else:
            return buf.decode("utf-16be")

    def read_list(self, element_type_or_func):
        length = self.read_uint32()
        l = []
        if hasattr(element_type_or_func, "__call__"):
            for i in range(length):
                l.append(element_type_or_func(self))
        else:
            for i in range(length):
                l.append(self.read(element_type_or_func))
        return l

    def read_map(self, key_type_or_func, value_type):
        length = self.read_uint32()
        d = {}
        for i in range(length):
            if hasattr(key_type_or_func, "__call__"):
                key = key_type_or_func(self)
            else:
                key = self.read(key_type_or_func)
            if hasattr(key_type_or_func, "__call__"):
                value = key_type_or_func(self)
            else:
                value = self.read(value_type)
            d[key] = value
        return d

    def read(self, vt):
        return {
            "int8": self.read_int8,
            "uint8": self.read_uint8,
            "int16": self.read_int16,
            "uint16": self.read_uint16,
            "int32": self.read_int32,
            "uint32": self.read_uint32,
            "int64": self.read_int64,
            "uint64": self.read_uint64,
            "float": self.read_float,
            "double": self.read_double,
            "str": self.read_string,
            "bytes": self.read_bytes,
            "bool": self.read_bool,
        }[vt]()


if __name__ == "__main__":
    try:
        from PyQt4.QtCore import QDataStream, QByteArray, QBuffer, QIODevice
    except ImportError:
        sys.exit(1)

    serializer = Serializer()
    #serializer.byte_order = "little"
    serializer.write_int8(64)
    serializer.write_uint8(0xee)
    serializer.write_int16(64)
    serializer.write_uint16(0xffee)
    serializer.write_int32(0x100020)
    serializer.write_uint32(0xffeeddcc)
    serializer.write_int64(0x1000200040)
    serializer.write_uint64(0xffeeddccbbaa9988)
    serializer.write_float(64)
    serializer.write_bytes(b"fish is here.")
    serializer.write_string("实在太帅了。")
    h1 = serializer.get_value()

    buf = QByteArray()
    d = QDataStream(buf, QIODevice.WriteOnly)
    #d.setByteOrder(QDataStream.LittleEndian)
    d.setVersion(QDataStream.Qt_4_6)
    d.writeInt8(chr(64))
    d.writeUInt8(chr(0xee))
    d.writeInt16(64)
    d.writeUInt16(0xffee)
    d.writeInt32(0x100020)
    d.writeUInt32(0xffeeddcc)
    d.writeInt64(0x1000200040)
    d.writeUInt64(0xffeeddccbbaa9988)
    d.writeFloat(64)
    d.writeBytes(b"fish is here.")
    d.writeQString("实在太帅了。")
    h2 = bytes(buf)

    print(repr(h1))
    print(repr(h2))
    print(h1 == h2)