caema-utils 0.1.2

System Dependency Manager - A collection of reusable Python modules for AI services

CAEMA Utils

System Dependency Manager - A collection of reusable Python modules for AI services.

Available Modules

Module	Description	Status
pose_estimation	AI-powered body pose analysis using YOLOv8	Stable (v0.1.2)

---

Installation

Quick Install (Recommended for servers/cloud)

pip install "caema-utils[all]" && pip uninstall opencv-python -y 2>/dev/null; pip install opencv-python-headless

This single command:

1. Installs caema-utils with all dependencies
2. Replaces opencv-python with opencv-python-headless (required for servers without display)

Standard Install

# Install with all dependencies
pip install "caema-utils[all]"

# For servers without display (Codespaces, Docker, cloud VMs), also run:
pip uninstall opencv-python -y && pip install opencv-python-headless

From GitHub

pip install "caema-utils[all] @ git+https://github.com/msorozabal/Utils.git"
pip uninstall opencv-python -y && pip install opencv-python-headless

For Development

git clone https://github.com/msorozabal/Utils.git
cd Utils
pip install -e ".[all,dev]"
pip uninstall opencv-python -y && pip install opencv-python-headless

---

Pose Estimation Module

AI-powered body pose analysis using YOLOv8 for biomechanical evaluation.

Features

• Detects 17 body keypoints (COCO format)
• Calculates biomechanical angles (neck, shoulders, spine, knees, etc.)
• Generates text analysis of postural findings
• Returns annotated images with skeleton overlay
• Multiple input formats: file path, bytes, numpy array
• Multiple output formats: bytes, base64, numpy array, file

Quick Start

from caema_utils.pose_estimation import PoseEstimator

# Initialize (downloads model on first run)
estimator = PoseEstimator(model_size="nano")

# Analyze from file
result = estimator.analyze("photo.jpg")

if result['success']:
    print(f"Confidence: {result['confidence']:.1%}")
    print(f"Analysis:\n{result['analysis']}")

    # Save annotated image
    with open("output.png", "wb") as f:
        f.write(result['annotated_image_bytes'])

Model Sizes

Size	Model	Speed	Accuracy	Use Case
nano	yolov8n-pose	Fastest	Good	Real-time, mobile
small	yolov8s-pose	Fast	Better	Balanced
medium	yolov8m-pose	Medium	High	Production
large	yolov8l-pose	Slow	Higher	Quality-critical
xlarge	yolov8x-pose	Slowest	Highest	Research

API Reference

PoseEstimator

class PoseEstimator:
    def __init__(self, model_size: str = "nano"):
        """
        Initialize pose estimator.

        Args:
            model_size: "nano", "small", "medium", "large", or "xlarge"
        """

    def analyze(
        self,
        image: Union[str, Path, bytes],
        output_format: str = "bytes",
        save_path: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Analyze body posture from image.

        Args:
            image: Path to image file or image bytes
            output_format: "bytes", "base64", "array", or "file"
            save_path: If output_format="file", save path for annotated image

        Returns:
            {
                'success': bool,
                'keypoints': Dict[str, List[float]],  # name -> [x, y, confidence]
                'angles': Dict[str, float],  # angle_name -> degrees
                'confidence': float,  # 0-1
                'analysis': str,  # Text analysis
                'annotated_image_*': ...,  # Image in requested format
            }
        """

Response Structure

{
    'success': True,
    'keypoints': {
        'nose': [256.5, 128.3, 0.95],
        'left_shoulder': [200.1, 180.2, 0.92],
        'right_shoulder': [312.8, 178.9, 0.93],
        # ... 17 keypoints total
    },
    'angles': {
        'neck_tilt': 5.2,
        'shoulder_angle': 3.1,
        'hip_angle': 2.8,
        'spine_angle': 4.5,
        'left_elbow': 165.3,
        'right_elbow': 168.7,
        'left_knee': 175.2,
        'right_knee': 176.8,
    },
    'confidence': 0.93,
    'analysis': 'OK: Body alignment within normal parameters\nOK: No significant postural deviations detected',
    'annotated_image_bytes': b'...',  # PNG image with skeleton overlay
}

---

HTTP API Server

The module includes a built-in FastAPI server for HTTP access.

Start the Server

# Check dependencies first
pose-server --check

# Start server (localhost only)
pose-server --port 8005

# Accept connections from other machines
pose-server --host 0.0.0.0 --port 8005

# With auto-reload for development
pose-server --host 0.0.0.0 --port 8005 --reload

API Endpoints

Endpoint	Method	Description
/	GET	API info
/health	GET	Health check
/analyze	POST	Full analysis (JSON + base64 image)
/analyze/image	POST	Returns annotated image directly
/analyze/json-only	POST	Analysis without image
/docs	GET	OpenAPI documentation

curl Examples

# Health check
curl http://localhost:8005/health

# Full analysis with annotated image (base64)
curl -X POST "http://localhost:8005/analyze" \
     -H "accept: application/json" \
     -F "file=@photo.jpg"

# Get annotated image directly (save to file)
curl -X POST "http://localhost:8005/analyze/image" \
     -F "file=@photo.jpg" \
     --output annotated.png

# JSON only (faster, no image)
curl -X POST "http://localhost:8005/analyze/json-only" \
     -F "file=@photo.jpg"

Complete Example: Test from Scratch (Codespaces/Cloud)

Copy and paste this entire block to install and test:

# Install (single command for headless servers)
pip install "caema-utils[all]" && pip uninstall opencv-python -y 2>/dev/null; pip install opencv-python-headless

# Download a test image
curl -o test.jpg "https://images.unsplash.com/photo-1544005313-94ddf0286df2?w=400"

# Start server in background
pose-server --host 0.0.0.0 --port 8005 &
sleep 5

# Test health endpoint
curl http://localhost:8005/health

# Analyze the test image (JSON response)
curl -X POST "http://localhost:8005/analyze/json-only" -F "file=@test.jpg"

# Get annotated image with skeleton overlay
curl -X POST "http://localhost:8005/analyze/image" -F "file=@test.jpg" --output result.png

# Check the result file was created
ls -la result.png

Testing from Another Machine

# On the SERVER machine:
pip install "caema-utils[all]" && pip uninstall opencv-python -y 2>/dev/null; pip install opencv-python-headless
pose-server --host 0.0.0.0 --port 8005

# On the CLIENT machine (replace <SERVER_IP> with actual IP):
curl http://<SERVER_IP>:8005/health
curl -X POST "http://<SERVER_IP>:8005/analyze" -F "file=@your_image.jpg"
curl -X POST "http://<SERVER_IP>:8005/analyze/image" -F "file=@your_image.jpg" --output result.png

Note: Ensure port 8005 is open in your firewall (sudo ufw allow 8005 on Ubuntu).

Response Example

{
  "success": true,
  "filename": "photo.jpg",
  "keypoints": {
    "nose": [256.5, 128.3, 0.95],
    "left_shoulder": [200.1, 180.2, 0.92]
  },
  "angles": {
    "neck_tilt": 5.2,
    "shoulder_angle": 3.1
  },
  "confidence": 0.93,
  "analysis": "OK: Body alignment within normal parameters",
  "annotated_image_base64": "iVBORw0KGgoAAAANS..."
}

---

Integration Examples

FastAPI Integration

from fastapi import FastAPI, File, UploadFile
from caema_utils.pose_estimation import PoseEstimator

app = FastAPI()
estimator = PoseEstimator(model_size="nano")

@app.post("/api/analyze")
async def analyze(file: UploadFile = File(...)):
    content = await file.read()
    result = estimator.analyze(content, output_format="base64")
    return result

Flask Integration

from flask import Flask, request, jsonify
from caema_utils.pose_estimation import PoseEstimator

app = Flask(__name__)
estimator = PoseEstimator(model_size="nano")

@app.route('/analyze', methods=['POST'])
def analyze():
    file = request.files['image']
    result = estimator.analyze(file.read())
    return jsonify(result)

Async Usage

import asyncio
from concurrent.futures import ThreadPoolExecutor
from caema_utils.pose_estimation import PoseEstimator

estimator = PoseEstimator()
executor = ThreadPoolExecutor(max_workers=4)

async def analyze_async(image_bytes):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        executor,
        estimator.analyze,
        image_bytes
    )

---

Keypoint Reference

The module detects 17 body keypoints in COCO format:

Index	Name	Description
0	nose	Nose tip
1	left_eye	Left eye
2	right_eye	Right eye
3	left_ear	Left ear
4	right_ear	Right ear
5	left_shoulder	Left shoulder
6	right_shoulder	Right shoulder
7	left_elbow	Left elbow
8	right_elbow	Right elbow
9	left_wrist	Left wrist
10	right_wrist	Right wrist
11	left_hip	Left hip
12	right_hip	Right hip
13	left_knee	Left knee
14	right_knee	Right knee
15	left_ankle	Left ankle
16	right_ankle	Right ankle

---

Calculated Angles

Angle	Description	Normal Range
neck_tilt	Head tilt from vertical	< 10°
shoulder_angle	Shoulder level asymmetry	< 5°
hip_angle	Hip level asymmetry	< 5°
spine_angle	Spinal lateral deviation	< 8°
left_elbow	Left elbow flexion	0-180°
right_elbow	Right elbow flexion	0-180°
left_knee	Left knee flexion	160-190°
right_knee	Right knee flexion	160-190°

---

Development

Running Tests

# Install dev dependencies
pip install -e ".[all,dev]"

# Run tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=caema_utils --cov-report=term-missing

Code Quality

# Format code
black src/ tests/

# Lint
ruff src/ tests/

# Type check
mypy src/

---

Requirements

• Python 3.9+
• For pose estimation:
  • numpy < 2.0.0
  • ultralytics >= 8.1.0
  • opencv-python-headless >= 4.9.0
• For HTTP server:
  • fastapi >= 0.100.0
  • uvicorn >= 0.23.0
  • python-multipart >= 0.0.6

---

License

MIT License - see LICENSE file.

---

Contributing

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit changes (git commit -m 'Add amazing feature')
4. Push to branch (git push origin feature/amazing-feature)
5. Open a Pull Request

---

Support

• Issues: GitHub Issues
• Documentation: GitHub Wiki