--- name: android-telemetry-pipeline description: | High-frequency sensor data pipeline for Android: hybrid storage (Room for session metadata + CSV/binary files for sample data), StateFlow-based streaming to UI, session lifecycle management, and offline-first write patterns. Covers atomic session creation, buffered CSV writes, telemetry normalization across sources (BLE device vs phone GPS+IMU), and post-session data analysis. Use this when: ingesting 10-100Hz sensor data, designing a session storage schema, streaming live telemetry to a Compose ViewModel, building a session list/review flow, handling source switching mid-session, or implementing post-session analysis pipelines. license: Apache-2.0 metadata: author: Drew Fairweather last-updated: '2026-05-10' based-on: https://github.com/android/skills keywords: - telemetry - sensor pipeline - Room database - CSV storage - hybrid storage - StateFlow - session management - offline-first - high-frequency data - GPS - IMU - post-session analysis --- > **Lineage:** Built upon and expanded from the official Android group skills repository at > [github.com/android/skills](https://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 ```kotlin // 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. ```kotlin // app/build.gradle.kts plugins { alias(libs.plugins.ksp) } dependencies { ksp(libs.room.compiler) implementation(libs.room.runtime) implementation(libs.room.ktx) } ``` ## Session Entity ```kotlin @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 ```kotlin @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> // 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 ```kotlin @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): ```kotlin 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: ```kotlin private val writeBuffer = ConcurrentLinkedQueue() 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() while (true) frames.add(writeBuffer.poll() ?: break) if (frames.isEmpty()) return val csv = buildCsvContent(frames) File(currentCsvPath).appendText(csv) } private fun buildCsvContent(frames: List): 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 ```kotlin // 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> = db.sessionDao().getAllSessions() ``` ## ViewModel Wiring ```kotlin class SessionListViewModel(private val repo: SessionRepository) : ViewModel() { val sessions: StateFlow> = 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 ```kotlin 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: ```kotlin 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 { // Read CSV, skip header line, parse each row into TelemetryFrame return file.readLines().drop(1).mapNotNull { line -> runCatching { parseLine(line) }.getOrNull() } } } ``` ## Anti-Patterns | Symptom | Root cause | Fix | |---|---|---| | `kapt` annotation errors on Room | Using `kapt` instead of KSP | Switch to `ksp(libs.room.compiler)` | | Database locked / WAL errors | Writing sample rows to Room at 25Hz | Move samples to file; only metadata in Room | | Lost samples on process kill | Flushing only in `onDestroy` | Flush buffer in the stop-recording action handler | | Session list doesn't update | Using `suspend` query instead of `Flow` | `getAllSessions()` must return `Flow>` | | CSV grows unbounded | No size/age pruning | Prune sessions older than N days on app start | | Concurrent write corruption | Multiple coroutines writing to same file | Confine all file writes to a single `Dispatchers.IO` context or use `Mutex` |