ethereum.fork_criteria

  1"""
  2Fork Criteria
  3^^^^^^^^^^^^^
  4
  5.. contents:: Table of Contents
  6    :backlinks: none
  7    :local:
  8
  9Introduction
 10------------
 11
 12Classes for specifying criteria for Mainnet forks.
 13"""
 14
 15import functools
 16from abc import ABC, abstractmethod
 17from typing import Final, Tuple
 18
 19
 20@functools.total_ordering
 21class ForkCriteria(ABC):
 22    """
 23    Type that represents the condition required for a fork to occur.
 24    """
 25
 26    BLOCK_NUMBER: Final[int] = 0
 27    TIMESTAMP: Final[int] = 1
 28    UNSCHEDULED: Final[int] = 2
 29
 30    _internal: Tuple[int, int]
 31
 32    def __eq__(self, other: object) -> bool:
 33        """
 34        Equality for fork criteria.
 35        """
 36        if isinstance(other, ForkCriteria):
 37            return self._internal == other._internal
 38        return NotImplemented
 39
 40    def __lt__(self, other: object) -> bool:
 41        """
 42        Ordering for fork criteria. Block number forks are before timestamp
 43        forks and scheduled forks are before unscheduled forks.
 44        """
 45        if isinstance(other, ForkCriteria):
 46            return self._internal < other._internal
 47        return NotImplemented
 48
 49    def __hash__(self) -> int:
 50        """
 51        `ForkCriteria` is hashable, so it can be stored in dictionaries.
 52        """
 53        return hash(self._internal)
 54
 55    @abstractmethod
 56    def check(self, block_number: int, timestamp: int) -> bool:
 57        """
 58        Check whether fork criteria have been met.
 59        """
 60        ...
 61
 62    @abstractmethod
 63    def __repr__(self) -> str:
 64        """
 65        String representation of this object.
 66        """
 67        raise NotImplementedError()
 68
 69
 70class ByBlockNumber(ForkCriteria):
 71    """
 72    Forks that occur when a specific block number has been reached.
 73    """
 74
 75    block_number: int
 76
 77    def __init__(self, block_number: int):
 78        self._internal = (ForkCriteria.BLOCK_NUMBER, block_number)
 79        self.block_number = block_number
 80
 81    def check(self, block_number: int, timestamp: int) -> bool:
 82        """
 83        Check whether the block number has been reached.
 84        """
 85        return block_number >= self.block_number
 86
 87    def __repr__(self) -> str:
 88        """
 89        String representation of this object.
 90        """
 91        return f"ByBlockNumber({self.block_number})"
 92
 93
 94class ByTimestamp(ForkCriteria):
 95    """
 96    Forks that occur when a specific timestamp has been reached.
 97    """
 98
 99    timestamp: int
100
101    def __init__(self, timestamp: int):
102        self._internal = (ForkCriteria.TIMESTAMP, timestamp)
103        self.timestamp = timestamp
104
105    def check(self, block_number: int, timestamp: int) -> bool:
106        """
107        Check whether the timestamp has been reached.
108        """
109        return timestamp >= self.timestamp
110
111    def __repr__(self) -> str:
112        """
113        String representation of this object.
114        """
115        return f"ByTimestamp({self.timestamp})"
116
117
118class Unscheduled(ForkCriteria):
119    """
120    Forks that have not been scheduled.
121    """
122
123    def __init__(self) -> None:
124        self._internal = (ForkCriteria.UNSCHEDULED, 0)
125
126    def check(self, block_number: int, timestamp: int) -> bool:
127        """
128        Unscheduled forks never occur.
129        """
130        return False
131
132    def __repr__(self) -> str:
133        """
134        String representation of this object.
135        """
136        return "Unscheduled()"