rostrum 14.0.1

from binascii import hexlify, unhexlify
import struct
import io

SER_DEFAULT = 0
SER_ID = 0


class CompactSize(int):
    def serialize(self, _stype=SER_DEFAULT):
        assert self >= 0
        if self < 253:
            return struct.pack("<B", self)
        if self < 2**16:
            return struct.pack("<B", 253) + struct.pack("<H", self)
        if self < 2**32:
            return struct.pack("<B", 254) + struct.pack("<I", self)
        if self < 2**64:
            return struct.pack("<B", 255) + struct.pack("<Q", self)
        raise Exception("Value too large for serializer")

    def deserialize(self, f):
        number = struct.unpack("<B", f.read(1))[0]
        if number == 253:
            number = struct.unpack("<H", f.read(2))[0]
        elif number == 254:
            number = struct.unpack("<I", f.read(4))[0]
        elif number == 255:
            number = struct.unpack("<Q", f.read(8))[0]
        return CompactSize(number)


def deser_uint256(f):
    if isinstance(f, bytes):
        f = io.BytesIO(f)
    r = 0
    for i in range(8):
        t = struct.unpack("<I", f.read(4))[0]
        r += t << (i * 32)
    return r


def ser_uint256(u, _stype=SER_DEFAULT):
    rs = b""
    for _ in range(8):
        rs += struct.pack("<I", u & 0xFFFFFFFF)
        u >>= 32
    return rs


def deser_string(f):
    """Convert an array of bytes in the bitcoin P2P protocol format into a string

    >>> import io
    >>> deser_string(io.BytesIO(ser_string("The grid bug bites!  You get zapped!".encode()))).decode()
    'The grid bug bites!  You get zapped!'
    """
    nit = struct.unpack("<B", f.read(1))[0]
    if nit == 253:
        nit = struct.unpack("<H", f.read(2))[0]
    elif nit == 254:
        nit = struct.unpack("<I", f.read(4))[0]
    elif nit == 255:
        nit = struct.unpack("<Q", f.read(8))[0]
    return f.read(nit)


def ser_string(s, _stype=SER_DEFAULT):
    """convert a string into an array of bytes (in the bitcoin network format)

    >>> ser_string("The grid bug bites!  You get zapped!".encode())
    b'$The grid bug bites!  You get zapped!'
    """
    if len(s) < 253:
        return struct.pack("B", len(s)) + s
    if len(s) < 0x10000:
        return struct.pack("<BH", 253, len(s)) + s
    if len(s) < 0x100000000:
        return struct.pack("<BI", 254, len(s)) + s
    return struct.pack("<BQ", 255, len(s)) + s


def uint256_from_str(s):
    """Decode a uint256 from a little-endian byte array or hex string (bitcoind strings are little-endian)"""
    if len(s) == 64:
        s = unhexlify(s)
    r = 0
    t = struct.unpack("<IIIIIIII", s[:32])
    for i in range(8):
        r += t[i] << (i * 32)
    return r


def uint256_from_bigendian(s):
    """Decode a uint256 from a big-endian byte array or hex string (lexical order is big-endian)"""
    if isinstance(s, str):
        s = unhexlify(s)
    r = 0
    t = struct.unpack(">QQQQ", s[:32])
    for i in t:
        r = (r << 64) | i
    return r


def uint256_from_compact(c):
    """Convert compact encoding to uint256

    Used for the nBits compact encoding of the target in the block header.
    """
    nbytes = (c >> 24) & 0xFF
    if nbytes <= 3:
        v = (c & 0xFFFFFF) >> 8 * (3 - nbytes)
    else:
        v = (c & 0xFFFFFF) << (8 * (nbytes - 3))
    return v


def compact_from_uint256(v):
    """Convert uint256 to compact encoding"""
    nbytes = (v.bit_length() + 7) >> 3
    compact = 0
    if nbytes <= 3:
        compact = (v & 0xFFFFFF) << 8 * (3 - nbytes)
    else:
        compact = v >> 8 * (nbytes - 3)
        compact = compact & 0xFFFFFF

    # If the sign bit (0x00800000) is set, divide the mantissa by 256 and
    # increase the exponent to get an encoding without it set.
    if compact & 0x00800000:
        compact >>= 8
        nbytes += 1

    return compact | nbytes << 24


def deser_double(f):
    return struct.unpack("<d", f.read(8))[0]


def ser_double(d):
    return struct.pack("<d", d)


def deser_vector(f, c):
    nit = struct.unpack("<B", f.read(1))[0]
    if nit == 253:
        nit = struct.unpack("<H", f.read(2))[0]
    elif nit == 254:
        nit = struct.unpack("<I", f.read(4))[0]
    elif nit == 255:
        nit = struct.unpack("<Q", f.read(8))[0]
    r = []
    for _ in range(nit):
        if isinstance(c, type(lambda: True)):
            t = c(f)
        else:  # class type
            t = c()
            t.deserialize(f)
            r.append(t)
    return r


def ser_vector(l, stype=SER_ID):
    r = b""
    if len(l) < 253:
        r = struct.pack("B", len(l))
    elif len(l) < 0x10000:
        r = struct.pack("<BH", 253, len(l))
    elif len(l) < 0x100000000:
        r = struct.pack("<BI", 254, len(l))
    else:
        r = struct.pack("<BQ", 255, len(l))
    for i in l:
        r += i.serialize(stype)
    return r


def deser_uint256_vector(f):
    nit = struct.unpack("<B", f.read(1))[0]
    if nit == 253:
        nit = struct.unpack("<H", f.read(2))[0]
    elif nit == 254:
        nit = struct.unpack("<I", f.read(4))[0]
    elif nit == 255:
        nit = struct.unpack("<Q", f.read(8))[0]
    r = []
    for _ in range(nit):
        t = deser_uint256(f)
        r.append(t)
    return r


def ser_uint256_vector(l):
    r = b""
    if len(l) < 253:
        r = struct.pack("B", len(l))
    elif len(l) < 0x10000:
        r = struct.pack("<BH", 253, len(l))
    elif len(l) < 0x100000000:
        r = struct.pack("<BI", 254, len(l))
    else:
        r = struct.pack("<BQ", 255, len(l))
    for i in l:
        r += ser_uint256(i)
    return r


def ser_hash32_vector(l):
    """Serializes a vector of 32 byte hashes supplied as a list of 32 byte 'bytes' objects"""
    r = b""
    if len(l) < 253:
        r = struct.pack("B", len(l))
    elif len(l) < 0x10000:
        r = struct.pack("<BH", 253, len(l))
    elif len(l) < 0x100000000:
        r = struct.pack("<BI", 254, len(l))
    else:
        r = struct.pack("<BQ", 255, len(l))
    for i in l:
        assert len(i) == 32
        r += i
    return r


def deser_hash32_vector(f):
    """Deserializes a vector of 32 byte hashes, into a list of 32 byte 'bytes' objects"""
    nit = struct.unpack("<B", f.read(1))[0]
    if nit == 253:
        nit = struct.unpack("<H", f.read(2))[0]
    elif nit == 254:
        nit = struct.unpack("<I", f.read(4))[0]
    elif nit == 255:
        nit = struct.unpack("<Q", f.read(8))[0]
    r = []
    for _ in range(nit):
        t = f.read(32)
        r.append(t)
    return r


def ser_compact_size(l):
    r = b""
    if l < 253:
        r = struct.pack("B", l)
    elif l < 0x10000:
        r = struct.pack("<BH", 253, l)
    elif l < 0x100000000:
        r = struct.pack("<BI", 254, l)
    else:
        r = struct.pack("<BQ", 255, l)
    return r


def deser_compact_size(f):
    nit = struct.unpack("<B", f.read(1))[0]
    if nit == 253:
        nit = struct.unpack("<H", f.read(2))[0]
    elif nit == 254:
        nit = struct.unpack("<I", f.read(4))[0]
    elif nit == 255:
        nit = struct.unpack("<Q", f.read(8))[0]
    return nit


def deser_varint(f):
    num = 0
    while True:
        b = struct.unpack("<B", f.read(1))[0]
        num = (num << 7) | (b & 0x7F)
        if b & 0x80:
            num += 1
        else:
            return num


def ser_varint(n):
    ret = bytearray()

    i = 0
    ret.append(0)
    while True:
        ret[i] = ret[i] | (n & 0x7F)
        if n <= 0x7F:
            break
        n = (n >> 7) - 1
        i += 1
        ret.append(0x80)

    return bytes(reversed(ret))


def deser_string_vector(f):
    nit = struct.unpack("<B", f.read(1))[0]
    if nit == 253:
        nit = struct.unpack("<H", f.read(2))[0]
    elif nit == 254:
        nit = struct.unpack("<I", f.read(4))[0]
    elif nit == 255:
        nit = struct.unpack("<Q", f.read(8))[0]
    r = []
    for _ in range(nit):
        t = deser_string(f)
        r.append(t)
    return r


def ser_string_vector(l):
    r = b""
    if len(l) < 253:
        r = struct.pack("B", len(l))
    elif len(l) < 0x10000:
        r = struct.pack("<BH", 253, len(l))
    elif len(l) < 0x100000000:
        r = struct.pack("<BI", 254, len(l))
    else:
        r = struct.pack("<BQ", 255, len(l))
    for sv in l:
        r += ser_string(sv)
    return r


def deser_int_vector(f):
    nit = struct.unpack("<B", f.read(1))[0]
    if nit == 253:
        nit = struct.unpack("<H", f.read(2))[0]
    elif nit == 254:
        nit = struct.unpack("<I", f.read(4))[0]
    elif nit == 255:
        nit = struct.unpack("<Q", f.read(8))[0]
    r = []
    for _ in range(nit):
        t = struct.unpack("<i", f.read(4))[0]
        r.append(t)
    return r


def ser_int_vector(l):
    r = b""
    if len(l) < 253:
        r = struct.pack("B", len(l))
    elif len(l) < 0x10000:
        r = struct.pack("<BH", 253, len(l))
    elif len(l) < 0x100000000:
        r = struct.pack("<BI", 254, len(l))
    else:
        r = struct.pack("<BQ", 255, len(l))
    for i in l:
        r += struct.pack("<i", i)
    return r


def serialize_script_num(value):
    r = bytearray(0)
    if value == 0:
        return r
    neg = value < 0
    absvalue = -value if neg else value
    while absvalue:
        r.append(int(absvalue & 0xFF))
        absvalue >>= 8
    if r[-1] & 0x80:
        r.append(0x80 if neg else 0)
    elif neg:
        r[-1] |= 0x80
    return r


# Deserialize from a hex string representation (eg from RPC)
def from_hex(obj, hex_string):
    obj.deserialize(io.BytesIO(unhexlify(hex_string.strip().encode("ascii"))))
    return obj


# Convert a binary-serializable object to hex (eg for submission via RPC)
def to_hex(obj):
    if isinstance(obj, bytes):
        return hexlify(obj)
    return hexlify(obj.serialize()).decode("ascii")