Android Telemetry Pipeline

Lineage: Built upon and expanded from the official Android group skills repository at github.com/android/skills. That repo covers areas where evaluations show LLMs underperform on standard Android patterns. This skill extends that foundation into high-frequency telemetry pipelines — an area the official set does not cover.

Overview

Core principle: Room stores metadata only. Raw samples go to files.

Storing 25Hz telemetry rows in Room causes excessive WAL churn and query overhead. The split — SessionEntity (metadata) in Room + session_{id}.csv (samples) on disk — gives fast session list queries, simple backup/export, and zero SQLite write contention.

Storage Architecture

app/files/sessions/
├── session_20260510_143022.csv     ← raw telemetry samples (append-only)
├── session_20260510_151847.csv
└── ...

Room (sessions table):
├── id, startTime, endTime, sampleCount, csvFilePath
├── locationId, locationName
├── analysisComplete, analysisResult
└── (index on startTime, locationId, analysisComplete)

Dependencies

// libs.versions.toml
[versions]
room = "2.6.1"
ksp = "2.0.0-1.0.22"

[libraries]
room-runtime  = { group = "androidx.room", name = "room-runtime",  version.ref = "room" }
room-ktx      = { group = "androidx.room", name = "room-ktx",      version.ref = "room" }
room-compiler = { group = "androidx.room", name = "room-compiler", version.ref = "room" }

[plugins]
ksp = { id = "com.google.devtools.ksp", version.ref = "ksp" }

Always use KSP for Room annotation processing — kapt is deprecated.

// app/build.gradle.kts
plugins {
    alias(libs.plugins.ksp)
}
dependencies {
    ksp(libs.room.compiler)
    implementation(libs.room.runtime)
    implementation(libs.room.ktx)
}

Session Entity

@Entity(
    tableName = "sessions",
    indices = [
        Index("startTime"),
        Index("locationId"),
        Index("analysisComplete")
    ]
)
data class SessionEntity(
    @PrimaryKey val id: String,                // UUID string
    val startTime: Long,                       // epoch millis
    val endTime: Long?,
    val sampleCount: Int,
    val csvFilePath: String,                   // absolute path
    val locationId: String?,
    val locationName: String?,
    val analysisComplete: Boolean = false,
    val analysisResult: String?                // summary output from post-session analysis
)

Never store sample data (lat, lon, speed, g-forces) in Room rows — use the CSV file. Only store aggregates in the entity if you query on them (e.g., maxSpeedKph for a leaderboard).

DAO

@Dao
interface SessionDao {
    @Insert(onConflict = OnConflictStrategy.REPLACE)
    suspend fun insert(session: SessionEntity)

    @Update
    suspend fun update(session: SessionEntity)

    @Query("SELECT * FROM sessions ORDER BY startTime DESC")
    fun getAllSessions(): Flow<List<SessionEntity>>   // Flow for reactive list

    @Query("SELECT * FROM sessions WHERE id = :id")
    suspend fun getById(id: String): SessionEntity?

    @Query("UPDATE sessions SET endTime = :endTime, sampleCount = :count WHERE id = :id")
    suspend fun finalizeSession(id: String, endTime: Long, count: Int)

    @Query("UPDATE sessions SET analysisComplete = 1, analysisResult = :result WHERE id = :id")
    suspend fun markAnalysisComplete(id: String, result: String?)
}

Database

@Database(entities = [SessionEntity::class], version = 2, exportSchema = false)
abstract class AppDatabase : RoomDatabase() {
    abstract fun sessionDao(): SessionDao

    companion object {
        @Volatile private var instance: AppDatabase? = null

        fun getInstance(context: Context): AppDatabase =
            instance ?: synchronized(this) {
                instance ?: Room.databaseBuilder(
                    context.applicationContext,
                    AppDatabase::class.java,
                    "sessions.db"
                )
                .fallbackToDestructiveMigration()   // dev only; use proper migrations in production
                .build().also { instance = it }
            }
    }
}

Use fallbackToDestructiveMigration() only during development. Before shipping, add a proper Migration object for each version bump.

Telemetry Normalization

Define a common data class regardless of source (BLE device vs phone sensors):

data class TelemetryFrame(
    val timestampMs: Long,          // epoch millis
    val source: TelemetrySource,    // BLE, PHONE
    val latDeg: Double,
    val lonDeg: Double,
    val speedKph: Float,
    val headingDeg: Float,
    val altitudeM: Float,
    val accelX: Float,              // m/s² lateral
    val accelY: Float,              // m/s² longitudinal
    val accelZ: Float,              // m/s² vertical
    val satellites: Int = 0,
    val hdop: Float = 0f            // horizontal dilution of precision
)

enum class TelemetrySource { BLE, PHONE }

Map all hardware-specific data types to TelemetryFrame in the source layer (BLE manager or sensor manager). The ViewModel, CSV writer, and analysis pipeline see only TelemetryFrame.

Buffered CSV Writer

Write samples to a ConcurrentLinkedQueue buffer, flush to disk periodically to avoid I/O on every sample at 25Hz:

private val writeBuffer = ConcurrentLinkedQueue<TelemetryFrame>()
private val sampleCounter = AtomicInteger(0)

fun onTelemetryReceived(frame: TelemetryFrame) {
    writeBuffer.add(frame)
    val count = sampleCounter.incrementAndGet()
    _sampleCount.value = count

    // Flush every 250 samples (~10 seconds at 25Hz)
    if (count % 250 == 0) {
        serviceScope.launch(Dispatchers.IO) { flushBuffer() }
    }
}

private fun flushBuffer() {
    val frames = mutableListOf<TelemetryFrame>()
    while (true) frames.add(writeBuffer.poll() ?: break)
    if (frames.isEmpty()) return

    val csv = buildCsvContent(frames)
    File(currentCsvPath).appendText(csv)
}

private fun buildCsvContent(frames: List<TelemetryFrame>): String =
    buildString {
        for (f in frames) {
            append("${f.timestampMs},${f.source},${f.latDeg},${f.lonDeg},")
            append("${f.speedKph},${f.headingDeg},${f.altitudeM},")
            append("${f.accelX},${f.accelY},${f.accelZ},")
            append("${f.satellites},${f.hdop}\n")
        }
    }

Write the CSV header once at session creation, not on every flush.

Session Lifecycle

// SessionRepository
suspend fun createSession(csvFilePath: String, sessionId: String? = null): SessionEntity {
    val id = sessionId ?: UUID.randomUUID().toString()
    val entity = SessionEntity(
        id = id,
        startTime = Instant.now().toEpochMilli(),
        endTime = null,
        sampleCount = 0,
        csvFilePath = csvFilePath,
        locationId = null,
        locationName = null
    )
    db.sessionDao().insert(entity)
    return entity
}

suspend fun finalizeSession(id: String, endTime: Instant, sampleCount: Int) {
    db.sessionDao().finalizeSession(id, endTime.toEpochMilli(), sampleCount)
}

suspend fun markAnalysisComplete(id: String, result: String?) {
    db.sessionDao().markAnalysisComplete(id, result)
}

fun getAllSessions(): Flow<List<SessionEntity>> = db.sessionDao().getAllSessions()

ViewModel Wiring

class SessionListViewModel(private val repo: SessionRepository) : ViewModel() {
    val sessions: StateFlow<List<SessionEntity>> = repo.getAllSessions()
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())
}

class RecordingViewModel : ViewModel() {
    val sampleCount = TelemetryRecordingService.sampleCount
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), 0)

    val activeSource = TelemetryRecordingService.activeSource
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), TelemetrySource.PHONE)
}

Use SharingStarted.WhileSubscribed(5_000) — the 5-second timeout keeps the flow alive through configuration changes (rotation) without leaking indefinitely.

CSV File Management

fun createCsvFile(context: Context, sessionId: String): File {
    val dir = File(context.filesDir, "sessions").apply { mkdirs() }
    val file = File(dir, "session_$sessionId.csv")
    // Write header
    file.writeText("timestampMs,source,lat,lon,speedKph,heading,altitude,accelX,accelY,accelZ,sats,hdop\n")
    return file
}

Use context.filesDir (internal storage) — no permissions required, automatically backed up if android:allowBackup="true", and excluded from media scans. Never use Environment.getExternalStorageDirectory() for telemetry data.

Post-Session Analysis Pipeline

Trigger analysis after finalizeSession. Run on a background dispatcher, update Room when done:

class PostSessionAnalyzer(private val repo: SessionRepository) {
    suspend fun analyze(session: SessionEntity) = withContext(Dispatchers.Default) {
        val frames = parseCsv(File(session.csvFilePath))
        val result = computeSummary(frames)   // app-specific: max speed, distance, etc.
        repo.markAnalysisComplete(session.id, result)
    }

    private fun parseCsv(file: File): List<TelemetryFrame> {
        // Read CSV, skip header line, parse each row into TelemetryFrame
        return file.readLines().drop(1).mapNotNull { line ->
            runCatching { parseLine(line) }.getOrNull()
        }
    }
}

Anti-Patterns