layerthree's block deserializer \ serializer

block.py

import binascii
from layerthree import ioc
from layerthree.utils import deserialize_header, header_to_hash, get_merkle_tree


class Block:
    TYPE = 1

    def __init__(self, blockhash: str):
        self.blockhash = blockhash
        self._verbose = None
        self._serialized_blockheader = None

    @property
    def verbose(self):
        if not self._verbose:
            self._verbose = ioc.bitcoin_rpc_client.getblock(self.blockhash, verbose=True)
        return self._verbose

    @verbose.setter
    def verbose(self, value):
        self._verbose = value

    @property
    def txids_blob(self):
        data = b''
        for tx in self.verbose.get('tx'):
            data += binascii.unhexlify(tx)
        return data

    @property
    def serialized_blockheader(self):
        if not self._serialized_blockheader:
            hex_header = ioc.bitcoin_rpc_client.getblockheader(self.blockhash, verbose=False)
            self._serialized_blockheader = hex_header and binascii.unhexlify(hex_header)
        return self._serialized_blockheader

    @serialized_blockheader.setter
    def serialized_blockheader(self, value):
        self._serialized_blockheader = value

    def serialize(self):
        header = self.serialized_blockheader
        data = self.txids_blob
        prefix = self.TYPE.to_bytes(2, 'little')
        block_size = self.verbose['size'].to_bytes(4, 'little')
        txs_count = len(self.verbose['tx']).to_bytes(2, 'little')
        return prefix + header + block_size + txs_count + data

    @classmethod
    def deserialize(cls, data: bytes, verify_merkle_tree=True):
        data_type = int.from_bytes(data[:2], 'little')
        assert data_type == cls.TYPE
        header = data[2:82]
        block_size = int.from_bytes(data[82:86], 'little')
        txs_count = int.from_bytes(data[86:88], 'little')
        block_hash = header_to_hash(header)
        instance = cls(block_hash)
        instance.serialized_blockheader = header
        i = 0
        txids = []
        while i < txs_count:
            txids.append(
                binascii.hexlify(data[88+(i*32):120+(i*32)]).decode()
            )
            i += 1
        _deserialized_header = deserialize_header(header)
        cls.verbose = {
            "hash": block_hash,
            "version": _deserialized_header['version'],
            "versionHex": "",
            "merkleroot": _deserialized_header['merkle_root'],
            "time": _deserialized_header['timestamp'],
            "mediantime": _deserialized_header['timestamp'],
            "nonce": _deserialized_header['nonce'],
            "bits": str(_deserialized_header['bits']),
            "difficulty": 0,
            "chainwork": '00'*32,
            "previousblockhash": _deserialized_header['prev_block_hash'],
            "size": block_size,
            "weight": 0,
            "tx": txids
        }
        if verify_merkle_tree:
            merkle_tree = get_merkle_tree(txids)
            if binascii.hexlify(merkle_tree.root.val[::-1]).decode() != instance.verbose['merkleroot']:
                raise ValueError('verify_merkle_tree failed')
        return instance