matrice-common 0.1.64

Common server utilities for Matrice.ai services

Matrice Common Library

matrice_common is a high-performance Python package providing reusable utilities for Matrice.ai services. It offers production-ready components for authentication, API communication, streaming, video processing, and more.

🚀 Quick Start

Installation

pip install --index-url https://test.pypi.org/simple/ matrice_common

Basic Usage

from matrice_common.rpc import RPC
from matrice_common.session import create_session

# Initialize RPC client
rpc = RPC(
    access_key="your_access_key",
    secret_key="your_secret_key"
)

# Make API requests
response = rpc.get("/v1/endpoint")
data = rpc.post("/v1/resource", payload={"key": "value"})

# Create a session
session = create_session(
    access_key="your_access_key",
    secret_key="your_secret_key"
)

# Create a project
project = session.create_classification_project(
    name="My Project",
    description="AI classification project"
)

✨ Key Features

🔐 Authentication & Security

• Token-based authentication with automatic refresh
• Secure credential management
• Environment-based configuration (prod/staging/dev)

🌐 RPC Client

• Synchronous and asynchronous HTTP methods
• Automatic token management
• Built-in error handling and retry logic
• Type-safe API interactions

📊 Streaming

• Unified Interface: Single API for Kafka and Redis streaming
• Async Support: Full async/await compatibility
• Metrics & Monitoring: Built-in performance tracking
• Auto-Reconnection: Resilient connection handling

🎥 Video Processing

• H.265 encoding and decoding
• Hardware acceleration support
• Stream processing capabilities
• Frame optimization algorithms

🔧 Utilities

• Comprehensive error logging (Sentry + Kafka)
• Error deduplication
• Automatic dependency installation
• Caching decorators
• Type hints throughout

📦 Session & Compute Management

• Project lifecycle management
• Compute instance orchestration
• Cluster management
• Resource allocation

📖 Documentation

Comprehensive documentation is available in DOCUMENTATION.md, including:

• Installation Guide - Setup and requirements
• Authentication - Token management and security
• RPC Client - API communication
• Session Management - Project lifecycle
• Compute Management - Instance and cluster control
• Streaming - Kafka and Redis streaming
• Video Processing - H.265 encoding/decoding
• Frame Optimization - Intelligent transmission
• Error Handling - Logging and monitoring
• Utilities - Helper functions
• API Reference - Complete API docs
• Testing - Test suite information
• Examples - Code examples

💡 Examples

Async API Calls

import asyncio
from matrice_common.rpc import RPC

async def fetch_data():
    rpc = RPC(access_key="...", secret_key="...")

    # Concurrent requests
    results = await asyncio.gather(
        rpc.get_async("/v1/users"),
        rpc.get_async("/v1/projects"),
        rpc.get_async("/v1/datasets")
    )

    return results

asyncio.run(fetch_data())

Streaming with Kafka

from matrice_common.stream.matrice_stream import MatriceStream, StreamType

# Create stream
stream = MatriceStream(
    stream_type=StreamType.KAFKA,
    access_key="...",
    secret_key="..."
)

# Setup and use
stream.setup(topic_or_stream_name="my-topic")
stream.add_message({"data": "value", "timestamp": "2025-01-01T12:00:00Z"})

# Receive messages
message = stream.get_message(timeout=30)
if message:
    print(f"Received: {message}")

stream.close()

Error Handling

from matrice_common.utils import log_errors, AppError, ErrorType, get_deduplication_config

# Configure deduplication (or use environment variables)
# export MATRICE_ERROR_DEDUPLICATION_ENABLED=true
# export MATRICE_ERROR_CACHE_TTL_SECONDS=1800  # 30 minutes

# Use service_name to properly track errors per service
@log_errors(service_name="my_service", raise_exception=False)
def process_data(data):
    """Automatically logs errors to Sentry and Kafka with deduplication."""
    if not data:
        raise ValueError("Data cannot be empty")

    # Process data
    return result

# Errors are automatically logged and deduplicated
result = process_data(my_data)

# Check deduplication config
print(get_deduplication_config())
# Output: {'enabled': True, 'ttl_seconds': 900, 'max_cache_size': 1000, 'current_cache_size': 0}

Video Encoding

from matrice_common.video.h265_processor import encode_frame_h265, decode_frame_h265
from PIL import Image

# Load and encode frame
frame = Image.open("frame.jpg")
encoded = encode_frame_h265(frame, quality=23, preset="medium")

# Decode frame
decoded = decode_frame_h265(encoded)
decoded.show()

🧪 Testing

The library includes a comprehensive test suite with high coverage.

Running Tests

# Install development dependencies
pip install -r requirements-dev.txt

# Run all tests
pytest

# Run with coverage report
pytest --cov=src/matrice_common --cov-report=html

# Run specific test module
pytest tests/unit/test_rpc.py -v

# Run with verbose output
pytest -v --tb=short

Test Coverage

• token_auth.py: 100% coverage
• utils.py: 52% coverage
• rpc.py: Comprehensive unit tests
• Integration tests for all major workflows

View detailed coverage:

pytest --cov --cov-report=html
open htmlcov/index.html  # View in browser

🏗️ Project Structure

py_common/
├── src/matrice_common/      # Source code
│   ├── rpc.py               # RPC client
│   ├── token_auth.py        # Authentication
│   ├── utils.py             # Utilities and error handling
│   ├── session.py           # Session management
│   ├── compute.py           # Compute management
│   ├── stream/              # Streaming modules
│   │   ├── matrice_stream.py
│   │   ├── kafka_stream.py
│   │   └── redis_stream.py
│   ├── optimize/            # Frame optimization
│   │   ├── cache_manager.py
│   │   ├── frame_comparators.py
│   │   ├── frame_difference.py
│   │   └── transmission.py
│   └── video/               # Video processing
│       ├── h265_processor.py
│       └── h265_video_processor.py
├── tests/                   # Test suite
│   ├── conftest.py         # Test fixtures
│   ├── unit/               # Unit tests
│   └── integration/        # Integration tests
├── DOCUMENTATION.md        # Comprehensive documentation
├── README.md              # This file
├── setup.py               # Package setup
├── pyproject.toml         # Project configuration
├── pytest.ini             # Pytest configuration
└── requirements-dev.txt   # Development dependencies

🔧 Development

Setup Development Environment

# Clone repository
git clone <repository-url>
cd py_common

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install development dependencies
pip install -r requirements-dev.txt

# Install package in editable mode
pip install -e .

Code Quality

# Format code
black src/

# Lint code
flake8 src/

# Type checking
mypy src/

Building the Package

# Build package
python setup.py bdist_wheel sdist

# Build with PyArmor obfuscation (optional)
python setup.py build

# Skip obfuscation
SKIP_PYARMOR_OBFUSCATION=true python setup.py bdist_wheel

🌍 Environment Variables

Core Configuration

Variable	Description	Required	Default
MATRICE_ACCESS_KEY_ID	API access key	Yes	-
MATRICE_SECRET_ACCESS_KEY	API secret key	Yes	-
ENV	Environment (prod/staging/dev)	No	prod
MATRICE_ACTION_ID	Current action ID	No	-
MATRICE_SESSION_ID	Current session ID	No	-
SKIP_PYARMOR_OBFUSCATION	Skip code obfuscation	No	false

Error Logging & Deduplication

Variable	Description	Required	Default
MATRICE_ERROR_DEDUPLICATION_ENABLED	Enable error deduplication	No	true
MATRICE_ERROR_CACHE_TTL_SECONDS	Deduplication time window (seconds)	No	900 (15 min)
MATRICE_ERROR_CACHE_MAX_SIZE	Max unique errors to track	No	1000

Setting Environment Variables

# Linux/Mac
export MATRICE_ACCESS_KEY_ID="your_access_key"
export MATRICE_SECRET_ACCESS_KEY="your_secret_key"
export ENV="staging"

# Windows PowerShell
$env:MATRICE_ACCESS_KEY_ID="your_access_key"
$env:MATRICE_SECRET_ACCESS_KEY="your_secret_key"
$env:ENV="staging"

# Python
import os
os.environ['MATRICE_ACCESS_KEY_ID'] = 'your_access_key'
os.environ['MATRICE_SECRET_ACCESS_KEY'] = 'your_secret_key'

📊 Module Overview

Core Modules

Module	Purpose	Key Features
rpc.py	API Communication	Sync/Async HTTP, auto-auth, retry logic
token_auth.py	Authentication	Token management, auto-refresh
utils.py	Utilities	Error logging, caching, helpers
session.py	Session Management	Project lifecycle, CRUD operations
compute.py	Compute Management	Instance/cluster management

Streaming Modules

Module	Purpose	Backend
matrice_stream.py	Unified Streaming	Kafka/Redis
kafka_stream.py	Kafka Streaming	Apache Kafka
redis_stream.py	Redis Streaming	Redis Streams

Video & Optimization

Module	Purpose	Features
h265_processor.py	Video Processing	H.265 encode/decode
frame_comparators.py	Frame Comparison	SSIM, perceptual hashing
frame_difference.py	Difference Detection	Change detection
transmission.py	Optimized Transfer	Smart transmission

🤝 Contributing

We welcome contributions! Please follow these guidelines:

1. Fork the repository
2. Create a feature branch: git checkout -b feature/my-feature
3. Write tests for new functionality
4. Ensure tests pass: pytest
5. Format code: black src/
6. Submit pull request

Contribution Guidelines

• Maintain test coverage above 80%
• Follow PEP 8 style guide
• Add type hints to all functions
• Write clear docstrings (Google style)
• Update documentation for new features

📝 License

This project is licensed under the MIT License - see the LICENSE.txt file for details.

🙏 Acknowledgments

• Built for Matrice.ai services
• Uses industry-standard libraries (requests, aiohttp, Kafka, Redis)
• Inspired by modern Python best practices

📞 Support

• Documentation: DOCUMENTATION.md
• Issues: GitHub Issues
• Email: [e-mail]

🗺️ Roadmap

• Additional streaming backends (RabbitMQ, NATS)
• Enhanced video codecs (H.264, VP9, AV1)
• GraphQL support
• WebSocket streaming
• Real-time metrics dashboard
• CLI tool for common operations

---

Made with ❤️ by the Matrice.ai Team

Last Updated: 2025-01-30 | Version: 0.0.2