Benchmark: Wear OS 5.0 vs watchOS 10 Step Counting Accuracy (2026)

// Wear OS 5.0 Step Counter Data Collector
// Hardware: Pixel Watch 2 (Wear OS 5.0 Build UTE2.230914.004)
// Dependencies: androidx.health:health-services-client:1.1.0
package com.benchmark.wearos.stepcounter

import android.content.Context
import android.hardware.Sensor
import android.hardware.SensorEvent
import android.hardware.SensorEventListener
import android.hardware.SensorManager
import android.util.Log
import androidx.health.services.client.HealthServices
import androidx.health.services.client.HealthServicesClient
import androidx.health.services.client.data.DataType
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import java.io.File
import java.io.FileWriter
import java.text.SimpleDateFormat
import java.util.Date
import java.util.Locale

class WearOS5StepCollector(private val context: Context) : SensorEventListener {
    private val sensorManager: SensorManager by lazy {
        context.getSystemService(Context.SENSOR_SERVICE) as SensorManager
    }
    private val healthClient: HealthServicesClient by lazy {
        HealthServices.getClient(context)
    }
    private var stepSensor: Sensor? = null
    private var totalSteps = 0
    private var lastStepTimestamp = 0L
    private val dateFormat = SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS", Locale.US)
    private val logFile = File(context.filesDir, "wearos_5_step_log.csv")

    init {
        // Initialize step counter sensor (TYPE_STEP_COUNTER returns cumulative steps since reboot)
        stepSensor = sensorManager.getDefaultSensor(Sensor.TYPE_STEP_COUNTER)
        if (stepSensor == null) {
            Log.e(TAG, "No step counter sensor available on device")
            throw UnsupportedOperationException("Step counter sensor not found")
        }
        // Register listener with normal sampling rate
        val registered = sensorManager.registerListener(
            this,
            stepSensor,
            SensorManager.SENSOR_DELAY_NORMAL
        )
        if (!registered) {
            Log.e(TAG, "Failed to register step sensor listener")
            throw IllegalStateException("Sensor registration failed")
        }
        // Initialize CSV log file with header
        if (!logFile.exists()) {
            logFile.writeText("timestamp,step_count,accuracy,device_build\n")
        }
    }

    override fun onSensorChanged(event: SensorEvent) {
        if (event.sensor.type != Sensor.TYPE_STEP_COUNTER) return
        val currentSteps = event.values[0].toInt()
        val eventTimestamp = event.timestamp / 1_000_000 // Convert ns to ms
        val accuracy = event.accuracy

        // Calculate delta steps (handle first reading as baseline)
        if (lastStepTimestamp == 0L) {
            totalSteps = currentSteps
            lastStepTimestamp = eventTimestamp
            logStep(eventTimestamp, currentSteps, accuracy)
            return
        }

        val delta = currentSteps - totalSteps
        if (delta < 0) {
            // Step counter reset (device reboot), reset baseline
            Log.w(TAG, "Step counter reset detected, resetting baseline")
            totalSteps = currentSteps
            lastStepTimestamp = eventTimestamp
            return
        }
        totalSteps = currentSteps
        lastStepTimestamp = eventTimestamp
        logStep(eventTimestamp, currentSteps, accuracy)
    }

    override fun onAccuracyChanged(sensor: Sensor, accuracy: Int) {
        Log.d(TAG, "Step sensor accuracy changed to: $accuracy")
    }

    private fun logStep(timestamp: Long, stepCount: Int, accuracy: Int) {
        val logLine = "${dateFormat.format(Date(timestamp))},$stepCount,$accuracy,UTE2.230914.004\n"
        try {
            FileWriter(logFile, true).use { it.write(logLine) }
        } catch (e: Exception) {
            Log.e(TAG, "Failed to write step log: ${e.message}")
        }
    }

    suspend fun getHealthServicesStepCount(): Int? = withContext(Dispatchers.IO) {
        try {
            val response = healthClient.aggregate(DataType.STEP_COUNT_TOTAL)
            response.dataPoints.firstOrNull()?.value?.toInt()
        } catch (e: Exception) {
            Log.e(TAG, "Health Services step count failed: ${e.message}")
            null
        }
    }

    fun cleanup() {
        sensorManager.unregisterListener(this)
        Log.d(TAG, "Step collector cleaned up, total steps logged: $totalSteps")
    }

    companion object {
        private const val TAG = "WearOS5StepCollector"
    }
}

// watchOS 10 Step Counter Data Collector
// Hardware: Apple Watch Series 9 (watchOS 10.0 Build 21R356)
// Dependencies: HealthKit, CoreMotion
import Foundation
import HealthKit
import CoreMotion

class WatchOS10StepCollector: NSObject, CMPedometerDelegate {
    private let pedometer = CMPedometer()
    private let healthStore = HKHealthStore()
    private var totalSteps: Int = 0
    private var startDate: Date?
    private let dateFormatter: DateFormatter = {
        let formatter = DateFormatter()
        formatter.dateFormat = "yyyy-MM-dd HH:mm:ss.SSS"
        formatter.locale = Locale(identifier: "en_US_POSIX")
        return formatter
    }()
    private let logFileURL: URL

    override init() {
        // Set up log file in app's documents directory
        let documentsDir = FileManager.default.urls(
            for: .documentDirectory,
            in: .userDomainMask
        ).first!
        logFileURL = documentsDir.appendingPathComponent("watchos_10_step_log.csv")
        super.init()
        // Initialize CSV header if file doesn't exist
        if !FileManager.default.fileExists(atPath: logFileURL.path) {
            do {
                try "timestamp,step_count,accuracy,device_build\n".write(
                    to: logFileURL,
                    atomically: true,
                    encoding: .utf8
                )
            } catch {
                print("Failed to initialize log file: \(error.localizedDescription)")
            }
        }
        // Request HealthKit authorization
        requestHealthKitAuth()
    }

    private func requestHealthKitAuth() {
        guard HKHealthStore.isHealthDataAvailable() else {
            print("HealthKit not available on device")
            return
        }
        let stepCountType = HKObjectType.quantityType(forIdentifier: .stepCount)!
        let readTypes: Set = [stepCountType]
        healthStore.requestAuthorization(toShare: nil, read: readTypes) { success, error in
            if let error = error {
                print("HealthKit auth failed: \(error.localizedDescription)")
                return
            }
            if success {
                print("HealthKit authorization granted")
                self.startPedometerUpdates()
            } else {
                print("HealthKit authorization denied")
            }
        }
    }

    func startPedometerUpdates() {
        guard CMPedometer.isStepCountingAvailable() else {
            print("Step counting not available on device")
            return
        }
        // Start pedometer updates from current date
        let now = Date()
        startDate = now
        pedometer.startUpdates(from: now) { [weak self] data, error in
            guard let self = self else { return }
            if let error = error {
                print("Pedometer update error: \(error.localizedDescription)")
                return
            }
            guard let data = data else { return }
            let steps = data.numberOfSteps.intValue
            let timestamp = data.startDate
            let accuracy = data.stepsAscending ? 3 : 1 // Map to Android accuracy scale (0-3)
            self.logStep(timestamp: timestamp, stepCount: steps, accuracy: accuracy)
            self.totalSteps = steps
        }
        pedometer.delegate = self
    }

    func pedometerDidUpdatePedometerData(_ pedometer: CMPedometer, data: CMPedometerData) {
        // Fallback delegate method for real-time updates
        let steps = data.numberOfSteps.intValue
        let timestamp = data.startDate
        logStep(timestamp: timestamp, stepCount: steps, accuracy: 3)
    }

    private func logStep(timestamp: Date, stepCount: Int, accuracy: Int) {
        let logLine = "\(dateFormatter.string(from: timestamp)),\(stepCount),\(accuracy),21R356\n"
        do {
            let fileHandle = try FileHandle(forWritingTo: logFileURL)
            fileHandle.seekToEndOfFile()
            fileHandle.write(logLine.data(using: .utf8)!)
            fileHandle.closeFile()
        } catch {
            print("Failed to write step log: \(error.localizedDescription)")
        }
    }

    func getHealthKitStepCount(completion: @escaping (Int?) -> Void) {
        guard let stepType = HKObjectType.quantityType(forIdentifier: .stepCount) else {
            completion(nil)
            return
        }
        let calendar = Calendar.current
        let endDate = Date()
        guard let startDate = calendar.date(byAdding: .day, value: -1, to: endDate) else {
            completion(nil)
            return
        }
        let predicate = HKQuery.predicateForSamples(withStart: startDate, end: endDate, options: .strictStartDate)
        let query = HKStatisticsQuery(
            quantityType: stepType,
            quantitySamplePredicate: predicate,
            options: .cumulativeSum
        ) { _, result, error in
            if let error = error {
                print("HealthKit query error: \(error.localizedDescription)")
                completion(nil)
                return
            }
            guard let sum = result?.sumQuantity() else {
                completion(nil)
                return
            }
            let steps = Int(sum.doubleValue(for: HKUnit.count()))
            completion(steps)
        }
        healthStore.execute(query)
    }

    func cleanup() {
        pedometer.stopUpdates()
        print("Step collector cleaned up, total steps logged: \(totalSteps)")
    }
}

# Step Count Benchmark Analysis Script
# Environment: Python 3.12, pandas 2.1.0, numpy 1.26.0
# Usage: python analyze_steps.py --wearos-log wearos_5_step_log.csv --watchos-log watchos_10_step_log.csv --ground-truth ground_truth.csv
import argparse
import pandas as pd
import numpy as np
from datetime import datetime
import logging
from typing import Tuple, Optional

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class StepBenchmarkAnalyzer:
    def __init__(self, wearos_log: str, watchos_log: str, ground_truth: str):
        self.wearos_log = wearos_log
        self.watchos_log = watchos_log
        self.ground_truth = ground_truth
        self.wearos_df: Optional[pd.DataFrame] = None
        self.watchos_df: Optional[pd.DataFrame] = None
        self.ground_truth_df: Optional[pd.DataFrame] = None

    def load_data(self) -> None:
        """Load and validate all input CSV files"""
        try:
            self.wearos_df = pd.read_csv(self.wearos_log, parse_dates=['timestamp'])
            logger.info(f"Loaded Wear OS log with {len(self.wearos_df)} entries")
        except Exception as e:
            logger.error(f"Failed to load Wear OS log: {e}")
            raise
        try:
            self.watchos_df = pd.read_csv(self.watchos_log, parse_dates=['timestamp'])
            logger.info(f"Loaded watchOS log with {len(self.watchos_df)} entries")
        except Exception as e:
            logger.error(f"Failed to load watchOS log: {e}")
            raise
        try:
            self.ground_truth_df = pd.read_csv(self.ground_truth, parse_dates=['timestamp'])
            logger.info(f"Loaded ground truth with {len(self.ground_truth_df)} entries")
        except Exception as e:
            logger.error(f"Failed to load ground truth: {e}")
            raise

        # Validate required columns
        required_cols = ['timestamp', 'step_count', 'accuracy', 'device_build']
        for df_name, df in [("Wear OS", self.wearos_df), ("watchOS", self.watchos_df)]:
            missing = [col for col in required_cols if col not in df.columns]
            if missing:
                raise ValueError(f"{df_name} log missing columns: {missing}")

        gt_required = ['timestamp', 'step_count']
        gt_missing = [col for col in gt_required if col not in self.ground_truth_df.columns]
        if gt_missing:
            raise ValueError(f"Ground truth missing columns: {gt_missing}")

    def align_timestamps(self, df: pd.DataFrame, reference_df: pd.DataFrame) -> pd.DataFrame:
        """Align device timestamps to ground truth using nearest 1-second window"""
        df = df.copy()
        # Round timestamps to nearest second for alignment
        df['aligned_ts'] = df['timestamp'].dt.round('1s')
        reference_df['aligned_ts'] = reference_df['timestamp'].dt.round('1s')
        # Merge on aligned timestamp
        merged = pd.merge(
            df,
            reference_df[['aligned_ts', 'step_count']],
            on='aligned_ts',
            how='inner',
            suffixes=('_device', '_ground')
        )
        logger.info(f"Aligned {len(merged)} entries for device data")
        return merged

    def calculate_accuracy(self, merged_df: pd.DataFrame) -> Tuple[float, float, float]:
        """Calculate accuracy metrics: MAE, RMSE, accuracy percentage"""
        if len(merged_df) == 0:
            logger.warning("No aligned entries to calculate accuracy")
            return 0.0, 0.0, 0.0
        # Calculate step delta (device - ground truth)
        merged_df['delta'] = merged_df['step_count_device'] - merged_df['step_count_ground']
        mae = np.mean(np.abs(merged_df['delta']))
        rmse = np.sqrt(np.mean(merged_df['delta'] ** 2))
        # Accuracy: 100 - (MAE / mean ground steps * 100)
        mean_ground = np.mean(merged_df['step_count_ground'])
        if mean_ground == 0:
            accuracy_pct = 0.0
        else:
            accuracy_pct = 100 - (mae / mean_ground * 100)
        return mae, rmse, max(0.0, accuracy_pct)

    def run_benchmark(self) -> dict:
        """Run full benchmark analysis and return results"""
        self.load_data()
        # Align both device datasets to ground truth
        wearos_aligned = self.align_timestamps(self.wearos_df, self.ground_truth_df)
        watchos_aligned = self.align_timestamps(self.watchos_df, self.ground_truth_df)
        # Calculate accuracy for each platform
        wearos_mae, wearos_rmse, wearos_acc = self.calculate_accuracy(wearos_aligned)
        watchos_mae, watchos_rmse, watchos_acc = self.calculate_accuracy(watchos_aligned)
        # Calculate battery impact (from separate battery log, mocked here for brevity)
        wearos_battery = 8.2  # % per 24h
        watchos_battery = 9.4  # % per 24h
        return {
            "wear_os_5": {
                "accuracy_pct": round(wearos_acc, 1),
                "mae": round(wearos_mae, 2),
                "rmse": round(wearos_rmse, 2),
                "battery_drain_pct": wearos_battery
            },
            "watch_os_10": {
                "accuracy_pct": round(watchos_acc, 1),
                "mae": round(watchos_mae, 2),
                "rmse": round(watchos_rmse, 2),
                "battery_drain_pct": watchos_battery
            },
            "sample_size": len(wearos_aligned) + len(watchos_aligned)
        }

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Analyze step count benchmark data")
    parser.add_argument("--wearos-log", required=True, help="Path to Wear OS step log CSV")
    parser.add_argument("--watchos-log", required=True, help="Path to watchOS step log CSV")
    parser.add_argument("--ground-truth", required=True, help="Path to ground truth step CSV")
    args = parser.parse_args()

    try:
        analyzer = StepBenchmarkAnalyzer(
            wearos_log=args.wearos_log,
            watchos_log=args.watchos_log,
            ground_truth=args.ground_truth
        )
        results = analyzer.run_benchmark()
        print("\n=== Benchmark Results ===")
        for platform, metrics in results.items():
            if platform == "sample_size":
                continue
            print(f"\n{platform.upper()}:")
            for k, v in metrics.items():
                print(f"  {k}: {v}")
        print(f"\nTotal aligned samples: {results['sample_size']}")
    except Exception as e:
        logger.error(f"Benchmark failed: {e}")
        exit(1)

// Register for multiple sensors for fusion
val accel = sensorManager.getDefaultSensor(Sensor.TYPE_ACCELEROMETER)
val gyro = sensorManager.getDefaultSensor(Sensor.TYPE_GYROSCOPE)
sensorManager.registerListener(fusionListener, accel, SensorManager.SENSOR_DELAY_GAME)
sensorManager.registerListener(fusionListener, gyro, SensorManager.SENSOR_DELAY_GAME)

// Batch HealthKit query for 24h steps
let startDate = Calendar.current.date(byAdding: .day, value: -1, to: Date())!
let predicate = HKQuery.predicateForSamples(withStart: startDate, end: Date())
let query = HKStatisticsQuery(quantityType: stepType, predicate: predicate, options: .cumulativeSum)

# Generate ground truth from treadmill speed
def generate_ground_truth(speed_mph: float, duration_min: int) -> int:
    steps_per_mile = 2000  # Average adult step length
    miles = speed_mph * (duration_min / 60)
    return int(miles * steps_per_mile)