Computer Vision

import tensorflow as tf

# Load your trained model
model = tf.keras.models.load_model('model.h5')  # (#1:Load Keras model)

# Create TFLite converter
converter = tf.lite.TFLiteConverter.from_keras_model(model)  # (#2:Initialize converter)

# Post-training quantization
converter.optimizations = [tf.lite.Optimize.DEFAULT]  # (#3:Enable optimization)

# Full integer quantization (requires representative dataset)
def representative_dataset():  # (#4:Calibration data)
    for i in range(100):
        yield [X_train[i:i+1].astype(np.float32)]

converter.representative_dataset = representative_dataset
converter.target_spec.supported_types = [tf.int8]  # (#5:INT8 precision)

# Convert and save
tflite_model = converter.convert()
with open('model_quantized.tflite', 'wb') as f:  # (#6:Save TFLite model)
    f.write(tflite_model)

import torch
from torch.quantization import quantize_dynamic, prepare, convert

# Dynamic quantization (easiest)
model_fp32 = load_model()
model_int8 = quantize_dynamic(  # (#1:Dynamic quant)
    model_fp32,
    {torch.nn.Linear, torch.nn.Conv2d},  # (#2:Layers to quantize)
    dtype=torch.qint8
)

# Static quantization (better accuracy)
model_fp32.qconfig = torch.quantization.get_default_qconfig('fbgemm')  # (#3:Set config)
model_prepared = prepare(model_fp32)  # (#4:Prepare model)

# Calibrate with representative data
with torch.no_grad():
    for batch in calibration_loader:  # (#5:Run calibration)
        model_prepared(batch)

model_quantized = convert(model_prepared)  # (#6:Convert to INT8)

# Save quantized model
torch.save(model_quantized.state_dict(), 'model_quantized.pt')

import tensorflow_model_optimization as tfmot

# Define pruning parameters
pruning_params = {
    'pruning_schedule': tfmot.sparsity.keras.PolynomialDecay(  # (#1:Gradual pruning)
        initial_sparsity=0.0,
        final_sparsity=0.5,  # (#2:50% weights pruned)
        begin_step=1000,
        end_step=5000
    )
}

# Apply pruning to model
model_for_pruning = tfmot.sparsity.keras.prune_low_magnitude(  # (#3:Wrap model)
    model, **pruning_params
)

# Compile and train with pruning callbacks
model_for_pruning.compile(optimizer='adam', loss='sparse_categorical_crossentropy')
callbacks = [tfmot.sparsity.keras.UpdatePruningStep()]  # (#4:Update pruning)

model_for_pruning.fit(X_train, y_train, epochs=10, callbacks=callbacks)

# Strip pruning wrappers for deployment
final_model = tfmot.sparsity.keras.strip_pruning(model_for_pruning)  # (#5:Clean model)

# PyTorch to ONNX
import torch
import torch.onnx

model = load_pytorch_model()
model.eval()

dummy_input = torch.randn(1, 3, 224, 224)  # (#1:Example input shape)

torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    export_params=True,  # (#2:Include weights)
    opset_version=13,  # (#3:ONNX version)
    input_names=['input'],
    output_names=['output'],
    dynamic_axes={  # (#4:Variable batch size)
        'input': {0: 'batch_size'},
        'output': {0: 'batch_size'}
    }
)

# Verify ONNX model
import onnx
model_onnx = onnx.load("model.onnx")
onnx.checker.check_model(model_onnx)  # (#5:Validate model)

import onnxruntime as ort
import numpy as np

# Create inference session
session = ort.InferenceSession(  # (#1:Load ONNX model)
    "model.onnx",
    providers=['CUDAExecutionProvider', 'CPUExecutionProvider']  # (#2:GPU first, CPU fallback)
)

# Get input/output names
input_name = session.get_inputs()[0].name  # (#3:Input tensor name)
output_name = session.get_outputs()[0].name

# Prepare input
image = preprocess_image("test.jpg")  # (#4:Your preprocessing)
input_data = np.expand_dims(image, axis=0).astype(np.float32)

# Run inference
outputs = session.run([output_name], {input_name: input_data})  # (#5:Run prediction)
predictions = outputs[0]

# Post-process
class_id = np.argmax(predictions)
confidence = np.max(predictions)

# Method 1: Using trtexec CLI
# trtexec --onnx=model.onnx --saveEngine=model.trt --fp16

# Method 2: Using Python API
import tensorrt as trt

logger = trt.Logger(trt.Logger.WARNING)  # (#1:TRT logger)
builder = trt.Builder(logger)
network = builder.create_network(
    1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)  # (#2:Explicit batch)
)
parser = trt.OnnxParser(network, logger)

# Parse ONNX model
with open("model.onnx", "rb") as f:
    parser.parse(f.read())  # (#3:Load ONNX)

# Configure builder
config = builder.create_builder_config()
config.set_flag(trt.BuilderFlag.FP16)  # (#4:Enable FP16)
config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 30)  # (#5:1GB workspace)

# Build engine
engine = builder.build_serialized_network(network, config)  # (#6:Build TRT engine)
with open("model.trt", "wb") as f:
    f.write(engine)

# Install dependencies
pip install fastapi uvicorn python-multipart pillow torch torchvision

# Run development server
uvicorn app:app --reload --host 0.0.0.0 --port 8000  # (#1:Dev server)

# Production with Gunicorn
pip install gunicorn
gunicorn app:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000  # (#2:Production)

# First, save model in SavedModel format
model.save('models/cv_model/1')  # (#1:Version 1)

# Pull TF Serving image
docker pull tensorflow/serving:latest-gpu  # (#2:GPU version)

# Run TF Serving container
docker run -d --name tf-serving \
  -p 8501:8501 \
  -p 8500:8500 \
  --gpus all \
  -v "$(pwd)/models:/models" \
  -e MODEL_NAME=cv_model \
  tensorflow/serving:latest-gpu  # (#3:Start serving)

# Test REST API
curl -X POST http://localhost:8501/v1/models/cv_model:predict \
  -H "Content-Type: application/json" \
  -d '{"instances": [[...image_data...]]}' # (#4:REST prediction)

# config.pbtxt
name: "image_classifier"
platform: "onnxruntime_onnx"  # (#1:Backend)
max_batch_size: 32  # (#2:Max batch)

input [
  {
    name: "input"
    data_type: TYPE_FP32
    dims: [ 3, 224, 224 ]  # (#3:Input shape)
  }
]

output [
  {
    name: "output"
    data_type: TYPE_FP32
    dims: [ 1000 ]  # (#4:Output classes)
  }
]

dynamic_batching {
  preferred_batch_size: [ 8, 16, 32 ]  # (#5:Batch sizes)
  max_queue_delay_microseconds: 100
}

instance_group [
  { count: 2, kind: KIND_GPU }  # (#6:GPU instances)
]

# Pull Triton image
docker pull nvcr.io/nvidia/tritonserver:23.10-py3  # (#1:Official image)

# Run Triton server
docker run --gpus all -d --name triton \
  -p 8000:8000 \
  -p 8001:8001 \
  -p 8002:8002 \
  -v $(pwd)/model_repository:/models \
  nvcr.io/nvidia/tritonserver:23.10-py3 \
  tritonserver --model-repository=/models  # (#2:Start server)

# Check model status
curl localhost:8000/v2/models/image_classifier  # (#3:Health check)

# Install NVIDIA Container Toolkit
distribution=$(. /etc/os-release;echo $ID$VERSION_ID)
curl -s -L https://nvidia.github.io/nvidia-docker/gpgkey | sudo apt-key add -
curl -s -L https://nvidia.github.io/nvidia-docker/$distribution/nvidia-docker.list \
  | sudo tee /etc/apt/sources.list.d/nvidia-docker.list  # (#1:Add repo)

sudo apt-get update
sudo apt-get install -y nvidia-container-toolkit  # (#2:Install toolkit)
sudo systemctl restart docker

# Run container with GPU access
docker run --gpus all nvidia/cuda:11.8-base nvidia-smi  # (#3:Test GPU)

# Run with specific GPUs
docker run --gpus '"device=0,1"' my-ml-app  # (#4:Select GPUs)

# Docker Compose with GPU
# In docker-compose.yml:
#   deploy:
#     resources:
#       reservations:
#         devices:
#           - capabilities: [gpu]  # (#5:Compose GPU)

# service.yaml
apiVersion: v1
kind: Service
metadata:
  name: cv-model-service
spec:
  selector:
    app: cv-model
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer  # (#1:External access)
---
# hpa.yaml - Horizontal Pod Autoscaler
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: cv-model-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: cv-model-api
  minReplicas: 2  # (#2:Min pods)
  maxReplicas: 10  # (#3:Max pods)
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70  # (#4:Scale at 70% CPU)

import mlflow
import mlflow.pytorch

# Set tracking URI (can be remote server)
mlflow.set_tracking_uri("http://mlflow-server:5000")  # (#1:MLflow server)
mlflow.set_experiment("cv-classification")  # (#2:Experiment name)

with mlflow.start_run(run_name="resnet50-v1"):  # (#3:Start run)
    # Log parameters
    mlflow.log_param("model", "resnet50")
    mlflow.log_param("learning_rate", 0.001)
    mlflow.log_param("batch_size", 32)
    mlflow.log_param("epochs", 50)  # (#4:Log params)

    # Train model...
    for epoch in range(epochs):
        train_loss, val_loss, val_acc = train_epoch(...)
        mlflow.log_metrics({  # (#5:Log metrics)
            "train_loss": train_loss,
            "val_loss": val_loss,
            "val_accuracy": val_acc
        }, step=epoch)

    # Log model
    mlflow.pytorch.log_model(model, "model")  # (#6:Save model)

    # Log artifacts
    mlflow.log_artifact("confusion_matrix.png")  # (#7:Save artifacts)

from mlflow.tracking import MlflowClient

client = MlflowClient()

# Register a model
model_uri = "runs:/abc123/model"  # (#1:Run ID + artifact path)
model_version = mlflow.register_model(
    model_uri,
    "image-classifier"  # (#2:Model name)
)

# Transition model stage
client.transition_model_version_stage(
    name="image-classifier",
    version=model_version.version,
    stage="Staging"  # (#3:Staging/Production/Archived)
)

# Load model by stage
model = mlflow.pytorch.load_model(
    "models:/image-classifier/Production"  # (#4:Load prod model)
)

# Add model description
client.update_model_version(
    name="image-classifier",
    version=model_version.version,
    description="ResNet50 trained on custom dataset, 95% accuracy"  # (#5:Document)

from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset

# Define column mapping
column_mapping = ColumnMapping(
    prediction='prediction',
    target='actual_label'
)

# Create drift report
report = Report(metrics=[
    DataDriftPreset(),  # (#1:Input drift)
    TargetDriftPreset()  # (#2:Prediction drift)
])

# Generate report
report.run(
    reference_data=training_data,  # (#3:Reference distribution)
    current_data=production_data,  # (#4:Current distribution)
    column_mapping=column_mapping
)

# Save report
report.save_html("drift_report.html")  # (#5:HTML report)

# Get results programmatically
results = report.as_dict()
if results['metrics'][0]['result']['dataset_drift']:  # (#6:Check drift)
    alert_team("Data drift detected!")

from prometheus_client import Counter, Histogram, generate_latest
from fastapi import FastAPI, Response

app = FastAPI()

# Define metrics
PREDICTIONS = Counter(
    'model_predictions_total',
    'Total predictions',
    ['model', 'class']  # (#1:Labels)
)

LATENCY = Histogram(
    'model_inference_latency_seconds',
    'Inference latency',
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0]  # (#2:Buckets)
)

@app.post("/predict")
async def predict(file: UploadFile):
    with LATENCY.time():  # (#3:Measure latency)
        result = model.predict(file)

    PREDICTIONS.labels(model='resnet50', class_=result['class']).inc()  # (#4:Count)
    return result

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

# TFLite inference on edge device
import tflite_runtime.interpreter as tflite
import numpy as np
from PIL import Image

# Load TFLite model
interpreter = tflite.Interpreter(model_path="model.tflite")  # (#1:Load model)
interpreter.allocate_tensors()  # (#2:Allocate memory)

# Get input/output details
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# Preprocess image
image = Image.open("test.jpg").resize((224, 224))
input_data = np.expand_dims(np.array(image), axis=0)
input_data = (input_data / 255.0).astype(np.float32)  # (#3:Normalize)

# Run inference
interpreter.set_tensor(input_details[0]['index'], input_data)  # (#4:Set input)
interpreter.invoke()  # (#5:Run inference)
output_data = interpreter.get_tensor(output_details[0]['index'])  # (#6:Get output)

predicted_class = np.argmax(output_data)

# Convert ONNX to TensorRT on Jetson
/usr/src/tensorrt/bin/trtexec \
  --onnx=model.onnx \
  --saveEngine=model.trt \
  --fp16 \
  --workspace=1024  # (#1:Build TRT engine)

# Coral Edge TPU inference
from pycoral.utils.edgetpu import make_interpreter
from pycoral.adapters import common, classify
from PIL import Image

# Load Edge TPU model (must be compiled for Edge TPU)
interpreter = make_interpreter('model_edgetpu.tflite')  # (#1:Edge TPU model)
interpreter.allocate_tensors()

# Get model input size
size = common.input_size(interpreter)  # (#2:Input size)

# Load and preprocess image
image = Image.open('test.jpg').resize(size, Image.LANCZOS)
common.set_input(interpreter, image)  # (#3:Set input)

# Run inference
interpreter.invoke()  # (#4:Run on TPU)

# Get classification results
classes = classify.get_classes(interpreter, top_k=3)  # (#5:Top 3 predictions)
for c in classes:
    print(f"Class {c.id}: {c.score:.4f}")