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 2696531e..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 @@ -1,10 +1,13 @@ 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 import java.util.Locale import java.util.TimeZone -import android.util.Log // ─── Column name mapping ────────────────────────────────────────────────────── @@ -27,21 +30,25 @@ private object ColumnMap { val SATELLITES = HeaderAliases(listOf("satellites")) } -// ─── CSV Ingest ─────────────────────────────────────────────────────────────── - -/** - * 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. - */ -internal object CsvIngest { +class CsvParseContext(var targetStartUs: Long = 0L, var totalSourceRows: Int = 0, var malformedRows: Int = 0) + +// ─── CSV Stream Pipeline ──────────────────────────────────────────────────────── +internal object CsvIngestParser { private const val TAG = "TelemetryCsv" + private const val MAX_INIT_BUFFER_SIZE = 5000 - fun read(file: File): Pair, List>> { + /** + * 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, + videoStartUsFromMetadata: Long?, + block: (CsvParseContext, Sequence) -> R + ): R { if (!file.exists()) throw TelemetryError.CsvNotFound(file.absolutePath) val ext = file.extension.lowercase() @@ -49,221 +56,187 @@ 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 + return file.inputStream().bufferedReader(Charsets.UTF_8).use { reader -> + var headerLine: Array? = null + + val formatBase = CSVFormat.DEFAULT.builder() + .setAllowMissingColumnNames(false) + .setIgnoreHeaderCase(true) + .setTrim(true) + .build() - file.inputStream().bufferedReader(Charsets.UTF_8).use { reader -> + // 1. Zero-Allocation Seeker: Consume lines until headers are naturally found 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 - } - // Locate the header row: first line that looks like a CSV header - // (contains a comma and at least one recognisable keyword). - if (headers == null) { - if (line.contains(',') && looksLikeHeader(line)) { - val headerLine = splitCsvLine(line) - headers = headerLine - Log.i(TAG, "CSV headers: ${headerLine.joinToString("|")}") + val isHeader = line.contains("latitude", ignoreCase = true) + || line.contains("datetime(utc)", ignoreCase = true) + + if (isHeader) { + val parsed = CSVParser.parse(line, formatBase) + val records = parsed.records + if (records.isNotEmpty()) { + headerLine = records.first().toList().toTypedArray() + break } - continue - } - - if (line.contains(',')) { - dataRows += splitCsvLine(line) } } - } + + if (headerLine == null) { + Log.e(TAG, "CSV: could not find header row. Malformed or purely preamble.") + throw TelemetryError.InsufficientRecords(0) + } - val headersFinal = headers ?: run { - Log.e(TAG, "CSV: could not find header row. First line: $firstNonEmpty") - throw TelemetryError.InsufficientRecords(0) - } + // 2. Main data tokenizer initialized dynamically + val dataFormat = formatBase.builder() + .setHeader(*headerLine) + .build() - return Pair(headersFinal, dataRows) - } + val csvParser = dataFormat.parse(reader) + val headerMap = csvParser.headerMap ?: emptyMap() + val rowParser = buildRowParser(headerMap) - // 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) - } + Log.i(TAG, "CSV headers detected successfully: ${headerLine.size} columns") + + val context = CsvParseContext(targetStartUs = videoStartUsFromMetadata ?: 0L) + var baseDateUs: Long? = null + + // 3. Dynamic sequence implementation with Peeker Buffer Constraints + val rowSequence = sequence { + var initializationBuffer: MutableList? = mutableListOf() + + for (record in csvParser) { + context.totalSourceRows++ + if (!record.isConsistent) { + Log.w(TAG, "Dropping malformed CSV row (inconsistent column size): line ${record.recordNumber}") + context.malformedRows++ + continue + } - 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) + val row = rowParser(record) + if (row == null) { + context.malformedRows++ + continue + } + + 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 { - add(current.toString().trim()) - current.setLength(0) + 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) } } - else -> current.append(ch) } - } + }.constrainOnce() - 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) { - throw TelemetryError.InsufficientRecords(0) + block(context, rowSequence) } } - fun parseStartTimeFromFilename(filename: String, rows: List): 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 - } + private fun buildRowParser(headerMap: Map): (CSVRecord) -> TelRow? { + val normalizedHeaders = headerMap.entries.associate { (k, v) -> + normalizeHeader(k) to v } - val baseDateUs = rows.firstOrNull { it.timestampUs > 0L }?.timestampUs ?: 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 parseInternal( - headers: List, - dataRows: List> - ): List { - val lower = headers.map { normalizeHeader(it) } - 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): 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 + } - return dataRows.mapNotNull { cols -> + 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(cols.getOrElse(iTs) { "" }) + val tsUs = parseLitchiDateTime(getColOrEmpty(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), - hdop = iHdop?.let { - parseHdop(cols.getOrElse(it) { "0" }, hdopHeader ?: "") - } ?: 1.0, - fixType = iFixType?.let { - parseFixType(cols.getOrElse(it) { "0" }, fixTypeHeader ?: "") - } ?: 3, - satellites = iSatellites?.let { - cols.getOrElse(it) { "0" }.toIntOrNull() ?: 0 - } ?: 0 + 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(getColOrEmpty(it), hdopHeader) } ?: 1.0, + fixType = iFixType?.let { parseFixType(getColOrEmpty(it), fixTypeHeader) } ?: 3, + satellites = iSatellites?.let { getColOrEmpty(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 @@ -276,7 +249,7 @@ internal object CsvParser { 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 @@ -288,12 +261,7 @@ internal object CsvParser { } } - /** - * 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 { @@ -305,17 +273,11 @@ internal object CsvParser { 0L } } -} -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 1c6b8939..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 @@ -21,15 +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) + 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) 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..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 @@ -123,28 +123,35 @@ 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) + + 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 { 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 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) @@ -212,7 +219,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, 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("""(?