iamai package¶

Subpackages¶

Submodules¶

Module contents¶

API Documentation

Comprehensive AI Toolkit for Multi Modal Learning and Cross-Platform Robotics.

This Module imports the following contents from the sub-module.

class iamai.Adapter(bot: Bot)[源代码]¶

基类：Generic[EventT, ConfigT], ABC

协议适配器基类。

name¶

适配器的名称。

Type:: str

bot¶

当前的机器人对象。

Type:: Bot

Config: Type[ConfigT]¶

bot: Bot¶

property config: ConfigT¶: 适配器配置。

获取满足指定条件的的事件，协程会等待直到适配器接收到满足条件的事件、超过最大事件数或超时。

类似 Bot 类的 get() 方法，但是隐含了判断产生事件的适配器是本适配器。等效于 Bot 类的 get() 方法传入 adapter_type 为本适配器类型。

参数:

func -- 协程或者函数，函数会被自动包装为协程执行。要求接受一个事件作为参数，返回布尔值。当协程返回 True 时返回当前事件。当为 None 时相当于输入对于任何事件均返回真的协程，即返回适配器接收到的下一个事件。
event_type -- 当指定时，只接受指定类型的事件，先于 func 条件生效。默认为 None。
max_try_times -- 最大事件数。
timeout -- 超时时间。

返回:

返回满足 func 条件的事件。

抛出:

GetEventTimeout -- 超过最大事件数或超时。

name: str¶

abstract async run() → None[源代码]¶

适配器运行方法，适配器开发者必须实现该方法。

适配器运行过程中保持保持运行，当此方法结束后， AliceBot 不会自动重新启动适配器。

final async safe_run() → None[源代码]¶: 附带有异常处理地安全运行适配器。

async shutdown() → None[源代码]¶

在适配器结束运行时运行的方法，用于安全地关闭适配器。

AliceBot 在接收到系统的结束信号后依次运行并等待所有适配器的 shutdown() 方法。当强制退出时此方法可能未被执行。

async startup() → None[源代码]¶

在适配器开始运行前运行的方法，用于初始化适配器。

AliceBot 依次运行并等待所有适配器的 startup() 方法，待运行完毕后再创建 run() 任务。

class iamai.Bot(*, config_file: str | None = 'config.toml', config_dict: Dict[str, Any] | None = None, hot_reload: bool = False)[源代码]¶

基类：object

iamai Bot class object, defines the basic functionality for interacting.

Read and save configuration Config, load adapters Adapter and plugins Plugin, then distribute the events.

config¶

Bot configuration.

Type:: iamai.config.MainConfig

should_exit¶

Whether the bot should enter the ready-to-exit state.

Type:: asyncio.locks.Event

adapters¶

List of currently loaded adapters.

Type:: List[iamai.adapter.Adapter[Any, Any]]

plugins_priority_dict¶

Plugin priority dictionary.

Type:: Dict[int, List[Type[iamai.plugin.Plugin[Any, Any, Any]]]]

plugin_state¶

Plugin status.

Type:: Dict[str, Any]

global_state¶

Global status.

Type:: Dict[Any, Any]

adapter_run_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) → Callable[[Adapter[Any, Any]], Awaitable[None]][源代码]¶

参数:: func -- the registered function.
返回:: The registered function.

adapter_shutdown_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) → Callable[[Adapter[Any, Any]], Awaitable[None]][源代码]¶

参数:: func -- the registered function.
返回:: The registered function.

adapter_startup_hook(func: Callable[[Adapter[Any, Any]], Awaitable[None]]) → Callable[[Adapter[Any, Any]], Awaitable[None]][源代码]¶

参数:: func -- the registered function.
返回:: The registered function.

adapters: List[Adapter[Any, Any]]¶

bot_exit_hook(func: Callable[[Bot], Awaitable[None]]) → Callable[[Bot], Awaitable[None]][源代码]¶

参数:: func -- the registered function.
返回:: The registered function.

bot_run_hook(func: Callable[[Bot], Awaitable[None]]) → Callable[[Bot], Awaitable[None]][源代码]¶

参数:: func -- the registered function.
返回:: The registered function.

config: MainConfig¶

error_or_exception(message: str, exception: Exception) → None[源代码]¶

Output error or exception logs based on the current Bot configuration.

参数:

message -- message.
exception -- Exception.

event_postprocessor_hook(func: Callable[[Event[Any]], Awaitable[None]]) → Callable[[Event[Any]], Awaitable[None]][源代码]¶

参数:: func -- the registered function.
返回:: The registered function.

event_preprocessor_hook(func: Callable[[Event[Any]], Awaitable[None]]) → Callable[[Event[Any]], Awaitable[None]][源代码]¶

参数:: func -- the registered function.
返回:: The registered function.

Get events that meet the specified conditions. The coroutine will wait until the adapter receives events that meet the conditions, exceeds the maximum number of events, or times out.

参数:

func -- Coroutine or function, the function will be automatically packaged as a coroutine for execution. Requires an event to be accepted as a parameter and returns a Boolean value. Returns the current event when the coroutine returns True. When None is equivalent to the input coroutine returning true for any event, that is, returning the next event received by the adapter.
event_type -- When specified, only events of the specified type are accepted, taking effect before the func condition. Defaults to None.
adapter_type -- When specified, only events generated by the specified adapter will be accepted, taking effect before the func condition. Defaults to None.
max_try_times -- Maximum number of events.
timeout -- timeout period.

返回:

Returns events that satisfy the condition of func.

抛出:

GetEventTimeout -- Maximum number of events exceeded or timeout.

get_adapter(adapter: str) → Adapter[Any, Any][源代码]¶

get_adapter(adapter: Type[AdapterT]) → AdapterT

Get the loaded adapter by name or adapter class.

参数:: adapter -- adapter name or adapter class.
返回:: The obtained adapter object.
抛出:: LookupError -- No adapter object with this name found.

get_plugin(name: str) → Type[Plugin[Any, Any, Any]][源代码]¶

Get the loaded plugin class by name.

参数:: name -- plugin name
返回:: The obtained plug-in class.
抛出:: LookupError -- The plugin class with this name cannot be found.

global_state: Dict[Any, Any]¶

async handle_event(current_event: Event[Any], *, handle_get: bool = True, show_log: bool = True) → None[源代码]¶

Called by the adapter object, distributes events to all plug-ins according to priority, and handles plug-in signals such as stop and skip.

This method should not be called manually by the user.

参数:

current_event -- The currently pending Event.
handle_get -- Whether the current event can be captured by the get method, the default is True.
show_log -- Whether to display in the log, the default is True.

load_adapters(*adapters: Type[Adapter[Any, Any]] | str) → None[源代码]¶

Load adapter.

参数:

*adapters --

Adapter class or adapter name, type can be Type[Adapter] or str. If it is of type Type[Adapter], it will be loaded as an adapter class. If it is of type str, it will be loaded as the adapter module name, and the format is the same as the Python import statement.

For example: path.of.adapter.

load_plugins(*plugins: Type[Plugin[Any, Any, Any]] | str | Path) → None[源代码]¶

Load the plugin.

参数:

*plugins --

Plugin class, plugin module name or plug-in module file path. Type can be Type[Plugin], str or pathlib.Path. If it is Type[Plugin], it will be loaded as a plug-in class. If it is of type str, it will be loaded as the plug-in module name, and the format is the same as the Python import statement.

For example: path.of.plugin.

If it is of type pathlib.Path, it will be loaded as the plug-in module file path.: For example: pathlib.Path("path/of/plugin").

load_plugins_from_dirs(*dirs: Path) → None[源代码]¶

Load plug-ins from the directory. Plug-ins in modules starting with _ will not be imported. The path can be a relative path or an absolute path.

参数:: *dirs -- Module paths that store modules containing plugins. For example: pathlib.Path("path/of/plugins/") .

property locale: List[str]¶: Get the current language of the bot.

property locale_dirs: List[Path]¶: Get the locale directories of the bot.

plugin_state: Dict[str, Any]¶

property plugins: List[Type[Plugin[Any, Any, Any]]]¶: List of currently loaded plugins.

plugins_priority_dict: Dict[int, List[Type[Plugin[Any, Any, Any]]]]¶

reload_plugins() → None[源代码]¶: Manually reload all plugins.

restart() → None[源代码]¶: Exit and rerun iamai.

run() → None[源代码]¶: Run iamai, monitor and intercept system exit signals, and update robot configuration.

should_exit: Event¶

class iamai.ConfigModel(**extra_data: Any)[源代码]¶

基类：BaseModel

iamai configuration model.

__config_name__¶

Configuration name.

Type:: str

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Subdependency decorator.

参数:

dependency -- dependency class. If not specified, it will be automatically determined based on the type annotation of the field.
use_cache -- whether to use cache. Defaults to True.

返回:

Returns the internal child dependency object.

class iamai.Event(*, adapter: Any, type: str | None, **extra_data: Any)[源代码]¶

基类：ABC, BaseModel, Generic[AdapterT]

The base class of event classes.

adapter¶

The adapter object that generated the current event.

Type:: Any

type¶

event type.

Type:: str | None

__handled__¶

Indicates whether the event has been handled, used for adapter processing. Warning: Do not change the value of this property manually.

Type:: bool

adapter: Any¶

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

type: str | None¶

class iamai.MessageEvent(*, adapter: Any, type: str | None, **extra_data: Any)[源代码]¶

基类：Event, Generic[AdapterT]

Base class for general message event classes.

async ask(message: str, max_try_times: int | None = None, timeout: int | float | None = None) → Self[源代码]¶

Ask for news.

Indicates getting the user's reply after replying to a message. Equivalent to executing get() after reply().

参数:

message -- The content of the reply message.
max_try_times -- Maximum number of events.
timeout -- timeout period.

返回:

Message event that the user replies to.

async get(*, max_try_times: int | None = None, timeout: int | float | None = None) → Self[源代码]¶

Get the user's reply message.

Equivalent to get() of Bot, the condition is that the adapter, event type and sender are the same.

参数:

max_try_times -- Maximum number of events.
timeout -- timeout period.

返回:

Message event that the user replies to.

抛出:

GetEventTimeout -- Maximum number of events exceeded or timeout.

abstract get_plain_text() → str[源代码]¶

Get the plain text content of the message.

返回:: The plain text content of the message.

abstract async is_same_sender(other: Self) → bool[源代码]¶

Determine whether itself and another event are the same sender.

参数:: other -- another event.
返回:: Is it the same sender?

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

abstract async reply(message: str) → Any[源代码]¶

Reply message.

参数:: message -- The content of the reply message.
返回:: The response to the reply message action.

class iamai.Plugin[源代码]¶

基类：ABC, Generic[EventT, StateT, ConfigT]

Base class for all iamai plugins.

event¶: The event currently being processed by this plugin.

priority¶

The priority of the plugin. The smaller the number, the higher the priority. The default is 0.

Type:: ClassVar[int]

block¶

Whether to prevent the propagation of events after the plug-in is executed. True means blocking.

Type:: ClassVar[bool]

__plugin_load_type__¶

Plugin load type, automatically set by iamai, reflects how this plugin is loaded.

Type:: ClassVar[iamai.plugin.PluginLoadType]

__plugin_file_path__¶

None when the plugin loading type is PluginLoadType.CLASS, Otherwise, the location of the Python module in which the plugin is defined.

Type:: ClassVar[str | None]

Config: Type[ConfigT]¶

block: ClassVar[bool] = False¶

property bot: Bot¶: bot object.

property config: ConfigT¶: plugin configuration.

event: EventT = InnerDepends(Event)¶

abstract async handle() → None[源代码]¶: Method to handle events. iamai will call this method when the rule() method returns True. Each plugin must implement this method.

property name: str¶: plugin class name.

priority: ClassVar[int] = 0¶

abstract async rule() → bool[源代码]¶: Method to match the event. When the event is processed, this method will be called in sequence according to the priority of the plugin. When this method returns True, the event will be handed over to this plugin for processing. Each plugin must implement this method.

备注

It is not recommended to implement event processing directly in this method. Please leave the specific processing of events to the handle() method.

final skip() → NoReturn[源代码]¶: Skips itself and continues propagation of the current event.

property state: StateT¶: plugin status.

final stop() → NoReturn[源代码]¶: Stop propagation of current events.