mirror of
https://github.com/Chia-Network/chia-blockchain.git
synced 2026-08-28 18:14:19 -05:00
Matt HauffandGitHub
40db4635a8
* Enable PEP604 Ruff rules * Fix harcoded signature in test * Hack CLVMStreamable test with note to fast follow
60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, ClassVar, cast
|
|
|
|
from chia_rs import BlockRecord
|
|
from chia_rs.sized_bytes import bytes32
|
|
from chia_rs.sized_ints import uint32
|
|
|
|
|
|
# implements BlockRecordsProtocol
|
|
class BlockCache:
|
|
if TYPE_CHECKING:
|
|
from chia.consensus.blockchain_interface import BlockRecordsProtocol
|
|
|
|
_protocol_check: ClassVar[BlockRecordsProtocol] = cast("BlockCache", None)
|
|
|
|
_block_records: dict[bytes32, BlockRecord]
|
|
_height_to_hash: dict[uint32, bytes32]
|
|
|
|
def __init__(
|
|
self,
|
|
blocks: dict[bytes32, BlockRecord],
|
|
):
|
|
self._block_records = blocks
|
|
self._height_to_hash = {block.height: hh for hh, block in blocks.items()}
|
|
|
|
def add_block(self, block: BlockRecord) -> None:
|
|
hh = block.header_hash
|
|
self._block_records[hh] = block
|
|
self._height_to_hash[block.height] = hh
|
|
|
|
def block_record(self, header_hash: bytes32) -> BlockRecord:
|
|
return self._block_records[header_hash]
|
|
|
|
def height_to_block_record(self, height: uint32) -> BlockRecord:
|
|
# Precondition: height is < peak height
|
|
header_hash: bytes32 | None = self.height_to_hash(height)
|
|
assert header_hash is not None
|
|
return self.block_record(header_hash)
|
|
|
|
def height_to_hash(self, height: uint32) -> bytes32 | None:
|
|
if height not in self._height_to_hash:
|
|
return None
|
|
return self._height_to_hash[height]
|
|
|
|
def contains_block(self, header_hash: bytes32, height: uint32) -> bool:
|
|
block_hash_from_hh = self.height_to_hash(height)
|
|
if block_hash_from_hh is None or block_hash_from_hh != header_hash:
|
|
return False
|
|
return True
|
|
|
|
def contains_height(self, height: uint32) -> bool:
|
|
return height in self._height_to_hash
|
|
|
|
def try_block_record(self, header_hash: bytes32) -> BlockRecord | None:
|
|
return self._block_records.get(header_hash)
|
|
|
|
async def prev_block_hash(self, header_hashes: list[bytes32]) -> list[bytes32]:
|
|
return [self._block_records[h].prev_hash for h in header_hashes]
|