# Imitator
[](https://badge.fury.io/py/imitator)
[](https://www.python.org/downloads/)
[](https://opensource.org/licenses/MIT)
A lightweight Python framework for monitoring and imitating function behavior with automatic I/O tracking and pattern learning. Perfect for collecting training data for machine learning models, debugging, performance analysis, and understanding function behavior in production systems with future capabilities for behavior imitation.
## âĻ Features
- ðŊ **Simple Decorator**: Just add `@monitor_function` to any function
- ð **Type Validation**: Uses Pydantic models for robust type handling
- ðū **Flexible Storage**: Local JSON/JSONL files with configurable backends
- ⥠**Performance Monitoring**: Tracks execution times and performance metrics
- ðĻ **Error Handling**: Captures and logs exceptions with full context
- ð **Async Support**: Full support for asynchronous functions
- ð **Sampling & Rate Limiting**: Control overhead with smart sampling
- ðïļ **Class Method Support**: Monitor class methods with proper handling
- ð **Modification Detection**: Detect in-place parameter modifications
- ðŠķ **Minimal Dependencies**: Only requires Pydantic (âĨ2.0.0)
## ð Installation
```bash
pip install imitator
```
**Requirements**: Python 3.8+, Pydantic âĨ2.0.0
## ⥠Quick Start
```python
from imitator import monitor_function
@monitor_function
def add_numbers(a: int, b: int) -> int:
return a + b
# Use the function normally
result = add_numbers(5, 3) # Automatically logged!
```
That's it! Your function calls are now being monitored and logged automatically.
## ð Usage Examples
### Basic Function Monitoring
```python
from imitator import monitor_function
from typing import List, Dict
@monitor_function
def process_data(data: List[float], multiplier: float = 1.0) -> Dict[str, float]:
"""Process a list of numbers and return statistics"""
if not data:
return {"mean": 0.0, "sum": 0.0, "count": 0}
total = sum(x * multiplier for x in data)
mean = total / len(data)
return {
"mean": mean,
"sum": total,
"count": len(data)
}
# Function calls are automatically logged
result = process_data([1.0, 2.0, 3.0], 2.0)
```
### Advanced Configuration
```python
from imitator import monitor_function, LocalStorage, FunctionMonitor
# Custom storage location
custom_storage = LocalStorage(log_dir="my_logs", format="json")
@monitor_function(storage=custom_storage)
def my_function(x: int) -> int:
return x * 2
# Rate limiting and sampling for high-frequency functions
monitor = FunctionMonitor(
sampling_rate=0.1, # Log 10% of calls
max_calls_per_minute=100 # Max 100 calls per minute
)
@monitor.monitor
def high_frequency_function(x: int) -> int:
return x ** 2
```
### Async Function Support
```python
import asyncio
from imitator import monitor_function
@monitor_function
async def fetch_data(url: str) -> dict:
"""Simulate async data fetching"""
await asyncio.sleep(0.1)
return {"data": f"Response from {url}"}
# Async functions work seamlessly
async def main():
result = await fetch_data("https://api.example.com")
print(result)
asyncio.run(main())
```
### Class Method Monitoring
```python
from imitator import monitor_function
class DataProcessor:
def __init__(self, name: str):
self.name = name
@monitor_function
def process_batch(self, items: List[dict]) -> dict:
"""Process a batch of items"""
processed = []
for item in items:
processed.append(self.process_item(item))
return {"processed": len(processed), "results": processed}
def process_item(self, item: dict) -> dict:
# Helper method (not monitored)
return {"id": item.get("id"), "processed_by": self.name}
processor = DataProcessor("BatchProcessor")
result = processor.process_batch([{"id": 1}, {"id": 2}])
```
### Examining Logged Data
```python
from imitator import LocalStorage
storage = LocalStorage()
# Get all monitored functions
functions = storage.get_all_functions()
print(f"Monitored functions: {functions}")
# Load calls for a specific function
calls = storage.load_calls("add_numbers")
for call in calls:
print(f"Input: {call.io_record.inputs}")
print(f"Output: {call.io_record.output}")
print(f"Execution time: {call.io_record.execution_time_ms}ms")
print(f"Timestamp: {call.io_record.timestamp}")
```
## ð Data Structure
The framework captures comprehensive information about each function call:
```python
class FunctionCall(BaseModel):
function_signature: FunctionSignature # Function name, parameters, return type
io_record: IORecord # Inputs, output, timestamp, execution time
call_id: str # Unique identifier
class IORecord(BaseModel):
inputs: Dict[str, Any] # Function input parameters
output: Any # Function return value
timestamp: str # ISO format timestamp
execution_time_ms: float # Execution time in milliseconds
input_modifications: Optional[Dict] # Detected in-place modifications
```
## ðū Storage Format
Logs are stored as JSON/JSONL files in the `logs/` directory:
```
logs/
âââ add_numbers_20241201.jsonl
âââ process_data_20241201.jsonl
âââ ...
```
Each log entry contains:
- **Function signature** with type annotations
- **Input parameters** with actual values
- **Output value** or exception details
- **Execution time** in milliseconds
- **Timestamp** in ISO format
- **Error information** (if applicable)
- **Input modifications** (if detected)
## ðĻ Error Handling
The framework gracefully handles and logs exceptions:
```python
@monitor_function
def divide_numbers(a: float, b: float) -> float:
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
try:
result = divide_numbers(10, 0)
except ValueError:
pass # Exception is logged with input parameters and full traceback
```
Exception logs include:
- **Input parameters** that caused the error
- **Exception type** and message
- **Full traceback** for debugging
- **Execution time** until the exception occurred
## ðïļ Framework Components
### Core Components
- **`monitor_function`**: Main decorator for function monitoring
- **`FunctionMonitor`**: Advanced monitoring with configuration options
- **`FunctionCall`**: Pydantic model for complete function call records
- **`IORecord`**: Pydantic model for input/output pairs
- **`LocalStorage`**: Local file-based storage backend
### Type System
The framework uses Pydantic for:
- **Runtime type validation** and serialization
- **Automatic schema generation** from function signatures
- **Type-safe data structures** with validation
- **JSON serialization** with complex type support
## ð Example Output
Running monitored functions generates structured logs like:
```json
{
"function_signature": {
"name": "add_numbers",
"parameters": {
"a": "<class 'int'>",
"b": "<class 'int'>"
},
"return_type": "<class 'int'>"
},
"io_record": {
"inputs": {
"a": 5,
"b": 3
},
"output": 8,
"timestamp": "2024-01-15T10:30:45.123456",
"execution_time_ms": 0.05,
"input_modifications": null
},
"call_id": "1705312245.123456"
}
```
## ðŊ Use Cases
### ðĪ Machine Learning
- **Training Data Collection**: Gather input-output pairs for model training
- **Model Inference Monitoring**: Track model performance and behavior
- **Feature Engineering**: Monitor data preprocessing pipelines
- **A/B Testing**: Compare different model versions
### ð§ Development & Debugging
- **Function Profiling**: Analyze performance bottlenecks
- **Debugging**: Track function calls and parameter values
- **Integration Testing**: Monitor system component interactions
- **Behavior Analysis**: Understand function usage patterns
### ð Production Monitoring
- **System Health**: Monitor critical business functions
- **Performance Tracking**: Track execution times and error rates
- **User Behavior**: Analyze how functions are used in production
- **Compliance**: Maintain audit trails for regulatory requirements
### ðŽ Research & Analysis
- **Algorithm Analysis**: Study algorithm behavior with real data
- **Performance Optimization**: Identify optimization opportunities
- **Data Quality**: Monitor data processing pipelines
- **Experimentation**: Support research and development workflows
## ð Examples
The package includes comprehensive examples demonstrating various use cases:
### Available Examples
- **`basic_usage.py`**: Getting started with core features
- **`advanced_monitoring.py`**: Advanced configuration and async support
- **`real_world_simulation.py`**: Practical applications and systems
### Run Examples
```bash
# Install the package
pip install imitator
# Clone repository for examples (if needed)
git clone https://github.com/yourusername/imitator.git
cd imitator/examples
# Run examples
python basic_usage.py
python advanced_monitoring.py
python real_world_simulation.py
```
Each example demonstrates:
- Different monitoring strategies
- Error handling scenarios
- Performance analysis
- Log inspection and analysis
## ð Getting Started
1. **Install**: `pip install imitator`
2. **Import**: `from imitator import monitor_function`
3. **Decorate**: Add `@monitor_function` to your functions
4. **Run**: Use your functions normally
5. **Analyze**: Check the generated logs in the `logs/` directory
## ð Documentation
- **Examples**: Comprehensive examples in the `examples/` directory
- **API Reference**: Detailed docstrings in all modules
- **Type Hints**: Full type annotation support
- **Error Handling**: Graceful handling of edge cases
## ðĪ Contributing
We welcome contributions! Please see our [Contributing Guidelines](CONTRIBUTING.md) for details.
## ð License
Apache License 2.0 - see [LICENSE](LICENSE) file for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "imitator",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "Imitator Team <contact@imitator.dev>",
"keywords": "monitoring, imitation, function, behavior, io, machine-learning, debugging, profiling, type-validation, decorator, pattern-learning",
"author": null,
"author_email": "Imitator Team <contact@imitator.dev>",
"download_url": "https://files.pythonhosted.org/packages/0f/ee/1a2e2fe0490ac8bfa021a65fd0f8d4c42c61d68087bd7127eb62648dc106/imitator-0.1.0.tar.gz",
"platform": null,
"description": "# Imitator\n\n[](https://badge.fury.io/py/imitator)\n[](https://www.python.org/downloads/)\n[](https://opensource.org/licenses/MIT)\n\nA lightweight Python framework for monitoring and imitating function behavior with automatic I/O tracking and pattern learning. Perfect for collecting training data for machine learning models, debugging, performance analysis, and understanding function behavior in production systems with future capabilities for behavior imitation.\n\n## \u2728 Features\n\n- \ud83c\udfaf **Simple Decorator**: Just add `@monitor_function` to any function\n- \ud83d\udcca **Type Validation**: Uses Pydantic models for robust type handling\n- \ud83d\udcbe **Flexible Storage**: Local JSON/JSONL files with configurable backends\n- \u26a1 **Performance Monitoring**: Tracks execution times and performance metrics\n- \ud83d\udea8 **Error Handling**: Captures and logs exceptions with full context\n- \ud83d\udd04 **Async Support**: Full support for asynchronous functions\n- \ud83d\udcc8 **Sampling & Rate Limiting**: Control overhead with smart sampling\n- \ud83c\udfd7\ufe0f **Class Method Support**: Monitor class methods with proper handling\n- \ud83d\udd0d **Modification Detection**: Detect in-place parameter modifications\n- \ud83e\udeb6 **Minimal Dependencies**: Only requires Pydantic (\u22652.0.0)\n\n## \ud83d\ude80 Installation\n\n```bash\npip install imitator\n```\n\n**Requirements**: Python 3.8+, Pydantic \u22652.0.0\n\n## \u26a1 Quick Start\n\n```python\nfrom imitator import monitor_function\n\n@monitor_function\ndef add_numbers(a: int, b: int) -> int:\n return a + b\n\n# Use the function normally\nresult = add_numbers(5, 3) # Automatically logged!\n```\n\nThat's it! Your function calls are now being monitored and logged automatically.\n\n## \ud83d\udcd6 Usage Examples\n\n### Basic Function Monitoring\n\n```python\nfrom imitator import monitor_function\nfrom typing import List, Dict\n\n@monitor_function\ndef process_data(data: List[float], multiplier: float = 1.0) -> Dict[str, float]:\n \"\"\"Process a list of numbers and return statistics\"\"\"\n if not data:\n return {\"mean\": 0.0, \"sum\": 0.0, \"count\": 0}\n \n total = sum(x * multiplier for x in data)\n mean = total / len(data)\n \n return {\n \"mean\": mean,\n \"sum\": total,\n \"count\": len(data)\n }\n\n# Function calls are automatically logged\nresult = process_data([1.0, 2.0, 3.0], 2.0)\n```\n\n### Advanced Configuration\n\n```python\nfrom imitator import monitor_function, LocalStorage, FunctionMonitor\n\n# Custom storage location\ncustom_storage = LocalStorage(log_dir=\"my_logs\", format=\"json\")\n\n@monitor_function(storage=custom_storage)\ndef my_function(x: int) -> int:\n return x * 2\n\n# Rate limiting and sampling for high-frequency functions\nmonitor = FunctionMonitor(\n sampling_rate=0.1, # Log 10% of calls\n max_calls_per_minute=100 # Max 100 calls per minute\n)\n\n@monitor.monitor\ndef high_frequency_function(x: int) -> int:\n return x ** 2\n```\n\n### Async Function Support\n\n```python\nimport asyncio\nfrom imitator import monitor_function\n\n@monitor_function\nasync def fetch_data(url: str) -> dict:\n \"\"\"Simulate async data fetching\"\"\"\n await asyncio.sleep(0.1)\n return {\"data\": f\"Response from {url}\"}\n\n# Async functions work seamlessly\nasync def main():\n result = await fetch_data(\"https://api.example.com\")\n print(result)\n\nasyncio.run(main())\n```\n\n### Class Method Monitoring\n\n```python\nfrom imitator import monitor_function\n\nclass DataProcessor:\n def __init__(self, name: str):\n self.name = name\n \n @monitor_function\n def process_batch(self, items: List[dict]) -> dict:\n \"\"\"Process a batch of items\"\"\"\n processed = []\n for item in items:\n processed.append(self.process_item(item))\n return {\"processed\": len(processed), \"results\": processed}\n \n def process_item(self, item: dict) -> dict:\n # Helper method (not monitored)\n return {\"id\": item.get(\"id\"), \"processed_by\": self.name}\n\nprocessor = DataProcessor(\"BatchProcessor\")\nresult = processor.process_batch([{\"id\": 1}, {\"id\": 2}])\n```\n\n### Examining Logged Data\n\n```python\nfrom imitator import LocalStorage\n\nstorage = LocalStorage()\n\n# Get all monitored functions\nfunctions = storage.get_all_functions()\nprint(f\"Monitored functions: {functions}\")\n\n# Load calls for a specific function\ncalls = storage.load_calls(\"add_numbers\")\n\nfor call in calls:\n print(f\"Input: {call.io_record.inputs}\")\n print(f\"Output: {call.io_record.output}\")\n print(f\"Execution time: {call.io_record.execution_time_ms}ms\")\n print(f\"Timestamp: {call.io_record.timestamp}\")\n```\n\n## \ud83d\udcca Data Structure\n\nThe framework captures comprehensive information about each function call:\n\n```python\nclass FunctionCall(BaseModel):\n function_signature: FunctionSignature # Function name, parameters, return type\n io_record: IORecord # Inputs, output, timestamp, execution time\n call_id: str # Unique identifier\n\nclass IORecord(BaseModel):\n inputs: Dict[str, Any] # Function input parameters\n output: Any # Function return value\n timestamp: str # ISO format timestamp\n execution_time_ms: float # Execution time in milliseconds\n input_modifications: Optional[Dict] # Detected in-place modifications\n```\n\n## \ud83d\udcbe Storage Format\n\nLogs are stored as JSON/JSONL files in the `logs/` directory:\n\n```\nlogs/\n\u251c\u2500\u2500 add_numbers_20241201.jsonl\n\u251c\u2500\u2500 process_data_20241201.jsonl\n\u2514\u2500\u2500 ...\n```\n\nEach log entry contains:\n- **Function signature** with type annotations\n- **Input parameters** with actual values\n- **Output value** or exception details\n- **Execution time** in milliseconds\n- **Timestamp** in ISO format\n- **Error information** (if applicable)\n- **Input modifications** (if detected)\n\n## \ud83d\udea8 Error Handling\n\nThe framework gracefully handles and logs exceptions:\n\n```python\n@monitor_function\ndef divide_numbers(a: float, b: float) -> float:\n if b == 0:\n raise ValueError(\"Cannot divide by zero\")\n return a / b\n\ntry:\n result = divide_numbers(10, 0)\nexcept ValueError:\n pass # Exception is logged with input parameters and full traceback\n```\n\nException logs include:\n- **Input parameters** that caused the error\n- **Exception type** and message\n- **Full traceback** for debugging\n- **Execution time** until the exception occurred\n\n## \ud83c\udfd7\ufe0f Framework Components\n\n### Core Components\n\n- **`monitor_function`**: Main decorator for function monitoring\n- **`FunctionMonitor`**: Advanced monitoring with configuration options\n- **`FunctionCall`**: Pydantic model for complete function call records\n- **`IORecord`**: Pydantic model for input/output pairs\n- **`LocalStorage`**: Local file-based storage backend\n\n### Type System\n\nThe framework uses Pydantic for:\n- **Runtime type validation** and serialization\n- **Automatic schema generation** from function signatures\n- **Type-safe data structures** with validation\n- **JSON serialization** with complex type support\n\n## \ud83d\udccb Example Output\n\nRunning monitored functions generates structured logs like:\n\n```json\n{\n \"function_signature\": {\n \"name\": \"add_numbers\",\n \"parameters\": {\n \"a\": \"<class 'int'>\",\n \"b\": \"<class 'int'>\"\n },\n \"return_type\": \"<class 'int'>\"\n },\n \"io_record\": {\n \"inputs\": {\n \"a\": 5,\n \"b\": 3\n },\n \"output\": 8,\n \"timestamp\": \"2024-01-15T10:30:45.123456\",\n \"execution_time_ms\": 0.05,\n \"input_modifications\": null\n },\n \"call_id\": \"1705312245.123456\"\n}\n```\n\n## \ud83c\udfaf Use Cases\n\n### \ud83e\udd16 Machine Learning\n- **Training Data Collection**: Gather input-output pairs for model training\n- **Model Inference Monitoring**: Track model performance and behavior\n- **Feature Engineering**: Monitor data preprocessing pipelines\n- **A/B Testing**: Compare different model versions\n\n### \ud83d\udd27 Development & Debugging\n- **Function Profiling**: Analyze performance bottlenecks\n- **Debugging**: Track function calls and parameter values\n- **Integration Testing**: Monitor system component interactions\n- **Behavior Analysis**: Understand function usage patterns\n\n### \ud83d\udcca Production Monitoring\n- **System Health**: Monitor critical business functions\n- **Performance Tracking**: Track execution times and error rates\n- **User Behavior**: Analyze how functions are used in production\n- **Compliance**: Maintain audit trails for regulatory requirements\n\n### \ud83d\udd2c Research & Analysis\n- **Algorithm Analysis**: Study algorithm behavior with real data\n- **Performance Optimization**: Identify optimization opportunities\n- **Data Quality**: Monitor data processing pipelines\n- **Experimentation**: Support research and development workflows\n\n## \ud83d\udcda Examples\n\nThe package includes comprehensive examples demonstrating various use cases:\n\n### Available Examples\n- **`basic_usage.py`**: Getting started with core features\n- **`advanced_monitoring.py`**: Advanced configuration and async support\n- **`real_world_simulation.py`**: Practical applications and systems\n\n### Run Examples\n```bash\n# Install the package\npip install imitator\n\n# Clone repository for examples (if needed)\ngit clone https://github.com/yourusername/imitator.git\ncd imitator/examples\n\n# Run examples\npython basic_usage.py\npython advanced_monitoring.py\npython real_world_simulation.py\n```\n\nEach example demonstrates:\n- Different monitoring strategies\n- Error handling scenarios\n- Performance analysis\n- Log inspection and analysis\n\n## \ud83d\ude80 Getting Started\n\n1. **Install**: `pip install imitator`\n2. **Import**: `from imitator import monitor_function`\n3. **Decorate**: Add `@monitor_function` to your functions\n4. **Run**: Use your functions normally\n5. **Analyze**: Check the generated logs in the `logs/` directory\n\n## \ud83d\udcd6 Documentation\n\n- **Examples**: Comprehensive examples in the `examples/` directory\n- **API Reference**: Detailed docstrings in all modules\n- **Type Hints**: Full type annotation support\n- **Error Handling**: Graceful handling of edge cases\n\n## \ud83e\udd1d Contributing\n\nWe welcome contributions! Please see our [Contributing Guidelines](CONTRIBUTING.md) for details.\n\n## \ud83d\udcc4 License\n\nApache License 2.0 - see [LICENSE](LICENSE) file for details. \n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "A lightweight Python framework for monitoring and imitating function behavior with automatic I/O tracking and pattern learning",
"version": "0.1.0",
"project_urls": {
"Changelog": "https://github.com/bhavikr2/imitator/blob/c4a45cfdaebccbd888ab12bdd2a7d3bced9ba61e/CHANGELOG.md",
"Documentation": "https://github.com/bhavikr2/imitator.git#readme",
"Homepage": "https://github.com/bhavikr2/imitator.git",
"Issues": "https://github.com/bhavikr2/imitator.git/issues",
"Repository": "https://github.com/bhavikr2/imitator.git"
},
"split_keywords": [
"monitoring",
" imitation",
" function",
" behavior",
" io",
" machine-learning",
" debugging",
" profiling",
" type-validation",
" decorator",
" pattern-learning"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "e232e74b7e249b57f11f9a241adca09a4ac71342747f8bd9bae84787bf0dc457",
"md5": "f57e12490284b5b7fa88a9dfb90aa8e9",
"sha256": "ad1182f3557555dd7e37a81505aa63a058d76bd0a43c5d8e19b9ab83955c7f0a"
},
"downloads": -1,
"filename": "imitator-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "f57e12490284b5b7fa88a9dfb90aa8e9",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 16317,
"upload_time": "2025-07-14T06:53:57",
"upload_time_iso_8601": "2025-07-14T06:53:57.700307Z",
"url": "https://files.pythonhosted.org/packages/e2/32/e74b7e249b57f11f9a241adca09a4ac71342747f8bd9bae84787bf0dc457/imitator-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "0fee1a2e2fe0490ac8bfa021a65fd0f8d4c42c61d68087bd7127eb62648dc106",
"md5": "ea64ca8738b921cf6124ecab792a740f",
"sha256": "01389ae37cd679c2259d7dd4cc0b91216d21d32ac2632b1e2a0b4181c984d105"
},
"downloads": -1,
"filename": "imitator-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "ea64ca8738b921cf6124ecab792a740f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 49656,
"upload_time": "2025-07-14T06:53:59",
"upload_time_iso_8601": "2025-07-14T06:53:59.435600Z",
"url": "https://files.pythonhosted.org/packages/0f/ee/1a2e2fe0490ac8bfa021a65fd0f8d4c42c61d68087bd7127eb62648dc106/imitator-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-14 06:53:59",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "bhavikr2",
"github_project": "imitator",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"requirements": [
{
"name": "pydantic",
"specs": [
[
">=",
"2.0.0"
]
]
},
{
"name": "pytest",
"specs": [
[
">=",
"7.0.0"
]
]
},
{
"name": "pytest-asyncio",
"specs": [
[
">=",
"0.20.0"
]
]
}
],
"lcname": "imitator"
}
JSON
