Overview The PluginManager class is responsible for discovering, loading, and managing plugins in Sakuya AC. It provides a system for registering hooks, commands, and managing plugin lifecycle.
Constructor PluginManager(connected_players)
Creates a new PluginManager instance and automatically loads all enabled plugins.
Reference to the connected players collection for plugin access
Behavior:
Initializes empty plugin registry
Initializes hook system
Registers default ‘help’ command
Automatically calls load_plugins() to discover and load plugins
Properties Dictionary mapping plugin names to plugin instances { "plugin_name" : plugin_instance, "another_plugin" : plugin_instance2 }
Dictionary mapping hook names to lists of callback functions { "on_packet_received" : [callback1, callback2], "on_player_join" : [callback3] }
commands
dict
default: "{'help': None}"
Dictionary mapping command names to callback functions { "help" : None , # Built-in help command "teleport" : teleport_callback, "tp" : teleport_callback # Alias }
help_message
str
default: "'List of Available Commands:\\n'"
Auto-generated help text containing all registered commands and descriptions
Reference to connected players collection
Methods load_plugins Scans the plugins directory and loads all enabled plugins.
Behavior:
Scans ../plugins directory relative to library location
Supports both single-file plugins (.py) and package plugins (directories with __init__.py)
Only loads plugins with ENABLED = True attribute
Requires plugins to have a Plugin class
Calls register_plugin() for each loaded plugin
Logs successful plugin loads
Plugin Requirements:
Must have ENABLED = True module-level attribute
Must have a Plugin class
Plugin class must have a register(plugin_manager) method
Example plugin structure: # plugins/my_plugin.py ENABLED = True class Plugin : def register ( self , plugin_manager ): plugin_manager.register_command( "mycommand" , self .my_command) def my_command ( self , full_message , player , msg_to_client , msg_to_server ): msg_to_client.append( "Command executed!" )
register_plugin register_plugin(plugin) -> None
Registers a plugin instance with the plugin manager.
Plugin instance with a register(plugin_manager) method
Behavior:
Calls the plugin’s register() method, passing self as argument
Plugin’s register method should call register_hook() and/or register_command() to integrate with the system
Example: class MyPlugin : def register ( self , pm ): pm.register_hook( "on_damage" , self .on_damage_hook) pm.register_command( "heal" , self .heal_command, "Heals your aircraft" ) def on_damage_hook ( self , data , * args , ** kwargs ): print ( f "Damage detected: { data } " ) return True def heal_command ( self , msg , player , to_client , to_server ): to_client.append( "Aircraft healed!" ) plugin = MyPlugin() plugin_manager.register_plugin(plugin)
register_hook register_hook(hook_name: str , callback) -> None
Registers a callback function for a specific hook.
Name of the hook to register (e.g., “on_packet_received”, “on_player_join”)
Callback function to execute when hook is triggered. Should accept (data, *args, **kwargs) and return bool
Callback Signature: def callback ( data , * args , ** kwargs ) -> bool : # Return True to keep original data # Return False to block/cancel the event pass
Example: def on_message_hook ( data , player , * args , ** kwargs ): print ( f "Message from { player.username } : { data } " ) if "blocked_word" in data: return False # Block message return True # Allow message plugin_manager.register_hook( "on_message" , on_message_hook)
register_command register_command(command_name: str , callback, help_text: str = "No Description" , alias: str = None ) -> None
Registers a chat command with optional alias.
Primary command name (without leading slash)
Callback function to execute when command is triggered
help_text
str
default: "'No Description'"
Description text shown in help command
Optional alias for the command (e.g., “tp” for “teleport”)
Callback Signature: def callback ( full_message : str , player : Player, message_to_client : list , message_to_server : list ) -> None : pass
Behavior:
Logs warning if command name already registered (ignores duplicate)
Logs warning if alias already registered (ignores duplicate)
Automatically updates help_message with command info
Format in help: /command_name [alias] : help_text
Example: def teleport_cmd ( full_msg , player , to_client , to_server ): # Parse coordinates from full_msg parts = full_msg.split() if len (parts) == 4 : x, y, z = float (parts[ 1 ]), float (parts[ 2 ]), float (parts[ 3 ]) player.aircraft.set_position([x, y, z]) to_client.append( f "Teleported to { x } , { y } , { z } " ) else : to_client.append( "Usage: /teleport <x> <y> <z>" ) plugin_manager.register_command( "teleport" , teleport_cmd, "Teleport to coordinates: /teleport <x> <y> <z>" , alias = "tp" ) # Now both /teleport and /tp work # help_message will show: "/teleport [tp] : Teleport to coordinates: /teleport <x> <y> <z>"
triggar_hook triggar_hook(hook_name: str , data, * args, ** kwargs) -> bool
Note: Method name has typo “triggar” instead of “trigger” in source code
Triggers all callbacks registered for a specific hook.
Name of the hook to trigger
Data to pass to hook callbacks
Additional positional arguments to pass to callbacks
Additional keyword arguments to pass to callbacks
True if all callbacks returned True or non-boolean, False if any callback returned False
Behavior:
Executes all callbacks for the hook in registration order
If any callback returns False, sets return value to False
Logs warning if callback returns non-boolean value
Useful for packet filtering: return False to block packet forwarding
Example: # Register hooks def validate_packet ( packet , player ): if packet.is_valid(): return True print ( "Invalid packet blocked" ) return False def log_packet ( packet , player ): print ( f "Packet from { player.username } " ) return True plugin_manager.register_hook( "on_packet" , validate_packet) plugin_manager.register_hook( "on_packet" , log_packet) # Trigger hook packet = get_packet() should_forward = plugin_manager.triggar_hook( "on_packet" , packet, player = current_player) if should_forward: forward_packet(packet) else : print ( "Packet blocked by plugin" )
trigger_command trigger_command(command: str , player: Player, full_message: str , message_to_client: list , message_to_server: list ) -> bool
Executes a registered command if it exists.
Command name to execute (without leading slash)
Player instance who issued the command
Complete command message including command name and arguments
List to append response messages for the client
List to append messages to forward to server
True if command was found and executed, False if command not registered
Example: # Setup def heal_command ( full_msg , player , to_client , to_server ): player.aircraft.life = 1000 to_client.append( "Aircraft fully repaired!" ) plugin_manager.register_command( "heal" , heal_command, "Repairs your aircraft" ) # Usage message = "/heal" command_name = message[ 1 :] # Remove leading slash to_client = [] to_server = [] if plugin_manager.trigger_command(command_name, player, message, to_client, to_server): # Send messages in to_client back to player for msg in to_client: send_message(player, msg) else : send_message(player, "Unknown command. Type /help for commands." )
Usage Example from lib.plugin_manager import PluginManager # Initialize with connected players reference connected_players = {} plugin_manager = PluginManager(connected_players) # Plugins are auto-loaded from plugins/ directory print ( f "Loaded plugins: { list (plugin_manager.plugins.keys()) } " ) # Manually register a quick plugin class QuickPlugin : def register ( self , pm ): pm.register_hook( "on_damage" , self .damage_hook) pm.register_command( "god" , self .god_mode, "Toggle god mode" , alias = "g" ) def damage_hook ( self , data , player ): if hasattr (player, 'god_mode' ) and player.god_mode: return False # Block damage return True def god_mode ( self , msg , player , to_client , to_server ): player.god_mode = not getattr (player, 'god_mode' , False ) status = "enabled" if player.god_mode else "disabled" to_client.append( f "God mode { status } " ) quick_plugin = QuickPlugin() plugin_manager.register_plugin(quick_plugin) # Trigger hooks keep_packet = plugin_manager.triggar_hook( "on_damage" , damage_data, player = player_obj) # Execute commands msg = "/god" to_client = [] to_server = [] if plugin_manager.trigger_command( "god" , player_obj, msg, to_client, to_server): for response in to_client: print (response) # Display help print (plugin_manager.help_message)
Hook System The hook system allows plugins to intercept and modify server behavior:
Common Hook Patterns: # Packet filtering hook def filter_hook ( packet , player ): if should_block(packet): return False # Block packet return True # Allow packet # Logging hook def log_hook ( data , * args , ** kwargs ): log_to_file(data) return True # Always allow, just observe # Modification hook def modify_hook ( data , player ): # Modify data in place data.value = transform(data.value) return True # Allow modified data
Command System Command Best Practices: def my_command ( full_message , player , message_to_client , message_to_server ): # Parse arguments args = full_message.split()[ 1 :] # Skip command name # Validate arguments if len (args) < 2 : message_to_client.append( "Usage: /mycommand <arg1> <arg2>" ) return # Execute command logic result = do_something(args[ 0 ], args[ 1 ]) # Send response to client message_to_client.append( f "Command executed: { result } " ) # Optionally send data to server # message_to_server.append(some_packet)