From 56cbce0d5b10efee0b1722b753f80eed8f64f849 Mon Sep 17 00:00:00 2001 From: r1ddh118 Date: Thu, 16 Apr 2026 13:37:19 +0530 Subject: [PATCH 1/3] refactor: replace two-stage CSV parsing with a single-pass streaming pipeline to reduce memory usage --- .../splats/app/telemetry/CsvIngestParser.kt | 162 ++++++++---------- .../splats/app/telemetry/KeyframePlanner.kt | 16 +- .../app/telemetry/TelemetryPreprocessor.kt | 22 ++- 3 files changed, 87 insertions(+), 113 deletions(-) diff --git a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt index 2696531e..78450c38 100644 --- a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt +++ b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt @@ -27,21 +27,22 @@ private object ColumnMap { val SATELLITES = HeaderAliases(listOf("satellites")) } -// ─── CSV Ingest ─────────────────────────────────────────────────────────────── - +// ─── CSV Stream Pipeline ──────────────────────────────────────────────────────── /** - * Stage 1 — Raw Ingest. - * - * Reads the file as a plain string matrix (rows × columns). - * Handles UTF-8 BOM, CRLF/LF line endings, and preamble rows that appear - * before the actual header row in older DJI firmware exports. - * - * No type parsing is performed here. + * Single-pass streaming parser. + * Reads line by line, identifies headers, resolves columns, parses into [TelRow], + * and applies the timestamp filter immediately, preventing OOM on large files. */ -internal object CsvIngest { +internal object CsvIngestParser { private const val TAG = "TelemetryCsv" + private val filenameDateTimeRegex = Regex("""(\d{4}-\d{2}-\d{2})[ _](\d{2})[:_-](\d{2})[:_-](\d{2})""") + private val filenameTimeRegex = Regex("""(?, List>> { + fun streamAndParse( + file: File, + videoFileName: String, + videoStartUsFromMetadata: Long? + ): Triple, Int> { if (!file.exists()) throw TelemetryError.CsvNotFound(file.absolutePath) val ext = file.extension.lowercase() @@ -49,13 +50,16 @@ internal object CsvIngest { val label = if (ext.isBlank()) "unknown" else ext throw TelemetryError.UnsupportedFormat(label) } - return readCsv(file) - } - private fun readCsv(file: File): Pair, List>> { - val dataRows = mutableListOf>() var headers: List? = null var firstNonEmpty: String? = null + var parser: ((List) -> TelRow?)? = null + + var baseDateUs: Long? = null + var targetStartUs: Long? = videoStartUsFromMetadata + + val rows = mutableListOf() + var dataRowsSize = 0 file.inputStream().bufferedReader(Charsets.UTF_8).use { reader -> while (true) { @@ -67,97 +71,48 @@ internal object CsvIngest { firstNonEmpty = line } - // Locate the header row: first line that looks like a CSV header - // (contains a comma and at least one recognisable keyword). + // Locate the header row if (headers == null) { if (line.contains(',') && looksLikeHeader(line)) { val headerLine = splitCsvLine(line) headers = headerLine + parser = buildRowParser(headerLine) Log.i(TAG, "CSV headers: ${headerLine.joinToString("|")}") } continue } if (line.contains(',')) { - dataRows += splitCsvLine(line) - } - } - } - - val headersFinal = headers ?: run { - Log.e(TAG, "CSV: could not find header row. First line: $firstNonEmpty") - throw TelemetryError.InsufficientRecords(0) - } - - return Pair(headersFinal, dataRows) - } - - // A line "looks like" a CSV header if it has at least one recognisable keyword. - private fun looksLikeHeader(line: String): Boolean { - val cells = splitCsvLine(line) - return looksLikeHeaderCells(cells) - } - - private fun splitCsvLine(line: String): List = - buildList { - val current = StringBuilder() - var inQuotes = false + val cols = splitCsvLine(line) + dataRowsSize++ + val row = parser?.invoke(cols) + if (row != null) { + if (baseDateUs == null && row.timestampUs > 0L) { + baseDateUs = row.timestampUs + } + + if (targetStartUs == null && baseDateUs != null) { + targetStartUs = parseStartTimeFromFilename(videoFileName, baseDateUs!!) + } - for (ch in line) { - when (ch) { - '"' -> inQuotes = !inQuotes - ',' -> { - if (inQuotes) { - current.append(ch) - } else { - add(current.toString().trim()) - current.setLength(0) + val actualStartUs = targetStartUs ?: 0L + if (actualStartUs == 0L || row.timestampUs >= actualStartUs) { + rows.add(row) } } - else -> current.append(ch) } } - - add(current.toString().trim()) } -} - -// ─── CSV Parser ─────────────────────────────────────────────────────────────── - -/** - * Stage 2 — Type-safe parsing. - * - * Resolves Litchi header names to column indices. - * Coerces raw strings into typed [TelRow] instances. - * Litchi "datetime(utc)" strings are converted to microseconds since epoch. - */ -internal object CsvParser { - private const val TAG = "TelemetryCsv" - private val filenameDateTimeRegex = Regex("""(\d{4}-\d{2}-\d{2})[ _](\d{2})[:_-](\d{2})[:_-](\d{2})""") - private val filenameTimeRegex = Regex("""(?, - dataRows: List>, - targetStartUs: Long = 0L - ): List { - Log.i(TAG, "Parse headers raw: ${headers.joinToString("|")}") - Log.i(TAG, "Parse headers norm: ${headers.map { normalizeHeader(it) }.joinToString("|")}") - return try { - val rows = parseInternal(headers, dataRows) - if (targetStartUs > 0) { - Log.i(TAG, "Filtering CSV to start at time $targetStartUs") - rows.filter { it.timestampUs >= targetStartUs } - } else { - rows - } - } catch (e: TelemetryError.InsufficientRecords) { + if (headers == null) { + Log.e(TAG, "CSV: could not find header row. First line: $firstNonEmpty") throw TelemetryError.InsufficientRecords(0) } + + return Triple(targetStartUs ?: 0L, rows, dataRowsSize) } - fun parseStartTimeFromFilename(filename: String, rows: List): Long { + private fun parseStartTimeFromFilename(filename: String, baseDateUs: Long): Long { val dateTimeMatch = filenameDateTimeRegex.find(filename) if (dateTimeMatch != null) { val (date, hr, min, sec) = dateTimeMatch.destructured @@ -170,7 +125,7 @@ internal object CsvParser { } } - val baseDateUs = rows.firstOrNull { it.timestampUs > 0L }?.timestampUs ?: return 0L + if (baseDateUs == 0L) return 0L val timeMatch = filenameTimeRegex.find(filename) ?: return 0L val (hr, min, sec) = timeMatch.destructured val sdf = SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.US) @@ -185,10 +140,7 @@ internal object CsvParser { } } - private fun parseInternal( - headers: List, - dataRows: List> - ): List { + private fun buildRowParser(headers: List): (List) -> TelRow? { val lower = headers.map { normalizeHeader(it) } fun resolve(headerAliases: HeaderAliases): Int? { @@ -220,7 +172,7 @@ internal object CsvParser { val hdopHeader = iHdop?.let { lower[it] } val fixTypeHeader = iFixType?.let { lower[it] } - return dataRows.mapNotNull { cols -> + return { cols: List -> runCatching { val tsUs = parseLitchiDateTime(cols.getOrElse(iTs) { "" }) TelRow( @@ -305,6 +257,34 @@ internal object CsvParser { 0L } } + // A line "looks like" a CSV header if it has at least one recognisable keyword. + private fun looksLikeHeader(line: String): Boolean { + val cells = splitCsvLine(line) + return looksLikeHeaderCells(cells) + } + + private fun splitCsvLine(line: String): List = + buildList { + val current = StringBuilder() + var inQuotes = false + + for (ch in line) { + when (ch) { + '"' -> inQuotes = !inQuotes + ',' -> { + if (inQuotes) { + current.append(ch) + } else { + add(current.toString().trim()) + current.setLength(0) + } + } + else -> current.append(ch) + } + } + + add(current.toString().trim()) + } } private fun looksLikeHeaderCells(cells: List): Boolean { diff --git a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/KeyframePlanner.kt b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/KeyframePlanner.kt index 1c6b8939..472bd71a 100644 --- a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/KeyframePlanner.kt +++ b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/KeyframePlanner.kt @@ -21,16 +21,12 @@ object KeyframePlanner { ): LongArray { require(maxOutputFrames >= 1) { "maxOutputFrames must be >= 1" } - val (headers, dataRows) = CsvIngest.read(csvFile) - val parsedRows = CsvParser.parse(headers, dataRows) - - val targetStartUs = readVideoFileStartTimeUs(videoFile) - ?: CsvParser.parseStartTimeFromFilename(videoFile.name, parsedRows) - val rawRows = if (targetStartUs > 0L) { - parsedRows.filter { it.timestampUs >= targetStartUs } - } else { - parsedRows - } + val startUsMetadata = readVideoFileStartTimeUs(videoFile) + val (targetStartUs, rawRows, _) = CsvIngestParser.streamAndParse( + csvFile, + videoFile.name, + startUsMetadata + ) val validRows = RowValidator.validate(rawRows) if (validRows.isEmpty()) return longArrayOf() diff --git a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/TelemetryPreprocessor.kt b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/TelemetryPreprocessor.kt index f92afcee..8c15c235 100644 --- a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/TelemetryPreprocessor.kt +++ b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/TelemetryPreprocessor.kt @@ -123,23 +123,21 @@ class TelemetryPreprocessor @JvmOverloads constructor( if (!csvFile.exists()) throw TelemetryError.CsvNotFound(csvFile.absolutePath) if (!videoFile.exists()) throw TelemetryError.VideoNotFound(videoFile.absolutePath) - // Stage 1–2: Ingest + single-pass parsing (strictly Litchi) + // Stage 1–2: Single-pass streaming ingest + parse (strictly Litchi) reportProgress(ProcessingStage.PARSING, 0.0f) - val (headers, dataRows) = CsvIngest.read(csvFile) - val parsedRows = CsvParser.parse(headers, dataRows) - val targetStartUs = readVideoFileStartTimeUs(videoFile) - ?: CsvParser.parseStartTimeFromFilename(videoFile.name, parsedRows) + val startUsMetadata = readVideoFileStartTimeUs(videoFile) + val (targetStartUs, rawRows, totalCsvRows) = CsvIngestParser.streamAndParse( + csvFile, + videoFile.name, + startUsMetadata + ) + if (targetStartUs > 0L) { Log.i(logTag, "Using telemetry start time ${targetStartUs}us for ${videoFile.name}") } else { Log.w(logTag, "Could not derive video start time from metadata or filename for ${videoFile.name}") } - val rawRows = if (targetStartUs > 0L) { - parsedRows.filter { it.timestampUs >= targetStartUs } - } else { - parsedRows - } - Log.i(logTag, "Parsed ${dataRows.size} rows, retained ${rawRows.size} after time filter.") + Log.i(logTag, "Parsed $totalCsvRows rows, retained ${rawRows.size} after time filter.") reportProgress(ProcessingStage.PARSING, 1.0f) // Stage 3: Row validation @@ -212,7 +210,7 @@ class TelemetryPreprocessor @JvmOverloads constructor( reportProgress(ProcessingStage.EMITTING, 1.0f) val report = TelemetryProcessingReport( - totalCsvRows = rawRows.size, + totalCsvRows = totalCsvRows, rejectedRows = rejectedRows, totalKeyframes = kfTimestampsUs.size, outputFrames = qResult.retained.size, From ffcad2ad3c09eef70a489605d3962e94d889c5ff Mon Sep 17 00:00:00 2001 From: r1ddh118 Date: Thu, 16 Apr 2026 14:06:34 +0530 Subject: [PATCH 2/3] refactor: replace custom CSV parser with Apache Commons CSV and extract video timestamp logic into a dedicated utility class. --- crates/brush-app/app/build.gradle | 3 + .../splats/app/telemetry/CsvIngestParser.kt | 302 ++++++++---------- .../splats/app/telemetry/KeyframePlanner.kt | 11 +- .../app/telemetry/TelemetryPreprocessor.kt | 23 +- .../app/telemetry/VideoTimeExtractor.kt | 45 +++ 5 files changed, 208 insertions(+), 176 deletions(-) create mode 100644 crates/brush-app/app/src/main/java/com/splats/app/telemetry/VideoTimeExtractor.kt diff --git a/crates/brush-app/app/build.gradle b/crates/brush-app/app/build.gradle index aee9bef1..cd091d83 100644 --- a/crates/brush-app/app/build.gradle +++ b/crates/brush-app/app/build.gradle @@ -118,6 +118,9 @@ dependencies { // JSON output for telemetry implementation "com.google.code.gson:gson:2.10.1" + // CSV parsing + implementation "org.apache.commons:commons-csv:1.10.0" + // To use the Games Controller Library //implementation "androidx.games:games-controller:1.1.0" diff --git a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt index 78450c38..4ed38c60 100644 --- a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt +++ b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt @@ -1,10 +1,12 @@ package com.splats.app.telemetry +import android.util.Log +import org.apache.commons.csv.CSVFormat +import org.apache.commons.csv.CSVRecord import java.io.File import java.text.SimpleDateFormat import java.util.Locale import java.util.TimeZone -import android.util.Log // ─── Column name mapping ────────────────────────────────────────────────────── @@ -27,22 +29,23 @@ private object ColumnMap { val SATELLITES = HeaderAliases(listOf("satellites")) } +class CsvParseContext(var targetStartUs: Long = 0L, var totalSourceRows: Int = 0, var malformedRows: Int = 0) + // ─── CSV Stream Pipeline ──────────────────────────────────────────────────────── /** - * Single-pass streaming parser. - * Reads line by line, identifies headers, resolves columns, parses into [TelRow], - * and applies the timestamp filter immediately, preventing OOM on large files. + * Single-pass streaming parser providing a safe scope for the CSV reader. + * Pre-scans for headers (bypassing preambles), initializes commons-csv with strict config, + * and maintains determinism using the first valid timestamp dynamically. */ internal object CsvIngestParser { private const val TAG = "TelemetryCsv" - private val filenameDateTimeRegex = Regex("""(\d{4}-\d{2}-\d{2})[ _](\d{2})[:_-](\d{2})[:_-](\d{2})""") - private val filenameTimeRegex = Regex("""(? streamAndParse( file: File, videoFileName: String, - videoStartUsFromMetadata: Long? - ): Triple, Int> { + videoStartUsFromMetadata: Long?, + block: (CsvParseContext, Sequence) -> R + ): R { if (!file.exists()) throw TelemetryError.CsvNotFound(file.absolutePath) val ext = file.extension.lowercase() @@ -51,171 +54,181 @@ internal object CsvIngestParser { throw TelemetryError.UnsupportedFormat(label) } - var headers: List? = null - var firstNonEmpty: String? = null - var parser: ((List) -> TelRow?)? = null - - var baseDateUs: Long? = null - var targetStartUs: Long? = videoStartUsFromMetadata - - val rows = mutableListOf() - var dataRowsSize = 0 - - file.inputStream().bufferedReader(Charsets.UTF_8).use { reader -> + // Bound to the entire function scope! No stream leakage allowed. + return file.inputStream().bufferedReader(Charsets.UTF_8).use { reader -> + var headerLine: Array? = null + + // 1. The Seeker Logic: Consume and discard preamble until header + reader.mark(1024 * 64) // 64kb max lookahead for preamble checking while (true) { val rawLine = reader.readLine() ?: break - var line = rawLine.trim() + val line = rawLine.trim().removePrefix("\uFEFF") if (line.isEmpty()) continue - if (firstNonEmpty == null) { - line = line.removePrefix("\uFEFF") // strip UTF-8 BOM on first line - firstNonEmpty = line + + val isHeader = line.contains("latitude", ignoreCase = true) + || line.contains("datetime(utc)", ignoreCase = true) + + if (isHeader) { + // Extract exactly what the file stated, keeping case. + // No spaces/bom trimming here; commons-csv format controls it. + val cells = splitPreCheckLine(line) + headerLine = cells.toTypedArray() + break } + } + + if (headerLine == null) { + Log.e(TAG, "CSV: could not find header row. Malformed or purely preamble.") + throw TelemetryError.InsufficientRecords(0) + } - // Locate the header row - if (headers == null) { - if (line.contains(',') && looksLikeHeader(line)) { - val headerLine = splitCsvLine(line) - headers = headerLine - parser = buildRowParser(headerLine) - Log.i(TAG, "CSV headers: ${headerLine.joinToString("|")}") + // 2. Format with strict consistency protocols + val format = CSVFormat.DEFAULT.builder() + .setHeader(*headerLine) + .setAllowMissingColumnNames(false) + .setIgnoreHeaderCase(true) + .setTrim(true) + .build() + + val csvParser = format.parse(reader) + + // Build the row parser exactly using mapped names dynamically + val headerMap = csvParser.headerMap ?: emptyMap() + val rowParser = buildRowParser(headerMap) + + Log.i(TAG, "CSV headers detected successfully: ${headerLine.size} columns") + + val context = CsvParseContext(targetStartUs = videoStartUsFromMetadata ?: 0L) + var baseDateUs: Long? = null + + // 3. Dynamic sequence implementation + val rowSequence = sequence { + for (record in csvParser) { + context.totalSourceRows++ + // Strict Leniency Rule - drop mismatched columns immediately + if (!record.isConsistent) { + Log.w(TAG, "Dropping malformed CSV row (inconsistent column size): line ${record.recordNumber}") + context.malformedRows++ + continue } - continue - } - if (line.contains(',')) { - val cols = splitCsvLine(line) - dataRowsSize++ - val row = parser?.invoke(cols) - if (row != null) { - if (baseDateUs == null && row.timestampUs > 0L) { - baseDateUs = row.timestampUs - } - - if (targetStartUs == null && baseDateUs != null) { - targetStartUs = parseStartTimeFromFilename(videoFileName, baseDateUs!!) - } + val row = rowParser(record) + if (row == null) { + context.malformedRows++ + continue + } - val actualStartUs = targetStartUs ?: 0L - if (actualStartUs == 0L || row.timestampUs >= actualStartUs) { - rows.add(row) - } + // Determine Target Start Time once statically + if (baseDateUs == null && row.timestampUs > 0L) { + baseDateUs = row.timestampUs + } + if (videoStartUsFromMetadata == null && context.targetStartUs == 0L && baseDateUs != null) { + context.targetStartUs = VideoTimeExtractor.extractTargetStartUs(videoFileName, baseDateUs!!) + } + + val activeFilterUs = context.targetStartUs + if (activeFilterUs == 0L || row.timestampUs >= activeFilterUs) { + yield(row) } } } - } - if (headers == null) { - Log.e(TAG, "CSV: could not find header row. First line: $firstNonEmpty") - throw TelemetryError.InsufficientRecords(0) + block(context, rowSequence) } - - return Triple(targetStartUs ?: 0L, rows, dataRowsSize) } - private fun parseStartTimeFromFilename(filename: String, baseDateUs: Long): Long { - val dateTimeMatch = filenameDateTimeRegex.find(filename) - if (dateTimeMatch != null) { - val (date, hr, min, sec) = dateTimeMatch.destructured - val sdf = SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.US) - sdf.timeZone = TimeZone.getTimeZone("UTC") - return try { - sdf.parse("$date $hr:$min:$sec")?.time?.times(1_000L) ?: 0L - } catch (e: Exception) { - 0L - } - } - - if (baseDateUs == 0L) return 0L - val timeMatch = filenameTimeRegex.find(filename) ?: return 0L - val (hr, min, sec) = timeMatch.destructured - val sdf = SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.US) - sdf.timeZone = TimeZone.getTimeZone("UTC") - return try { - val dayStartUs = (baseDateUs / 86_400_000_000L) * 86_400_000_000L - val dayStart = sdf.format(java.util.Date(dayStartUs / 1_000L)) - .substring(0, 10) - sdf.parse("$dayStart $hr:$min:$sec")?.time?.times(1_000L) ?: 0L - } catch (e: Exception) { - 0L - } + private fun splitPreCheckLine(line: String): List { + // Fast split just to identify the initial header array schema + return line.split(',').map { it.trim().removePrefix("\"").removeSuffix("\"") } } - private fun buildRowParser(headers: List): (List) -> TelRow? { - val lower = headers.map { normalizeHeader(it) } + private fun buildRowParser(headerMap: Map): (CSVRecord) -> TelRow? { + // Prepare normalized mapping table + val normalizedHeaders = headerMap.entries.associate { (k, v) -> + normalizeHeader(k) to v + } fun resolve(headerAliases: HeaderAliases): Int? { for (alias in headerAliases.aliases) { - val idx = lower.indexOf(normalizeHeader(alias)) - if (idx >= 0) return idx + val normAlias = normalizeHeader(alias) + if (normalizedHeaders.containsKey(normAlias)) return normalizedHeaders[normAlias] } return null } - val iTs = resolve(ColumnMap.TIMESTAMP) - ?: throw TelemetryError.InsufficientRecords(0) - val iLat = resolve(ColumnMap.LAT) ?: throw TelemetryError.InsufficientRecords(0) - val iLon = resolve(ColumnMap.LON) ?: throw TelemetryError.InsufficientRecords(0) - val iAlt = resolve(ColumnMap.ALT) ?: throw TelemetryError.InsufficientRecords(0) - val iHeading = resolve(ColumnMap.HEADING) ?: throw TelemetryError.InsufficientRecords(0) + val iTs = resolve(ColumnMap.TIMESTAMP) ?: throw TelemetryError.InsufficientRecords(0) + val iLat = resolve(ColumnMap.LAT) ?: throw TelemetryError.InsufficientRecords(0) + val iLon = resolve(ColumnMap.LON) ?: throw TelemetryError.InsufficientRecords(0) + val iAlt = resolve(ColumnMap.ALT) ?: throw TelemetryError.InsufficientRecords(0) + val iHeading = resolve(ColumnMap.HEADING) ?: throw TelemetryError.InsufficientRecords(0) val iPitch = resolve(ColumnMap.GIMBAL_PITCH)?: throw TelemetryError.InsufficientRecords(0) - val iVelN = resolve(ColumnMap.VEL_N) ?: throw TelemetryError.InsufficientRecords(0) - val iVelE = resolve(ColumnMap.VEL_E) ?: throw TelemetryError.InsufficientRecords(0) - val iVelD = resolve(ColumnMap.VEL_D) ?: throw TelemetryError.InsufficientRecords(0) + val iVelN = resolve(ColumnMap.VEL_N) ?: throw TelemetryError.InsufficientRecords(0) + val iVelE = resolve(ColumnMap.VEL_E) ?: throw TelemetryError.InsufficientRecords(0) + val iVelD = resolve(ColumnMap.VEL_D) ?: throw TelemetryError.InsufficientRecords(0) val iHdop = resolve(ColumnMap.HDOP) val iFixType = resolve(ColumnMap.FIX_TYPE) val iSatellites = resolve(ColumnMap.SATELLITES) - val altHeader = lower[iAlt] - val velNHeader = lower[iVelN] - val velEHeader = lower[iVelE] - val velDHeader = lower[iVelD] - val hdopHeader = iHdop?.let { lower[it] } - val fixTypeHeader = iFixType?.let { lower[it] } + val revMap = headerMap.entries.associate { it.value to it.key } + val altHeader = revMap[iAlt] ?: "" + val velNHeader = revMap[iVelN] ?: "" + val velEHeader = revMap[iVelE] ?: "" + val velDHeader = revMap[iVelD] ?: "" + val hdopHeader = iHdop?.let { revMap[it] } ?: "" + val fixTypeHeader = iFixType?.let { revMap[it] } ?: "" + + return { cols: CSVRecord -> + fun getCol(idx: Int, default: String = "0"): String { + return if (idx >= 0 && idx < cols.size()) { + val raw = cols.get(idx).trim() + if (raw.isBlank()) default else raw + } else default + } - return { cols: List -> runCatching { - val tsUs = parseLitchiDateTime(cols.getOrElse(iTs) { "" }) + val tsUs = parseLitchiDateTime(getCol(iTs, "")) TelRow( timestampUs = tsUs, - lat = cols.getOrElse(iLat) { "0" }.toDouble(), - lon = cols.getOrElse(iLon) { "0" }.toDouble(), - altM = parseAltitudeMeters(cols.getOrElse(iAlt) { "0" }, altHeader), - headingDeg = cols.getOrElse(iHeading) { "0" }.toDouble(), - gimbalPitch = cols.getOrElse(iPitch) { "0" }.toDouble(), - velN = parseSpeedMps(cols.getOrElse(iVelN) { "0" }, velNHeader), - velE = parseSpeedMps(cols.getOrElse(iVelE) { "0" }, velEHeader), - velD = parseSpeedMps(cols.getOrElse(iVelD) { "0" }, velDHeader), + lat = getCol(iLat).toDouble(), + lon = getCol(iLon).toDouble(), + altM = parseAltitudeMeters(getCol(iAlt), altHeader), + headingDeg = getCol(iHeading).toDouble(), + gimbalPitch = getCol(iPitch).toDouble(), + velN = parseSpeedMps(getCol(iVelN), velNHeader), + velE = parseSpeedMps(getCol(iVelE), velEHeader), + velD = parseSpeedMps(getCol(iVelD), velDHeader), hdop = iHdop?.let { - parseHdop(cols.getOrElse(it) { "0" }, hdopHeader ?: "") + parseHdop(getCol(it), hdopHeader) } ?: 1.0, fixType = iFixType?.let { - parseFixType(cols.getOrElse(it) { "0" }, fixTypeHeader ?: "") + parseFixType(getCol(it), fixTypeHeader) } ?: 3, satellites = iSatellites?.let { - cols.getOrElse(it) { "0" }.toIntOrNull() ?: 0 + getCol(it).toIntOrNull() ?: 0 } ?: 0 ) - }.getOrNull() // malformed rows are dropped; row validator counts them + }.getOrNull() } } private fun parseAltitudeMeters(raw: String, header: String): Double { val value = raw.toDoubleOrNull() ?: return 0.0 - return if (header.contains("[ft]") || header.contains("_ft")) value * 0.3048 else value + return if (header.contains("[ft]", ignoreCase = true) || header.contains("_ft", ignoreCase = true)) + value * 0.3048 else value } private fun parseSpeedMps(raw: String, header: String): Double { val value = raw.toDoubleOrNull() ?: return 0.0 - return if (header.contains("[mph]")) value * 0.44704 else value + return if (header.contains("[mph]", ignoreCase = true)) value * 0.44704 else value } private fun parseHdop(raw: String, header: String): Double { val value = raw.toDoubleOrNull() ?: return 99.0 return when { - header.contains("gpslevel") -> { + header.contains("gpslevel", ignoreCase = true) -> { if (value <= 0.0) 99.0 else (6.0 - value).coerceIn(0.8, 5.0) } - header.contains("gpsnum") -> when { + header.contains("gpsnum", ignoreCase = true) -> when { value >= 20.0 -> 0.9 value >= 15.0 -> 1.5 value >= 10.0 -> 2.5 @@ -228,7 +241,7 @@ internal object CsvIngestParser { private fun parseFixType(raw: String, header: String): Int { val value = raw.toIntOrNull() ?: raw.toDoubleOrNull()?.toInt() ?: return 3 - return if (header.contains("gpslevel")) { + return if (header.contains("gpslevel", ignoreCase = true)) { when { value >= 4 -> 3 value >= 2 -> 2 @@ -240,12 +253,7 @@ internal object CsvIngestParser { } } - /** - * Parse Litchi datetime strings of the form "2024-06-01 10:23:45.456" - * into microseconds since Unix epoch. - */ private fun parseLitchiDateTime(raw: String): Long { - // Format: yyyy-MM-dd HH:mm:ss.SSS or yyyy-MM-dd HH:mm:ss val sdf = SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.US) sdf.timeZone = TimeZone.getTimeZone("UTC") return try { @@ -257,45 +265,11 @@ internal object CsvIngestParser { 0L } } - // A line "looks like" a CSV header if it has at least one recognisable keyword. - private fun looksLikeHeader(line: String): Boolean { - val cells = splitCsvLine(line) - return looksLikeHeaderCells(cells) - } - private fun splitCsvLine(line: String): List = - buildList { - val current = StringBuilder() - var inQuotes = false - - for (ch in line) { - when (ch) { - '"' -> inQuotes = !inQuotes - ',' -> { - if (inQuotes) { - current.append(ch) - } else { - add(current.toString().trim()) - current.setLength(0) - } - } - else -> current.append(ch) - } - } - - add(current.toString().trim()) - } -} - -private fun looksLikeHeaderCells(cells: List): Boolean { - val norm = cells.map { normalizeHeader(it) } - return norm.any { it.contains("latitude") } - || norm.any { it.contains("datetime(utc)") } + private fun normalizeHeader(raw: String): String = + raw.lowercase(Locale.US) + .replace("\uFEFF", "") + .replace("\"", "") + .replace(" ", "") + .trim() } - -private fun normalizeHeader(raw: String): String = - raw.lowercase() - .replace("\uFEFF", "") - .replace("\"", "") - .replace(" ", "") - .trim() diff --git a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/KeyframePlanner.kt b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/KeyframePlanner.kt index 472bd71a..7f3cc710 100644 --- a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/KeyframePlanner.kt +++ b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/KeyframePlanner.kt @@ -22,11 +22,12 @@ object KeyframePlanner { require(maxOutputFrames >= 1) { "maxOutputFrames must be >= 1" } val startUsMetadata = readVideoFileStartTimeUs(videoFile) - val (targetStartUs, rawRows, _) = CsvIngestParser.streamAndParse( - csvFile, - videoFile.name, - startUsMetadata - ) + var targetStartUs: Long = 0L + val rawRows = CsvIngestParser.streamAndParse(csvFile, videoFile.name, startUsMetadata) { context, sequence -> + val list = sequence.toList() + targetStartUs = context.targetStartUs + list + } val validRows = RowValidator.validate(rawRows) if (validRows.isEmpty()) return longArrayOf() diff --git a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/TelemetryPreprocessor.kt b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/TelemetryPreprocessor.kt index 8c15c235..1fef02c4 100644 --- a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/TelemetryPreprocessor.kt +++ b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/TelemetryPreprocessor.kt @@ -126,12 +126,21 @@ class TelemetryPreprocessor @JvmOverloads constructor( // Stage 1–2: Single-pass streaming ingest + parse (strictly Litchi) reportProgress(ProcessingStage.PARSING, 0.0f) val startUsMetadata = readVideoFileStartTimeUs(videoFile) - val (targetStartUs, rawRows, totalCsvRows) = CsvIngestParser.streamAndParse( - csvFile, - videoFile.name, - startUsMetadata - ) - + + var targetStartUs: Long = 0L + var totalCsvRows: Int = 0 + var malformedRows: Int = 0 + val rawRows = CsvIngestParser.streamAndParse(csvFile, videoFile.name, startUsMetadata) { context, sequence -> + // The sequence is strictly bound inside this closure + val filteredList = sequence.toList() + + targetStartUs = context.targetStartUs + totalCsvRows = context.totalSourceRows + malformedRows = context.malformedRows + + filteredList + } + if (targetStartUs > 0L) { Log.i(logTag, "Using telemetry start time ${targetStartUs}us for ${videoFile.name}") } else { @@ -142,7 +151,7 @@ class TelemetryPreprocessor @JvmOverloads constructor( // Stage 3: Row validation val validRows = RowValidator.validate(rawRows) - val rejectedRows = rawRows.size - validRows.size + val rejectedRows = malformedRows + (rawRows.size - validRows.size) // Stage 4: ENU conversion reportProgress(ProcessingStage.CONVERTING, 0.0f) diff --git a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/VideoTimeExtractor.kt b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/VideoTimeExtractor.kt new file mode 100644 index 00000000..48953551 --- /dev/null +++ b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/VideoTimeExtractor.kt @@ -0,0 +1,45 @@ +package com.splats.app.telemetry + +import java.text.SimpleDateFormat +import java.util.Locale +import java.util.TimeZone + +/** + * Extracts and interpolates video start timestamps from filenames and metadata. + */ +object VideoTimeExtractor { + private val filenameDateTimeRegex = Regex("""(\d{4}-\d{2}-\d{2})[ _](\d{2})[:_-](\d{2})[:_-](\d{2})""") + private val filenameTimeRegex = Regex("""(? Date: Thu, 16 Apr 2026 14:13:53 +0530 Subject: [PATCH 3/3] refactor: improve CSV header parsing and add initialization buffering to CsvIngestParser for robust telemetry synchronization --- .../splats/app/telemetry/CsvIngestParser.kt | 122 ++++++++++-------- 1 file changed, 65 insertions(+), 57 deletions(-) diff --git a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt index 4ed38c60..2652e6ac 100644 --- a/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt +++ b/crates/brush-app/app/src/main/java/com/splats/app/telemetry/CsvIngestParser.kt @@ -2,6 +2,7 @@ package com.splats.app.telemetry import android.util.Log import org.apache.commons.csv.CSVFormat +import org.apache.commons.csv.CSVParser import org.apache.commons.csv.CSVRecord import java.io.File import java.text.SimpleDateFormat @@ -32,14 +33,16 @@ private object ColumnMap { class CsvParseContext(var targetStartUs: Long = 0L, var totalSourceRows: Int = 0, var malformedRows: Int = 0) // ─── CSV Stream Pipeline ──────────────────────────────────────────────────────── -/** - * Single-pass streaming parser providing a safe scope for the CSV reader. - * Pre-scans for headers (bypassing preambles), initializes commons-csv with strict config, - * and maintains determinism using the first valid timestamp dynamically. - */ internal object CsvIngestParser { private const val TAG = "TelemetryCsv" + private const val MAX_INIT_BUFFER_SIZE = 5000 + /** + * Single-pass streaming parser providing a safe scope for the CSV reader. + * The Reader remains open ONLY while the provided lambda evaluates the `Sequence`. + * Caller MUST ensure terminal evaluations (like `.toList()`) are synchronously contained + * before the lambda returns. + */ fun streamAndParse( file: File, videoFileName: String, @@ -54,12 +57,16 @@ internal object CsvIngestParser { throw TelemetryError.UnsupportedFormat(label) } - // Bound to the entire function scope! No stream leakage allowed. return file.inputStream().bufferedReader(Charsets.UTF_8).use { reader -> var headerLine: Array? = null - // 1. The Seeker Logic: Consume and discard preamble until header - reader.mark(1024 * 64) // 64kb max lookahead for preamble checking + val formatBase = CSVFormat.DEFAULT.builder() + .setAllowMissingColumnNames(false) + .setIgnoreHeaderCase(true) + .setTrim(true) + .build() + + // 1. Zero-Allocation Seeker: Consume lines until headers are naturally found while (true) { val rawLine = reader.readLine() ?: break val line = rawLine.trim().removePrefix("\uFEFF") @@ -69,11 +76,12 @@ internal object CsvIngestParser { || line.contains("datetime(utc)", ignoreCase = true) if (isHeader) { - // Extract exactly what the file stated, keeping case. - // No spaces/bom trimming here; commons-csv format controls it. - val cells = splitPreCheckLine(line) - headerLine = cells.toTypedArray() - break + val parsed = CSVParser.parse(line, formatBase) + val records = parsed.records + if (records.isNotEmpty()) { + headerLine = records.first().toList().toTypedArray() + break + } } } @@ -82,30 +90,26 @@ internal object CsvIngestParser { throw TelemetryError.InsufficientRecords(0) } - // 2. Format with strict consistency protocols - val format = CSVFormat.DEFAULT.builder() + // 2. Main data tokenizer initialized dynamically + val dataFormat = formatBase.builder() .setHeader(*headerLine) - .setAllowMissingColumnNames(false) - .setIgnoreHeaderCase(true) - .setTrim(true) .build() - val csvParser = format.parse(reader) - - // Build the row parser exactly using mapped names dynamically + val csvParser = dataFormat.parse(reader) val headerMap = csvParser.headerMap ?: emptyMap() val rowParser = buildRowParser(headerMap) - + Log.i(TAG, "CSV headers detected successfully: ${headerLine.size} columns") val context = CsvParseContext(targetStartUs = videoStartUsFromMetadata ?: 0L) var baseDateUs: Long? = null - // 3. Dynamic sequence implementation + // 3. Dynamic sequence implementation with Peeker Buffer Constraints val rowSequence = sequence { + var initializationBuffer: MutableList? = mutableListOf() + for (record in csvParser) { context.totalSourceRows++ - // Strict Leniency Rule - drop mismatched columns immediately if (!record.isConsistent) { Log.w(TAG, "Dropping malformed CSV row (inconsistent column size): line ${record.recordNumber}") context.malformedRows++ @@ -118,32 +122,37 @@ internal object CsvIngestParser { continue } - // Determine Target Start Time once statically - if (baseDateUs == null && row.timestampUs > 0L) { - baseDateUs = row.timestampUs - } - if (videoStartUsFromMetadata == null && context.targetStartUs == 0L && baseDateUs != null) { - context.targetStartUs = VideoTimeExtractor.extractTargetStartUs(videoFileName, baseDateUs!!) - } - - val activeFilterUs = context.targetStartUs - if (activeFilterUs == 0L || row.timestampUs >= activeFilterUs) { - yield(row) + if (context.targetStartUs == 0L) { + if (row.timestampUs > 0L) { + baseDateUs = row.timestampUs + context.targetStartUs = VideoTimeExtractor.extractTargetStartUs(videoFileName, baseDateUs!!) + + // Flush micro-buffer safely + initializationBuffer?.forEach { b -> + if (b.timestampUs >= context.targetStartUs) yield(b) + } + initializationBuffer = null + + if (row.timestampUs >= context.targetStartUs) yield(row) + } else { + initializationBuffer?.add(row) + if (initializationBuffer != null && initializationBuffer!!.size > MAX_INIT_BUFFER_SIZE) { + throw TelemetryError.InsufficientRecords(MAX_INIT_BUFFER_SIZE) + } + } + } else { + if (row.timestampUs >= context.targetStartUs) { + yield(row) + } } } - } + }.constrainOnce() block(context, rowSequence) } } - private fun splitPreCheckLine(line: String): List { - // Fast split just to identify the initial header array schema - return line.split(',').map { it.trim().removePrefix("\"").removeSuffix("\"") } - } - private fun buildRowParser(headerMap: Map): (CSVRecord) -> TelRow? { - // Prepare normalized mapping table val normalizedHeaders = headerMap.entries.associate { (k, v) -> normalizeHeader(k) to v } @@ -178,15 +187,20 @@ internal object CsvIngestParser { val fixTypeHeader = iFixType?.let { revMap[it] } ?: "" return { cols: CSVRecord -> - fun getCol(idx: Int, default: String = "0"): String { - return if (idx >= 0 && idx < cols.size()) { - val raw = cols.get(idx).trim() - if (raw.isBlank()) default else raw - } else default + fun getCol(idx: Int): String { + if (idx < 0 || idx >= cols.size()) throw IllegalArgumentException("Missing coordinate column index") + val raw = cols.get(idx).trim() + if (raw.isBlank()) throw IllegalArgumentException("Empty required coordinate cell") + return raw + } + + fun getColOrEmpty(idx: Int): String { + return if (idx >= 0 && idx < cols.size()) cols.get(idx).trim() else "" } + // Explicit localized boundary tracking kinematics strictly runCatching { - val tsUs = parseLitchiDateTime(getCol(iTs, "")) + val tsUs = parseLitchiDateTime(getColOrEmpty(iTs)) TelRow( timestampUs = tsUs, lat = getCol(iLat).toDouble(), @@ -197,15 +211,9 @@ internal object CsvIngestParser { velN = parseSpeedMps(getCol(iVelN), velNHeader), velE = parseSpeedMps(getCol(iVelE), velEHeader), velD = parseSpeedMps(getCol(iVelD), velDHeader), - hdop = iHdop?.let { - parseHdop(getCol(it), hdopHeader) - } ?: 1.0, - fixType = iFixType?.let { - parseFixType(getCol(it), fixTypeHeader) - } ?: 3, - satellites = iSatellites?.let { - getCol(it).toIntOrNull() ?: 0 - } ?: 0 + hdop = iHdop?.let { parseHdop(getColOrEmpty(it), hdopHeader) } ?: 1.0, + fixType = iFixType?.let { parseFixType(getColOrEmpty(it), fixTypeHeader) } ?: 3, + satellites = iSatellites?.let { getColOrEmpty(it).toIntOrNull() ?: 0 } ?: 0 ) }.getOrNull() }