CoreML Deployment Guide: Training to Production on Mac Mini M4

# Create a virtual environment
python3 -m venv ~/coreml-env
source ~/coreml-env/bin/activate

# Install coremltools and PyTorch
pip install coremltools torch torchvision

# Verify installation
python3 -c "import coremltools as ct; print(ct.__version__)"
# 8.1

import torch
import torchvision
import coremltools as ct

# Load a pretrained ResNet50 model
model = torchvision.models.resnet50(weights=torchvision.models.ResNet50_Weights.DEFAULT)
model.eval()

# Create a sample input (batch=1, channels=3, height=224, width=224)
example_input = torch.randn(1, 3, 224, 224)

# Trace the model with TorchScript
traced_model = torch.jit.trace(model, example_input)

# Convert to CoreML
coreml_model = ct.convert(
    traced_model,
    inputs=[ct.ImageType(
        name="image",
        shape=(1, 3, 224, 224),
        scale=1.0 / (255.0 * 0.226),
        bias=[-0.485 / 0.226, -0.456 / 0.226, -0.406 / 0.226],
        color_layout=ct.colorlayout.RGB
    )],
    classifier_config=ct.ClassifierConfig("imagenet_classes.txt"),
    compute_units=ct.ComputeUnit.ALL,  # Use Neural Engine + GPU + CPU
    minimum_deployment_target=ct.target.macOS15,
)

# Save the model
coreml_model.save("ResNet50.mlpackage")
print("Model saved: ResNet50.mlpackage")

import torch
import torch.nn as nn
import coremltools as ct

# Define a custom text embedding model
class TextEncoder(nn.Module):
    def __init__(self, vocab_size=30000, embed_dim=512, num_heads=8, num_layers=6):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.pos_encoding = nn.Parameter(torch.randn(1, 512, embed_dim))
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=embed_dim, nhead=num_heads, batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        self.output_proj = nn.Linear(embed_dim, 256)

    def forward(self, input_ids):
        x = self.embedding(input_ids) + self.pos_encoding[:, :input_ids.shape[1], :]
        x = self.transformer(x)
        x = x.mean(dim=1)  # Global average pooling
        return self.output_proj(x)

# Initialize and load your trained weights
model = TextEncoder()
# model.load_state_dict(torch.load("text_encoder_weights.pt"))
model.eval()

# Trace with example input
example_input = torch.randint(0, 30000, (1, 128))
traced_model = torch.jit.trace(model, example_input)

# Convert to CoreML
coreml_model = ct.convert(
    traced_model,
    inputs=[ct.TensorType(name="input_ids", shape=(1, 128), dtype=int)],
    outputs=[ct.TensorType(name="embedding")],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

# Add metadata
coreml_model.author = "My Remote Mac"
coreml_model.short_description = "Text embedding model for semantic search"
coreml_model.version = "1.0.0"

coreml_model.save("TextEncoder.mlpackage")
print("Model saved: TextEncoder.mlpackage")

import coremltools as ct
import numpy as np

# Load the converted model
model = ct.models.MLModel("ResNet50.mlpackage")

# Inspect model metadata
spec = model.get_spec()
print(f"Model type: {spec.WhichOneof('Type')}")
print(f"Inputs: {[inp.name for inp in spec.description.input]}")
print(f"Outputs: {[out.name for out in spec.description.output]}")

# Run a test prediction
from PIL import Image
img = Image.open("test_image.jpg").resize((224, 224))
prediction = model.predict({"image": img})
print(f"Top prediction: {prediction}")

# Check which compute units are available
print(f"Compute units: {model.compute_unit}")

import coremltools as ct
import tensorflow as tf

# Load a TensorFlow SavedModel (e.g., EfficientNet trained on your data)
tf_model = tf.keras.applications.EfficientNetV2S(
    weights="imagenet",
    input_shape=(384, 384, 3)
)

# Convert to CoreML
coreml_model = ct.convert(
    tf_model,
    inputs=[ct.ImageType(
        name="image",
        shape=(1, 384, 384, 3),
        scale=1.0 / 255.0,
        color_layout=ct.colorlayout.RGB
    )],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

coreml_model.save("EfficientNetV2S.mlpackage")
print("Saved EfficientNetV2S.mlpackage")

import coremltools as ct
import tensorflow as tf

# Load your custom Keras model
model = tf.keras.models.load_model("my_custom_model.h5")

# Print model summary to understand input/output shapes
model.summary()

# Convert with explicit input/output specifications
coreml_model = ct.convert(
    model,
    inputs=[ct.TensorType(name="features", shape=(1, 128))],
    outputs=[ct.TensorType(name="prediction")],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

# Add model metadata
coreml_model.author = "ML Team"
coreml_model.license = "Proprietary"
coreml_model.short_description = "Customer churn prediction model v2.1"
coreml_model.version = "2.1.0"

coreml_model.save("ChurnPredictor.mlpackage")

import coremltools as ct

# Convert directly from a .tflite file
coreml_model = ct.convert(
    "object_detector.tflite",
    source="tensorflow",
    inputs=[ct.ImageType(
        name="image",
        shape=(1, 320, 320, 3),
        scale=1.0 / 255.0,
        color_layout=ct.colorlayout.RGB
    )],
    outputs=[
        ct.TensorType(name="boxes"),
        ct.TensorType(name="scores"),
        ct.TensorType(name="classes"),
    ],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

coreml_model.save("ObjectDetector.mlpackage")
print("Saved ObjectDetector.mlpackage")

# Install ONNX and onnxruntime for validation
pip install onnx onnxruntime coremltools

import coremltools as ct
import onnx

# Load and validate the ONNX model
onnx_model = onnx.load("yolov8n.onnx")
onnx.checker.check_model(onnx_model)
print("ONNX model is valid")

# Inspect input/output shapes
for inp in onnx_model.graph.input:
    print(f"Input: {inp.name}, shape: {[d.dim_value for d in inp.type.tensor_type.shape.dim]}")
for out in onnx_model.graph.output:
    print(f"Output: {out.name}, shape: {[d.dim_value for d in out.type.tensor_type.shape.dim]}")

# Convert ONNX to CoreML
coreml_model = ct.converters.convert(
    "yolov8n.onnx",
    inputs=[ct.ImageType(
        name="images",
        shape=(1, 3, 640, 640),
        scale=1.0 / 255.0,
        color_layout=ct.colorlayout.RGB
    )],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

coreml_model.short_description = "YOLOv8 Nano object detection model"
coreml_model.save("YOLOv8n.mlpackage")
print("Saved YOLOv8n.mlpackage")

import torch
import coremltools as ct

# When direct PyTorch conversion fails, use ONNX as an intermediate step
model = torch.hub.load('pytorch/vision', 'detr_resnet50', pretrained=True)
model.eval()

dummy_input = torch.randn(1, 3, 800, 800)

# Step 1: Export to ONNX
torch.onnx.export(
    model,
    dummy_input,
    "detr_resnet50.onnx",
    input_names=["image"],
    output_names=["pred_logits", "pred_boxes"],
    opset_version=17,
    dynamic_axes={"image": {0: "batch"}}
)
print("Exported to ONNX")

# Step 2: Convert ONNX to CoreML
coreml_model = ct.converters.convert(
    "detr_resnet50.onnx",
    inputs=[ct.ImageType(name="image", shape=(1, 3, 800, 800), scale=1.0/255.0)],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

coreml_model.save("DETR_ResNet50.mlpackage")
print("Saved DETR_ResNet50.mlpackage")

import coremltools as ct
from coremltools.models.neural_network import quantization_utils

# Load the full-precision model
model = ct.models.MLModel("ResNet50.mlpackage")

# Quantize to float16 -- halves model size with virtually no quality loss
model_fp16 = quantization_utils.quantize_weights(model, nbits=16)
model_fp16.save("ResNet50_fp16.mlpackage")

# Check file sizes
import os
original_size = sum(
    os.path.getsize(os.path.join(dp, f))
    for dp, dn, fn in os.walk("ResNet50.mlpackage") for f in fn
)
optimized_size = sum(
    os.path.getsize(os.path.join(dp, f))
    for dp, dn, fn in os.walk("ResNet50_fp16.mlpackage") for f in fn
)
print(f"Original: {original_size / 1e6:.1f} MB")
print(f"Float16:  {optimized_size / 1e6:.1f} MB")
print(f"Reduction: {(1 - optimized_size/original_size)*100:.1f}%")

import coremltools as ct
import coremltools.optimize as cto
import numpy as np

# Load the model
model = ct.models.MLModel("ResNet50.mlpackage")

# Configure linear (int8) quantization with calibration data
op_config = cto.coreml.OpLinearQuantizerConfig(
    mode="linear_symmetric",
    dtype="int8",
    granularity="per_channel"
)

config = cto.coreml.OptimizationConfig(global_config=op_config)

# Prepare calibration data (representative samples from your dataset)
def load_calibration_data():
    """Load 100-200 representative samples for calibration."""
    calibration_samples = []
    for i in range(100):
        # Replace with your actual data loading
        sample = np.random.randn(1, 3, 224, 224).astype(np.float32)
        calibration_samples.append({"image": sample})
    return calibration_samples

# Apply post-training quantization
model_int8 = cto.coreml.linear_quantize_weights(
    model,
    config=config,
    sample_data=load_calibration_data()
)

model_int8.save("ResNet50_int8.mlpackage")
print("Saved int8 quantized model")

import coremltools as ct
import coremltools.optimize as cto

# Palettization clusters weights into a small lookup table
# 4-bit palettization = 16 unique weight values per tensor
# Achieves ~4x compression with minimal accuracy loss

model = ct.models.MLModel("ResNet50.mlpackage")

# Configure palettization
op_config = cto.coreml.OpPalettizerConfig(
    mode="kmeans",
    nbits=4,              # 4-bit = 16 clusters, 2-bit = 4 clusters
    granularity="per_tensor"
)

config = cto.coreml.OptimizationConfig(global_config=op_config)

# Apply palettization
model_palettized = cto.coreml.palettize_weights(model, config=config)
model_palettized.save("ResNet50_palettized_4bit.mlpackage")

print("4-bit palettized model saved")
print("This model runs optimally on the Neural Engine")

import coremltools as ct
import coremltools.optimize as cto

# Pruning sets small weights to zero, enabling sparse computation
# The Neural Engine can skip zero-weight operations for speed gains

model = ct.models.MLModel("ResNet50.mlpackage")

# Configure magnitude-based pruning
op_config = cto.coreml.OpMagnitudePrunerConfig(
    target_sparsity=0.75,              # Remove 75% of smallest weights
    granularity="per_channel",
    block_size=None                     # Unstructured pruning
)

config = cto.coreml.OptimizationConfig(global_config=op_config)

# Apply pruning
model_pruned = cto.coreml.prune_weights(model, config=config)
model_pruned.save("ResNet50_pruned_75.mlpackage")

print("75% sparse model saved")

import coremltools as ct
import coremltools.optimize as cto

# For maximum optimization, combine pruning + palettization + quantization
# This can achieve 8-16x compression with 1-2% accuracy loss

model = ct.models.MLModel("ResNet50.mlpackage")

# Step 1: Prune (set small weights to zero)
prune_config = cto.coreml.OptimizationConfig(
    global_config=cto.coreml.OpMagnitudePrunerConfig(target_sparsity=0.5)
)
model = cto.coreml.prune_weights(model, config=prune_config)
print("Step 1: Pruning complete (50% sparsity)")

# Step 2: Palettize (cluster remaining weights)
palette_config = cto.coreml.OptimizationConfig(
    global_config=cto.coreml.OpPalettizerConfig(mode="kmeans", nbits=4)
)
model = cto.coreml.palettize_weights(model, config=palette_config)
print("Step 2: Palettization complete (4-bit)")

# Save the fully optimized model
model.save("ResNet50_optimized.mlpackage")
print("Fully optimized model saved -- ready for Neural Engine deployment")

# flask_coreml_server.py
# pip install flask pillow coremltools gunicorn

import io
import time
import coremltools as ct
from flask import Flask, request, jsonify
from PIL import Image

app = Flask(__name__)

# Load the CoreML model at startup (runs on Neural Engine)
print("Loading CoreML model...")
model = ct.models.MLModel(
    "ResNet50_optimized.mlpackage",
    compute_units=ct.ComputeUnit.ALL
)
print("Model loaded successfully")

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "healthy", "model": "ResNet50"})

@app.route("/predict", methods=["POST"])
def predict():
    if "image" not in request.files:
        return jsonify({"error": "No image file provided"}), 400

    # Read and preprocess the image
    image_file = request.files["image"]
    image = Image.open(io.BytesIO(image_file.read())).resize((224, 224))

    # Run inference with timing
    start = time.perf_counter()
    prediction = model.predict({"image": image})
    latency_ms = (time.perf_counter() - start) * 1000

    return jsonify({
        "prediction": prediction,
        "latency_ms": round(latency_ms, 2),
        "compute_unit": "neural_engine+gpu+cpu"
    })

@app.route("/predict/batch", methods=["POST"])
def predict_batch():
    """Process multiple images in a single request."""
    if "images" not in request.files:
        return jsonify({"error": "No image files provided"}), 400

    results = []
    files = request.files.getlist("images")

    start = time.perf_counter()
    for image_file in files:
        image = Image.open(io.BytesIO(image_file.read())).resize((224, 224))
        prediction = model.predict({"image": image})
        results.append(prediction)
    total_ms = (time.perf_counter() - start) * 1000

    return jsonify({
        "predictions": results,
        "total_latency_ms": round(total_ms, 2),
        "images_processed": len(results),
        "avg_latency_ms": round(total_ms / len(results), 2)
    })

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

# Production: gunicorn flask_coreml_server:app -w 2 -b 0.0.0.0:5000 --timeout 120

# fastapi_coreml_server.py
# pip install fastapi uvicorn python-multipart pillow coremltools

import io
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor
import coremltools as ct
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from PIL import Image
from typing import List

app = FastAPI(
    title="CoreML Inference API",
    description="Production CoreML model serving on Mac Mini M4",
    version="1.0.0"
)

# Load model at startup
model = ct.models.MLModel(
    "ResNet50_optimized.mlpackage",
    compute_units=ct.ComputeUnit.ALL
)

# Thread pool for blocking CoreML calls
executor = ThreadPoolExecutor(max_workers=4)

def run_prediction(image_bytes: bytes) -> dict:
    """Run CoreML prediction in a thread (blocking call)."""
    image = Image.open(io.BytesIO(image_bytes)).resize((224, 224))
    start = time.perf_counter()
    result = model.predict({"image": image})
    latency = (time.perf_counter() - start) * 1000
    return {"prediction": result, "latency_ms": round(latency, 2)}

@app.get("/health")
async def health():
    return {"status": "healthy", "model": "ResNet50_optimized", "engine": "CoreML"}

@app.post("/predict")
async def predict(image: UploadFile = File(...)):
    if not image.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="File must be an image")

    image_bytes = await image.read()
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, run_prediction, image_bytes)
    return result

@app.post("/predict/batch")
async def predict_batch(images: List[UploadFile] = File(...)):
    loop = asyncio.get_event_loop()
    tasks = []
    for img in images:
        image_bytes = await img.read()
        tasks.append(loop.run_in_executor(executor, run_prediction, image_bytes))

    results = await asyncio.gather(*tasks)
    return {
        "predictions": list(results),
        "total_images": len(results)
    }

# Run: uvicorn fastapi_coreml_server:app --host 0.0.0.0 --port 8000 --workers 2

# Test single prediction
curl -X POST http://localhost:8000/predict \
  -F "image=@test_image.jpg"

# Test batch prediction
curl -X POST http://localhost:8000/predict/batch \
  -F "images=@image1.jpg" \
  -F "images=@image2.jpg" \
  -F "images=@image3.jpg"

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

# Python client example
import requests

with open("test_image.jpg", "rb") as f:
    response = requests.post(
        "http://your-mac-mini:8000/predict",
        files={"image": f}
    )
print(response.json())
# {"prediction": {"classLabel": "golden_retriever", "confidence": 0.94}, "latency_ms": 3.2}

# Create launchd plist for auto-start on boot
cat <<EOF > ~/Library/LaunchAgents/com.coreml.api.plist
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.coreml.api</string>
    <key>ProgramArguments</key>
    <array>
        <string>/Users/admin/coreml-env/bin/uvicorn</string>
        <string>fastapi_coreml_server:app</string>
        <string>--host</string>
        <string>0.0.0.0</string>
        <string>--port</string>
        <string>8000</string>
        <string>--workers</string>
        <string>2</string>
    </array>
    <key>WorkingDirectory</key>
    <string>/Users/admin/coreml-api</string>
    <key>RunAtLoad</key>
    <true/>
    <key>KeepAlive</key>
    <true/>
    <key>StandardOutPath</key>
    <string>/var/log/coreml-api.log</string>
    <key>StandardErrorPath</key>
    <string>/var/log/coreml-api-error.log</string>
</dict>
</plist>
EOF

# Load the service
launchctl load ~/Library/LaunchAgents/com.coreml.api.plist

# Verify
curl http://localhost:8000/health

import coremltools as ct
import numpy as np
import time

model = ct.models.MLModel("ResNet50_optimized.mlpackage", compute_units=ct.ComputeUnit.ALL)

# Warmup (first inference compiles the model for the Neural Engine)
from PIL import Image
dummy = Image.new("RGB", (224, 224))
for _ in range(10):
    model.predict({"image": dummy})

# Benchmark
latencies = []
for _ in range(1000):
    start = time.perf_counter()
    model.predict({"image": dummy})
    latencies.append((time.perf_counter() - start) * 1000)

latencies = np.array(latencies)
print(f"Mean latency:   {latencies.mean():.2f} ms")
print(f"Median latency: {np.median(latencies):.2f} ms")
print(f"P95 latency:    {np.percentile(latencies, 95):.2f} ms")
print(f"P99 latency:    {np.percentile(latencies, 99):.2f} ms")
print(f"Throughput:     {1000 / latencies.mean():.0f} images/sec")

# multi_model_server.py
import io
import time
import coremltools as ct
from fastapi import FastAPI, File, UploadFile, HTTPException
from PIL import Image

app = FastAPI(title="Multi-Model CoreML Server")

# Load multiple models at startup
models = {}

@app.on_event("startup")
async def load_models():
    print("Loading models...")
    models["resnet50"] = ct.models.MLModel(
        "ResNet50_optimized.mlpackage", compute_units=ct.ComputeUnit.ALL
    )
    models["yolov8"] = ct.models.MLModel(
        "YOLOv8n.mlpackage", compute_units=ct.ComputeUnit.ALL
    )
    models["efficientnet"] = ct.models.MLModel(
        "EfficientNetV2S.mlpackage", compute_units=ct.ComputeUnit.ALL
    )
    print(f"Loaded {len(models)} models: {list(models.keys())}")

@app.get("/models")
async def list_models():
    return {"models": list(models.keys())}

@app.post("/predict/{model_name}")
async def predict(model_name: str, image: UploadFile = File(...)):
    if model_name not in models:
        raise HTTPException(404, f"Model '{model_name}' not found. Available: {list(models.keys())}")

    image_data = Image.open(io.BytesIO(await image.read()))

    # Resize based on model requirements
    input_sizes = {"resnet50": (224, 224), "yolov8": (640, 640), "efficientnet": (384, 384)}
    image_data = image_data.resize(input_sizes.get(model_name, (224, 224)))

    start = time.perf_counter()
    result = models[model_name].predict({"image": image_data})
    latency = (time.perf_counter() - start) * 1000

    return {"model": model_name, "prediction": result, "latency_ms": round(latency, 2)}

# Run: uvicorn multi_model_server:app --host 0.0.0.0 --port 8000

# /etc/nginx/nginx.conf
# Install nginx: brew install nginx

upstream coreml_backend {
    # Round-robin across multiple Mac Mini M4 instances
    server 10.0.1.10:8000 weight=1;   # Mac Mini M4 #1
    server 10.0.1.11:8000 weight=1;   # Mac Mini M4 #2
    server 10.0.1.12:8000 weight=1;   # Mac Mini M4 #3

    # Health check: remove unhealthy backends
    keepalive 32;
}

server {
    listen 80;
    server_name api.yourdomain.com;

    # Rate limiting
    limit_req_zone $binary_remote_addr zone=api:10m rate=100r/s;

    location / {
        limit_req zone=api burst=50 nodelay;

        proxy_pass http://coreml_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # Timeout settings for ML inference
        proxy_connect_timeout 10s;
        proxy_send_timeout 30s;
        proxy_read_timeout 30s;

        # Enable keepalive to backend
        proxy_http_version 1.1;
        proxy_set_header Connection "";
    }

    location /health {
        proxy_pass http://coreml_backend;
        access_log off;
    }
}

# Test config and start
# nginx -t
# nginx

# docker-compose.yml
# Note: CoreML requires macOS -- Docker containers run CPU-only inference
# For production, use launchd services directly on macOS

version: "3.8"
services:
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - coreml-api

  coreml-api:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - ./models:/app/models
    environment:
      - MODEL_PATH=/app/models/ResNet50_optimized.mlpackage
      - WORKERS=2
    deploy:
      replicas: 2

# pip install prometheus-client prometheus-fastapi-instrumentator

import io
import time
import coremltools as ct
from fastapi import FastAPI, File, UploadFile
from PIL import Image
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from starlette.responses import Response

app = FastAPI(title="CoreML API with Monitoring")

# Prometheus metrics
PREDICTIONS_TOTAL = Counter(
    "coreml_predictions_total",
    "Total number of predictions",
    ["model", "status"]
)
PREDICTION_LATENCY = Histogram(
    "coreml_prediction_latency_seconds",
    "Prediction latency in seconds",
    ["model"],
    buckets=[0.001, 0.002, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
)
MODEL_LOAD_TIME = Gauge(
    "coreml_model_load_time_seconds",
    "Time taken to load the model",
    ["model"]
)
ACTIVE_REQUESTS = Gauge(
    "coreml_active_requests",
    "Number of currently active requests"
)

# Load model with timing
load_start = time.perf_counter()
model = ct.models.MLModel("ResNet50_optimized.mlpackage", compute_units=ct.ComputeUnit.ALL)
MODEL_LOAD_TIME.labels(model="resnet50").set(time.perf_counter() - load_start)

@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

@app.post("/predict")
async def predict(image: UploadFile = File(...)):
    ACTIVE_REQUESTS.inc()
    try:
        img = Image.open(io.BytesIO(await image.read())).resize((224, 224))

        start = time.perf_counter()
        result = model.predict({"image": img})
        latency = time.perf_counter() - start

        PREDICTION_LATENCY.labels(model="resnet50").observe(latency)
        PREDICTIONS_TOTAL.labels(model="resnet50", status="success").inc()

        return {"prediction": result, "latency_ms": round(latency * 1000, 2)}
    except Exception as e:
        PREDICTIONS_TOTAL.labels(model="resnet50", status="error").inc()
        raise
    finally:
        ACTIVE_REQUESTS.dec()

# Run: uvicorn monitored_server:app --host 0.0.0.0 --port 8000

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: "coreml-api"
    static_configs:
      - targets:
        - "10.0.1.10:8000"   # Mac Mini #1
        - "10.0.1.11:8000"   # Mac Mini #2
        - "10.0.1.12:8000"   # Mac Mini #3
    metrics_path: /metrics
    scrape_interval: 5s

  - job_name: "node-exporter"
    static_configs:
      - targets:
        - "10.0.1.10:9100"
        - "10.0.1.11:9100"
        - "10.0.1.12:9100"

#!/bin/bash
# monitor_coreml.sh -- System health monitoring for CoreML inference servers
# Run with: ./monitor_coreml.sh

echo "=== CoreML Server Health Monitor ==="
echo "$(date)"
echo ""

# Memory usage (critical for CoreML model loading)
echo "--- Memory Usage ---"
vm_stat | head -10
echo ""
memory_pressure
echo ""

# CPU and GPU utilization
echo "--- CPU Usage ---"
top -l 1 -n 5 -stats pid,command,cpu,mem | head -10
echo ""

# GPU/Neural Engine power (indicates compute unit activity)
echo "--- GPU/Neural Engine Power ---"
sudo powermetrics --samplers gpu_power,ane_power -n 1 -i 2000 2>/dev/null | grep -E "(GPU|ANE|Neural)"
echo ""

# Disk usage (model files can be large)
echo "--- Disk Usage ---"
df -h / | tail -1
echo ""

# Network connections to API
echo "--- Active API Connections ---"
netstat -an | grep ":8000" | wc -l | xargs echo "Active connections on port 8000:"
echo ""

# API health check
echo "--- API Health Check ---"
curl -s -w "\nHTTP Status: %{http_code}\nResponse Time: %{time_total}s\n" \
  http://localhost:8000/health 2>/dev/null || echo "API is DOWN"

# Useful PromQL queries for your Grafana dashboard:

# Average prediction latency (last 5 minutes)
rate(coreml_prediction_latency_seconds_sum[5m]) / rate(coreml_prediction_latency_seconds_count[5m])

# Predictions per second
rate(coreml_predictions_total[1m])

# P99 latency
histogram_quantile(0.99, rate(coreml_prediction_latency_seconds_bucket[5m]))

# Error rate percentage
rate(coreml_predictions_total{status="error"}[5m]) / rate(coreml_predictions_total[5m]) * 100

# Active concurrent requests
coreml_active_requests