74 lines
3.3 KiB
Python
74 lines
3.3 KiB
Python
import logging
|
|
from typing import List, Optional, Union, Tuple
|
|
from chia.types.blockchain_format.program import Program, SerializedProgram
|
|
from chia.types.generator_types import BlockGenerator, GeneratorArg, GeneratorBlockCacheInterface, CompressorArg
|
|
from chia.util.ints import uint32, uint64
|
|
from chia.wallet.puzzles.load_clvm import load_clvm
|
|
from chia.wallet.puzzles.rom_bootstrap_generator import get_generator
|
|
|
|
GENERATOR_MOD = get_generator()
|
|
|
|
DECOMPRESS_BLOCK = load_clvm("block_program_zero.clvm", package_or_requirement="chia.wallet.puzzles")
|
|
DECOMPRESS_PUZZLE = load_clvm("decompress_puzzle.clvm", package_or_requirement="chia.wallet.puzzles")
|
|
# DECOMPRESS_CSE = load_clvm("decompress_coin_solution_entry.clvm", package_or_requirement="chia.wallet.puzzles")
|
|
|
|
DECOMPRESS_CSE_WITH_PREFIX = load_clvm(
|
|
"decompress_coin_solution_entry_with_prefix.clvm", package_or_requirement="chia.wallet.puzzles"
|
|
)
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def create_block_generator(
|
|
generator: SerializedProgram, block_heights_list: List[uint32], generator_block_cache: GeneratorBlockCacheInterface
|
|
) -> Optional[BlockGenerator]:
|
|
""" `create_block_generator` will returns None if it fails to look up any referenced block """
|
|
generator_arg_list: List[GeneratorArg] = []
|
|
for i in block_heights_list:
|
|
previous_generator = generator_block_cache.get_generator_for_block_height(i)
|
|
if previous_generator is None:
|
|
log.error(f"Failed to look up generator for block {i}. Ref List: {block_heights_list}")
|
|
return None
|
|
generator_arg_list.append(GeneratorArg(i, previous_generator))
|
|
return BlockGenerator(generator, generator_arg_list)
|
|
|
|
|
|
def create_generator_args(generator_ref_list: List[SerializedProgram]) -> Program:
|
|
"""
|
|
`create_generator_args`: The format and contents of these arguments affect consensus.
|
|
"""
|
|
gen_ref_list = [bytes(g) for g in generator_ref_list]
|
|
return Program.to([gen_ref_list])
|
|
|
|
|
|
def create_compressed_generator(
|
|
original_generator: CompressorArg,
|
|
compressed_cse_list: List[List[Union[List[uint64], List[Union[bytes, None, Program]]]]],
|
|
) -> BlockGenerator:
|
|
"""
|
|
Bind the generator block program template to a particular reference block,
|
|
template bytes offsets, and SpendBundle.
|
|
"""
|
|
start = original_generator.start
|
|
end = original_generator.end
|
|
program = DECOMPRESS_BLOCK.curry(
|
|
DECOMPRESS_PUZZLE, DECOMPRESS_CSE_WITH_PREFIX, Program.to(start), Program.to(end), compressed_cse_list
|
|
)
|
|
generator_arg = GeneratorArg(original_generator.block_height, original_generator.generator)
|
|
return BlockGenerator(program, [generator_arg])
|
|
|
|
|
|
def setup_generator_args(self: BlockGenerator) -> Tuple[SerializedProgram, Program]:
|
|
args = create_generator_args(self.generator_refs())
|
|
return self.program, args
|
|
|
|
|
|
def run_generator(self: BlockGenerator, max_cost: int) -> Tuple[int, SerializedProgram]:
|
|
program, args = setup_generator_args(self)
|
|
return GENERATOR_MOD.run_safe_with_cost(max_cost, program, args)
|
|
|
|
|
|
def run_generator_unsafe(self: BlockGenerator, max_cost: int) -> Tuple[int, SerializedProgram]:
|
|
"""This mode is meant for accepting possibly soft-forked transactions into the mempool"""
|
|
program, args = setup_generator_args(self)
|
|
return GENERATOR_MOD.run_with_cost(max_cost, program, args)
|