resource_monitor.py

#!/usr/bin/env python3
"""
Claude Resource Monitor
Tracks actual memory, CPU, and network usage for Claude services
"""

import psutil
import subprocess
import time
import json
from datetime import datetime, timedelta
from pathlib import Path
import threading
import sys

class ClaudeResourceMonitor:
    def __init__(self):
        self.tracking_data = {
            'claude.ai': {
                'memory_mb': 0,
                'cpu_percent': 0,
                'network_mb': 0,
                'processes': []
            },
            'claude-code': {
                'memory_mb': 0,
                'cpu_percent': 0,
                'network_mb': 0,
                'processes': []
            },
            'system_total': {
                'memory_mb': 0,
                'cpu_percent': 0,
                'memory_available_mb': 0
            }
        }

        self.browser_processes = ['Google Chrome', 'Safari', 'Arc', 'Firefox', 'Microsoft Edge']
        self.claude_processes = ['claude', 'Claude', 'node']  # Node for potential Claude Code extensions

        # Network tracking
        self.last_network_io = psutil.net_io_counters()
        self.claude_network_usage = 0

        # Data file for persistence
        self.data_file = Path.home() / 'claude-usage-tracker' / 'resource_usage.json'
        self.load_historical_data()

    def load_historical_data(self):
        """Load historical resource usage data"""
        try:
            if self.data_file.exists():
                with open(self.data_file, 'r') as f:
                    self.historical_data = json.load(f)
            else:
                self.historical_data = {
                    'daily_stats': {},
                    'peak_usage': {
                        'memory_mb': 0,
                        'cpu_percent': 0,
                        'timestamp': None
                    }
                }
        except:
            self.historical_data = {'daily_stats': {}, 'peak_usage': {}}

    def save_data(self):
        """Save resource usage data"""
        try:
            self.data_file.parent.mkdir(exist_ok=True)
            with open(self.data_file, 'w') as f:
                json.dump(self.historical_data, f, indent=2, default=str)
        except Exception as e:
            print(f"Error saving data: {e}")

    def get_browser_memory_for_claude(self):
        """Get memory usage for browser tabs running Claude"""
        claude_memory = 0
        claude_cpu = 0

        try:
            # Check all browser processes
            for proc in psutil.process_iter(['pid', 'name', 'memory_info', 'cpu_percent']):
                try:
                    pname = proc.info['name']

                    # Check if it's a browser process
                    if any(browser in pname for browser in self.browser_processes):
                        # Get memory info
                        memory_mb = proc.info['memory_info'].rss / 1024 / 1024
                        cpu = proc.cpu_percent(interval=0.1)

                        # Check if this process is related to Claude
                        # This is approximate - we check window titles and process details
                        if self.is_claude_related_browser(proc):
                            claude_memory += memory_mb
                            claude_cpu += cpu

                            self.tracking_data['claude.ai']['processes'].append({
                                'name': pname,
                                'pid': proc.info['pid'],
                                'memory_mb': round(memory_mb, 2),
                                'cpu_percent': round(cpu, 2)
                            })
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue

        except Exception as e:
            print(f"Error checking browser memory: {e}")

        return claude_memory, claude_cpu

    def is_claude_related_browser(self, process):
        """Check if a browser process is Claude-related"""
        try:
            # Try to get window title using AppleScript
            pid = process.info['pid']
            script = f'''
            tell application "System Events"
                set windowList to every window of (processes whose unix id is {pid})
                if windowList is not {{}} then
                    return name of item 1 of windowList
                end if
            end tell
            '''

            result = subprocess.run(['osascript', '-e', script],
                                  capture_output=True, text=True, timeout=1)

            if result.returncode == 0:
                window_title = result.stdout.strip().lower()
                return 'claude' in window_title or 'anthropic' in window_title

        except:
            pass

        # Fallback: Check if browser is on Claude URL
        try:
            # Get active browser tab URL
            for browser in self.browser_processes:
                if browser in process.info['name']:
                    script = f'tell application "{browser}" to return URL of active tab of front window'
                    result = subprocess.run(['osascript', '-e', script],
                                          capture_output=True, text=True, timeout=1)
                    if result.returncode == 0:
                        url = result.stdout.strip().lower()
                        return 'claude.ai' in url or 'anthropic.com' in url
        except:
            pass

        return False

    def get_terminal_claude_memory(self):
        """Get memory usage for Claude Code and terminal processes"""
        claude_memory = 0
        claude_cpu = 0

        try:
            for proc in psutil.process_iter(['pid', 'name', 'cmdline', 'memory_info', 'cpu_percent']):
                try:
                    pname = proc.info['name'].lower()

                    # Check for Claude-related processes
                    if any(cp.lower() in pname for cp in self.claude_processes):
                        memory_mb = proc.info['memory_info'].rss / 1024 / 1024
                        cpu = proc.cpu_percent(interval=0.1)

                        claude_memory += memory_mb
                        claude_cpu += cpu

                        self.tracking_data['claude-code']['processes'].append({
                            'name': proc.info['name'],
                            'pid': proc.info['pid'],
                            'memory_mb': round(memory_mb, 2),
                            'cpu_percent': round(cpu, 2)
                        })

                    # Check command line for Claude references
                    elif proc.info['cmdline']:
                        cmdline = ' '.join(proc.info['cmdline']).lower()
                        if 'claude' in cmdline or 'anthropic' in cmdline:
                            memory_mb = proc.info['memory_info'].rss / 1024 / 1024
                            cpu = proc.cpu_percent(interval=0.1)

                            claude_memory += memory_mb
                            claude_cpu += cpu

                            self.tracking_data['claude-code']['processes'].append({
                                'name': proc.info['name'],
                                'pid': proc.info['pid'],
                                'memory_mb': round(memory_mb, 2),
                                'cpu_percent': round(cpu, 2)
                            })

                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue

        except Exception as e:
            print(f"Error checking terminal memory: {e}")

        return claude_memory, claude_cpu

    def estimate_network_usage(self):
        """Estimate network usage for Claude services"""
        try:
            current_io = psutil.net_io_counters()

            # Calculate bytes transferred since last check
            bytes_sent = current_io.bytes_sent - self.last_network_io.bytes_sent
            bytes_recv = current_io.bytes_recv - self.last_network_io.bytes_recv

            # Total MB transferred
            total_mb = (bytes_sent + bytes_recv) / 1024 / 1024

            # Estimate Claude's portion (this is approximate)
            # We check if Claude is actively being used
            if self.tracking_data['claude.ai']['processes'] or self.tracking_data['claude-code']['processes']:
                # Assume 30% of network traffic during active Claude usage is Claude-related
                self.claude_network_usage += total_mb * 0.3

            self.last_network_io = current_io

        except Exception as e:
            print(f"Error tracking network: {e}")

    def get_system_resources(self):
        """Get overall system resource usage"""
        try:
            # CPU usage
            cpu_percent = psutil.cpu_percent(interval=1)

            # Memory usage
            memory = psutil.virtual_memory()
            memory_used_mb = (memory.total - memory.available) / 1024 / 1024
            memory_available_mb = memory.available / 1024 / 1024

            self.tracking_data['system_total'] = {
                'memory_mb': round(memory_used_mb, 2),
                'memory_available_mb': round(memory_available_mb, 2),
                'cpu_percent': round(cpu_percent, 2),
                'memory_percent': round(memory.percent, 2)
            }

        except Exception as e:
            print(f"Error getting system resources: {e}")

    def update_tracking(self):
        """Update all tracking data"""
        # Clear previous process lists
        self.tracking_data['claude.ai']['processes'] = []
        self.tracking_data['claude-code']['processes'] = []

        # Get browser memory for Claude
        browser_memory, browser_cpu = self.get_browser_memory_for_claude()
        self.tracking_data['claude.ai']['memory_mb'] = round(browser_memory, 2)
        self.tracking_data['claude.ai']['cpu_percent'] = round(browser_cpu, 2)

        # Get terminal/Claude Code memory
        terminal_memory, terminal_cpu = self.get_terminal_claude_memory()
        self.tracking_data['claude-code']['memory_mb'] = round(terminal_memory, 2)
        self.tracking_data['claude-code']['cpu_percent'] = round(terminal_cpu, 2)

        # Get system resources
        self.get_system_resources()

        # Estimate network usage
        self.estimate_network_usage()
        self.tracking_data['claude.ai']['network_mb'] = round(self.claude_network_usage, 2)

        # Update peak usage
        total_claude_memory = browser_memory + terminal_memory
        total_claude_cpu = browser_cpu + terminal_cpu

        if total_claude_memory > self.historical_data['peak_usage'].get('memory_mb', 0):
            self.historical_data['peak_usage']['memory_mb'] = round(total_claude_memory, 2)
            self.historical_data['peak_usage']['timestamp'] = datetime.now().isoformat()

        if total_claude_cpu > self.historical_data['peak_usage'].get('cpu_percent', 0):
            self.historical_data['peak_usage']['cpu_percent'] = round(total_claude_cpu, 2)

        # Update daily stats
        today = datetime.now().strftime('%Y-%m-%d')
        if today not in self.historical_data['daily_stats']:
            self.historical_data['daily_stats'][today] = {
                'total_memory_mb_seconds': 0,
                'total_cpu_percent_seconds': 0,
                'samples': 0,
                'peak_memory_mb': 0,
                'peak_cpu_percent': 0
            }

        daily = self.historical_data['daily_stats'][today]
        daily['total_memory_mb_seconds'] += total_claude_memory
        daily['total_cpu_percent_seconds'] += total_claude_cpu
        daily['samples'] += 1
        daily['peak_memory_mb'] = max(daily['peak_memory_mb'], total_claude_memory)
        daily['peak_cpu_percent'] = max(daily['peak_cpu_percent'], total_claude_cpu)

        # Save data periodically
        if daily['samples'] % 60 == 0:  # Save every 60 samples
            self.save_data()

    def display_status(self):
        """Display current resource usage"""
        print("\n" + "="*60)
        print("📊 CLAUDE RESOURCE USAGE MONITOR")
        print("-"*60)
        print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print()

        # Claude.ai usage
        print("🌐 Claude.ai (Browser):")
        if self.tracking_data['claude.ai']['memory_mb'] > 0:
            print(f"   Memory: {self.tracking_data['claude.ai']['memory_mb']:.1f} MB")
            print(f"   CPU: {self.tracking_data['claude.ai']['cpu_percent']:.1f}%")
            print(f"   Network: {self.tracking_data['claude.ai']['network_mb']:.1f} MB total")
            if self.tracking_data['claude.ai']['processes']:
                print("   Processes:")
                for proc in self.tracking_data['claude.ai']['processes'][:3]:  # Show top 3
                    print(f"     • {proc['name']} (PID {proc['pid']}): "
                          f"{proc['memory_mb']:.1f} MB, {proc['cpu_percent']:.1f}% CPU")
        else:
            print("   Not active")

        print()

        # Claude Code usage
        print("💻 Claude Code (Terminal):")
        if self.tracking_data['claude-code']['memory_mb'] > 0:
            print(f"   Memory: {self.tracking_data['claude-code']['memory_mb']:.1f} MB")
            print(f"   CPU: {self.tracking_data['claude-code']['cpu_percent']:.1f}%")
            if self.tracking_data['claude-code']['processes']:
                print("   Processes:")
                for proc in self.tracking_data['claude-code']['processes'][:3]:
                    print(f"     • {proc['name']} (PID {proc['pid']}): "
                          f"{proc['memory_mb']:.1f} MB, {proc['cpu_percent']:.1f}% CPU")
        else:
            print("   Not active")

        print()

        # Total Claude usage
        total_memory = (self.tracking_data['claude.ai']['memory_mb'] +
                       self.tracking_data['claude-code']['memory_mb'])
        total_cpu = (self.tracking_data['claude.ai']['cpu_percent'] +
                    self.tracking_data['claude-code']['cpu_percent'])

        print("📈 Total Claude Usage:")
        print(f"   Memory: {total_memory:.1f} MB")
        print(f"   CPU: {total_cpu:.1f}%")

        print()

        # System resources
        print("💻 System Resources:")
        print(f"   Total Memory Used: {self.tracking_data['system_total']['memory_mb']:.1f} MB")
        print(f"   Memory Available: {self.tracking_data['system_total']['memory_available_mb']:.1f} MB")
        print(f"   Total CPU Usage: {self.tracking_data['system_total']['cpu_percent']:.1f}%")
        print(f"   Claude % of System: {(total_memory / self.tracking_data['system_total']['memory_mb'] * 100):.1f}% of memory")

        print()

        # Historical data
        if self.historical_data['peak_usage'].get('memory_mb'):
            print("📊 Peak Usage (All Time):")
            print(f"   Memory: {self.historical_data['peak_usage']['memory_mb']:.1f} MB")
            print(f"   CPU: {self.historical_data['peak_usage']['cpu_percent']:.1f}%")
            if self.historical_data['peak_usage'].get('timestamp'):
                print(f"   Recorded: {self.historical_data['peak_usage']['timestamp']}")

        today = datetime.now().strftime('%Y-%m-%d')
        if today in self.historical_data['daily_stats']:
            daily = self.historical_data['daily_stats'][today]
            if daily['samples'] > 0:
                avg_memory = daily['total_memory_mb_seconds'] / daily['samples']
                avg_cpu = daily['total_cpu_percent_seconds'] / daily['samples']
                print()
                print(f"📅 Today's Average:")
                print(f"   Memory: {avg_memory:.1f} MB")
                print(f"   CPU: {avg_cpu:.1f}%")
                print(f"   Peak Memory: {daily['peak_memory_mb']:.1f} MB")
                print(f"   Peak CPU: {daily['peak_cpu_percent']:.1f}%")

        print("="*60)

    def export_data(self, format='json'):
        """Export resource usage data"""
        export_data = {
            'current': self.tracking_data,
            'historical': self.historical_data,
            'timestamp': datetime.now().isoformat()
        }

        if format == 'json':
            export_file = Path.home() / 'claude-usage-tracker' / f'resource_export_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json'
            with open(export_file, 'w') as f:
                json.dump(export_data, f, indent=2, default=str)
            print(f"📁 Data exported to: {export_file}")
            return export_file

        elif format == 'csv':
            import csv
            export_file = Path.home() / 'claude-usage-tracker' / f'resource_export_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv'

            with open(export_file, 'w', newline='') as f:
                writer = csv.writer(f)
                writer.writerow(['Timestamp', 'Service', 'Memory (MB)', 'CPU (%)', 'Network (MB)'])

                timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                for service in ['claude.ai', 'claude-code']:
                    writer.writerow([
                        timestamp,
                        service,
                        self.tracking_data[service]['memory_mb'],
                        self.tracking_data[service]['cpu_percent'],
                        self.tracking_data[service].get('network_mb', 0)
                    ])

            print(f"📁 Data exported to: {export_file}")
            return export_file

    def run(self, interval=5):
        """Run the resource monitor"""
        print("🚀 Claude Resource Monitor Started")
        print(f"📊 Tracking memory, CPU, and network usage every {interval} seconds")
        print("Press Ctrl+C to stop\n")

        display_counter = 0

        try:
            while True:
                # Update tracking data
                self.update_tracking()

                # Display status every 30 seconds
                display_counter += interval
                if display_counter >= 30:
                    self.display_status()
                    display_counter = 0

                time.sleep(interval)

        except KeyboardInterrupt:
            print("\n⏹️ Stopping resource monitor...")
            self.save_data()
            self.display_status()

            # Export final data
            self.export_data('json')
            print("\n✅ Resource monitoring stopped")

if __name__ == "__main__":
    monitor = ClaudeResourceMonitor()
    monitor.run(interval=5)  # Check every 5 seconds