Skip to content

packages.valory.skills.abstract_round_abci.base

This module contains the base classes for the models classes of the skill.

consensus_threshold

def consensus_threshold(n: int) -> int

Get consensus threshold.

Arguments:

  • n: the number of participants

Returns:

the consensus threshold

ABCIAppException Objects

class ABCIAppException(Exception)

A parent class for all exceptions related to the ABCIApp.

SignatureNotValidError Objects

class SignatureNotValidError(ABCIAppException)

Error raised when a signature is invalid.

AddBlockError Objects

class AddBlockError(ABCIAppException)

Exception raised when a block addition is not valid.

ABCIAppInternalError Objects

class ABCIAppInternalError(ABCIAppException)

Internal error due to a bad implementation of the ABCIApp.

__init__

def __init__(message: str, *args: Any) -> None

Initialize the error object.

TransactionTypeNotRecognizedError Objects

class TransactionTypeNotRecognizedError(ABCIAppException)

Error raised when a transaction type is not recognized.

TransactionNotValidError Objects

class TransactionNotValidError(ABCIAppException)

Error raised when a transaction is not valid.

LateArrivingTransaction Objects

class LateArrivingTransaction(ABCIAppException)

Error raised when the transaction belongs to previous round.

_MetaPayload Objects

class _MetaPayload(ABCMeta)

Payload metaclass.

The purpose of this metaclass is to remember the association between the type of payload and the payload class to build it. This is necessary to recover the right payload class to instantiate at decoding time.

Each class that has this class as metaclass must have a class attribute 'transaction_type', which for simplicity is required to be convertible to string, for serialization purposes.

__new__

def __new__(mcs, name: str, bases: Tuple, namespace: Dict, **kwargs: Any) -> Type

Create a new class object.

BaseTxPayload Objects

class BaseTxPayload(ABC, metaclass=_MetaPayload)

This class represents a base class for transaction payload classes.

__init__

def __init__(sender: str, id_: Optional[str] = None, round_count: int = ROUND_COUNT_DEFAULT) -> None

Initialize a transaction payload.

Arguments:

  • sender: the sender (Ethereum) address
  • id_: the id of the transaction
  • round_count: the count of the round in which the payload was sent

round_count

@property
def round_count() -> int

Get the round count.

round_count

@round_count.setter
def round_count(round_count: int) -> None

Set the round count.

encode

def encode() -> bytes

Encode the payload.

decode

@classmethod
def decode(cls, obj: bytes) -> "BaseTxPayload"

Decode the payload.

from_json

@classmethod
def from_json(cls, obj: Dict) -> "BaseTxPayload"

Decode the payload.

json

@property
def json() -> Dict

Get the JSON representation of the payload.

data

@property
def data() -> Dict

Get the dictionary data.

The returned dictionary is required to be used as keyword constructor initializer, i.e. these two should have the same effect:

sender = "..."
some_kwargs = {...}
p1 = SomePayloadClass(sender, **some_kwargs)
p2 = SomePayloadClass(sender, **p1.data)

Returns:

a dictionary which contains the payload data

with_new_id

def with_new_id() -> "BaseTxPayload"

Create a new payload with the same content but new id.

__eq__

def __eq__(other: Any) -> bool

Check equality.

__hash__

def __hash__() -> int

Hash the payload.

Transaction Objects

class Transaction(ABC)

Class to represent a transaction for the ephemeral chain of a period.

__init__

def __init__(payload: BaseTxPayload, signature: str) -> None

Initialize a transaction object.

encode

def encode() -> bytes

Encode the transaction.

decode

@classmethod
def decode(cls, obj: bytes) -> "Transaction"

Decode the transaction.

verify

def verify(ledger_id: str) -> None

Verify the signature is correct.

Arguments:

:raises: SignatureNotValidError: if the signature is not valid. - ledger_id: the ledger id of the address

__eq__

def __eq__(other: Any) -> bool

Check equality.

Block Objects

class Block()

Class to represent (a subset of) data of a Tendermint block.

__init__

def __init__(header: Header, transactions: Sequence[Transaction]) -> None

Initialize the block.

transactions

@property
def transactions() -> Tuple[Transaction, ...]

Get the transactions.

timestamp

@property
def timestamp() -> datetime.datetime

Get the block timestamp.

Blockchain Objects

class Blockchain()

Class to represent a (naive) Tendermint blockchain.

The consistency of the data in the blocks is guaranteed by Tendermint.

__init__

def __init__(height_offset: int = 0) -> None

Initialize the blockchain.

add_block

def add_block(block: Block) -> None

Add a block to the list.

height

@property
def height() -> int

Get the height.

Tendermint's height starts from 1. A return value equal to 0 means empty blockchain.

Returns:

the height.

length

@property
def length() -> int

Get the blockchain length.

blocks

@property
def blocks() -> Tuple[Block, ...]

Get the blocks.

last_block

@property
def last_block() -> Block

Returns the last stored block.

BlockBuilder Objects

class BlockBuilder()

Helper class to build a block.

__init__

def __init__() -> None

Initialize the block builder.

reset

def reset() -> None

Reset the temporary data structures.

@property
def header() -> Header

Get the block header.

Returns:

the block header

header

@header.setter
def header(header: Header) -> None

Set the header.

transactions

@property
def transactions() -> Tuple[Transaction, ...]

Get the sequence of transactions.

add_transaction

def add_transaction(transaction: Transaction) -> None

Add a transaction.

get_block

def get_block() -> Block

Get the block.

ConsensusParams Objects

class ConsensusParams()

Represent the consensus parameters.

__init__

def __init__(max_participants: int)

Initialize the consensus parameters.

max_participants

@property
def max_participants() -> int

Get the maximum number of participants.

consensus_threshold

@property
def consensus_threshold() -> int

Get the consensus threshold.

from_json

@classmethod
def from_json(cls, obj: Dict) -> "ConsensusParams"

Get from JSON.

__eq__

def __eq__(other: Any) -> bool

Check equality.

AbciAppDB Objects

class AbciAppDB()

Class to represent all data replicated across agents.

This class stores all the data in self._data. Every entry on this dict represents an optional "period" within your app execution. The concept of period is user-defined, so it might be something like a sequence of rounds that together conform a logical cycle of its execution, or it might have no sense at all (thus its optionality) and therefore only period 0 will be used.

Every "period" entry stores a dict where every key is a saved parameter and its corresponding value a list containing the history of the parameter values. For instance, for period 0:

0: {"parameter_name": [parameter_history]}

A complete database could look like this:

data = { 0: { "participants": [ {"participant_a", "participant_b", "participant_c", "participant_d"}, {"participant_a", "participant_b", "participant_c"}, {"participant_a", "participant_b", "participant_c", "participant_d"}, ] }, "other_parameter": [0, 2, 8] }, 1: { "participants": [ {"participant_a", "participant_c", "participant_d"}, {"participant_a", "participant_b", "participant_c", "participant_d"}, {"participant_a", "participant_b", "participant_c"}, {"participant_a", "participant_b", "participant_d"}, {"participant_a", "participant_b", "participant_c", "participant_d"}, ], "other_parameter": [3, 19, 10, 32, 6] }, 2: ... }

Adding and removing data from the current period


To update the current period entry, just call update() on the class. The new values will be appended to the current list for each updated parameter.

To clean up old data from the current period entry, call cleanup_current_histories(cleanup_history_depth_current), where cleanup_history_depth_current is the amount of data that you want to keep after the cleanup. The newest cleanup_history_depth_current values will be kept for each parameter in the DB.

Creating and removing old periods


To create a new period entry, call create() on the class. The new values will be stored in a new list for each updated parameter.

To remove old periods, call cleanup(cleanup_history_depth, [cleanup_history_depth_current]), where cleanup_history_depth is the amount of periods that you want to keep after the cleanup. The newest cleanup_history_depth periods will be kept. If you also specify cleanup_history_depth_current, cleanup_current_histories will be also called (see previous point).

The parameters cleanup_history_depth and cleanup_history_depth_current can also be configured in skill.yaml so they are used automatically when the cleanup method is called from AbciApp.cleanup().

__init__

def __init__(setup_data: Dict[str, List[Any]], cross_period_persisted_keys: Optional[List[str]] = None) -> None

Initialize the AbciApp database.

setup_data must be passed as a Dict[str, List[Any]] (the database internal format). The class method 'data_to_lists' can be used to convert from Dict[str, Any] to Dict[str, List[Any]] before instantiating this class.

Arguments:

  • setup_data: the setup data
  • cross_period_persisted_keys: data keys that will be kept after a new period starts

setup_data

@property
def setup_data() -> Dict[str, Any]

Get the setup_data without entries which have empty values.

Returns:

the setup_data

reset_index

@property
def reset_index() -> int

Get the current reset index.

round_count

@property
def round_count() -> int

Get the round count.

cross_period_persisted_keys

@property
def cross_period_persisted_keys() -> List[str]

Keys in the database which are persistent across periods.

get

def get(key: str, default: Any = VALUE_NOT_PROVIDED) -> Optional[Any]

Given a key, get its last for the current reset index.

get_strict

def get_strict(key: str) -> Any

Get a value from the data dictionary and raise if it is None.

update

def update(**kwargs: Any) -> None

Update the current data.

create

def create(**kwargs: List[Any]) -> None

Add a new entry to the data.

get_latest_from_reset_index

def get_latest_from_reset_index(reset_index: int) -> Dict[str, Any]

Get the latest key-value pairs from the data dictionary for the specified period.

get_latest

def get_latest() -> Dict[str, Any]

Get the latest key-value pairs from the data dictionary for the current period.

increment_round_count

def increment_round_count() -> None

Increment the round count.

__repr__

def __repr__() -> str

Return a string representation of the data.

cleanup

def cleanup(cleanup_history_depth: int, cleanup_history_depth_current: Optional[int] = None) -> None

Reset the db, keeping only the latest entries (periods).

If cleanup_history_depth_current has been also set, also clear oldest historic values in the current entry.

Arguments:

  • cleanup_history_depth: depth to clean up history
  • cleanup_history_depth_current: whether or not to clean up current entry too.

cleanup_current_histories

def cleanup_current_histories(cleanup_history_depth_current: int) -> None

Reset the parameter histories for the current entry (period), keeping only the latest values for each parameter.

data_to_lists

@staticmethod
def data_to_lists(data: Dict[str, Any]) -> Dict[str, List[Any]]

Convert Dict[str, Any] to Dict[str, List[Any]].

BaseSynchronizedData Objects

class BaseSynchronizedData()

Class to represent the synchronized data.

This is the relevant data constructed and replicated by the agents.

__init__

def __init__(db: AbciAppDB) -> None

Initialize the synchronized data.

db

@property
def db() -> AbciAppDB

Get DB.

round_count

@property
def round_count() -> int

Get the round count.

period_count

@property
def period_count() -> int

Get the period count.

Periods are executions between calls to AbciAppDB.create(), so as soon as it is called, a new period begins. It is useful to have a logical subdivision of the FSM execution. For example, if AbciAppDB.create() is called during reset, then a period will be the execution between resets.

Returns:

the period count

participants

@property
def participants() -> FrozenSet[str]

Get the currently active participants.

all_participants

@property
def all_participants() -> FrozenSet[str]

Get all registered participants.

sorted_participants

@property
def sorted_participants() -> Sequence[str]

Get the sorted participants' addresses.

The addresses are sorted according to their hexadecimal value; this is the reason we use key=str.lower as comparator.

This property is useful when interacting with the Safe contract.

Returns:

the sorted participants' addresses

nb_participants

@property
def nb_participants() -> int

Get the number of participants.

update

def update(synchronized_data_class: Optional[Type] = None, **kwargs: Any, ,) -> "BaseSynchronizedData"

Copy and update the current data.

create

def create(synchronized_data_class: Optional[Type] = None, **kwargs: Any, ,) -> "BaseSynchronizedData"

Copy and update with new data.

__repr__

def __repr__() -> str

Return a string representation of the data.

keeper_randomness

@property
def keeper_randomness() -> float

Get the keeper's random number [0-1].

most_voted_randomness

@property
def most_voted_randomness() -> str

Get the most_voted_randomness.

most_voted_keeper_address

@property
def most_voted_keeper_address() -> str

Get the most_voted_keeper_address.

is_keeper_set

@property
def is_keeper_set() -> bool

Check whether keeper is set.

blacklisted_keepers

@property
def blacklisted_keepers() -> Set[str]

Get the current cycle's blacklisted keepers who cannot submit a transaction.

participant_to_selection

@property
def participant_to_selection() -> Mapping

Check whether keeper is set.

participant_to_randomness

@property
def participant_to_randomness() -> Mapping

Check whether keeper is set.

participant_to_votes

@property
def participant_to_votes() -> Mapping

Check whether keeper is set.

AbstractRound Objects

class AbstractRound(Generic[EventType, TransactionType],  ABC)

This class represents an abstract round.

A round is a state of the FSM App execution. It usually involves interactions between participants in the FSM App, although this is not enforced at this level of abstraction.

Concrete classes must set: - round_id: the identifier for the concrete round class; - allowed_tx_type: the transaction type that is allowed for this round.

__init__

def __init__(synchronized_data: BaseSynchronizedData, consensus_params: ConsensusParams, previous_round_tx_type: Optional[TransactionType] = None) -> None

Initialize the round.

synchronized_data

@property
def synchronized_data() -> BaseSynchronizedData

Get the synchronized data.

check_transaction

def check_transaction(transaction: Transaction) -> None

Check transaction against the current state.

Arguments:

  • transaction: the transaction

process_transaction

def process_transaction(transaction: Transaction) -> None

Process a transaction.

By convention, the payload handler should be a method of the class that is named '{payload_name}'.

Arguments:

  • transaction: the transaction.

end_block

@abstractmethod
def end_block() -> Optional[Tuple[BaseSynchronizedData, Enum]]

Process the end of the block.

The role of this method is check whether the round is considered ended.

If the round is ended, the return value is - the final result of the round. - the event that triggers a transition. If None, the period in which the round was executed is considered ended.

This is done after each block because we consider the consensus engine's block, and not the transaction, as the smallest unit on which the consensus is reached; in other words, each read operation on the state should be done only after each block, and not after each transaction.

check_allowed_tx_type

def check_allowed_tx_type(transaction: Transaction) -> None

Check the transaction is of the allowed transaction type.

Arguments:

:raises: TransactionTypeNotRecognizedError if the transaction can be applied to the current state. - transaction: the transaction

check_majority_possible_with_new_voter

@classmethod
def check_majority_possible_with_new_voter(cls, votes_by_participant: Dict[str, BaseTxPayload], new_voter: str, new_vote: BaseTxPayload, nb_participants: int, exception_cls: Type[ABCIAppException] = ABCIAppException) -> None

Check that a Byzantine majority is achievable, once a new vote is added.

Arguments:

    before the new vote is added
                   check fails.

:raises: exception_cls: in case the check does not pass. - votes_by_participant: a mapping from a participant to its vote, - new_voter: the new voter - new_vote: the new vote - nb_participants: the total number of participants - exception_cls: the class of the exception to raise in case the

check_majority_possible

@classmethod
def check_majority_possible(cls, votes_by_participant: Dict[str, BaseTxPayload], nb_participants: int, exception_cls: Type[ABCIAppException] = ABCIAppException) -> None

Check that a Byzantine majority is still achievable.

The idea is that, even if all the votes have not been delivered yet, it can be deduced whether a quorum cannot be reached due to divergent preferences among the voters and due to a too small number of other participants whose vote has not been delivered yet.

The check fails iff:

nb_remaining_votes + largest_nb_votes < quorum

That is, if the number of remaining votes is not enough to make the most voted item so far to exceed the quorum.

Preconditions on the input: - the size of votes_by_participant should not be greater than "nb_participants - 1" voters - new voter must not be in the current votes_by_participant

Arguments:

                  check fails.
  • votes_by_participant: a mapping from a participant to its vote
  • nb_participants: the total number of participants
  • exception_cls: the class of the exception to raise in case the

Raises:

  • exception_cls: in case the check does not pass.

is_majority_possible

@classmethod
def is_majority_possible(cls, votes_by_participant: Dict[str, BaseTxPayload], nb_participants: int) -> bool

Return true if a Byzantine majority is achievable, false otherwise.

Arguments:

  • votes_by_participant: a mapping from a participant to its vote
  • nb_participants: the total number of participants

Returns:

True if the majority is still possible, false otherwise.

consensus_threshold

@property
def consensus_threshold() -> int

Consensus threshold

check_payload

@abstractmethod
def check_payload(payload: BaseTxPayload) -> None

Check payload.

process_payload

@abstractmethod
def process_payload(payload: BaseTxPayload) -> None

Process payload.

DegenerateRound Objects

class DegenerateRound(AbstractRound)

This class represents the finished round during operation.

It is a sink round.

check_payload

def check_payload(payload: BaseTxPayload) -> None

Check payload.

process_payload

def process_payload(payload: BaseTxPayload) -> None

Process payload.

end_block

def end_block() -> Optional[Tuple[BaseSynchronizedData, Enum]]

End block.

CollectionRound Objects

class CollectionRound(AbstractRound)

CollectionRound.

This class represents abstract logic for collection based rounds where the round object needs to collect data from different agents. The data might for example be from a voting round or estimation round.

__init__

def __init__(*args: Any, **kwargs: Any)

Initialize the collection round.

accepting_payloads_from

@property
def accepting_payloads_from() -> FrozenSet[str]

Accepting from the active set, or also from (re)joiners

payloads

@property
def payloads() -> List[BaseTxPayload]

Get all agent payloads

payloads_count

@property
def payloads_count() -> Counter

Get count of payload attributes

process_payload

def process_payload(payload: BaseTxPayload) -> None

Process payload.

check_payload

def check_payload(payload: BaseTxPayload) -> None

Check Payload

_CollectUntilAllRound Objects

class _CollectUntilAllRound(CollectionRound,  ABC)

_CollectUntilAllRound

This class represents logic for when rounds need to collect payloads from all agents.

This round should only be used for registration of new agents.

check_payload

def check_payload(payload: BaseTxPayload) -> None

Check Payload

process_payload

def process_payload(payload: BaseTxPayload) -> None

Process payload.

collection_threshold_reached

@property
def collection_threshold_reached() -> bool

Check that the collection threshold has been reached.

CollectDifferentUntilAllRound Objects

class CollectDifferentUntilAllRound(_CollectUntilAllRound,  ABC)

CollectDifferentUntilAllRound

This class represents logic for rounds where a round needs to collect different payloads from each agent.

This round should only be used for registration of new agents when there is synchronization of the db.

check_payload

def check_payload(payload: BaseTxPayload) -> None

Check Payload

CollectSameUntilAllRound Objects

class CollectSameUntilAllRound(_CollectUntilAllRound,  ABC)

This class represents logic for when a round needs to collect the same payload from all the agents.

This round should only be used for registration of new agents when there is no synchronization of the db.

check_payload

def check_payload(payload: BaseTxPayload) -> None

Check Payload

common_payload

@property
def common_payload() -> Any

Get the common payload among the agents.

CollectSameUntilThresholdRound Objects

class CollectSameUntilThresholdRound(CollectionRound)

CollectSameUntilThresholdRound

This class represents logic for rounds where a round needs to collect same payload from k of n agents.

threshold_reached

@property
def threshold_reached() -> bool

Check if the threshold has been reached.

most_voted_payload

@property
def most_voted_payload() -> Any

Get the most voted payload.

end_block

def end_block() -> Optional[Tuple[BaseSynchronizedData, Enum]]

Process the end of the block.

OnlyKeeperSendsRound Objects

class OnlyKeeperSendsRound(AbstractRound)

OnlyKeeperSendsRound

This class represents logic for rounds where only one agent sends a payload

__init__

def __init__(*args: Any, **kwargs: Any)

Initialize the 'collect-observation' round.

process_payload

def process_payload(payload: BaseTxPayload) -> None

Handle a deploy safe payload.

check_payload

def check_payload(payload: BaseTxPayload) -> None

Check a deploy safe payload can be applied to the current state.

has_keeper_sent_payload

@property
def has_keeper_sent_payload() -> bool

Check if keeper has sent the payload.

end_block

def end_block() -> Optional[Tuple[BaseSynchronizedData, Enum]]

Process the end of the block.

VotingRound Objects

class VotingRound(CollectionRound)

VotingRound

This class represents logic for rounds where a round needs votes from agents, pass if k same votes of n agents

vote_count

@property
def vote_count() -> Counter

Get agent payload vote count

positive_vote_threshold_reached

@property
def positive_vote_threshold_reached() -> bool

Check that the vote threshold has been reached.

negative_vote_threshold_reached

@property
def negative_vote_threshold_reached() -> bool

Check that the vote threshold has been reached.

none_vote_threshold_reached

@property
def none_vote_threshold_reached() -> bool

Check that the vote threshold has been reached.

end_block

def end_block() -> Optional[Tuple[BaseSynchronizedData, Enum]]

Process the end of the block.

CollectDifferentUntilThresholdRound Objects

class CollectDifferentUntilThresholdRound(CollectionRound)

CollectDifferentUntilThresholdRound

This class represents logic for rounds where a round needs to collect different payloads from k of n agents

collection_threshold_reached

@property
def collection_threshold_reached() -> bool

Check if the threshold has been reached.

end_block

def end_block() -> Optional[Tuple[BaseSynchronizedData, Enum]]

Process the end of the block.

CollectNonEmptyUntilThresholdRound Objects

class CollectNonEmptyUntilThresholdRound(CollectDifferentUntilThresholdRound)

Collect all the data among agents.

This class represents logic for rounds where we need to collect payloads from each agent which will contain optional, different data and only keep the non-empty.

This round may be used for cases that we want to collect all the agent's data, such as late-arriving messages.

end_block

def end_block() -> Optional[Tuple[BaseSynchronizedData, Enum]]

Process the end of the block.

TimeoutEvent Objects

@dataclass(order=True)
class TimeoutEvent(Generic[EventType])

Timeout event.

Timeouts Objects

class Timeouts(Generic[EventType])

Class to keep track of pending timeouts.

__init__

def __init__() -> None

Initialize.

size

@property
def size() -> int

Get the size of the timeout queue.

add_timeout

def add_timeout(deadline: datetime.datetime, event: EventType) -> int

Add a timeout.

cancel_timeout

def cancel_timeout(entry_count: int) -> None

Remove a timeout.

Arguments:

:raises: KeyError: if the entry count is not found. - entry_count: the entry id to remove.

pop_earliest_cancelled_timeouts

def pop_earliest_cancelled_timeouts() -> None

Pop earliest cancelled timeouts.

get_earliest_timeout

def get_earliest_timeout() -> Tuple[datetime.datetime, Any]

Get the earliest timeout-event pair.

pop_timeout

def pop_timeout() -> Tuple[datetime.datetime, Any]

Remove and return the earliest timeout-event pair.

_MetaAbciApp Objects

class _MetaAbciApp(ABCMeta)

A metaclass that validates AbciApp's attributes.

__new__

def __new__(mcs, name: str, bases: Tuple, namespace: Dict, **kwargs: Any) -> Type

Initialize the class.

AbciApp Objects

class AbciApp(
    Generic[EventType],  ABC, metaclass=_MetaAbciApp)

Base class for ABCI apps.

Concrete classes of this class implement the ABCI App.

__init__

def __init__(synchronized_data: BaseSynchronizedData, consensus_params: ConsensusParams, logger: logging.Logger)

Initialize the AbciApp.

is_abstract

@classmethod
def is_abstract(cls) -> bool

Return if the abci app is abstract.

synchronized_data

@property
def synchronized_data() -> BaseSynchronizedData

Return the current synchronized data.

reset_index

@property
def reset_index() -> int

Return the reset index.

reset_index

@reset_index.setter
def reset_index(reset_index: int) -> None

Set the reset index.

get_all_rounds

@classmethod
def get_all_rounds(cls) -> Set[AppState]

Get all the round states.

get_all_events

@classmethod
def get_all_events(cls) -> Set[EventType]

Get all the events.

get_all_round_classes

@classmethod
def get_all_round_classes(cls) -> Set[AppState]

Get all round classes.

last_timestamp

@property
def last_timestamp() -> datetime.datetime

Get last timestamp.

setup

def setup() -> None

Set up the behaviour.

current_round

@property
def current_round() -> AbstractRound

Get the current round.

current_round_id

@property
def current_round_id() -> Optional[str]

Get the current round id.

current_round_height

@property
def current_round_height() -> int

Get the current round height.

last_round_id

@property
def last_round_id() -> Optional[str]

Get the last round id.

is_finished

@property
def is_finished() -> bool

Check whether the AbciApp execution has finished.

latest_result

@property
def latest_result() -> Optional[BaseSynchronizedData]

Get the latest result of the round.

check_transaction

def check_transaction(transaction: Transaction) -> None

Check a transaction.

Forward the call to the current round object.

Arguments:

  • transaction: the transaction.

process_transaction

def process_transaction(transaction: Transaction) -> None

Process a transaction.

Forward the call to the current round object.

Arguments:

  • transaction: the transaction.

process_event

def process_event(event: EventType, result: Optional[BaseSynchronizedData] = None) -> None

Process a round event.

update_time

def update_time(timestamp: datetime.datetime) -> None

Observe timestamp from last block.

Arguments:

  • timestamp: the latest block's timestamp.

cleanup

def cleanup(cleanup_history_depth: int, cleanup_history_depth_current: Optional[int] = None) -> None

Clear data.

cleanup_current_histories

def cleanup_current_histories(cleanup_history_depth_current: int) -> None

Reset the parameter histories for the current entry (period), keeping only the latest values for each parameter.

RoundSequence Objects

class RoundSequence()

This class represents a sequence of rounds

It is a generic class that keeps track of the current round of the consensus period. It receives 'deliver_tx' requests from the ABCI handlers and forwards them to the current active round instance, which implements the ABCI app logic. It also schedules the next round (if any) whenever a round terminates.

__init__

def __init__(abci_app_cls: Type[AbciApp])

Initialize the round.

setup

def setup(*args: Any, **kwargs: Any) -> None

Set up the round sequence.

Arguments:

  • args: the arguments to pass to the round constructor.
  • kwargs: the keyword-arguments to pass to the round constructor.

start_sync

def start_sync() -> None

Set _syncing_up flag to true.

if the _syncing_up flag is set to true, the async_act method won't be executed. For more details refer to https://github.com/valory-xyz/open-autonomy/issues/247#issuecomment-1012268656

end_sync

def end_sync() -> None

Set _syncing_up flag to false.

syncing_up

@property
def syncing_up() -> bool

Return if the app is in sync mode.

abci_app

@property
def abci_app() -> AbciApp

Get the AbciApp.

height

@property
def height() -> int

Get the height.

is_finished

@property
def is_finished() -> bool

Check if a round sequence has finished.

check_is_finished

def check_is_finished() -> None

Check if a round sequence has finished.

current_round

@property
def current_round() -> AbstractRound

Get current round.

current_round_id

@property
def current_round_id() -> Optional[str]

Get the current round id.

current_round_height

@property
def current_round_height() -> int

Get the current round height.

last_round_id

@property
def last_round_id() -> Optional[str]

Get the last round id.

last_timestamp

@property
def last_timestamp() -> datetime.datetime

Get the last timestamp.

last_round_transition_timestamp

@property
def last_round_transition_timestamp() -> datetime.datetime

Returns the timestamp for last round transition.

last_round_transition_height

@property
def last_round_transition_height() -> int

Returns the height for last round transition.

last_round_transition_root_hash

@property
def last_round_transition_root_hash() -> bytes

Returns the root hash for last round transition.

last_round_transition_tm_height

@property
def last_round_transition_tm_height() -> int

Returns the Tendermint height for last round transition.

latest_synchronized_data

@property
def latest_synchronized_data() -> BaseSynchronizedData

Get the latest synchronized_data.

root_hash

@property
def root_hash() -> bytes

Get the Merkle root hash of the application state.

Create an app hash that always increases in order to avoid conflicts between resets. Eventually, we do not necessarily need to have a value that increases, but we have to generate a hash that is always different among the resets, since our abci's state is different even thought we have reset the chain! For example, if we are in height 11, reset and then reach height 11 again, if we end up using the same hash at height 11 between the resets, then this is problematic.

Returns:

the root hash to be included as the Header.AppHash in the next block.

tm_height

@property
def tm_height() -> int

Get Tendermint's current height.

tm_height

@tm_height.setter
def tm_height(_tm_height: int) -> None

Set Tendermint's current height.

block_stall_deadline_expired

@property
def block_stall_deadline_expired() -> bool

Get if the deadline for not having received any begin block requests from the Tendermint node has expired.

init_chain

def init_chain(initial_height: int) -> None

Init chain.

begin_block

def begin_block(header: Header) -> None

Begin block.

deliver_tx

def deliver_tx(transaction: Transaction) -> None

Deliver a transaction.

Appends the transaction to build the block on 'end_block' later.

Arguments:

:raises: an Error otherwise. - transaction: the transaction.

end_block

def end_block() -> None

Process the 'end_block' request.

commit

def commit() -> None

Process the 'commit' request.

reset_blockchain

def reset_blockchain(is_replay: bool = False) -> None

Reset blockchain after tendermint reset.

Back to top