82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
|
# Python API for AnySystem 0.3.0 (IDE/type checking only).
|
||
|
|
# Source: https://github.com/osukhoroslov/anysystem/blob/v0.3.0/python/anysystem.py
|
||
|
|
# The executable module is embedded in the Rust crate; keep these declarations
|
||
|
|
# in sync with the pinned crate when upgrading AnySystem.
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from abc import ABC, abstractmethod
|
||
|
|
from typing import Any, Dict, List, Tuple, Union
|
||
|
|
|
||
|
|
JSON = Union[Dict[str, "JSON"], List["JSON"], str, int, float, bool, None]
|
||
|
|
|
||
|
|
class Message:
|
||
|
|
_type: str
|
||
|
|
_data: Dict[str, Any]
|
||
|
|
|
||
|
|
def __init__(self, message_type: str, data: Dict[str, Any]) -> None: ...
|
||
|
|
@property
|
||
|
|
def type(self) -> str: ...
|
||
|
|
def get(self, key: str, default: Any = None) -> Any: ...
|
||
|
|
def remove(self, key: str) -> None: ...
|
||
|
|
def __contains__(self, key: str) -> bool: ...
|
||
|
|
def __getitem__(self, key: str) -> Any: ...
|
||
|
|
def __setitem__(self, key: str, value: Any) -> None: ...
|
||
|
|
@staticmethod
|
||
|
|
def from_json(message_type: str, json_str: str) -> Message: ...
|
||
|
|
|
||
|
|
class Context:
|
||
|
|
_time: float
|
||
|
|
_sent_messages: List[Tuple[str, str, str]]
|
||
|
|
_sent_local_messages: List[tuple[str, str]]
|
||
|
|
_timer_actions: List[Tuple[str, float, bool]]
|
||
|
|
|
||
|
|
def __init__(self, time: float) -> None: ...
|
||
|
|
def send(self, msg: Message, to: str) -> None:
|
||
|
|
"""Sends a message to the specified process."""
|
||
|
|
...
|
||
|
|
def send_local(self, msg: Message) -> None:
|
||
|
|
"""Sends a _local_ message."""
|
||
|
|
...
|
||
|
|
def set_timer(self, timer_name: str, delay: float) -> None:
|
||
|
|
"""
|
||
|
|
Sets a timer that will trigger on_timer callback after the specified delay.
|
||
|
|
If there is an active timer with this name, its delay is overridden.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
def set_timer_once(self, timer_name: str, delay: float) -> None:
|
||
|
|
"""
|
||
|
|
Sets a timer that will trigger on_timer callback after the specified delay.
|
||
|
|
If there is an active timer with this name, this call is ignored.
|
||
|
|
"""
|
||
|
|
...
|
||
|
|
def cancel_timer(self, timer_name: str) -> None:
|
||
|
|
"""Cancels timer with the specified name."""
|
||
|
|
...
|
||
|
|
def time(self) -> float:
|
||
|
|
"""Returns the current system time."""
|
||
|
|
...
|
||
|
|
|
||
|
|
class Process(ABC):
|
||
|
|
@abstractmethod
|
||
|
|
def on_start(self, ctx: Context) -> None:
|
||
|
|
"""This method is called when the process is started on a node."""
|
||
|
|
...
|
||
|
|
@abstractmethod
|
||
|
|
def on_local_message(self, msg: Message, ctx: Context) -> None:
|
||
|
|
"""This method is called when a _local_ message is received."""
|
||
|
|
...
|
||
|
|
@abstractmethod
|
||
|
|
def on_message(self, msg: Message, sender: str, ctx: Context) -> None:
|
||
|
|
"""This method is called when a message is received."""
|
||
|
|
...
|
||
|
|
@abstractmethod
|
||
|
|
def on_timer(self, timer_name: str, ctx: Context) -> None:
|
||
|
|
"""This method is called when a timer fires."""
|
||
|
|
...
|
||
|
|
def get_state(self) -> str:
|
||
|
|
"""This method returns the string representation of process state."""
|
||
|
|
...
|
||
|
|
def set_state(self, state_encoded: str) -> None:
|
||
|
|
"""This method restores the process state by its string representation."""
|
||
|
|
...
|