The AbciApp class
Note
For clarity, the snippets of code presented here are a simplified version of the actual implementation. We refer the reader to the Open Autonomy API for the complete details.
The AbciApp
abstract class provides the necessary interface for implementation of FSM Apps. Concrete implementations of the AbciApp
class requires that the
developer implement the class attributes initial_round_cls
,
transition_function
and final_states
. The internal
_MetaRoundBehaviour
metaclass is used to enforce this during implementation by the developer.
An overview of the code looks as follows:
# skills.abstract_round_behaviour.base.py
AppState = Type[AbstractRound]
AbciAppTransitionFunction = Dict[AppState, Dict[EventType, AppState]]
EventToTimeout = Dict[EventType, float]
class AbciApp(
Generic[EventType], ABC, metaclass=_MetaAbciApp
):
"""
Base class for ABCI apps.
Concrete classes of this class implement the ABCI App.
"""
initial_round_cls: AppState
initial_states: Set[AppState] = set()
transition_function: AbciAppTransitionFunction
final_states: Set[AppState] = set()
event_to_timeout: EventToTimeout = {}
cross_period_persisted_keys: FrozenSet[str] = frozenset()
def __init__(
self,
synchronized_data: BaseSynchronizedData,
logger: logging.Logger,
):
"""Initialize the AbciApp."""
def process_transaction(self, transaction: Transaction, dry: bool = False) -> None:
"""
Process a transaction.
The background rounds run concurrently with other (normal) rounds.
First we check if the transaction is meant for a background round,
if not we forward it to the current round object.
:param transaction: the transaction.
:param dry: whether the transaction should only be checked and not processed.
"""
def process_event(
self, event: EventType, result: Optional[BaseSynchronizedData] = None
) -> None:
"""Process a round event."""
def update_time(self, timestamp: datetime.datetime) -> None:
"""
Observe timestamp from last block.
:param timestamp: the latest block's timestamp.
"""
# (...)
Some of its methods relate to concepts discussed in the FSM section:
process_transaction( )
: Processes the payload generated by the AEAs during a round.process_event( )
: Allows for the execution of transitions to the next round based on the output of the current round.update_time( )
: Allows for resetting of timeouts based on the timestamp from last block. This is the only form of time synchronization that exists in this system of asynchronously operating AEAs, an understanding of which is indispensable to a developer that needs to implement any sort of time-based logic as part of their agents' behaviour.
A concrete implementation of a subclass of AbciApp
looks as follows:
class MyAbciApp(AbciApp):
"""My ABCI-based Finite-State Machine Application execution behaviour"""
initial_round_cls: AppState = RoundA
initial_states: Set[AppState] = set()
transition_function: AbciAppTransitionFunction = {
RoundA: {
Event.DONE: RoundB,
Event.ROUND_TIMEOUT: RoundA,
Event.NO_MAJORITY: RoundA,
},
RoundB: {
Event.DONE: FinalRound,
Event.ROUND_TIMEOUT: RoundA,
Event.NO_MAJORITY: RoundA,
},
FinalRound: {},
}
final_states: Set[AppState] = {FinalRound}
event_to_timeout: EventToTimeout = {
Event.ROUND_TIMEOUT: 30.0,
}
db_pre_conditions: Dict[AppState, List[str]] = {
RoundA: {
get_name(BaseSynchronizedData.required_value),
},
}
db_post_conditions: Dict[AppState, List[str]] = {
FinalRound: {
get_name(BaseSynchronizedData.generated_value),
},
}
# (...)
NO_MAJORITY
NO_MAJORITY
The mandatory field initial_round_cls
indicates the round associated to the initial state of the FSM.
The set of initial_states
is optionally provided by the developer. If none is provided,
provided a set containing the initial_round_cls
is inferred automatically.
When the FSM App processes an Event
it schedules the round associated to the next state by looking at the corresponding transition from the transition_function
and sets the associated timeouts, if
any.
The db_pre_conditions
and db_post_conditions
are conditions that need to be met when entering and when leaving
the AbciApp
. These are taken into consideration when chaining FSMs, in order to make sure that
the required data exist in the synchronized data. Therefore, an application can fail early, before running any rounds,
and inform the user about an incorrect chaining attempt.
The pre- and post- conditions on the synchronized data need to be defined for each initial and final state
in an AbciApp
. If there are no conditions required for an app, they can be mapped to an empty list.
Otherwise, the list should contain the names of all the required properties in the synchronized data.
The suggested way to do this is to use the get_name
function, defined in the abstract_round_abci
,
so that strings are avoided as they can get out of sync.
In addition to the AbciApp
class, the FSM App also requires that the AbstractRoundBehaviour
class be implemented in order to run the state transition logic contained in it.