15. Model Updates OTA

Over-the-air (OTA) model updates deliver improved ML models to deployed devices without physical access. Production OTA pipelines must handle version management, rollback capability, encryption, and bandwidth efficiency.

Version manifest schema:

import json
import hashlib

def create_model_manifest(model_path, metadata=None):
    """Create version manifest for OTA deployment"""
    with open(model_path, 'rb') as f:
        content = f.read()
        checksum = hashlib.sha256(content).hexdigest()
        size_bytes = len(content)
    
    manifest = {
        "version": metadata.get("version", "1.0.0"),
        "checksum_sha256": checksum,
        "size_bytes": size_bytes,
        "model_type": metadata.get("type", "classifier"),
        "min_compatible_version": metadata.get("min_version", "0.9.0"),
        "release_date": metadata.get("date"),
        "accuracy_metrics": metadata.get("metrics", {}),
        "deprecation_date": metadata.get("deprecates_after"),
    }
    
    return manifest

def sign_manifest(manifest, private_key_path):
    """Sign manifest with Ed25519 for authenticity"""
    from cryptography.hazmat.primitives import ed25519
    
    with open(private_key_path, 'rb') as key_file:
        private_key = ed25519.Ed25519PrivateKey.from_private_bytes(key_file.read())
    
    message = json.dumps(manifest, sort_keys=True).encode()
    signature = private_key.sign(message)
    
    signed_manifest = {**manifest, "signature": signature.hex()}
    return signed_manifest

Delta updates for bandwidth efficiency:

import bsdiff4

def compute_delta(old_model_path, new_model_path, delta_path):
    """Compute binary delta between model versions"""
    with open(old_model_path, 'rb') as old:
        with open(new_model_path, 'rb') as new:
            with open(delta_path, 'wb') as delta:
                delta.write(bsdiff4.diff(old.read(), new.read()))
    
    # Report savings
    new_size = os.path.getsize(new_model_path)
    delta_size = os.path.getsize(delta_path)
    print(f"Full model: {new_size/1024:.1f}KB, Delta model: {delta_size/1024:.1f}KB")

def apply_delta(model_path, delta_path, output_path):
    """Apply delta to produce new model"""
    with open(model_path, 'rb') as old:
        with open(delta_path, 'rb') as delta:
            with open(output_path, 'wb') as output:
                output.write(bsdiff4.patch(old.read(), delta.read()))

iOS model update via App Group:

import Foundation
import BackgroundTasks

class ModelUpdateManager {
    static let modelUpdateTask = "com.app.modelupdate"
    
    func registerBackgroundTask() {
        BGTaskScheduler.shared.register(forTaskWithIdentifier: Self.modelUpdateTask, using: nil) { task in
            self.handleModelUpdate(task: task as! BGProcessingTask)
        }
    }
    
    func scheduleModelUpdate() {
        let request = BGProcessingTaskRequest(identifier: Self.modelUpdateTask)
        request.requiresNetworkConnectivity = true
        request.requiresExternalPower = false
        
        try? BGTaskScheduler.shared.submit(request)
    }
    
    private func handleModelUpdate(task: BGProcessingTask) {
        task.expirationHandler = {
            task.setTaskCompleted(success: false)
        }
        
        Task {
            do {
                let newModel = try await downloadModel()
                try installModel(newModel)
                task.setTaskCompleted(success: true)
            } catch {
                task.setTaskCompleted(success: false)
            }
        }
    }
    
    private func installModel(_ model: URL) throws {
        let containerURL = FileManager.default.containerURL(
            forSecurityApplicationGroupIdentifier: "group.com.app"
        )!
        
        let destURL = containerURL.appendingPathComponent("model.mlmodel")
        try? FileManager.default.removeItem(at: destURL)
        try FileManager.default.moveItem(at: model, to: destURL)
    }
}

Android Play ML Model delivery:

// Use Play ML Model Delivery API
val modelDownloadConditions = Conditions.Builder()
    .require(Condition.INTERNET)
    .require(Condition.DEVICE_CHARGING)
    .require(Condition.NOT_LOW_BATTERY)
    .build()

val downloadTask = ModelManager.getInstance()
    .update(Constants.MODEL_NAME)
    .addConditions(modelDownloadConditions)
    .addOnFailureListener { e ->
        Timber.e(e, "Model download failed")
    }
    .addOnSuccessListener { model ->
        // Model downloaded successfully
        updateActiveClassifier(model)
    }

// Continue app operation while model downloads
downloadTask.addOnProgressListener { status ->
    progressBar.progress = status
}