enrobie.plugins package#

Subpackages#

Module contents#

Functions and routines associated with Enasis Network Chatting Robie.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class enrobie.plugins.AinswerPlugin(robie: Robie, name: str, params: RobieChildParams)[source]#

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin allows for interacting with an LLM model.

property agent: Agent[AinswerDepends, str]#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

ainswer(mitem: RobieMessage, prompt: str, respond: Type[AinswerResponse]) → str[source]#

Submit the question to the LLM and return the response.

Parameters:

mitem – Item containing information for operation.
prompt – Additional prompt insert before question.
respond – Model to describe the expected response.

Returns:

Response adhering to provided specifications.

property history: AinswerHistory#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property memory: AinswerMemory#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property models: AinswerModels#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

operate() → None[source]#: Perform the operation related to Robie service threads.

property params: AinswerPluginParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

printer(source: dict[str, Any], color: int = 6) → None[source]#

Print the contents for the object within Robie instance.

Parameters:

source – Content which will be shown after header.
color – Override the color used for box character.

property question: AinswerQuestion#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

classmethod schema() → Type[AinswerPluginParams][source]#

Return the configuration parameters relevant for class.

Returns:: Configuration parameters relevant for class.

property toolset: AinswerToolset#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

validate() → None[source]#: Perform advanced validation on the parameters provided.

class enrobie.plugins.AinswerPluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, database: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)] = 'sqlite:///:memory:', histories: ~typing.Annotated[int, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=1000)] = 10, memories: ~typing.Annotated[int, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=1000)] = 10, clients: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], ainswer: ~enrobie.plugins.ainswer.params.AinswerPluginAinswerParams, prompt: ~enrobie.plugins.ainswer.params.AinswerPluginPromptParams = <factory>, plugins: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, logger: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, status: ~enrobie.plugins.status.params.StatusPluginIconParams = <factory>)[source]#

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

ainswer: Annotated[AinswerPluginAinswerParams, FieldInfo(annotation=NoneType, required=True, description='Parameters for the AI platforms')]#

clients: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, description='List of clients to enable plugin', metadata=[MinLen(min_length=1)])]#

database: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='sqlite:///:memory:', description='Database connection string', metadata=[MinLen(min_length=1)])]#

histories: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=10, description='Number of messages per anchor', metadata=[Ge(ge=1), Le(le=1000)])]#

logger: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Logger for including recents', metadata=[MinLen(min_length=1)])]#

memories: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=10, description='Number of messages per person', metadata=[Ge(ge=1), Le(le=1000)])]#

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

plugins: Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='List of plugins to load tools', metadata=[MinLen(min_length=1)])]#

prompt: Annotated[AinswerPluginPromptParams, FieldInfo(annotation=NoneType, required=False, default_factory=AinswerPluginPromptParams, description='Override the agent system prompt')]#

status: Annotated[StatusPluginIconParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconParams, description='Icon used per chat platform')]#

class enrobie.plugins.AutoJoinPlugin(robie: Robie, name: str, params: RobieChildParams)[source]#

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin maintains joined for configured channels.

operate() → None[source]#: Perform the operation related to Robie service threads.

property params: AutoJoinPluginParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

classmethod schema() → Type[AutoJoinPluginParams][source]#

Return the configuration parameters relevant for class.

Returns:: Configuration parameters relevant for class.

validate() → None[source]#: Perform advanced validation on the parameters provided.

class enrobie.plugins.AutoJoinPluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, channels: ~typing.Annotated[list[~enrobie.plugins.autojoin.params.AutoJoinPluginChannelParams], ~annotated_types.MinLen(min_length=1)], interval: ~typing.Annotated[int, ~annotated_types.Ge(ge=5), ~annotated_types.Le(le=300)] = 5, status: ~enrobie.plugins.status.params.StatusPluginIconParams = <factory>)[source]#

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

channels: Annotated[list[AutoJoinPluginChannelParams], FieldInfo(annotation=NoneType, required=True, description='Which channels to maintain join', metadata=[MinLen(min_length=1)])]#

interval: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=5, description='Interval when channels are joined', metadata=[Ge(ge=5), Le(le=300)])]#

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

status: Annotated[StatusPluginIconParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconParams, description='Icon used per chat platform')]#

class enrobie.plugins.AutoNickPlugin(robie: Robie, name: str, params: RobieChildParams)[source]#

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin maintains configured nickname on server.

operate() → None[source]#: Perform the operation related to Robie service threads.

property params: AutoNickPluginParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

classmethod schema() → Type[AutoNickPluginParams][source]#

Return the configuration parameters relevant for class.

Returns:: Configuration parameters relevant for class.

validate() → None[source]#: Perform advanced validation on the parameters provided.

class enrobie.plugins.AutoNickPluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, clients: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], interval: ~typing.Annotated[int, ~annotated_types.Ge(ge=5), ~annotated_types.Le(le=300)] = 5, services: list[~enrobie.plugins.autonick.params.AutoNickPluginServiceParams] | None = None, status: ~enrobie.plugins.status.params.StatusPluginIconParams = <factory>)[source]#

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

clients: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, description='List of clients to enable plugin', metadata=[MinLen(min_length=1)])]#

interval: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=5, description='Interval when nick is validated', metadata=[Ge(ge=5), Le(le=300)])]#

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

services: Annotated[list[AutoNickPluginServiceParams] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='How to identify with services')]#

status: Annotated[StatusPluginIconParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconParams, description='Icon used per chat platform')]#

class enrobie.plugins.HomiePlugin(robie: Robie, name: str, params: RobieChildParams)[source]#

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin allows for connecting to Homie Automate.

ainswer() → list[Callable[[...], Any]][source]#

Return the Ainswer tools that are related to the plugin.

Returns:: Ainswer tools that are related to the plugin.

operate() → None[source]#

Perform the operation related to Robie service threads.

Parameters:: thread – Child class instance for Chatting Robie.

property params: HomiePluginParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

property persist: HomiePersist#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

classmethod schema() → Type[HomiePluginParams][source]#

Return the configuration parameters relevant for class.

Returns:: Configuration parameters relevant for class.

validate() → None[source]#: Perform advanced validation on the parameters provided.

class enrobie.plugins.HomiePluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, command: ~enrobie.plugins.enhomie.params.HomiePluginCommandParams = <factory>, restful: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], username: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, password: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, timeout: ~typing.Annotated[int, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=300)] = 30, ssl_verify: bool = True, ssl_capem: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, clients: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], status: ~enrobie.plugins.status.params.StatusPluginIconParams = <factory>)[source]#

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

clients: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, description='List of clients to enable plugin', metadata=[MinLen(min_length=1)])]#

command: Annotated[HomiePluginCommandParams, FieldInfo(annotation=NoneType, required=False, default_factory=HomiePluginCommandParams, description='Command name per chat platform')]#

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

password: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Authenticate with the service', metadata=[MinLen(min_length=1)])]#

restful: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Where to find the RESTful API', examples=['http://localhost:8420'], metadata=[MinLen(min_length=1)])]#

ssl_capem: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Verify the ceritifcate valid', metadata=[MinLen(min_length=1)])]#

ssl_verify: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Verify the ceritifcate valid')]#

status: Annotated[StatusPluginIconParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconParams, description='Icon used per chat platform')]#

timeout: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=30, description='Timeout connecting to server', metadata=[Ge(ge=1), Le(le=300)])]#

username: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Authenticate with the service', metadata=[MinLen(min_length=1)])]#

class enrobie.plugins.LoggerPlugin(robie: Robie, name: str, params: RobieChildParams)[source]#

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin allows for interacting with an LLM model.

property history: LoggerHistory#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

operate() → None[source]#: Perform the operation related to Robie service threads.

property params: LoggerPluginParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

classmethod schema() → Type[LoggerPluginParams][source]#

Return the configuration parameters relevant for class.

Returns:: Configuration parameters relevant for class.

validate() → None[source]#: Perform advanced validation on the parameters provided.

class enrobie.plugins.LoggerPluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, database: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)] = 'sqlite:///:memory:', histories: ~typing.Annotated[int, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=10000)] = 100, clients: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], output: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=4)] = None, status: ~enrobie.plugins.status.params.StatusPluginIconParams = <factory>)[source]#

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

clients: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, description='List of clients to enable plugin', metadata=[MinLen(min_length=1)])]#

database: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='sqlite:///:memory:', description='Database connection string', metadata=[MinLen(min_length=1)])]#

histories: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=100, description='Number of messages per anchor', metadata=[Ge(ge=1), Le(le=10000)])]#

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

output: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Optional path where logs append', metadata=[MinLen(min_length=4)])]#

status: Annotated[StatusPluginIconParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconParams, description='Icon used per chat platform')]#

class enrobie.plugins.NagiosPlugin(robie: Robie, name: str, params: RobieChildParams)[source]#

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin allows for connecting to Nagios Console.

ainswer() → list[Callable[[...], Any]][source]#

Return the Ainswer tools that are related to the plugin.

Returns:: Ainswer tools that are related to the plugin.

property current: NagiosCurrent#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

operate() → None[source]#

Perform the operation related to Robie service threads.

Parameters:: thread – Child class instance for Chatting Robie.

property params: NagiosPluginParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

classmethod schema() → Type[NagiosPluginParams][source]#

Return the configuration parameters relevant for class.

Returns:: Configuration parameters relevant for class.

validate() → None[source]#: Perform advanced validation on the parameters provided.

class enrobie.plugins.NagiosPluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, command: ~enrobie.plugins.nagios.params.NagiosPluginCommandParams = <factory>, restful: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)], username: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, password: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, timeout: ~typing.Annotated[int, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=300)] = 30, ssl_verify: bool = True, ssl_capem: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, clients: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], status: ~enrobie.plugins.status.params.StatusPluginIconParams = <factory>)[source]#

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

clients: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, description='List of clients to enable plugin', metadata=[MinLen(min_length=1)])]#

command: Annotated[NagiosPluginCommandParams, FieldInfo(annotation=NoneType, required=False, default_factory=NagiosPluginCommandParams, description='Command name per chat platform')]#

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

password: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Authenticate with the service', metadata=[MinLen(min_length=1)])]#

restful: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Where to find the RESTful API', examples=['http://localhost:8420'], metadata=[MinLen(min_length=1)])]#

ssl_capem: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Verify the ceritifcate valid', metadata=[MinLen(min_length=1)])]#

ssl_verify: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Verify the ceritifcate valid')]#

status: Annotated[StatusPluginIconParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconParams, description='Icon used per chat platform')]#

timeout: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=30, description='Timeout connecting to server', metadata=[Ge(ge=1), Le(le=300)])]#

username: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Authenticate with the service', metadata=[MinLen(min_length=1)])]#

class enrobie.plugins.StatusPlugin(robie: Robie, name: str, params: RobieChildParams)[source]#

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin responds to inquiries about Robie status.

operate() → None[source]#: Perform the operation related to Robie service threads.

property params: StatusPluginParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

report() → None[source]#: Perform the operation related to Robie service threads.

classmethod schema() → Type[StatusPluginParams][source]#

Return the configuration parameters relevant for class.

Returns:: Configuration parameters relevant for class.

property status: StatusPluginItems#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

update(unique: str, group: str, title: str, icon: StatusPluginIconParams, state: Literal['pending', 'normal', 'failure', 'unknown']) → None[source]#

Update or insert the status of the Robie child instance.

Parameters:

unique – Unique identifier to use for the status.
group – Name for the group the status is membered.
title – Friendly name of the related unique entry.
icon – Optional icon object if supported platform.
state – One of several possible value for status.

validate() → None[source]#: Perform advanced validation on the parameters provided.

class enrobie.plugins.StatusPluginIconParams(*, irc: Annotated[str | None, MinLen(min_length=1)] = None, dsc: Annotated[str | None, MinLen(min_length=1)] = None, mtm: Annotated[str | None, MinLen(min_length=1)] = None)[source]#

Bases: RobieParamsModel

Contain information for constructing the chat messages.

dsc: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Icon used for the chat platform', metadata=[MinLen(min_length=1)])]#

irc: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Icon used for the chat platform', metadata=[MinLen(min_length=1)])]#

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

mtm: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Icon used for the chat platform', metadata=[MinLen(min_length=1)])]#

class enrobie.plugins.StatusPluginItem(time: Time, unique: str, group: str, title: str, icon: StatusPluginIconParams, state: Literal['pending', 'normal', 'failure', 'unknown'])[source]#

Bases: object

Contain the relevant status information for the entry.

group: str#

icon: StatusPluginIconParams#

state: Literal['pending', 'normal', 'failure', 'unknown']#

time: Time#

title: str#

unique: str#

class enrobie.plugins.StatusPluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, reports: ~typing.Annotated[list[~enrobie.plugins.status.params.StatusPluginReportParams] | None, ~annotated_types.MinLen(min_length=1)] = None, command: ~enrobie.plugins.status.params.StatusPluginCommandParams = <factory>, icons: ~enrobie.plugins.status.params.StatusPluginIconsParams = <factory>)[source]#

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

command: Annotated[StatusPluginCommandParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginCommandParams, description='Command name per chat platform')]#

icons: Annotated[StatusPluginIconsParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconsParams, description='Icon used per the chat platform')]#

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

reports: Annotated[list[StatusPluginReportParams] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Where to send status updates', metadata=[MinLen(min_length=1)])]#

enrobie.plugins package#

Subpackages#

Module contents#

This Page