taskmon 0.1.1

Function-level CPU and Memory monitoring with visual terminal output

📊 taskmon - Function-Level Performance Monitoring

Simple, powerful Python function monitoring with beautiful terminal output.

Track CPU, memory, and performance of any Python function with a simple decorator. Perfect for finding memory leaks, bottlenecks, and understanding your code's resource usage.

✨ Features

• 🎯 Function-level tracking - Monitor any function with a simple @monitor() decorator
• 📊 Real-time metrics - CPU, memory, execution time, and throughput
• 🎨 Beautiful output - Color-coded, visual terminal display
• 🔍 Memory leak detection - Automatic detection of memory growth
• 📈 Nested function support - Track call hierarchies with visual indentation
• 🚀 Zero configuration - Works out of the box
• 📝 Aggregated statistics - See summaries across all function calls
• 🧵 Thread-safe - Works with multi-threaded applications

🚀 Quick Start

Installation

pip install taskmon

Or install from source:

git clone https://github.com/kadirtutenn/taskmon
cd taskmon
pip install -e .

Basic Usage

from taskmon import monitor

@monitor()
def process_data(items):
    result = []
    for item in items:
        result.append(item * 2)
    return result

# Call your function normally
data = process_data([1, 2, 3, 4, 5])

Output:

┌─ 🚀 process_data() started [call_1_1738574400123]
│  📊 Baseline: CPU 2.3% | MEM 156.2MB | THR 4
  └─ ✅ process_data() completed in 0.02s
     📊 CPU: 3.1% [███░░░░░░░] | MEM: 🟢 +0.8MB (peak: 157.0MB)
     ⚡ Processed: 5 items (250.0 items/s)

📚 Usage Examples

1. Monitor Any Function

from taskmon import monitor

@monitor()
def calculate_fibonacci(n):
    if n <= 1:
        return n
    return calculate_fibonacci(n-1) + calculate_fibonacci(n-2)

result = calculate_fibonacci(10)

2. Track Code Sections

from taskmon import monitor, track

@monitor()
def data_pipeline():
    with track("load_data"):
        data = load_from_database()
    
    with track("process"):
        processed = process_data(data)
    
    with track("save"):
        save_to_database(processed)

data_pipeline()

Output shows nested structure:

┌─ 🚀 data_pipeline() started
│  📊 Baseline: CPU 2.1% | MEM 145.3MB | THR 4
  ├─ 🚀 load_data started
  │  📊 Baseline: CPU 2.3% | MEM 145.5MB | THR 4
    └─ ✅ load_data completed in 1.23s
       📊 CPU: 15.2% [█████░░░░░] | MEM: 🟢 +12.3MB (peak: 157.8MB)
  
  ├─ 🚀 process started
  │  📊 Baseline: CPU 3.1% | MEM 157.8MB | THR 4
    └─ ✅ process completed in 2.45s
       📊 CPU: 45.6% [████████░░] | MEM: 🟡 +45.2MB (peak: 203.0MB)
  
  └─ ✅ data_pipeline() completed in 3.89s
     📊 CPU: 21.3% [██████░░░░] | MEM: 🟢 +9.1MB (peak: 203.0MB)

3. Track Items Processed

from taskmon import monitor

@monitor()
def process_batch(items):
    results = []
    for item in items:
        results.append(transform_item(item))
    return results

# Automatically tracks len(results)
items = ["item1", "item2", "item3"]
process_batch(items)

Output:

└─ ✅ process_batch() completed in 5.23s
   📊 CPU: 12.3% [████░░░░░░] | MEM: 🟢 +2.1MB (peak: 158.4MB)
   ⚡ Processed: 3 items (0.6 items/s)

4. View Statistics

from taskmon import monitor, print_summary

@monitor()
def expensive_function(n):
    time.sleep(n)
    return n * 2

# Call multiple times
for i in range(5):
    expensive_function(i)

# Print summary
print_summary()

Output:

================================================================================
📊 FUNCTION MONITORING SUMMARY
================================================================================

Function                                    Calls   Total Time   Avg Time   Failures
--------------------------------------------------------------------------------
__main__.expensive_function                     5       10.00s      2.000s      0 (0.0%)

================================================================================

5. Find Memory Leaks

from taskmon import monitor

@monitor()
def potential_leak():
    data = []
    for i in range(1000000):
        data.append(i)
    # Oops, forgot to clean up!
    return len(data)

potential_leak()

Output (notice the 🔴 red indicator):

└─ ✅ potential_leak() completed in 0.15s
   📊 CPU: 85.2% [█████████░] | MEM: 🔴 +156.3MB (peak: 312.5MB)
   ⚡ Processed: 1,000,000 items (6666666.7 items/s)

6. Real-World Example: Space Mission Data Pipeline

from taskmon import monitor, track

@monitor()
def plan_mission_data_collection():
    with track("Load Configuration"):
        satellites = get_satellite_config_cached()
    
    with track("Count Signals"):
        total_signals = get_total_signal_count()
    
    # Process in chunks
    chunk_size = 50000
    for offset in range(0, total_signals, chunk_size):
        with track(f"Process Chunk {offset//chunk_size + 1}"):
            process_signal_chunk(offset, chunk_size, satellites)

plan_mission_data_collection()

Output shows where memory leaks occur:

┌─ 🚀 plan_mission_data_collection() started
  ├─ 🚀 Load Configuration started
    └─ ✅ Load Configuration completed in 0.05s
       📊 CPU: 2.1% | MEM: 🟢 +1.2MB
  
  ├─ 🚀 Count Signals started
    └─ ✅ Count Signals completed in 0.23s
       📊 CPU: 5.3% | MEM: 🟢 +0.1MB
  
  ├─ 🚀 Process Chunk 1 started
    └─ ✅ Process Chunk 1 completed in 12.34s
       📊 CPU: 45.2% | MEM: 🟢 +2.3MB
  
  ├─ 🚀 Process Chunk 2 started
    └─ ✅ Process Chunk 2 completed in 11.89s
       📊 CPU: 43.1% | MEM: 🔴 +125.6MB  <-- Memory leak here!
  
  └─ ✅ plan_mission_data_collection() completed in 45.67s
     📊 CPU: 38.5% | MEM: 🔴 +128.4MB

🎨 Visual Indicators

Status Icons

• 🚀 Function started
• ✅ Successfully completed
• ❌ Failed with error
• ⚠️ Warning

Memory Status

• 🟢 Green: < 10MB change (normal)
• 🟡 Yellow: 10-100MB change (monitor)
• 🔴 Red: > 100MB change (memory leak!)

CPU Bars

CPU: 15.2% [█████░░░░░]  <- Visual bar
CPU: 45.6% [████████░░]
CPU: 85.2% [█████████░]

📊 API Reference

@monitor(items_attr=None, print_summary=False, logger=None, use_color=None)

Decorator to monitor function performance.

Parameters:

• items_attr (str, optional): Attribute name to track as items_processed
• print_summary (bool): Print summary after function completes
• logger (optional): Logger instance or callable (e.g. logging.Logger, Celery logger)
• use_color (bool, optional): Enable/disable ANSI colors for this function

Example:

@monitor(items_attr='count', print_summary=True)
def process_data():
    return {"count": 100, "data": [...]}

@monitor(logger=my_logger, use_color=False)
def task():
    pass

track(section_name, items=0, logger=None, use_color=None)

Context manager for monitoring code sections.

Parameters:

• section_name (str): Name of the section
• items (int): Number of items processed
• logger (optional): Logger instance or callable
• use_color (bool, optional): Enable/disable ANSI colors for this section

Example:

with track("database_query", items=100):
    results = db.query(...)

with track("cache_warmup", logger=my_logger, use_color=False):
    warmup()

get_stats(function_name=None)

Get function statistics.

Parameters:

• function_name (str, optional): Specific function name, or None for all

Returns: Dict with statistics

Example:

stats = get_stats("my_module.my_function")
print(f"Total calls: {stats['total_calls']}")
print(f"Avg CPU: {stats['avg_cpu']:.1f}%")

print_summary(logger=None)

Print summary of all monitored functions.

set_logger(logger, use_color=None)

Set default logger and color behavior for all outputs.

clear_stats()

Clear all statistics.

get_active_calls()

Get list of currently active function calls.

Returns: List of FunctionMetrics objects

🔧 Advanced Usage

Custom Items Tracking

@monitor()
def process_batch(items):
    count = 0
    for item in items:
        process(item)
        count += 1
    return count  # Automatically tracked

Nested Function Monitoring

@monitor()
def parent_function():
    child_function_1()
    child_function_2()

@monitor()
def child_function_1():
    pass

@monitor()
def child_function_2():
    pass

Output shows hierarchy:

┌─ 🚀 parent_function() started
  ├─ 🚀 child_function_1() started
    └─ ✅ child_function_1() completed
  ├─ 🚀 child_function_2() started
    └─ ✅ child_function_2() completed
  └─ ✅ parent_function() completed

Integration with Celery

from celery import shared_task
from celery.utils.log import get_task_logger
from taskmon import monitor

celery_logger = get_task_logger(__name__)

@shared_task
@monitor(logger=celery_logger, use_color=False)
def celery_task(data):
    process(data)

Logger Integration

import logging
from taskmon import set_logger, monitor, track

logger = logging.getLogger("myapp.taskmon")
set_logger(logger, use_color=False)

@monitor()
def pipeline():
    with track("step_1"):
        pass

Programmatic Access

from taskmon import get_active_calls, get_stats

# Get currently running functions
active = get_active_calls()
for call in active:
    print(f"{call.function_name}: {call.memory_delta_mb:.1f}MB")

# Get historical stats
stats = get_stats()
for func_name, metrics in stats.items():
    if metrics['avg_cpu'] > 50:
        print(f"High CPU function: {func_name}")

🎯 Use Cases

1. Find Memory Leaks

@monitor()
def suspect_function():
    # Your code
    pass

Look for 🔴 indicators showing large memory growth.

2. Identify Bottlenecks

@monitor()
def pipeline():
    with track("step1"):
        slow_step()
    with track("step2"):
        fast_step()

See which sections take the most time.

3. Optimize Batch Processing

@monitor()
def process_chunks():
    for chunk in chunks:
        with track(f"chunk_{i}"):
            process(chunk)

Identify which chunks are slow or leak memory.

4. Monitor Production Code

@monitor()
def critical_api_endpoint():
    # Monitor in production
    pass

Track performance metrics over time.

📈 Performance Overhead

taskmon is designed to be lightweight:

• CPU overhead: < 1% in most cases
• Memory overhead: < 1MB
• Sampling: Background thread samples every 1 second
• Thread-safe: Safe for multi-threaded applications

🐛 Troubleshooting

Output not showing colors?

Set your terminal to support ANSI colors:

export TERM=xterm-256color

Or disable colors:

from taskmon import set_logger
set_logger(my_logger, use_color=False)

Too much output?

Disable monitoring for specific functions:

# Just remove the decorator
def my_function():
    pass

Want quieter output?

The library respects standard output - redirect if needed:

import sys
sys.stdout = open('monitor.log', 'w')

🤝 Contributing

Contributions welcome! Please:

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Submit a pull request

📝 License

MIT License - see LICENSE file for details.

🙏 Acknowledgments

Built with:

• psutil - Process and system utilities

📞 Support

• 📧 Email: [e-mail]
• 🐛 Issues: GitHub Issues
• 💬 Discussions: GitHub Discussions

---

Made with ❤️ for Python developers who care about performance