Phase 12: Additional plugin examples (File Tools and System Utils)

plugins/file_tools_plugin.py

@@ -0,0 +1,441 @@
+"""
+File tools plugin for MaxClaw Agent
+Provides additional file management tools
+"""
+import os
+import sys
+import shutil
+from typing import List, Dict, Any
+from pathlib import Path
+from langchain.tools import Tool
+
+# Add parent directory to path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from plugins.base import BasePlugin
+
+
+class FileToolsPlugin(BasePlugin):
+    """Plugin providing additional file management tools"""
+    
+    def __init__(self):
+        """Initialize file tools plugin"""
+        super().__init__(
+            name="file_tools",
+            version="1.0.0",
+            description="Additional file management tools for MaxClaw Agent"
+        )
+    
+    def get_tools(self) -> List[Tool]:
+        """Get tools provided by this plugin"""
+        return [
+            Tool(
+                name="file_find",
+                func=self.find_files,
+                description="Find files by name pattern"
+            ),
+            Tool(
+                name="file_size",
+                func=self.get_file_size,
+                description="Get file or directory size"
+            ),
+            Tool(
+                name="file_copy",
+                func=self.copy_file,
+                description="Copy a file to a new location"
+            ),
+            Tool(
+                name="file_move",
+                func=self.move_file,
+                description="Move a file to a new location"
+            ),
+            Tool(
+                name="file_delete",
+                func=self.delete_file,
+                description="Delete a file or directory"
+            ),
+            Tool(
+                name="file_info",
+                func=self.get_file_info,
+                description="Get detailed file information"
+            ),
+            Tool(
+                name="dir_list",
+                func=self.list_directory,
+                description="List directory contents with details"
+            ),
+            Tool(
+                name="dir_create",
+                func=self.create_directory,
+                description="Create a new directory"
+            )
+        ]
+    
+    def get_commands(self) -> Dict[str, callable]:
+        """Get commands provided by this plugin"""
+        return {
+            "/find": self.find_command,
+            "/size": self.size_command,
+            "/copy": self.copy_command,
+            "/move": self.move_command,
+            "/delete": self.delete_command,
+            "/info": self.info_command,
+            "/dir": self.dir_command,
+            "/mkdir": self.mkdir_command
+        }
+    
+    def find_files(self, pattern: str, directory: str = ".") -> str:
+        """Find files by name pattern"""
+        try:
+            path = Path(directory)
+            if not path.exists():
+                return f"❌ Directory not found: {directory}"
+            
+            matches = list(path.rglob(pattern))
+            
+            if not matches:
+                return f"No files found matching '{pattern}' in {directory}"
+            
+            output = f"🔍 Found {len(matches)} files matching '{pattern}':\n"
+            output += "=" * 50 + "\n"
+            
+            for match in matches[:20]:  # Limit to 20 results
+                if match.is_file():
+                    size = match.stat().st_size
+                    if size > 1024 * 1024:
+                        size_str = f"{size / (1024 * 1024):.1f} MB"
+                    elif size > 1024:
+                        size_str = f"{size / 1024:.1f} KB"
+                    else:
+                        size_str = f"{size} B"
+                    
+                    output += f"📄 {match.relative_to(path)} ({size_str})\n"
+            
+            if len(matches) > 20:
+                output += f"\n... and {len(matches) - 20} more files"
+            
+            return output
+            
+        except Exception as e:
+            return f"❌ Error finding files: {str(e)}"
+    
+    def get_file_size(self, path: str) -> str:
+        """Get file or directory size"""
+        try:
+            path_obj = Path(path)
+            
+            if not path_obj.exists():
+                return f"❌ Path not found: {path}"
+            
+            if path_obj.is_file():
+                size = path_obj.stat().st_size
+                if size > 1024 * 1024 * 1024:
+                    size_str = f"{size / (1024 * 1024 * 1024):.2f} GB"
+                elif size > 1024 * 1024:
+                    size_str = f"{size / (1024 * 1024):.2f} MB"
+                elif size > 1024:
+                    size_str = f"{size / 1024:.2f} KB"
+                else:
+                    size_str = f"{size} bytes"
+                
+                return f"📄 {path}: {size_str}"
+            
+            elif path_obj.is_dir():
+                total_size = 0
+                file_count = 0
+                
+                for item in path_obj.rglob('*'):
+                    if item.is_file():
+                        total_size += item.stat().st_size
+                        file_count += 1
+                
+                if total_size > 1024 * 1024 * 1024:
+                    size_str = f"{total_size / (1024 * 1024 * 1024):.2f} GB"
+                elif total_size > 1024 * 1024:
+                    size_str = f"{total_size / (1024 * 1024):.2f} MB"
+                elif total_size > 1024:
+                    size_str = f"{total_size / 1024:.2f} KB"
+                else:
+                    size_str = f"{total_size} bytes"
+                
+                return f"📁 {path}: {size_str} ({file_count} files)"
+            
+            else:
+                return f"❌ Unknown path type: {path}"
+                
+        except Exception as e:
+            return f"❌ Error getting size: {str(e)}"
+    
+    def copy_file(self, source: str, destination: str) -> str:
+        """Copy a file to a new location"""
+        try:
+            source_path = Path(source)
+            dest_path = Path(destination)
+            
+            if not source_path.exists():
+                return f"❌ Source not found: {source}"
+            
+            # If destination is a directory, copy into it
+            if dest_path.is_dir():
+                dest_path = dest_path / source_path.name
+            
+            # Create parent directories if needed
+            dest_path.parent.mkdir(parents=True, exist_ok=True)
+            
+            if source_path.is_file():
+                shutil.copy2(source_path, dest_path)
+                return f"✅ File copied: {source} → {destination}"
+            elif source_path.is_dir():
+                shutil.copytree(source_path, dest_path)
+                return f"✅ Directory copied: {source} → {destination}"
+            else:
+                return f"❌ Unknown source type: {source}"
+                
+        except Exception as e:
+            return f"❌ Error copying: {str(e)}"
+    
+    def move_file(self, source: str, destination: str) -> str:
+        """Move a file to a new location"""
+        try:
+            source_path = Path(source)
+            dest_path = Path(destination)
+            
+            if not source_path.exists():
+                return f"❌ Source not found: {source}"
+            
+            # If destination is a directory, move into it
+            if dest_path.is_dir():
+                dest_path = dest_path / source_path.name
+            
+            # Create parent directories if needed
+            dest_path.parent.mkdir(parents=True, exist_ok=True)
+            
+            shutil.move(str(source_path), str(dest_path))
+            return f"✅ Moved: {source} → {destination}"
+                
+        except Exception as e:
+            return f"❌ Error moving: {str(e)}"
+    
+    def delete_file(self, path: str, confirm: bool = True) -> str:
+        """Delete a file or directory"""
+        try:
+            path_obj = Path(path)
+            
+            if not path_obj.exists():
+                return f"❌ Path not found: {path}"
+            
+            if confirm:
+                return f"⚠️  To delete {path}, use the command with confirm=True"
+            
+            if path_obj.is_file():
+                path_obj.unlink()
+                return f"✅ File deleted: {path}"
+            elif path_obj.is_dir():
+                shutil.rmtree(path_obj)
+                return f"✅ Directory deleted: {path}"
+            else:
+                return f"❌ Unknown path type: {path}"
+                
+        except Exception as e:
+            return f"❌ Error deleting: {str(e)}"
+    
+    def get_file_info(self, path: str) -> str:
+        """Get detailed file information"""
+        try:
+            path_obj = Path(path)
+            
+            if not path_obj.exists():
+                return f"❌ Path not found: {path}"
+            
+            stat = path_obj.stat()
+            
+            # Format size
+            size = stat.st_size
+            if size > 1024 * 1024 * 1024:
+                size_str = f"{size / (1024 * 1024 * 1024):.2f} GB"
+            elif size > 1024 * 1024:
+                size_str = f"{size / (1024 * 1024):.2f} MB"
+            elif size > 1024:
+                size_str = f"{size / 1024:.2f} KB"
+            else:
+                size_str = f"{size} bytes"
+            
+            # Format dates
+            from datetime import datetime
+            created = datetime.fromtimestamp(stat.st_ctime).strftime('%Y-%m-%d %H:%M:%S')
+            modified = datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S')
+            accessed = datetime.fromtimestamp(stat.st_atime).strftime('%Y-%m-%d %H:%M:%S')
+            
+            # File type
+            if path_obj.is_file():
+                file_type = "File"
+            elif path_obj.is_dir():
+                file_type = "Directory"
+            elif path_obj.is_symlink():
+                file_type = "Symbolic Link"
+            else:
+                file_type = "Unknown"
+            
+            output = f"📋 File Information: {path}\n"
+            output += "=" * 50 + "\n"
+            output += f"Type: {file_type}\n"
+            output += f"Size: {size_str}\n"
+            output += f"Created: {created}\n"
+            output += f"Modified: {modified}\n"
+            output += f"Accessed: {accessed}\n"
+            output += f"Permissions: {oct(stat.st_mode)[-3:]}\n"
+            
+            if path_obj.is_file():
+                output += f"Extension: {path_obj.suffix}\n"
+                output += f"Name: {path_obj.name}\n"
+                output += f"Parent: {path_obj.parent}\n"
+            
+            return output
+            
+        except Exception as e:
+            return f"❌ Error getting file info: {str(e)}"
+    
+    def list_directory(self, path: str = ".", show_hidden: bool = False) -> str:
+        """List directory contents with details"""
+        try:
+            path_obj = Path(path)
+            
+            if not path_obj.exists():
+                return f"❌ Directory not found: {path}"
+            
+            if not path_obj.is_dir():
+                return f"❌ Not a directory: {path}"
+            
+            items = list(path_obj.iterdir())
+            
+            if not show_hidden:
+                items = [item for item in items if not item.name.startswith('.')]
+            
+            if not items:
+                return f"📁 Directory is empty: {path}"
+            
+            output = f"📁 Directory: {path}\n"
+            output += "=" * 50 + "\n"
+            
+            # Sort items: directories first, then files
+            dirs = [item for item in items if item.is_dir()]
+            files = [item for item in items if item.is_file()]
+            
+            # List directories
+            for dir_item in sorted(dirs, key=lambda x: x.name.lower()):
+                try:
+                    size = sum(f.stat().st_size for f in dir_item.rglob('*') if f.is_file())
+                    if size > 1024 * 1024:
+                        size_str = f"{size / (1024 * 1024):.1f} MB"
+                    elif size > 1024:
+                        size_str = f"{size / 1024:.1f} KB"
+                    else:
+                        size_str = f"{size} B"
+                    
+                    output += f"📁 {dir_item.name}/ ({size_str})\n"
+                except:
+                    output += f"📁 {dir_item.name}/\n"
+            
+            # List files
+            for file_item in sorted(files, key=lambda x: x.name.lower()):
+                try:
+                    size = file_item.stat().st_size
+                    if size > 1024 * 1024:
+                        size_str = f"{size / (1024 * 1024):.1f} MB"
+                    elif size > 1024:
+                        size_str = f"{size / 1024:.1f} KB"
+                    else:
+                        size_str = f"{size} B"
+                    
+                    output += f"📄 {file_item.name} ({size_str})\n"
+                except:
+                    output += f"📄 {file_item.name}\n"
+            
+            output += f"\nTotal: {len(dirs)} directories, {len(files)} files"
+            
+            return output
+            
+        except Exception as e:
+            return f"❌ Error listing directory: {str(e)}"
+    
+    def create_directory(self, path: str, parents: bool = True) -> str:
+        """Create a new directory"""
+        try:
+            path_obj = Path(path)
+            
+            if path_obj.exists():
+                return f"⚠️  Path already exists: {path}"
+            
+            path_obj.mkdir(parents=parents, exist_ok=True)
+            return f"✅ Directory created: {path}"
+            
+        except Exception as e:
+            return f"❌ Error creating directory: {str(e)}"
+    
+    # Command implementations
+    def find_command(self, args: List[str] = None) -> str:
+        """Find command implementation"""
+        if not args:
+            return "❌ Usage: /find <pattern> [directory]"
+        
+        pattern = args[0]
+        directory = args[1] if len(args) > 1 else "."
+        
+        return self.find_files(pattern, directory)
+    
+    def size_command(self, args: List[str] = None) -> str:
+        """Size command implementation"""
+        if not args:
+            return "❌ Usage: /size <path>"
+        
+        return self.get_file_size(args[0])
+    
+    def copy_command(self, args: List[str] = None) -> str:
+        """Copy command implementation"""
+        if not args or len(args) < 2:
+            return "❌ Usage: /copy <source> <destination>"
+        
+        return self.copy_file(args[0], args[1])
+    
+    def move_command(self, args: List[str] = None) -> str:
+        """Move command implementation"""
+        if not args or len(args) < 2:
+            return "❌ Usage: /move <source> <destination>"
+        
+        return self.move_file(args[0], args[1])
+    
+    def delete_command(self, args: List[str] = None) -> str:
+        """Delete command implementation"""
+        if not args:
+            return "❌ Usage: /delete <path>"
+        
+        return self.delete_file(args[0], confirm=False)
+    
+    def info_command(self, args: List[str] = None) -> str:
+        """Info command implementation"""
+        if not args:
+            return "❌ Usage: /info <path>"
+        
+        return self.get_file_info(args[0])
+    
+    def dir_command(self, args: List[str] = None) -> str:
+        """Dir command implementation"""
+        path = args[0] if args else "."
+        show_hidden = "-a" in args if args else False
+        
+        return self.list_directory(path, show_hidden)
+    
+    def mkdir_command(self, args: List[str] = None) -> str:
+        """Mkdir command implementation"""
+        if not args:
+            return "❌ Usage: /mkdir <path>"
+        
+        return self.create_directory(args[0])
+    
+    def on_load(self):
+        """Called when plugin is loaded"""
+        print(f"📦 File tools plugin loaded: {self.name} v{self.version}")
+    
+    def on_unload(self):
+        """Called when plugin is unloaded"""
+        print(f"🗑️  File tools plugin unloaded: {self.name}")

plugins/system_utils_plugin.py

@@ -0,0 +1,421 @@
+"""
+System utilities plugin for MaxClaw Agent
+Provides system information and utility tools
+"""
+import os
+import sys
+import platform
+import socket
+import subprocess
+from typing import List, Dict, Any
+from datetime import datetime
+from langchain.tools import Tool
+
+# Add parent directory to path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from plugins.base import BasePlugin
+
+
+class SystemUtilsPlugin(BasePlugin):
+    """Plugin providing system utilities and information"""
+    
+    def __init__(self):
+        """Initialize system utilities plugin"""
+        super().__init__(
+            name="system_utils",
+            version="1.0.0",
+            description="System information and utility tools"
+        )
+    
+    def get_tools(self) -> List[Tool]:
+        """Get tools provided by this plugin"""
+        return [
+            Tool(
+                name="system_info",
+                func=self.get_system_info,
+                description="Get comprehensive system information"
+            ),
+            Tool(
+                name="system_uptime",
+                func=self.get_system_uptime,
+                description="Get system uptime"
+            ),
+            Tool(
+                name="system_ip",
+                func=self.get_ip_address,
+                description="Get system IP addresses"
+            ),
+            Tool(
+                name="system_hostname",
+                func=self.get_hostname,
+                description="Get system hostname"
+            ),
+            Tool(
+                name="system_platform",
+                func=self.get_platform_info,
+                description="Get platform information"
+            ),
+            Tool(
+                name="system_users",
+                func=self.get_logged_users,
+                description="Get logged in users"
+            ),
+            Tool(
+                name="system_processes",
+                func=self.get_running_processes,
+                description="Get running processes"
+            ),
+            Tool(
+                name="system_disk_usage",
+                func=self.get_disk_usage,
+                description="Get disk usage information"
+            )
+        ]
+    
+    def get_commands(self) -> Dict[str, callable]:
+        """Get commands provided by this plugin"""
+        return {
+            "/sysinfo": self.sysinfo_command,
+            "/uptime": self.uptime_command,
+            "/ip": self.ip_command,
+            "/hostname": self.hostname_command,
+            "/platform": self.platform_command,
+            "/users": self.users_command,
+            "/ps": self.ps_command,
+            "/df": self.df_command
+        }
+    
+    def get_system_info(self) -> str:
+        """Get comprehensive system information"""
+        try:
+            info = {
+                "System": platform.system(),
+                "Node Name": platform.node(),
+                "Release": platform.release(),
+                "Version": platform.version(),
+                "Machine": platform.machine(),
+                "Processor": platform.processor(),
+                "Python Version": platform.python_version(),
+                "Hostname": socket.gethostname(),
+                "IP Address": socket.gethostbyname(socket.gethostname()),
+                "Current User": os.getlogin(),
+                "Current Directory": os.getcwd(),
+                "Home Directory": os.path.expanduser("~"),
+                "Timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            }
+            
+            output = "💻 System Information:\n"
+            output += "=" * 50 + "\n"
+            
+            for key, value in info.items():
+                output += f"{key}: {value}\n"
+            
+            return output
+            
+        except Exception as e:
+            return f"❌ Error getting system info: {str(e)}"
+    
+    def get_system_uptime(self) -> str:
+        """Get system uptime"""
+        try:
+            if platform.system() == "Linux":
+                with open('/proc/uptime', 'r') as f:
+                    uptime_seconds = float(f.readline().split()[0])
+                
+                days = int(uptime_seconds // (24 * 3600))
+                hours = int((uptime_seconds % (24 * 3600)) // 3600)
+                minutes = int((uptime_seconds % 3600) // 60)
+                seconds = int(uptime_seconds % 60)
+                
+                if days > 0:
+                    return f"⏱️  System uptime: {days} days, {hours} hours, {minutes} minutes, {seconds} seconds"
+                elif hours > 0:
+                    return f"⏱️  System uptime: {hours} hours, {minutes} minutes, {seconds} seconds"
+                elif minutes > 0:
+                    return f"⏱️  System uptime: {minutes} minutes, {seconds} seconds"
+                else:
+                    return f"⏱️  System uptime: {seconds} seconds"
+            
+            elif platform.system() == "Darwin":  # macOS
+                result = subprocess.run(['uptime'], capture_output=True, text=True)
+                return f"⏱️  {result.stdout.strip()}"
+            
+            elif platform.system() == "Windows":
+                result = subprocess.run(['net', 'stats', 'srv'], capture_output=True, text=True)
+                # Parse Windows uptime from output
+                for line in result.stdout.split('\n'):
+                    if 'Statistics since' in line:
+                        return f"⏱️  {line.strip()}"
+                
+                return "⏱️  Uptime information not available"
+            
+            else:
+                return "⏱️  Uptime information not available for this platform"
+                
+        except Exception as e:
+            return f"❌ Error getting uptime: {str(e)}"
+    
+    def get_ip_address(self) -> str:
+        """Get system IP addresses"""
+        try:
+            hostname = socket.gethostname()
+            
+            # Get all IP addresses
+            ip_addresses = []
+            
+            # Get hostname IP
+            try:
+                host_ip = socket.gethostbyname(hostname)
+                ip_addresses.append(("Hostname", host_ip))
+            except:
+                pass
+            
+            # Get all network interfaces (if available)
+            try:
+                import netifaces
+                for interface in netifaces.interfaces():
+                    addrs = netifaces.ifaddresses(interface)
+                    if netifaces.AF_INET in addrs:
+                        for addr in addrs[netifaces.AF_INET]:
+                            ip_addresses.append((interface, addr['addr']))
+            except ImportError:
+                # netifaces not available, use alternative method
+                pass
+            
+            if not ip_addresses:
+                return "❌ No IP addresses found"
+            
+            output = "🌐 IP Addresses:\n"
+            output += "=" * 30 + "\n"
+            
+            for interface, ip in ip_addresses:
+                output += f"{interface}: {ip}\n"
+            
+            return output
+            
+        except Exception as e:
+            return f"❌ Error getting IP addresses: {str(e)}"
+    
+    def get_hostname(self) -> str:
+        """Get system hostname"""
+        try:
+            hostname = socket.gethostname()
+            fqdn = socket.getfqdn()
+            
+            output = "🏠 Hostname Information:\n"
+            output += "=" * 30 + "\n"
+            output += f"Hostname: {hostname}\n"
+            output += f"FQDN: {fqdn}\n"
+            
+            return output
+            
+        except Exception as e:
+            return f"❌ Error getting hostname: {str(e)}"
+    
+    def get_platform_info(self) -> str:
+        """Get platform information"""
+        try:
+            info = {
+                "System": platform.system(),
+                "Node": platform.node(),
+                "Release": platform.release(),
+                "Version": platform.version(),
+                "Machine": platform.machine(),
+                "Processor": platform.processor(),
+                "Python Build": platform.python_build(),
+                "Python Compiler": platform.python_compiler(),
+                "Python Implementation": platform.python_implementation(),
+                "Python Version": platform.python_version()
+            }
+            
+            output = "🖥️  Platform Information:\n"
+            output += "=" * 50 + "\n"
+            
+            for key, value in info.items():
+                output += f"{key}: {value}\n"
+            
+            return output
+            
+        except Exception as e:
+            return f"❌ Error getting platform info: {str(e)}"
+    
+    def get_logged_users(self) -> str:
+        """Get logged in users"""
+        try:
+            if platform.system() in ["Linux", "Darwin"]:
+                result = subprocess.run(['who'], capture_output=True, text=True)
+                users = result.stdout.strip()
+                
+                if not users:
+                    return "👥 No users currently logged in"
+                
+                output = "👥 Logged in users:\n"
+                output += "=" * 30 + "\n"
+                output += users
+                
+                return output
+            
+            elif platform.system() == "Windows":
+                result = subprocess.run(['query', 'user'], capture_output=True, text=True)
+                users = result.stdout.strip()
+                
+                if not users:
+                    return "👥 No users currently logged in"
+                
+                output = "👥 Logged in users:\n"
+                output += "=" * 30 + "\n"
+                output += users
+                
+                return output
+            
+            else:
+                return "👥 User information not available for this platform"
+                
+        except Exception as e:
+            return f"❌ Error getting logged users: {str(e)}"
+    
+    def get_running_processes(self, limit: int = 20) -> str:
+        """Get running processes"""
+        try:
+            if platform.system() in ["Linux", "Darwin"]:
+                result = subprocess.run(['ps', 'aux'], capture_output=True, text=True)
+                processes = result.stdout.strip().split('\n')
+                
+                if len(processes) <= 1:
+                    return "🔄 No processes found"
+                
+                output = f"🔄 Running processes (top {limit}):\n"
+                output += "=" * 50 + "\n"
+                
+                # Header
+                output += processes[0] + "\n"
+                output += "-" * 50 + "\n"
+                
+                # Processes (limit to specified number)
+                for process in processes[1:limit+1]:
+                    output += process + "\n"
+                
+                if len(processes) > limit+1:
+                    output += f"\n... and {len(processes) - limit - 1} more processes"
+                
+                return output
+            
+            elif platform.system() == "Windows":
+                result = subprocess.run(['tasklist'], capture_output=True, text=True)
+                processes = result.stdout.strip().split('\n')
+                
+                if len(processes) <= 3:
+                    return "🔄 No processes found"
+                
+                output = f"🔄 Running processes (top {limit}):\n"
+                output += "=" * 50 + "\n"
+                
+                # Header (first 3 lines in Windows)
+                for i in range(3):
+                    if i < len(processes):
+                        output += processes[i] + "\n"
+                
+                output += "-" * 50 + "\n"
+                
+                # Processes (limit to specified number)
+                for process in processes[3:limit+3]:
+                    output += process + "\n"
+                
+                if len(processes) > limit+3:
+                    output += f"\n... and {len(processes) - limit - 3} more processes"
+                
+                return output
+            
+            else:
+                return "🔄 Process information not available for this platform"
+                
+        except Exception as e:
+            return f"❌ Error getting processes: {str(e)}"
+    
+    def get_disk_usage(self, path: str = "/") -> str:
+        """Get disk usage information"""
+        try:
+            if platform.system() == "Windows":
+                # For Windows, use different drive letters
+                if path == "/":
+                    path = "C:\\"
+            
+            import shutil
+            total, used, free = shutil.disk_usage(path)
+            
+            # Convert to human readable
+            def format_bytes(bytes):
+                for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
+                    if bytes < 1024.0:
+                        return f"{bytes:.2f} {unit}"
+                    bytes /= 1024.0
+                return f"{bytes:.2f} PB"
+            
+            total_str = format_bytes(total)
+            used_str = format_bytes(used)
+            free_str = format_bytes(free)
+            percent_used = (used / total) * 100
+            
+            output = f"💾 Disk Usage for {path}:\n"
+            output += "=" * 40 + "\n"
+            output += f"Total: {total_str}\n"
+            output += f"Used: {used_str} ({percent_used:.1f}%)\n"
+            output += f"Free: {free_str}\n"
+            
+            # Create a simple progress bar
+            bar_length = 30
+            filled_length = int(bar_length * percent_used // 100)
+            bar = '█' * filled_length + '░' * (bar_length - filled_length)
+            output += f"\n[{bar}] {percent_used:.1f}%\n"
+            
+            return output
+            
+        except Exception as e:
+            return f"❌ Error getting disk usage: {str(e)}"
+    
+    # Command implementations
+    def sysinfo_command(self, args: List[str] = None) -> str:
+        """Sysinfo command implementation"""
+        return self.get_system_info()
+    
+    def uptime_command(self, args: List[str] = None) -> str:
+        """Uptime command implementation"""
+        return self.get_system_uptime()
+    
+    def ip_command(self, args: List[str] = None) -> str:
+        """IP command implementation"""
+        return self.get_ip_address()
+    
+    def hostname_command(self, args: List[str] = None) -> str:
+        """Hostname command implementation"""
+        return self.get_hostname()
+    
+    def platform_command(self, args: List[str] = None) -> str:
+        """Platform command implementation"""
+        return self.get_platform_info()
+    
+    def users_command(self, args: List[str] = None) -> str:
+        """Users command implementation"""
+        return self.get_logged_users()
+    
+    def ps_command(self, args: List[str] = None) -> str:
+        """PS command implementation"""
+        limit = 20
+        if args and args[0].isdigit():
+            limit = int(args[0])
+        
+        return self.get_running_processes(limit)
+    
+    def df_command(self, args: List[str] = None) -> str:
+        """DF command implementation"""
+        path = args[0] if args else "/"
+        return self.get_disk_usage(path)
+    
+    def on_load(self):
+        """Called when plugin is loaded"""
+        print(f"📦 System utils plugin loaded: {self.name} v{self.version}")
+    
+    def on_unload(self):
+        """Called when plugin is unloaded"""
+        print(f"🗑️  System utils plugin unloaded: {self.name}")