thundra package

Subpackages

• thundra.core package
  • Submodules
  • thundra.core.graph module
    • Graph
      • Graph.combine_graph()
      • Graph.get_all_names()
      • Graph.graph()
  • thundra.core.llm module
    • LLM
      • LLM.chat_models
      • LLM.llm_available
      • LLM.available
      • LLM.chat_models
      • LLM.llm
      • LLM.remove_llm()
  • thundra.core.memory module
    • AIMemory
      • AIMemory.clear_history()
      • AIMemory.delete_memory()
      • AIMemory.get_memory()
    • UserMemory
      • UserMemory.clear_history()
      • UserMemory.create_ai_instance()
      • UserMemory.get_memory()
      • UserMemory.k
      • UserMemory.memory
      • UserMemory.system_message
    • build_system_message()
  • Module contents
    • Graph
      • Graph.combine_graph()
      • Graph.get_all_names()
      • Graph.graph()
• thundra.profiler package
  • Module contents
    • Env
      • Env.env
      • Env.workspace
    • EnvNotExist
    • Profile
      • Profile.db_exist()
      • Profile.db_path()
      • Profile.duplicate_count
      • Profile.get_id()
      • Profile.get_name()
      • Profile.phonenumber
      • Profile.pushname
      • Profile.workspace
    • ProfileNotExist
    • Profiler
      • Profiler.add_profile()
      • Profiler.delete_profile()
      • Profiler.dir_path
      • Profiler.get_profile()
      • Profiler.get_profiler()
      • Profiler.path
      • Profiler.save()
    • VirtualEnv
      • VirtualEnv.activate()
      • VirtualEnv.get_env()
      • VirtualEnv.save()
• thundra.storage package
  • Submodules
  • thundra.storage.file module
    • File
      • File.url
      • File.mimetype
      • File.fileSha256
      • File.fileLength
      • File.mediaKey
      • File.fileEncSha256
      • File.directPath
      • File.mediaKeyTimestamp
      • File.type
      • File.viewOnce
      • File.caption
      • File.height
      • File.width
      • File.seconds
      • File.ptt
      • File.pageCount
      • File.isAnimated
      • File.URL
      • File.caption
      • File.directPath
      • File.download()
      • File.fileEncSHA256
      • File.fileLength
      • File.fileSHA256
      • File.from_message()
      • File.height
      • File.isAnimated
      • File.mediaKey
      • File.mediaKeyTimestamp
      • File.mimetype
      • File.pageCount
      • File.ptt
      • File.seconds
      • File.type
      • File.viewOnce
      • File.width
    • FileRegistry
      • FileRegistry.update()
    • StorageRegistry
      • StorageRegistry.max_files
      • StorageRegistry.get_file()
      • StorageRegistry.get_files()
      • StorageRegistry.get_files_by_type()
      • StorageRegistry.max_files
      • StorageRegistry.save()
  • Module contents
    • File
      • File.url
      • File.mimetype
      • File.fileSha256
      • File.fileLength
      • File.mediaKey
      • File.fileEncSha256
      • File.directPath
      • File.mediaKeyTimestamp
      • File.type
      • File.viewOnce
      • File.caption
      • File.height
      • File.width
      • File.seconds
      • File.ptt
      • File.pageCount
      • File.isAnimated
      • File.URL
      • File.caption
      • File.directPath
      • File.download()
      • File.fileEncSHA256
      • File.fileLength
      • File.fileSHA256
      • File.from_message()
      • File.height
      • File.isAnimated
      • File.mediaKey
      • File.mediaKeyTimestamp
      • File.mimetype
      • File.pageCount
      • File.ptt
      • File.seconds
      • File.type
      • File.viewOnce
      • File.width
    • FileRegistry
      • FileRegistry.update()

Submodules

thundra.agents module

class thundra.agents.Agent(message_types, agent)

Bases: object

A class representing an agent with specific message types and a callable agent function.

message_types

The message types that the agent can handle.

Type:

Sequence[Type[MessageType]]

agent

The function to be executed by the agent.

Type:

Callable[[NewClient, Message], Any]

agent: Callable[[NewClient, Message], Any]

message_types: Sequence[Type[Union[ImageMessage, AudioMessage, VideoMessage, StickerMessage, DocumentMessage, ExtendedTextMessage, str]]]

class thundra.agents.AgentRegister(iterable=(), /)

Bases: list[Agent], Graph

A registry for agents, allowing filtering and management of agents.

get_all_names()

Generate the names of all agents in the register.

filter_tools(message_type

Type[MessageType]): Generate agents that can handle the given message type.

tool(*message_types

Type[MessageType]): Decorator to register an agent with specific message types.

filter_tools(message_type)

Generate agents that can handle the given message type.

Parameters:

message_type (Type[MessageType]) – The type of message to filter agents by.

Yield:

Agents that can handle the given message type.

Return type:

Generator[Agent, None, None]

get_all_names()

Generate the names of all agents in the register.

Yield:

The name of each agent.

Return type:

Generator[str, None, None]

tool(*message_types)

Decorator to register an agent with specific message types.

Parameters:

message_types (Type[MessageType]) – The message types the agent can handle.

Returns:

A decorator to register the agent function.

Return type:

Callable[[Callable[[NewClient, Message], Callable[[str], str]]], None]

thundra.chain module

thundra.chain.execute_agent(memory, client, message)

Execute an agent based on the incoming message.

This function initializes and executes an agent based on the provided message and client.

Parameters:

• memory (Any) – Memory object for the agent execution.

• client (NewClient) – Client object for communication.

• message (Message) – Incoming message to process.

Returns:

AgentExecutor object for the executed agent.

Return type:

AgentExecutor

thundra.cli module

thundra.cli.main()

thundra.command module

class thundra.command.Command(command, prefix=None)

Bases: Filter

filter(client, message)

Checks whether the provided message starts with the command this instance represents.

Parameters:

• client (NewClient) – The client that received the message.

• message (Message) – The message to check.

Returns:

True if the message starts with the command, False otherwise.

Return type:

bool

class thundra.command.CommandFunc(name, filter, description, func, category, allow_broadcast, on_error=None)

Bases: object

A class representing a command function with its associated metadata.

name

The name of the command.

Type:

str

filter

The filter to determine when this command should be executed.

Type:

Filter | FilterOP

description

A description of the command.

Type:

str

func

The function to execute for this command.

Type:

Callable[[NewClient, Message]]

category

Categories this command belongs to.

Type:

Sequence[str]

allow_broadcast

Whether this command can be used in broadcast messages.

Type:

bool

on_error

Error handler for the command.

Type:

Optional[Union[str, Callable[[NewClient, Message, Exception], None]]]

allow_broadcast: bool

category: Sequence[str]

description: str

filter: Filter | FilterOP

func: Callable[[NewClient, Message]]

name: str

on_error: Optional[Union[str, Callable[[NewClient, Message, Exception], None]]] = None

class thundra.command.Filter

Bases: ABC

abstract filter(client, message)

Return type:

bool

invert = False

class thundra.command.FilterOP(left, op, right, invert=False)

Bases: Filter

filter(client, message)

Return type:

bool

left: Filter

op: OP

right: Filter

class thundra.command.GlobalCommand

Bases: dict[str, CommandFunc], Graph

A class representing a global command registry that integrates with a graph. Commands are stored in a dictionary with their names as keys, and can be executed based on incoming messages.

start_point

The starting point for generating command names.

Type:

int

add(command)

Add a command to the registry.

Parameters:

command (CommandFunc) – The command function to be added.

execute(client, message)

Execute the appropriate command based on the given message.

Parameters:

• client (NewClient) – The client instance to use for executing the command.

• message (Message) – The message to process and execute the command.

Raises:

e – If an error occurs during command execution and no error handler is provided.

Returns:

True if a command is executed successfully, otherwise False.

Return type:

bool

classmethod generate_name(start_point)

Generate a unique name based on the start point.

Parameters:

start_point (int) – The starting point for generating the name.

Returns:

The generated unique name.

Return type:

str

get_all_names()

Generate the names of all commands in the registry.

Yield:

The name of each command.

Return type:

Generator[str, None, None]

register(filter, name='', description='', category=['all'], allow_broadcast=False, on_error=None)

Register a new command with the provided parameters.

Parameters:

• filter (Filterable) – The filter to apply to messages for this command.

• name (str, optional) – The name of the command, defaults to the function name if not provided.

• description (str, optional) – A description of the command, defaults to an empty string.

• category (Sequence[str], optional) – The categories this command belongs to, defaults to [“all”].

• allow_broadcast (bool, optional) – Whether this command can be used in broadcast messages, defaults to False.

• on_error (Optional[Union[str, Callable[[NewClient, Message, Exception], None]]], optional) – An error handler for this command, defaults to None.

Returns:

A decorator that registers the command.

Return type:

Callable[[Callable[[NewClient, Message], Any]], Callable[[NewClient, Message], Any]]

start_point: int = 1

class thundra.command.MessageType(*types)

Bases: Filter

filter(client, message)

Filter messages based on their types.

Parameters:

• client (NewClient) – The client object.

• message (Message) – The message object.

Returns:

True if the message type matches any of the specified types, False otherwise.

Return type:

bool

class thundra.command.OP(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

AND = '__and__'

OR = '__or__'

class thundra.command.Owner

Bases: Filter

filter(client, message)

Filter messages based on whether the sender is the owner.

Parameters:

• client (NewClient) – The client object.

• message (Message) – The message object.

Returns:

True if the sender is the owner, False otherwise.

Return type:

bool

thundra.config module

class thundra.config.Config(prefix, owner)

Bases: object

owner: List[str]

prefix: str

class thundra.config.Workdir(db, config_path, workspace_dir=None, config=None)

Bases: object

config: Optional[Config] = None

config_format(config=None, path='')

Recursively formats a configuration dictionary into a flat dictionary with keys representing the hierarchical structure of the original dictionary.

Parameters:

• config (Optional[Dict], optional) – The configuration dictionary to format. If None, the method uses self.read_config. Defaults to None.

• path (str, optional) – The base path for nested keys in the formatted dictionary. Defaults to an empty string.

Returns:

A flat dictionary with keys representing the hierarchical structure of the original dictionary.

Return type:

dict

config_path: Path

db: Path

property db_workspace: Path

Returns the path to the database workspace directory.

Returns:

The path to the database workspace directory.

Return type:

Path

classmethod find(path)

Finds the working directory by searching for specific configuration files (‘thundra-dev.toml’ or ‘thundra.toml’) in the provided path and its parent directories.

This method traverses up the directory tree starting from the provided path, looking for either ‘thundra-dev.toml’ or ‘thundra.toml’ configuration files. If one of these files is found, a Workdir instance is created with the path of the directory containing the configuration file.

Parameters:

path (Path) – The starting path from which to search for the configuration file.

Raises:

TypeError – If no configuration file is found in the directory tree.

Returns:

A Workdir instance with the found configuration file path.

Return type:

Workdir

get_config()

Reads the configuration from a TOML file and returns it as a Config object. The result is cached to improve performance on subsequent calls.

Returns:

A Config object containing the configuration data.

Return type:

Config

property read_config: dict

Reads the configuration from a TOML file and returns it as a dictionary.

Returns:

The configuration dictionary read from the TOML file.

Return type:

dict

property workspace: Path

Returns the workspace directory if it is set; otherwise, returns the database workspace directory.

Returns:

The path to the workspace directory.

Return type:

Path

workspace_dir: Optional[Path] = None

write_config_toml(config)

Writes the provided configuration dictionary to a TOML file.

Parameters:

config (dict) – The configuration dictionary to write to the file.

thundra.evaluater module

thundra.evaluater.evaluate_module(root_path, workspace=None)

Recursively imports Python modules found within the specified root directory.

Parameters:

• root_path (Path) – The root directory from which to start searching for Python modules.

• workspace (Optional[Path], optional) – The optional workspace directory. If provided, module paths will be relative to this directory. Defaults to None.

thundra.middleware module

class thundra.middleware.Middleware

Bases: ABC

name: Optional[str] = None

abstract run(client, message)

Return type:

None | bool

stop: bool = True

class thundra.middleware.MiddlewareRegister(iterable=(), /)

Bases: list[Middleware], Graph

add(middleware)

This method adds a middleware to the middleware register.

Parameters:

middleware (Middleware instance or Middleware class (Type[Middleware])) – The middleware instance or class to be added. If a class is provided, it will be instantiated before being added.

execute(client, message)

This method executes each middleware in the current instance with the provided client and message. If a middleware’s run method returns True and its stop attribute is set to True, the method will return that middleware and no further middlewares or global commands will be executed. If the run method returns False, the global command will run after the Middleware. If the run method returns None, it will continue to the next middleware.

Parameters:

• client (NewClient) – The client that the middleware will be executed with.

• message (Message) – The message that the middleware will process.

Returns:

The middleware that stopped the execution, if any. If no middleware stopped the execution, returns None.

Return type:

None | Middleware

get_all_names()

This method is a generator that yields the name of each middleware in the current instance.

Yield:

The name of each middleware in the current instance, as a string.

Return type:

Generator[str, None, None]

thundra.plugins module

class thundra.plugins.Plugin(name, author, sha, dependencies, version, includes, path)

Bases: object

author: str

dependencies: List[str]

classmethod find_by_author(author)

Finds plugins by author.

Parameters:

author (str) – The author of the plugins.

Returns:

An iterable of Plugin objects.

Return type:

Iterable[Self]

Yield:

Each found Plugin object.

Return type:

Iterator[Iterable[Self]]

classmethod find_by_author_and_name(author, name)

Finds plugins by author and name.

Parameters:

• author (str) – The author of the plugins.

• name (str) – The name of the plugins.

Returns:

An iterable of Plugin objects.

Return type:

Iterable[Self]

classmethod find_full_args(author, name, sha)

Finds a plugin with the specified author, name, and SHA hash.

Parameters:

• author (str) – The author of the plugin.

• name (str) – The name of the plugin.

• sha (str) – The SHA hash of the plugin.

Returns:

A Plugin object.

Return type:

Self

classmethod get_all_plugins()

Retrieves all plugins.

Yield:

Each found Plugin object.

Return type:

_type_

includes: List[PluginDirType]

name: str

path: Path

property repo_name: str

Retrieves the name of the plugin’s repository.

Returns:

The name of the repository.

Return type:

str

sha: str

stringify()

Converts the plugin information into a string.

Returns:

A string representation of the plugin.

Return type:

str

uninstall()

Uninstalls the plugin by removing its directory and updating the configuration file.

version: str

class thundra.plugins.PluginDirType(type, dirs)

Bases: object

dirs: List[str]

type: Union[Literal['command', 'agent', 'middleware'], str]

class thundra.plugins.PluginSource(username, repository)

Bases: Session

branch()

Retrieves a list of branches in the repository.

Returns:

A list of branches.

Return type:

List

download_from_sha(sha, branch)

Downloads a plugin from a specific commit SHA on a branch.

Parameters:

• sha (str) – The commit SHA.

• branch (str) – The branch name.

Returns:

A PluginZip object representing the downloaded plugin.

Return type:

PluginZip

download_head(branch)

Downloads the latest version of the repository from the specified branch.

Parameters:

branch (str) – The branch name.

Raises:

Response404 – If the branch does not exist.

Returns:

A PluginZip object representing the downloaded plugin.

Return type:

PluginZip

get_last_commit_sha(branch)

Retrieves the SHA hash of the last commit on the specified branch.

Parameters:

branch (str) – The branch name.

Returns:

The SHA hash of the last commit.

Return type:

str

latest_version()

Retrieves information about the latest version of the repository.

Raises:

Response404 – If the repository does not exist or has no releases.

Returns:

Information about the latest version.

Return type:

dict

class thundra.plugins.PluginZip(fp, username, repos, branch, sha)

Bases: object

get_info()

Retrieves information about the plugin from its ZIP file.

Returns:

Information about the plugin.

Return type:

dict

install()

Installs the plugin by installing dependencies and extracting its contents.

install_dependencies()

Installs the plugin’s dependencies using pip.

install_extract()

Extracts the contents of the plugin ZIP file.

exception thundra.plugins.Response404

Bases: Exception

Raised when a 404 response is encountered.

thundra.test module

thundra.test.ptree(start, tree, indent_width=4)

Prints a tree structure starting from the specified node.

Parameters:

• start (Any) – The starting node of the tree.

• tree (Dict[Any, Any]) – The tree structure represented as a dictionary.

• indent_width (int, optional) – The width of each indentation level, defaults to 4.

thundra.test.tree_test()

Test function to generate and print a tree structure.

This function creates a tree structure representing different types of items and then prints the tree.

thundra.types module

thundra.utils module

class thundra.utils.ChainMessage(message, neonize_message=None)

Bases: object

Represents a chain message.

Parameters:

• message (MessageProto) – The original message.

• neonize_message (Optional[Message], optional) – Optional neonize message.

classmethod extract_text(message)

Extracts text from the message.

Parameters:

message (MessageProto) – The message.

Returns:

The extracted text.

Return type:

str

message: Message

neonize_message: Optional[Message] = None

property text: str

Extracts the text from the message.

Returns:

The extracted text.

Return type:

str

to_json()

Converts the chain message to a JSON representation.

Returns:

JSON representation of the chain message.

Return type:

Dict[str, Dict[str, Any]]

property type: str

Returns the type of the message.

Returns:

The type of the message.

Return type:

str

class thundra.utils.MediaTypeToMMS(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

MediaAppState = 'md-app-state'

MediaAudio = 'audio'

MediaDocument = 'document'

MediaHistory = 'md-msg-hist'

MediaImage = 'image'

MediaLinkThumbnail = 'thumbnail-link'

MediaVideo = 'video'

thundra.utils.convert_size(size_bytes)

Converts a size in bytes to a human-readable string representation.

Parameters:

size_bytes (int) – The size in bytes to convert.

Returns:

The human-readable representation of the size.

Return type:

str

thundra.utils.download_media(client, message, types)

Downloads media from a message if it matches the specified types.

Parameters:

• client (NewClient) – The client used for downloading.

• message (MessageProto) – The message containing the media.

• types (Iterable[Type[MessageType]] | Type[MessageType]) – The types of media to download.

Returns:

The downloaded media, or None if no media of the specified types is found.

Return type:

bytes | None

thundra.utils.get_message_type(message)

Gets the message type.

Parameters:

message (MessageProto) – The message.

Raises:

IndexError – If the message type cannot be determined.

Returns:

The message type.

Return type:

MessageType

thundra.utils.get_tag(message)

Gets the tag from the message.

Parameters:

message (MessageProto) – The message.

Returns:

The tag from the message.

Return type:

List[str]

thundra.utils.get_user_id(message)

Gets the user ID from the message.

Parameters:

message (Message) – The message.

Returns:

The user ID from the message.

Return type:

str

thundra.utils.hidder(parent)

_summary_

Parameters:

parent (dict) – _description_

Returns:

_description_

Return type:

_type_

thundra.workdir module

Module contents