Post-Processing Module - Refactored Architecture

Overview

This module provides a comprehensive, refactored post-processing system for the Matrice Python SDK. The system has been completely redesigned to be more pythonic, maintainable, and extensible while providing powerful analytics capabilities for various use cases.

🚀 Key Features

✅ Unified Architecture

Single Entry Point: PostProcessor class handles all processing needs
Standardized Results: All operations return ProcessingResult objects
Consistent Configuration: Type-safe configuration system with validation
Registry Pattern: Easy registration and discovery of use cases

✅ Separate Use Case Classes

People Counting: Advanced people counting with zone analysis and tracking
Customer Service: Comprehensive customer service analytics with business intelligence
Extensible Design: Easy to add new use cases

✅ Pythonic Configuration Management

Dataclass-based: Type-safe configurations using dataclasses
Nested Configurations: Support for complex nested config structures
File Support: JSON/YAML configuration file loading and saving
Validation: Built-in validation with detailed error messages

✅ Comprehensive Error Handling

Standardized Errors: All errors return structured ProcessingResult objects
Detailed Information: Error messages include type, context, and debugging info
Graceful Degradation: System continues operating even with partial failures

✅ Processing Statistics

Performance Tracking: Automatic processing time measurement
Success Metrics: Success/failure rates and statistics
Insights Generation: Automatic generation of actionable insights

📁 Architecture

post_processing/
├── __init__.py              # Main exports and convenience functions
├── processor.py             # Main PostProcessor class
├── README.md               # This documentation
│
├── core/                   # Core system components
│   ├── __init__.py
│   ├── base.py            # Base classes, enums, and protocols
│   ├── config.py          # Configuration system
│   └── advanced_usecases.py # Advanced use case implementations
│
├── usecases/              # Separate use case implementations
│   ├── __init__.py
│   ├── people_counting.py # People counting use case
│   └── customer_service.py # Customer service use case
│
└── utils/                 # Utility functions organized by category
    ├── __init__.py
    ├── geometry_utils.py  # Geometric calculations
    ├── format_utils.py    # Format detection and conversion
    ├── filter_utils.py    # Filtering and cleaning operations
    ├── counting_utils.py  # Counting and aggregation
    └── tracking_utils.py  # Tracking and movement analysis

🛠 Quick Start

Basic Usage

from matrice_analytics.post_processing import PostProcessor, process_simple

# Method 1: Simple processing (recommended for quick tasks)
result = process_simple(
    raw_results,
    usecase="people_counting",
    confidence_threshold=0.5
)

# Method 2: Using PostProcessor class (recommended for complex workflows)
processor = PostProcessor()
result = processor.process_simple(
    raw_results,
    usecase="people_counting",
    confidence_threshold=0.5,
    enable_tracking=True
)

print(f"Status: {result.status.value}")
print(f"Summary: {result.summary}")
print(f"Insights: {len(result.insights)} generated")

Advanced Configuration

# Create complex configuration
config = processor.create_config(
    'people_counting',
    confidence_threshold=0.6,
    enable_tracking=True,
    person_categories=['person', 'people', 'human'],
    zone_config={
        'zones': {
            'entrance': [[0, 0], [100, 0], [100, 100], [0, 100]],
            'checkout': [[200, 200], [300, 200], [300, 300], [200, 300]]
        }
    },
    alert_config={
        'count_thresholds': {'all': 10},
        'occupancy_thresholds': {'entrance': 5}
    }
)

# Process with configuration
result = processor.process(raw_results, config)

Configuration File Support

# Save configuration to file
processor.save_config(config, "people_counting_config.json")

# Load and use configuration from file
result = processor.process_from_file(raw_results, "people_counting_config.json")

📊 Use Cases

1. People Counting (`people_counting`)

Advanced people counting with comprehensive analytics:

result = process_simple(
    raw_results,
    usecase="people_counting",
    confidence_threshold=0.5,
    enable_tracking=True,
    person_categories=['person', 'people'],
    zone_config={
        'zones': {
            'entrance': [[0, 0], [100, 0], [100, 100], [0, 100]]
        }
    }
)

Features:

Multi-category person detection
Zone-based counting and analysis
Unique person tracking
Occupancy analysis
Alert generation based on thresholds
Temporal analysis and trends

2. Customer Service (`customer_service`)

Comprehensive customer service analytics:

result = process_simple(
    raw_results,
    usecase="customer_service",
    confidence_threshold=0.6,
    service_proximity_threshold=50.0,
    staff_categories=['staff', 'employee'],
    customer_categories=['customer', 'person']
)

Features:

Staff utilization analysis
Customer-staff interaction detection
Service quality metrics
Area occupancy analysis
Queue management insights
Business intelligence metrics

🔧 Configuration System

Configuration Classes

All configurations are type-safe dataclasses with built-in validation:

from matrice_analytics.post_processing import PeopleCountingConfig, ZoneConfig

# Create configuration programmatically
config = PeopleCountingConfig(
    confidence_threshold=0.5,
    enable_tracking=True,
    zone_config=ZoneConfig(
        zones={
            'entrance': [[0, 0], [100, 0], [100, 100], [0, 100]]
        }
    )
)

# Validate configuration
errors = config.validate()
if errors:
    print(f"Configuration errors: {errors}")

Configuration Templates

# Get configuration template for a use case
template = processor.get_config_template('people_counting')
print(f"Available options: {list(template.keys())}")

# List all available use cases
use_cases = processor.list_available_usecases()
print(f"Available use cases: {use_cases}")

📈 Processing Results

All processing operations return a standardized ProcessingResult object:

class ProcessingResult:
    data: Any                           # Processed data
    status: ProcessingStatus           # SUCCESS, ERROR, WARNING, PARTIAL
    usecase: str                       # Use case name
    category: str                      # Use case category
    processing_time: float             # Processing time in seconds
    summary: str                       # Human-readable summary
    insights: List[str]                # Generated insights
    warnings: List[str]                # Warning messages
    error_message: Optional[str]       # Error message if failed
    predictions: List[Dict[str, Any]]  # Detailed predictions
    metrics: Dict[str, Any]            # Performance metrics

Working with Results

result = processor.process_simple(data, "people_counting")

# Check status
if result.is_success():
    print(f"✅ {result.summary}")

    # Access insights
    for insight in result.insights:
        print(f"💡 {insight}")

    # Access metrics
    print(f"📊 Metrics: {result.metrics}")

    # Access processed data
    processed_data = result.data
else:
    print(f"❌ Processing failed: {result.error_message}")

📊 Statistics and Monitoring

# Get processing statistics
stats = processor.get_statistics()
print(f"Total processed: {stats['total_processed']}")
print(f"Success rate: {stats['success_rate']:.2%}")
print(f"Average processing time: {stats['average_processing_time']:.3f}s")

# Reset statistics
processor.reset_statistics()

🔌 Extensibility

Adding New Use Cases

Create Use Case Class:

from matrice_analytics.post_processing.core.base import BaseProcessor

class MyCustomUseCase(BaseProcessor):
    def __init__(self):
        super().__init__("my_custom_usecase")
        self.category = "custom"

    def process(self, data, config, context=None):
        # Implement your processing logic
        return self.create_result(processed_data, "my_custom_usecase", "custom")

Register Use Case:

from matrice_analytics.post_processing.core.base import registry

registry.register_use_case("custom", "my_custom_usecase", MyCustomUseCase)

Adding New Utility Functions

Add utility functions to the appropriate module in the utils/ directory and export them in utils/__init__.py.

🧪 Testing

The system includes comprehensive error handling and validation. Here's how to test your implementations:

# Test configuration validation
errors = processor.validate_config({
    'usecase': 'people_counting',
    'confidence_threshold': 0.5
})

# Test with sample data
sample_data = [
    {'category': 'person', 'confidence': 0.8, 'bbox': [10, 10, 50, 50]}
]

result = process_simple(sample_data, 'people_counting')
assert result.is_success()

🔄 Migration from Old System

If you're migrating from the old post-processing system:

Update Imports:

# Old
from matrice_analytics.old_post_processing import some_function

# New
from matrice_analytics.post_processing import PostProcessor, process_simple

Update Processing Calls:

# Old
result = old_process_function(data, config_dict)

# New
result = process_simple(data, "usecase_name", **config_dict)

Update Configuration:

# Old
config = {"threshold": 0.5, "enable_tracking": True}

# New
config = processor.create_config("people_counting",
                                confidence_threshold=0.5,
                                enable_tracking=True)

🐛 Troubleshooting

Common Issues

Use Case Not Found:

# Check available use cases
print(processor.list_available_usecases())

Configuration Validation Errors:

# Validate configuration
errors = processor.validate_config(config)
if errors:
    print(f"Validation errors: {errors}")

Processing Failures:

# Check result status and error details
if not result.is_success():
    print(f"Error: {result.error_message}")
    print(f"Error type: {result.error_type}")
    print(f"Error details: {result.error_details}")

📝 API Reference

Main Classes

PostProcessor: Main processing class
ProcessingResult: Standardized result container
BaseConfig: Base configuration class
PeopleCountingConfig: People counting configuration
CustomerServiceConfig: Customer service configuration

Convenience Functions

process_simple(): Simple processing function
create_config_template(): Get configuration template
list_available_usecases(): List available use cases
validate_config(): Validate configuration

Utility Functions

The system provides comprehensive utility functions organized by category:

Geometry: Point-in-polygon, distance calculations, IoU
Format: Format detection and conversion
Filter: Confidence filtering, deduplication
Counting: Object counting, zone analysis
Tracking: Movement analysis, line crossing detection

🎯 Best Practices

Use Simple Processing for Quick Tasks:

result = process_simple(data, "people_counting", confidence_threshold=0.5)

Use PostProcessor Class for Complex Workflows:

processor = PostProcessor()
config = processor.create_config("people_counting", **params)
result = processor.process(data, config)

Always Check Result Status:

if result.is_success():
    # Process successful result
else:
    # Handle error

Use Configuration Files for Complex Setups:

processor.save_config(config, "config.json")
result = processor.process_from_file(data, "config.json")

Monitor Processing Statistics:

stats = processor.get_statistics()
# Monitor success rates and performance

🔮 Future Enhancements

The refactored system is designed for easy extension. Planned enhancements include:

Additional use cases (security monitoring, retail analytics)
Advanced tracking algorithms
Real-time processing capabilities
Integration with external analytics platforms
Machine learning-based insights generation

The refactored post-processing system provides a solid foundation for scalable, maintainable, and powerful analytics capabilities. The clean architecture makes it easy to extend and customize for specific use cases while maintaining consistency and reliability.

Release files for matrice-analytics 0.3.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for matrice-analytics 0.3.0
File	Size	Uploaded
matrice_analytics-0.3.0.tar.gz	2.9 MB	Sep 1, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for matrice-analytics 0.3.0
File	Interpreter	ABI	Platform	Reset
matrice_analytics-0.3.0-py3-none-any.whl 3.4 MB Sep 1, 2026	Python 3	none	any	Details

Total release size: 6.3 MB

Release files / matrice_analytics-0.3.0.tar.gz

Download URL	matrice_analytics-0.3.0.tar.gz
Size	2.9 MB
Tags	Source
SHA-256 checksum How to use checksums	`b2cfec85077fb06c13fd52bbc3d1bbede13a794e47458fefb19a542301173f3e`
BLAKE2b-256 checksum How to use checksums	`6a343be4cb977d4bb22912abcac393d48cc406d885f525b8164971ef7d91d2ea`
Upload date	Sep 1, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.12.14`

Release files / matrice_analytics-0.3.0-py3-none-any.whl

Download URL	matrice_analytics-0.3.0-py3-none-any.whl
Size	3.4 MB
Tags	Python 3
SHA-256 checksum How to use checksums	`11e25b7d8fc391b7a07a4009b22feb923a0be0cdc0aa3e6959f5f6b9e331a50a`
BLAKE2b-256 checksum How to use checksums	`716dcc7b43f9cb727595b12e6325e63a36baa500c0a398f4226328f3d52dccc2`
Upload date	Sep 1, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.12.14`