diff --git a/README.md b/README.md index 2df1b63..ff7cb66 100644 --- a/README.md +++ b/README.md @@ -199,13 +199,16 @@ Configure notification providers and alert preferences under **Settings → Noti Bring your history across from another app under **Settings → Import Data**: choose **New import**, pick the source, add its export files, and check the preview before importing. -| Source | Getting the export | Files | -| --------- | ------------------------------------------------------------------------ | ----------------- | -| ElectraFi | **Settings → Account → Advanced → Download Data**, then **Download All** | `.csv`, `.csv.gz` | +| Source | Getting the export | Files | +| ------------- | ------------------------------------------------------------------------------------------------------------------- | ----------------- | +| Rivian Roamer | **Account** (from the menu by your name) → **Data → Download a copy**; request it, then **Download ZIP** once ready | one `.zip` | +| ElectraFi | **Settings → Account → Advanced → Download Data**, then **Download All** | `.csv`, `.csv.gz` | - Imports add drives, charging sessions, and vehicle history. +- A Rivian Roamer copy is matched to your Rivian by VIN, and brings each charge's curve and energy + breakdown. - Anything already recorded stays as it is, and an existing charge only gets what it's missing. -- **Undo** removes what an import added. +- **Undo** removes what an import added, apart from drivers and saved places, which stay. ### TeslaMate diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 235aa7e..225ae56 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -62,6 +62,9 @@ importers: ws: specifier: ^8.22.0 version: 8.22.0 + yauzl: + specifier: ^3.4.0 + version: 3.4.0 zod: specifier: ^4.6.7 version: 4.6.7 @@ -72,6 +75,9 @@ importers: '@types/ws': specifier: ^8.18.2 version: 8.18.2 + '@types/yauzl': + specifier: ^3.4.0 + version: 3.4.0 drizzle-kit: specifier: ^0.31.11 version: 0.31.11 @@ -980,6 +986,9 @@ packages: '@types/ws@8.18.2': resolution: {integrity: sha512-67MQl+fpWKVTT1NYdnmo3U4sc/xPo/zQBncVnI74qmQa0z/b+1g6iYqNmGCPbxO+zz2aklb08a0oHfegiVd0/w==} + '@types/yauzl@3.4.0': + resolution: {integrity: sha512-NRPn5w6h8dhcnmx3YIRQcqMywY/+nND/uOkJessedcrowO3C0AssHp3tMJpxKAwOhFOo0OV1y9VtsC5hbKKBAw==} + '@typescript-eslint/eslint-plugin@8.71.0': resolution: {integrity: sha512-pqcS9c1HxZTHt7End4nXqd0s5lJrrFzrgCkKFJrsbUnaL6M3+6oBFZaslg6Gjsl3argl2DDRFROnXARaZ2e4Nw==} engines: {node: ^18.18.0 || ^20.9.0 || >=21.1.0} @@ -1890,6 +1899,9 @@ packages: resolution: {integrity: sha512-mnvGdvOrIvJOBGUEdGkrVXjN8E/VkIJCkf2eS1DH2yv82ORUlLttmDt0rWY38yYZmVwciZwBUvHM20qxBZf40w==} hasBin: true + pend@1.2.0: + resolution: {integrity: sha512-F3asv42UuXchdzt+xXqfW1OGlVBe+mxa2mqI0pg5yAHZPvFmY3Y6drSf/GQ1A86WgWEN9Kzh/WrgKa6iGcHXLg==} + picocolors@1.1.1: resolution: {integrity: sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==} @@ -2322,6 +2334,10 @@ packages: yallist@3.1.1: resolution: {integrity: sha512-a4UGQaWPH59mOXUYnAG2ewncQS4i4F43Tv3JoAM+s2VDAmS9NsK8GpDMLrCHPksFT7h3K6TOoUNn2pb7RoXx4g==} + yauzl@3.4.0: + resolution: {integrity: sha512-jIH9yLR9wqr0wOS0TpBvo/g/2UgZH5qePVbjgRliiF0BYvOZyaBknKsF+x9Iht0O6sqgnB93rCICdOZFecJuDw==} + engines: {node: '>=12'} + yocto-queue@0.1.0: resolution: {integrity: sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q==} engines: {node: '>=10'} @@ -2993,6 +3009,10 @@ snapshots: dependencies: '@types/node': 22.20.4 + '@types/yauzl@3.4.0': + dependencies: + '@types/node': 22.20.4 + '@typescript-eslint/eslint-plugin@8.71.0(@typescript-eslint/parser@8.71.0(eslint@10.11.0(jiti@2.7.0))(typescript@6.0.3))(eslint@10.11.0(jiti@2.7.0))(typescript@6.0.3)': dependencies: '@eslint-community/regexpp': 4.12.2 @@ -3816,6 +3836,8 @@ snapshots: dependencies: resolve-protobuf-schema: 2.1.0 + pend@1.2.0: {} + picocolors@1.1.1: {} picomatch@4.0.7: {} @@ -4211,6 +4233,10 @@ snapshots: yallist@3.1.1: {} + yauzl@3.4.0: + dependencies: + pend: 1.2.0 + yocto-queue@0.1.0: {} zod-validation-error@4.0.2(zod@4.6.7): diff --git a/server/package.json b/server/package.json index d6b5883..3dabdfc 100644 --- a/server/package.json +++ b/server/package.json @@ -24,11 +24,13 @@ "pino-pretty": "^13.2.0", "postgres": "^3.4.9", "ws": "^8.22.0", + "yauzl": "^3.4.0", "zod": "^4.6.7" }, "devDependencies": { "@types/node": "^22.20.4", "@types/ws": "^8.18.2", + "@types/yauzl": "^3.4.0", "drizzle-kit": "^0.31.11", "tsx": "^4.23.15", "typescript": "^6.0.3", diff --git a/server/src/api-types.ts b/server/src/api-types.ts index a3356cc..60ac617 100644 --- a/server/src/api-types.ts +++ b/server/src/api-types.ts @@ -388,9 +388,9 @@ export interface ChargingSessionDto { socLimit: number | null; /** * live = recorded by RivianMate; rivian = imported from Rivian's history; - * electrafi = imported from an ElectraFi export; teslamate = read from TeslaMate. + * electrafi, roamer = imported from that app's export; teslamate = read from TeslaMate. */ - source: 'live' | 'rivian' | 'live+rivian' | 'electrafi' | 'teslamate'; + source: 'live' | 'rivian' | 'live+rivian' | 'electrafi' | 'roamer' | 'teslamate'; vendor: string | null; city: string | null; isPublic: boolean | null; @@ -585,17 +585,25 @@ export interface ImportCounts { sessionsEnriched: number; locationPoints: number; stateChanges: number; + /** Saved places added; absent on imports from before places were. */ + places?: number; } -/** An ElectraFi upload waiting to be imported. */ +/** The kinds of export that can be imported. */ +export type ImportFormat = 'electrafi' | 'roamer'; + +/** An upload waiting to be imported. */ export interface ImportDraftDto { id: string; + format: ImportFormat; files: string[]; + /** The vehicles the export has data for, matched by VIN; null when it doesn't say. */ + vehicleIds: string[] | null; } export interface ImportSelection { vehicleId: string; - /** IANA time zone the export's local times are in. */ + /** IANA time zone the dates are in, and an ElectraFi export's local times. */ timeZone: string; /** Both included, as YYYY-MM-DD in `timeZone`. */ from: string; @@ -619,11 +627,20 @@ export interface ImportDrivePreview { /** Saved locations it starts and ends at. */ startPlace: string | null; endPlace: string | null; + /** Who the export says drove. */ + driver: string | null; } /** What a charge can add to a session that's missing it. */ export type ImportSessionAddition = - 'chargingTime' | 'batteryLevels' | 'energy' | 'range' | 'power' | 'location' | 'curve'; + | 'chargingTime' + | 'batteryLevels' + | 'energy' + | 'breakdown' + | 'range' + | 'power' + | 'location' + | 'curve'; export interface ImportChargePreview { startedAt: string; @@ -655,6 +672,11 @@ export interface ImportPreviewDto { sessionsEnriched: number; readingsNew: number; readingsRecorded: number; + /** Past what live recording last reported, so left for it. */ + readingsNewer: number; + /** Names of the saved places it adds, and how many are saved already. */ + placesNew: string[]; + placesRecorded: number; warnings: ('odometer' | 'future')[]; /** Everything starting in the chosen dates, in time order. */ drives: ImportDrivePreview[]; @@ -664,6 +686,7 @@ export interface ImportPreviewDto { export interface ImportBatchDto { id: number; vehicleId: string; + format: ImportFormat; files: string[]; timeZone: string; from: string; diff --git a/server/src/context.ts b/server/src/context.ts index c52fe7d..26cf204 100644 --- a/server/src/context.ts +++ b/server/src/context.ts @@ -7,7 +7,7 @@ import type { RivianApi } from './rivian/client.js'; import type { RivianGovernor } from './rivian/governor.js'; import type { RivianTokens } from './rivian/types.js'; import type { PlaceSearch } from './services/drive-places.js'; -import type { ElectraFiImports } from './services/electrafi/imports.js'; +import type { Imports } from './services/imports/imports.js'; import type { LiveBus } from './services/live-bus.js'; import type { LoginThrottle } from './services/login-throttle.js'; import type { NotificationService } from './services/notifications.js'; @@ -46,7 +46,7 @@ export interface AppContext { /** Vehicles read from a TeslaMate database, when one is set up. */ teslamate: TeslaMateSource; /** Uploads and runs ElectraFi imports. */ - imports: ElectraFiImports; + imports: Imports; /** Shared throttle/cooldown/counters for all Rivian traffic. */ governor: RivianGovernor; /** Address search for placing saved locations. */ diff --git a/server/src/db/schema.ts b/server/src/db/schema.ts index b5fb41a..2971955 100644 --- a/server/src/db/schema.ts +++ b/server/src/db/schema.ts @@ -73,7 +73,7 @@ export const importBatches = pgTable('import_batches', { .notNull() .references(() => vehicles.id), /** What kind of export it came from. */ - format: text('format', { enum: ['electrafi'] }).notNull(), + format: text('format', { enum: ['electrafi', 'roamer'] }).notNull(), files: jsonb('files').$type().notNull(), timeZone: text('time_zone').notNull(), /** The dates chosen, in `time_zone`, both included. */ @@ -274,8 +274,11 @@ export const chargingSessions = pgTable( lat: doublePrecision('lat'), lon: doublePrecision('lon'), rawFinal: jsonb('raw_final'), - /** live = recorded by RivianMate; rivian = imported; live+rivian = both; electrafi = from an ElectraFi export. */ - source: text('source', { enum: ['live', 'rivian', 'live+rivian', 'electrafi'] }) + /** + * live = recorded by RivianMate; rivian = imported; live+rivian = both; + * electrafi, roamer = from that app's export. + */ + source: text('source', { enum: ['live', 'rivian', 'live+rivian', 'electrafi', 'roamer'] }) .notNull() .default('live'), rivianTransactionId: text('rivian_transaction_id'), @@ -328,11 +331,21 @@ export const chargingCurvePoints = pgTable( * feed's chart; * push_live = the push feed's current reading; forecast = the vehicle's * projection to the limit for a charge in progress (replaced by each - * graph, never a reading); electrafi = from an ElectraFi export; legacy = - * recorded before sources were kept apart. + * graph, never a reading); roamer = the push feed's readings as Roamer + * kept them; electrafi = from an ElectraFi export; legacy = recorded + * before sources were kept apart. */ source: text('source', { - enum: ['observed', 'graph', 'push_chart', 'push_live', 'forecast', 'electrafi', 'legacy'], + enum: [ + 'observed', + 'graph', + 'push_chart', + 'push_live', + 'forecast', + 'roamer', + 'electrafi', + 'legacy', + ], }) .notNull() .default('legacy'), diff --git a/server/src/index.ts b/server/src/index.ts index e3ab92e..86bdde8 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -16,7 +16,7 @@ import { seedAppPassword } from './routes/auth.js'; import { getLookupSettings } from './routes/settings.js'; import { DrivePlaces, nominatimGeocoder, nominatimSearch } from './services/drive-places.js'; import { DriveWeather, openMeteoTemperatures } from './services/drive-weather.js'; -import { ElectraFiImports } from './services/electrafi/imports.js'; +import { Imports } from './services/imports/imports.js'; import { LiveBus } from './services/live-bus.js'; import { NotificationService } from './services/notifications.js'; import { LoginThrottle } from './services/login-throttle.js'; @@ -104,7 +104,7 @@ async function main(): Promise { ); notifications.start(bus); - const imports = new ElectraFiImports(db, historyKeys, logger.child({ component: 'import' }), () => + const imports = new Imports(db, historyKeys, logger.child({ component: 'import' }), () => monitor.queueDriveLookups(), ); await imports.recover(); diff --git a/server/src/routes/imports.ts b/server/src/routes/imports.ts index fef4ebd..17c6a43 100644 --- a/server/src/routes/imports.ts +++ b/server/src/routes/imports.ts @@ -2,7 +2,7 @@ import type { FastifyInstance } from 'fastify'; import type { Readable } from 'node:stream'; import { z } from 'zod'; import type { AppContext } from '../context.js'; -import { ImportError } from '../services/electrafi/imports.js'; +import { ImportError } from '../services/imports/format.js'; const isoDate = z.string().regex(/^\d{4}-\d{2}-\d{2}$/); const selectionSchema = z.object({ @@ -14,7 +14,8 @@ const selectionSchema = z.object({ }); const uploadQuery = z.object({ draft: z.string().min(1).max(64).optional(), - name: z.string().max(1000).default('export.csv'), + name: z.string().max(1000).default('export'), + format: z.enum(['electrafi', 'roamer']).default('electrafi'), }); export async function importRoutes(app: FastifyInstance, ctx: AppContext): Promise { @@ -38,11 +39,11 @@ export async function importRoutes(app: FastifyInstance, ctx: AppContext): Promi app.get('/api/imports', async () => ctx.imports.list()); app.post('/api/imports/files', async (request) => { - const { draft, name } = uploadQuery.parse(request.query); + const { draft, name, format } = uploadQuery.parse(request.query); if (request.headers['content-type'] !== 'application/octet-stream') { throw new ImportError('Upload the file as a binary body.', 415); } - return ctx.imports.addFile(draft, name, request.body as Readable); + return ctx.imports.addFile(draft, name, request.body as Readable, format); }); app.delete<{ Params: { id: string } }>('/api/imports/drafts/:id', async (request, reply) => { diff --git a/server/src/services/charging-curve.ts b/server/src/services/charging-curve.ts index 9d6ef8c..8ef03b4 100644 --- a/server/src/services/charging-curve.ts +++ b/server/src/services/charging-curve.ts @@ -2,7 +2,14 @@ import type { LiveSessionData } from '../rivian/types.js'; export type CurveSource = - 'observed' | 'graph' | 'push_chart' | 'push_live' | 'forecast' | 'electrafi' | 'legacy'; + | 'observed' + | 'graph' + | 'push_chart' + | 'push_live' + | 'forecast' + | 'roamer' + | 'electrafi' + | 'legacy'; export interface CurvePoint { source: CurveSource; @@ -14,7 +21,8 @@ export interface CurvePoint { /** * Recorded series, most trusted first: Parallax's charging graph is the * vehicle's own record; the push feed's chart is Rivian's; its current - * readings are snapshots; an ElectraFi export is another app's samples. Rows + * readings are snapshots, which a Roamer export keeps too; an ElectraFi + * export is another app's samples. Rows * from before sources were kept apart come last. The forecast is never a * recorded series. */ @@ -22,6 +30,7 @@ const SOURCE_ORDER: readonly CurveSource[] = [ 'graph', 'push_chart', 'push_live', + 'roamer', 'electrafi', 'legacy', ]; diff --git a/server/src/services/drive-places.ts b/server/src/services/drive-places.ts index c1124d1..bd1f49a 100644 --- a/server/src/services/drive-places.ts +++ b/server/src/services/drive-places.ts @@ -1,4 +1,4 @@ -import { and, desc, eq, isNotNull, isNull } from 'drizzle-orm'; +import { and, desc, eq, inArray, isNotNull, isNull } from 'drizzle-orm'; import { nearbySpot, wallboxSpots } from './locations.js'; import type { Db } from '../db/client.js'; import { type Logger, silentLogger } from '../logger.js'; @@ -192,7 +192,7 @@ export class DrivePlaces { .from(chargingSessions) .where( and( - eq(chargingSessions.source, 'electrafi'), + inArray(chargingSessions.source, ['electrafi', 'roamer']), isNull(chargingSessions.placesCheckedAt), isNull(chargingSessions.vendor), isNull(chargingSessions.city), diff --git a/server/src/services/drivers.db.test.ts b/server/src/services/drivers.db.test.ts index e747681..8ba07f6 100644 --- a/server/src/services/drivers.db.test.ts +++ b/server/src/services/drivers.db.test.ts @@ -174,6 +174,28 @@ describe.skipIf(!url)('Drivers with Postgres', () => { { profileId: BEN, reportedName: 'Ben', lastSeenAt: at(2) }, ]); }); + + it('takes over a driver an import made from the name, with its drives', async () => { + const [imported] = await handle.db + .insert(drivers) + .values({ vehicleId: 'v1', reportedName: 'Ben V.', name: 'Benny' }) + .returning({ id: drivers.id }); + const id = await drive(0); + await handle.db + .update(drives) + .set({ driverId: imported!.id, driverSource: 'profile' }) + .where(eq(drives.id, id)); + + await service.noteProfile('v1', BEN, at(10)); + service.noteName('v1', ' ben v. '); + await service.flushNames(); + + const rows = await handle.db + .select({ id: drivers.id, profileId: drivers.profileId, name: drivers.name }) + .from(drivers); + expect(rows).toEqual([{ id: await driverId(BEN), profileId: BEN, name: 'Benny' }]); + expect(await driveDriver(id)).toEqual({ driverId: await driverId(BEN), source: 'profile' }); + }); }); async function waitFor(check: () => Promise, timeoutMs = 5000): Promise { diff --git a/server/src/services/drivers.ts b/server/src/services/drivers.ts index a6030c4..b72aeb1 100644 --- a/server/src/services/drivers.ts +++ b/server/src/services/drivers.ts @@ -1,4 +1,16 @@ -import { and, asc, desc, eq, isNotNull, isNull, ne, notInArray, or, sql } from 'drizzle-orm'; +import { + and, + asc, + desc, + eq, + inArray, + isNotNull, + isNull, + ne, + notInArray, + or, + sql, +} from 'drizzle-orm'; import type { Db } from '../db/client.js'; import { drivers, drives } from '../db/schema.js'; import { type Logger, silentLogger } from '../logger.js'; @@ -142,11 +154,44 @@ export class Drivers { this.pendingNames.delete(vehicleId); const profileId = this.current.get(vehicleId)?.profileId; if (!name || !profileId) return; - await this.driverFor(vehicleId, profileId, null); + const driverId = await this.driverFor(vehicleId, profileId, null); await this.db .update(drivers) .set({ reportedName: name }) .where(and(eq(drivers.vehicleId, vehicleId), eq(drivers.profileId, profileId))); + await this.adoptImported(vehicleId, driverId, name); + } + + /** + * An import makes a driver from a name when the vehicle hasn't reported + * that profile yet: the only kind with a reported name and no profile. + * Once the vehicle reports the name, the profile's driver takes over its + * drives, and the name the user gave it if the profile has none. + */ + private async adoptImported(vehicleId: string, driverId: number, name: string): Promise { + await this.db.transaction(async (tx) => { + const imported = await tx + .select({ id: drivers.id, name: drivers.name }) + .from(drivers) + .where( + and( + eq(drivers.vehicleId, vehicleId), + isNull(drivers.profileId), + sql`lower(trim(${drivers.reportedName})) = ${name.trim().toLowerCase()}`, + ), + ); + if (imported.length === 0) return; + const ids = imported.map((d) => d.id); + await tx.update(drives).set({ driverId }).where(inArray(drives.driverId, ids)); + const given = imported.find((d) => d.name)?.name; + if (given) { + await tx + .update(drivers) + .set({ name: given }) + .where(and(eq(drivers.id, driverId), isNull(drivers.name))); + } + await tx.delete(drivers).where(inArray(drivers.id, ids)); + }); } /** The driver for a profile, created the first time it's seen. */ diff --git a/server/src/services/electrafi/electrafi.test.ts b/server/src/services/electrafi/electrafi.test.ts index 7f77775..684bd07 100644 --- a/server/src/services/electrafi/electrafi.test.ts +++ b/server/src/services/electrafi/electrafi.test.ts @@ -2,16 +2,17 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { TimeZoneClock } from '../imports/time-zone.js'; import { NotAnExportError, - TimeZoneClock, firstDataId, parseLocalTime, parseRow, readRows, rowState, } from './parse.js'; -import { type Existing, buildPlan, odometerFits, rangeOf, scanFiles } from './plan.js'; +import { type Existing, buildPlan, odometerFits, rangeOf } from '../imports/plan.js'; +import { scanFiles } from './format.js'; import { asleep, awake, chargeRows, csvOf, driveRows } from '../../testing/electrafi-rows.js'; let dir: string; @@ -206,6 +207,26 @@ describe('finding drives', () => { expect(new Date(drives[0]!.endedAt).toISOString()).toBe('2025-12-16T19:30:00.000Z'); }); + it('takes the driver whose profile was active as the drive set off', async () => { + const { drives } = await scanOf([ + awake('2026-10-30 10:00:00', { autopark_state_v2: 'Mitch V.' }), + awake('2026-10-30 10:01:00', { + shift_state: 'D', + odometer: 1, + autopark_state_v2: 'Laura V.', + }), + awake('2026-10-30 10:02:00', { + shift_state: 'D', + odometer: 2, + autopark_state_v2: 'Mitch V.', + }), + awake('2026-10-30 10:03:00', { odometer: 3 }), + awake('2026-10-30 10:30:00', { shift_state: 'D', odometer: 4 }), + awake('2026-10-30 10:31:00', { odometer: 5 }), + ]); + expect(drives.map((d) => d.driverName)).toEqual(['Laura V.', null]); + }); + it('ignores asleep rows and ends a drive open at the last row', async () => { const { drives } = await scanOf([ awake('2026-10-30 10:00:00', { shift_state: 'D', odometer: 1 }), @@ -249,7 +270,10 @@ describe('finding charges', () => { ...chargeRows(new Date('2026-01-02T20:00:00Z'), 5, 5), ...chargeRows(new Date('2026-01-03T20:00:00Z'), 5, 6), ]); - expect(charges.map((c) => c.chargeNumber)).toEqual([5, 6]); + expect(charges.map((c) => new Date(c.startedAt).toISOString())).toEqual([ + '2026-01-02T20:00:00.000Z', + '2026-01-03T20:00:00.000Z', + ]); }); }); @@ -261,6 +285,7 @@ describe('planning', () => { drives: [], sessions: [], mileage: [], + places: [], }; const all = { from: at('2026-01-01T00:00:00Z'), to: at('2027-01-01T00:00:00Z') }; const now = at('2026-10-31T00:00:00Z'); diff --git a/server/src/services/electrafi/format.ts b/server/src/services/electrafi/format.ts new file mode 100644 index 0000000..7ecdd65 --- /dev/null +++ b/server/src/services/electrafi/format.ts @@ -0,0 +1,119 @@ +import { createReadStream, createWriteStream } from 'node:fs'; +import { open, rm } from 'node:fs/promises'; +import { pipeline } from 'node:stream/promises'; +import { createGunzip } from 'node:zlib'; +import { + type ExportFormat, + ImportError, + type Reading, + type Scan, + capped, +} from '../imports/format.js'; +import { + type ElectraFiRow, + NotAnExportError, + ROW_STATE_FIELDS, + firstDataId, + readRows, + rowState, +} from './parse.js'; +import { ChargeSegmenter, DriveSegmenter } from './segment.js'; + +async function isGzip(path: string): Promise { + const file = await open(path, 'r'); + try { + const { bytesRead, buffer } = await file.read(Buffer.alloc(2), 0, 2, 0); + return bytesRead === 2 && buffer[0] === 0x1f && buffer[1] === 0x8b; + } finally { + await file.close(); + } +} + +export async function scanFiles(paths: readonly string[], timeZone: string): Promise { + const scan: Scan = { + rows: 0, + unreadable: 0, + firstAt: null, + lastAt: null, + firstMileage: null, + lastMileage: null, + drives: [], + charges: [], + places: [], + readingTimes: [], + }; + const driveFinder = new DriveSegmenter(); + const chargeFinder = new ChargeSegmenter(); + for await (const { row, at } of readRows(paths, timeZone, scan)) { + scan.rows += 1; + scan.firstAt = Math.min(scan.firstAt ?? at, at); + scan.lastAt = Math.max(scan.lastAt ?? at, at); + if (row.mileageM != null) { + const reading = { at, value: row.mileageM }; + scan.firstMileage ??= reading; + scan.lastMileage = reading; + scan.readingTimes.push(at); + } + scan.drives.push(...driveFinder.push(row, at)); + scan.charges.push(...chargeFinder.push(row, at)); + } + scan.drives.push(...driveFinder.flush()); + scan.charges.push(...chargeFinder.flush()); + return scan; +} + +export function readingOf(row: ElectraFiRow, at: number): Reading { + return { + at, + state: rowState(row, at), + // Asleep rows have no state in them, and a sleeping car sends no heartbeats live either. + reporting: row.mileageM != null, + fix: + row.lat != null && row.lon != null + ? { + lat: row.lat, + lon: row.lon, + speedKmh: row.speedKmh, + altitudeM: row.altitudeM, + bearing: null, + } + : null, + }; +} + +/** ElectraFi's CSV exports: one account per car, with local times and no zone. */ +export const electraFiFormat: ExportFormat = { + id: 'electrafi', + singleFile: false, + unreadable: "Couldn't read this file as an ElectraFi export.", + + async store(upload, stem, maxBytes) { + const path = `${stem}.csv`; + try { + const stored = createWriteStream(path); + const uploaded = createReadStream(upload); + if (await isGzip(upload)) { + await pipeline(uploaded, createGunzip(), capped(maxBytes), stored); + } else { + await pipeline(uploaded, stored); + } + // Reads the header and first row, so a wrong file is refused here. + await firstDataId(path); + } catch (err) { + await rm(path, { force: true }); + if (err instanceof NotAnExportError) throw new ImportError(err.message); + throw err; + } + return { path, vins: null }; + }, + + scanKey: ({ timeZone }) => timeZone, + scan: (paths, { timeZone }) => scanFiles(paths, timeZone), + + async *readings(paths, { timeZone }) { + for await (const { row, at } of readRows(paths, timeZone)) yield readingOf(row, at); + }, + + stateFields: ROW_STATE_FIELDS, + repeatsFixes: true, +}; diff --git a/server/src/services/electrafi/parse.ts b/server/src/services/electrafi/parse.ts index 87a1b12..d166126 100644 --- a/server/src/services/electrafi/parse.ts +++ b/server/src/services/electrafi/parse.ts @@ -1,6 +1,7 @@ import { parse } from 'csv-parse'; import { createReadStream } from 'node:fs'; import type { VehicleState } from '../../rivian/types.js'; +import { type LocalTime, TimeZoneClock } from '../imports/time-zone.js'; const KM_PER_MILE = 1.609344; const MPH_TO_KMH = KM_PER_MILE; @@ -10,7 +11,6 @@ const M_PER_MILE = 1609.344; * its range (255 steps of 0.025 bar), not a pressure. */ const TIRE_NO_READING_BAR = 6.37; -const OFFSET_SLOT_MS = 15 * 60_000; /** Columns that show a file is an ElectraFi export. */ const REQUIRED_COLUMNS = [ @@ -63,6 +63,8 @@ export interface ElectraFiRow { chargeLimit: number | null; softwareVersion: string | null; driveMode: string | null; + /** The active driver profile's name, which ElectraFi keeps in Tesla's `autopark_state_v2`. */ + driverName: string | null; charging: ChargingState; /** ElectraFi's number for the charge; null outside one. */ chargeNumber: number | null; @@ -89,15 +91,6 @@ const round = (n: number | null, places: number): number | null => n == null ? null : Math.round(n * 10 ** places) / 10 ** places; const times = (n: number | null, by: number): number | null => (n == null ? null : n * by); -export interface LocalTime { - year: number; - month: number; - day: number; - hour: number; - minute: number; - second: number; -} - /** Exports write `2026-10-29 8:02:08`, with the hour unpadded in the morning. */ export function parseLocalTime(value: string | undefined): LocalTime | null { const m = /^(\d{4})-(\d{1,2})-(\d{1,2})[ T](\d{1,2}):(\d{2}):(\d{2})$/.exec(value?.trim() ?? ''); @@ -157,6 +150,7 @@ export function parseRow(cells: Record): ElectraFiRow | null { chargeLimit: num(cells['charge_limit_soc']), softwareVersion: text(cells['car_version']), driveMode: text(cells['spoiler_type']), + driverName: text(cells['autopark_state_v2']), charging: charging === 'Charging' || charging === 'Complete' || charging === 'Disconnected' ? charging @@ -226,105 +220,6 @@ export const ROW_STATE_FIELDS: readonly string[] = [ 'chargerStatus', ]; -/** - * Local times to instants. An hour that clocks repeat is ambiguous, so a row - * takes the first reading that isn't before the row ahead of it; rows are in - * order, so this walks through the repeat as the export did. - */ -export class TimeZoneClock { - private readonly format: Intl.DateTimeFormat; - private previous = -Infinity; - private readonly offsets = new Map(); - - constructor(timeZone: string) { - this.format = new Intl.DateTimeFormat('en-US', { - timeZone, - hourCycle: 'h23', - year: 'numeric', - month: 'numeric', - day: 'numeric', - hour: 'numeric', - minute: 'numeric', - second: 'numeric', - }); - } - - /** The zone's offset from UTC at an instant (ms). */ - private offsetAt(at: number): number { - const parts = Object.fromEntries( - this.format.formatToParts(at).map((p) => [p.type, Number(p.value)]), - ) as Record; - const wall = Date.UTC( - parts['year']!, - parts['month']! - 1, - parts['day']!, - parts['hour']!, - parts['minute']!, - parts['second']!, - ); - return wall - Math.floor(at / 1000) * 1000; - } - - /** The instant of a local time, keeping rows in order. */ - next(local: LocalTime): number { - const at = this.candidates(local).find((t) => t >= this.previous) ?? this.first(local); - this.previous = at; - return at; - } - - /** The instant of a local time, on its own (the earlier one if clocks repeat it). */ - first(local: LocalTime): number { - return this.candidates(local)[0]!; - } - - private candidates(local: LocalTime): number[] { - const wall = Date.UTC( - local.year, - local.month - 1, - local.day, - local.hour, - local.minute, - local.second, - ); - return this.offsetsNear(wall).map((offset) => wall - offset); - } - - /** - * The offsets a local time can have. Read for every row of an export and - * slow to work out, so kept per 15 minutes, the finest clocks change on. - */ - private offsetsNear(wall: number): number[] { - const slot = Math.floor(wall / OFFSET_SLOT_MS); - const known = this.offsets.get(slot); - if (known) return known; - const before = this.offsetAt(wall - 86_400_000); - const found = [...new Set([before, this.offsetAt(wall + 86_400_000)])] - .filter((offset) => this.offsetAt(wall - offset) === offset) - .sort((a, b) => b - a); - // A time clocks skip doesn't exist; it reads as the offset before the skip. - const offsets = found.length > 0 ? found : [before]; - this.offsets.set(slot, offsets); - return offsets; - } - - /** An instant's calendar date in this zone, as YYYY-MM-DD. */ - dateOf(at: number): string { - const parts = Object.fromEntries( - this.format.formatToParts(at).map((p) => [p.type, p.value]), - ) as Record; - return `${parts['year']}-${parts['month']!.padStart(2, '0')}-${parts['day']!.padStart(2, '0')}`; - } -} - -export function isTimeZone(value: string): boolean { - try { - new Intl.DateTimeFormat('en-US', { timeZone: value }); - return true; - } catch { - return false; - } -} - /** Raises NotAnExportError unless the header has the columns an export has. */ export function assertExportHeader(columns: readonly string[]): void { if (!REQUIRED_COLUMNS.every((c) => columns.includes(c))) throw new NotAnExportError(); diff --git a/server/src/services/electrafi/segment.ts b/server/src/services/electrafi/segment.ts index efc7de0..97034f2 100644 --- a/server/src/services/electrafi/segment.ts +++ b/server/src/services/electrafi/segment.ts @@ -1,4 +1,5 @@ import { elevationChange } from '../drive-metrics.js'; +import type { ChargeSegment, DriveSegment, Readings } from '../imports/format.js'; import { haversineKm } from '../state-utils.js'; import type { ElectraFiRow } from './parse.js'; @@ -10,48 +11,6 @@ const CHARGE_GAP_MS = 5 * 60_000; /** A charge's first row with more energy than this was already under way. */ const MID_CHARGE_KWH = 0.5; -export interface Readings { - mileageM: number | null; - battery: number | null; - rangeKm: number | null; - lat: number | null; - lon: number | null; -} - -export interface DriveSegment { - startedAt: number; - endedAt: number; - start: Readings; - end: Readings; - batteryCapacityKwh: number | null; - driveMode: string | null; - distanceKm: number | null; - elevationGainM: number | null; - elevationLossM: number | null; -} - -export interface CurveSample { - at: number; - powerKw: number | null; - soc: number | null; -} - -export interface ChargeSegment { - chargeNumber: number; - startedAt: number; - endedAt: number; - chargingSeconds: number; - startSoc: number | null; - endSoc: number | null; - energyKwh: number | null; - rangeAddedKm: number | null; - avgPowerKw: number | null; - maxPowerKw: number | null; - lat: number | null; - lon: number | null; - curve: CurveSample[]; -} - interface Timed { row: ElectraFiRow; at: number; @@ -184,11 +143,15 @@ export class DriveSegmenter { distanceKm: byOdometer || (open.gpsKm > 0 ? open.gpsKm : byOdometer), elevationGainM: elevation?.gainM ?? null, elevationLossM: elevation?.lossM ?? null, + // As live: the profile active as it set off, not one switched to on the way. + driverName: open.first.row.driverName ?? open.start.row.driverName, + destination: null, }); } } interface OpenCharge extends ChargeSegment { + chargeNumber: number; firstEnergy: number | null; powerSum: number; powerCount: number; @@ -216,6 +179,8 @@ export class ChargeSegmenter { startSoc: null, endSoc: null, energyKwh: null, + packKwh: null, + thermalKwh: null, rangeAddedKm: null, avgPowerKw: null, maxPowerKw: null, @@ -261,13 +226,14 @@ export class ChargeSegmenter { this.open = null; if (!open) return; done.push({ - chargeNumber: open.chargeNumber, startedAt: open.startedAt, endedAt: open.endedAt, chargingSeconds: Math.round(open.chargingSeconds), startSoc: (open.firstEnergy ?? 0) > MID_CHARGE_KWH ? null : open.startSoc, endSoc: open.endSoc, energyKwh: open.energyKwh, + packKwh: null, + thermalKwh: null, rangeAddedKm: open.rangeAddedKm, avgPowerKw: open.powerCount > 0 ? open.powerSum / open.powerCount : null, maxPowerKw: open.maxPowerKw, diff --git a/server/src/services/imports/format.ts b/server/src/services/imports/format.ts new file mode 100644 index 0000000..1f40151 --- /dev/null +++ b/server/src/services/imports/format.ts @@ -0,0 +1,160 @@ +import { Transform } from 'node:stream'; +import type { ImportFormat } from '../../api-types.js'; +import type { VehicleState } from '../../rivian/types.js'; + +/** Something the user can act on; the route answers with its message. */ +export class ImportError extends Error { + constructor( + message: string, + readonly status = 400, + ) { + super(message); + } +} + +/** Stops a stream past `limit` bytes, so a small archive can't fill the disk. */ +export function capped(limit: number): Transform { + let seen = 0; + return new Transform({ + transform(chunk: Buffer, _encoding, done) { + seen += chunk.length; + if (seen > limit) done(new ImportError('This file is too large to import.', 413)); + else done(null, chunk); + }, + }); +} + +export interface Readings { + mileageM: number | null; + battery: number | null; + rangeKm: number | null; + lat: number | null; + lon: number | null; +} + +export interface Destination { + name: string; + lat: number; + lon: number; +} + +export interface DriveSegment { + startedAt: number; + endedAt: number; + start: Readings; + end: Readings; + batteryCapacityKwh: number | null; + driveMode: string | null; + distanceKm: number | null; + elevationGainM: number | null; + elevationLossM: number | null; + /** The name of the driver profile the vehicle reported, where the export has it. */ + driverName: string | null; + /** Where the vehicle's navigation was headed. */ + destination: Destination | null; +} + +/** A place the owner named in the other app. */ +export interface ImportPlace { + name: string; + lat: number; + lon: number; + radiusM: number; + isHome: boolean; + /** Price per kWh charged there; null = not set. */ + ratePerKwh: number | null; +} + +export interface CurveSample { + at: number; + powerKw: number | null; + soc: number | null; +} + +export interface ChargeSegment { + startedAt: number; + endedAt: number; + chargingSeconds: number; + startSoc: number | null; + endSoc: number | null; + energyKwh: number | null; + /** Rivian's breakdown, where the export has it: into the pack, and heating or cooling it. */ + packKwh: number | null; + thermalKwh: number | null; + rangeAddedKm: number | null; + avgPowerKw: number | null; + maxPowerKw: number | null; + lat: number | null; + lon: number | null; + curve: CurveSample[]; +} + +/** What an export's files hold, read once and kept until the files or choices change. */ +export interface Scan { + rows: number; + /** Rows that couldn't be read, and are left out. */ + unreadable: number; + firstAt: number | null; + lastAt: number | null; + firstMileage: { at: number; value: number } | null; + lastMileage: { at: number; value: number } | null; + drives: DriveSegment[]; + charges: ChargeSegment[]; + /** The account's saved places; not tied to dates, so offered whatever the range. */ + places: ImportPlace[]; + /** When each row with readings was taken, so a new date range needn't read the files again. */ + readingTimes: number[]; +} + +export interface Fix { + lat: number; + lon: number; + speedKmh: number | null; + altitudeM: number | null; + bearing: number | null; +} + +/** One moment of an export, as RivianMate would have recorded it live. */ +export interface Reading { + at: number; + /** Rivian's field names, with only what the export has a value for. */ + state: VehicleState; + /** Whether the vehicle was sending state then, so a heartbeat can be due. */ + reporting: boolean; + fix: Fix | null; +} + +/** The choices that decide what an export's files are read as. */ +export interface ReadOptions { + timeZone: string; + /** The vehicle being imported to, for an export that holds several. */ + vin: string; +} + +/** One kind of export the import can read. */ +export interface ExportFormat { + id: ImportFormat; + /** A whole export comes as one file, so another upload replaces it. */ + singleFile: boolean; + /** Says why an upload that isn't this kind of export can't be read. */ + unreadable: string; + /** + * Turns an upload into the file kept for import, or throws if it isn't + * this kind of export. Returns the VINs it has data for, or null when it + * doesn't say. + */ + store( + upload: string, + stem: string, + maxBytes: number, + ): Promise<{ path: string; vins: string[] | null }>; + /** Which of the choices the scan depends on. */ + scanKey(options: ReadOptions): string; + scan(paths: readonly string[], options: ReadOptions): Promise; + /** The readings in time order, each once. */ + readings(paths: readonly string[], options: ReadOptions): AsyncIterable; + /** Every state field `readings` can hold. */ + stateFields: readonly string[]; + /** Repeats the last fix between new ones, speed included, so one that hasn't moved is dropped. */ + repeatsFixes: boolean; +} diff --git a/server/src/services/electrafi/imports.db.test.ts b/server/src/services/imports/imports.db.test.ts similarity index 71% rename from server/src/services/electrafi/imports.db.test.ts rename to server/src/services/imports/imports.db.test.ts index 2103044..25733f9 100644 --- a/server/src/services/electrafi/imports.db.test.ts +++ b/server/src/services/imports/imports.db.test.ts @@ -12,6 +12,7 @@ import { createDb, runMigrations } from '../../db/client.js'; import { chargingCurvePoints, chargingSessions, + drivers, drives, importBatches, importSessionFills, @@ -25,15 +26,16 @@ import { import { HistoryKeys } from '../history-keys.js'; import { changesBetween, latestChanges } from '../state-history.js'; import { loadStateSpans } from '../state-spans.js'; -import { ElectraFiImports, ImportError } from './imports.js'; +import { ImportError, Imports } from './imports.js'; import { asleep, awake, chargeRows, csvOf, driveRows } from '../../testing/electrafi-rows.js'; +import { roamerExport } from '../../testing/roamer-archive.js'; const url = process.env.TEST_DATABASE_URL; const here = dirname(fileURLToPath(import.meta.url)); describe.skipIf(!url)('ElectraFi import with Postgres', () => { let handle: ReturnType; - let imports: ElectraFiImports; + let imports: Imports; let lookups: ReturnType; const NOW = Date.parse('2026-11-15T00:00:00Z'); const selection = { @@ -80,13 +82,7 @@ describe.skipIf(!url)('ElectraFi import with Postgres', () => { await handle.sql`TRUNCATE vehicles, state_fields, import_batches, saved_locations RESTART IDENTITY CASCADE`; await handle.db.insert(vehicles).values({ id: 'v1', vin: 'VIN1' }); lookups = vi.fn(); - imports = new ElectraFiImports( - handle.db, - new HistoryKeys(handle.db), - undefined, - lookups, - () => NOW, - ); + imports = new Imports(handle.db, new HistoryKeys(handle.db), undefined, lookups, () => NOW); }); it('imports a drive with its route and history, and queues its lookups', async () => { @@ -121,6 +117,22 @@ describe.skipIf(!url)('ElectraFi import with Postgres', () => { expect(tires[0]!.value).toBe(3.35); }); + it('gives a drive the driver ElectraFi named, matching one RivianMate knows', async () => { + const [laura] = await handle.db + .insert(drivers) + .values({ vehicleId: 'v1', profileId: 'p-laura', reportedName: 'Laura V.' }) + .returning(); + const preview = await imports.preview( + (await upload(csvOf(driveRows().map((r) => ({ ...r, autopark_state_v2: 'Laura V.' }))))).id, + selection, + ); + expect(preview.drives.map((d) => d.driver)).toEqual(['Laura V.']); + await importCsv(csvOf(driveRows().map((r) => ({ ...r, autopark_state_v2: 'Laura V.' })))); + const [drive] = await handle.db.select().from(drives); + expect(drive).toMatchObject({ driverId: laura!.id, driverSource: 'profile' }); + expect(await handle.db.select().from(drivers)).toHaveLength(1); + }); + it('gives History what it needs to show awake, asleep, driving and charging', async () => { await importCsv( csvOf([ @@ -164,7 +176,7 @@ describe.skipIf(!url)('ElectraFi import with Postgres', () => { }); it('refuses a file over the limit, gzipped or not', async () => { - const small = new ElectraFiImports( + const small = new Imports( handle.db, new HistoryKeys(handle.db), undefined, @@ -771,4 +783,263 @@ describe.skipIf(!url)('ElectraFi import with Postgres', () => { await expect(imports.start(second.id, selection)).rejects.toThrow(/already running/); await imports.idle(); }); + + describe('a Roamer export', () => { + const exported = (vin = 'VIN1') => + roamerExport( + { + drive_sessions: [ + { + started_at: '2026-03-01T10:00:00', + ended_at: '2026-03-01T10:30:00', + start_odometer: 100, + end_odometer: 120, + start_battery_level: 80, + end_battery_level: 75, + total_distance: 20, + primary_drive_mode: 'everyday', + }, + ], + charging_sessions: [ + { + started_at: '2026-03-01T12:00:00', + ended_at: '2026-03-01T13:00:00', + periods: [{ started_at: '2026-03-01T12:00:00Z', ended_at: '2026-03-01T12:50:00Z' }], + start_battery_level: 50, + end_battery_level: 60, + total_energy_kwh: 11, + pack_kwh: 10.5, + thermal_kwh: 0.2, + latitude: 33.1, + longitude: -117.2, + }, + ], + vehicle_metrics: [ + { + created_at: '2026-03-01T09:50:00+00:00', + vehicle_mileage: 100000, + gear_status: 'park', + power_state: 'ready', + battery_capacity: 110, + }, + { created_at: '2026-03-01T10:01:00+00:00', gear_status: 'drive', power_state: 'go' }, + { + created_at: '2026-03-01T10:30:00+00:00', + gear_status: 'park', + vehicle_mileage: 120000, + }, + { + created_at: '2026-03-01T12:10:00+00:00', + battery_level: 52, + charger_state: 'charging_active', + }, + ], + location_metrics: [ + { + created_at: '2026-03-01T10:05:00+00:00', + latitude: 33.1, + longitude: -117.2, + speed: 10, + }, + { + created_at: '2026-03-01T10:15:00+00:00', + latitude: 33.2, + longitude: -117.2, + speed: 12, + }, + // Parked: not a new place, and not moving. + { + created_at: '2026-03-01T10:40:00+00:00', + latitude: 33.2, + longitude: -117.2, + speed: 0, + }, + ], + charging_metrics: [ + { created_at: '2026-03-01T12:05:00+00:00', state: 3, power_kw: 7.4 }, + { created_at: '2026-03-01T12:15:00+00:00', state: 3, power_kw: 7.5 }, + ], + }, + vin, + ); + const uploadArchive = (bytes: Buffer, draft?: string) => + imports.addFile(draft, 'roamer.zip', Readable.from(bytes), 'roamer'); + + it('imports its drives, charges, routes and history', async () => { + const draft = await uploadArchive(exported()); + expect(draft).toMatchObject({ format: 'roamer', files: ['roamer.zip'], vehicleIds: ['v1'] }); + const batch = await imports.start(draft.id, selection); + await imports.idle(); + const [row] = await handle.db.select().from(importBatches); + expect(row).toMatchObject({ id: batch.id, format: 'roamer', status: 'done' }); + expect(row!.counts).toMatchObject({ drives: 1, sessions: 1, locationPoints: 2 }); + + const [drive] = await handle.db.select().from(drives); + expect(drive).toMatchObject({ + startMileageM: 100000, + endMileageM: 120000, + distanceKm: 20, + batteryCapacityKwh: 110, + driveMode: 'everyday', + }); + const points = await handle.db.select().from(locationPoints).orderBy(asc(locationPoints.ts)); + expect(points.map((p) => p.driveId)).toEqual([drive!.id, drive!.id]); + expect(points[0]!.speedKmh).toBeCloseTo(36); + + const [session] = await handle.db.select().from(chargingSessions); + expect(session).toMatchObject({ + source: 'roamer', + chargingSeconds: 50 * 60, + energyKwh: 11, + packKwh: 10.5, + startSoc: 50, + }); + expect(session!.thermalKwh).toBeCloseTo(0.2); + const curve = await handle.db + .select() + .from(chargingCurvePoints) + .orderBy(asc(chargingCurvePoints.ts)); + expect(curve.map((p) => [p.source, p.powerKw, p.soc])).toEqual([ + ['roamer', expect.closeTo(7.4), 50], + ['roamer', expect.closeTo(7.5), 52], + ]); + + const gear = await changesBetween( + handle.db, + 'v1', + ['gearStatus'], + new Date(0), + new Date('2027-01-01'), + ); + expect(gear.map((c) => c.value)).toEqual(['park', 'drive', 'park']); + }); + + it('brings who drove, where navigation was headed, and saved places, which undo keeps', async () => { + const [mitch] = await handle.db + .insert(drivers) + .values({ vehicleId: 'v1', profileId: 'p-mitch', reportedName: 'Mitch V.' }) + .returning(); + await handle.db + .insert(savedLocations) + .values({ name: 'My house', lat: 33.0715, lon: -117.2566, radiusM: 150, isHome: true }); + const trip = (hour: number, driver: string) => ({ + started_at: `2026-03-01T${hour}:00:00`, + ended_at: `2026-03-01T${hour}:30:00`, + start_odometer: hour, + end_odometer: hour + 10, + total_distance: 10, + primary_driver_profile: driver, + }); + const draft = await uploadArchive( + roamerExport( + { + drive_sessions: [trip(10, 'Mitch V.'), trip(12, 'Laura V.')], + navigation_metrics: [ + { + observed_at: '2026-03-01T12:05:00+00:00', + navigation_active: true, + destination_name: 'Beach', + destination_latitude: 32.7, + destination_longitude: -117.1, + }, + ], + saved_places: [ + { + id: 'h', + name: 'Home', + latitude: 33.0715, + longitude: -117.2566, + radius_meters: 100, + }, + { id: 'w', name: 'Work', latitude: 32.88, longitude: -117.22, radius_meters: 100 }, + ], + rivian_favorites: [{ key: 'home', tagged_location_id: 'h' }], + }, + 'VIN1', + ), + ); + const preview = await imports.preview(draft.id, selection); + expect(preview).toMatchObject({ placesNew: ['Work'], placesRecorded: 1 }); + expect(preview.drives.map((d) => d.driver)).toEqual(['Mitch V.', 'Laura V.']); + + const batch = await imports.start(draft.id, selection); + await imports.idle(); + const [row] = await handle.db.select().from(importBatches); + expect(row!.counts).toMatchObject({ drives: 2, places: 1 }); + + const imported = await handle.db.select().from(drives).orderBy(asc(drives.startedAt)); + const laura = (await handle.db.select().from(drivers)).find((d) => d.id !== mitch!.id)!; + expect(laura).toMatchObject({ reportedName: 'Laura V.', profileId: null }); + expect(imported.map((d) => [d.driverId, d.driverSource])).toEqual([ + [mitch!.id, 'profile'], + [laura.id, 'profile'], + ]); + expect(imported.map((d) => d.destinationName)).toEqual([null, 'Beach']); + const places = await handle.db.select().from(savedLocations).orderBy(asc(savedLocations.id)); + expect(places.map((p) => [p.name, p.isHome])).toEqual([ + ['My house', true], + ['Work', false], + ]); + + // The drives go; the people and places stay. + await imports.undo(batch.id); + expect(await handle.db.select().from(drives)).toHaveLength(0); + expect((await handle.db.select().from(drivers)).map((d) => d.id)).toEqual([ + mitch!.id, + laura.id, + ]); + expect((await handle.db.select().from(savedLocations)).map((p) => p.name)).toEqual([ + 'My house', + 'Work', + ]); + }); + + it('has only the vehicles it holds data for to import to', async () => { + const draft = await uploadArchive(exported('SOMEONE-ELSE')); + expect(draft.vehicleIds).toEqual([]); + await expect(imports.preview(draft.id, selection)).rejects.toThrow( + /no data for that vehicle/, + ); + }); + + it('is one file: another upload replaces it, and nothing is added to it', async () => { + const first = await uploadArchive(exported()); + await expect(uploadArchive(exported(), first.id)).rejects.toThrow(/single file/); + await expect( + imports.addFile(first.id, 'export.csv', Readable.from(csvOf(driveRows())), 'electrafi'), + ).rejects.toThrow(/one kind of export/); + const second = await uploadArchive(exported()); + expect((await imports.list()).draft?.id).toBe(second.id); + await expect(uploadArchive(Buffer.from(csvOf(driveRows())))).rejects.toThrow(/Roamer export/); + }); + + it('adds the breakdown to a charge Rivian recorded without one', async () => { + await handle.db.insert(chargingSessions).values({ + vehicleId: 'v1', + source: 'rivian', + rivianTransactionId: 'tx1', + startedAt: new Date('2026-03-01T12:00:30Z'), + endedAt: new Date('2026-03-01T13:00:00Z'), + energyKwh: 11.2, + }); + const [before] = await handle.db.select().from(chargingSessions); + const draft = await uploadArchive(exported()); + const preview = await imports.preview(draft.id, selection); + expect(preview.charges[0]).toMatchObject({ status: 'enrich' }); + expect(preview.charges[0]!.adds).toContain('breakdown'); + expect(preview.charges[0]!.adds).not.toContain('energy'); + const batch = await imports.start(draft.id, selection); + await imports.idle(); + const [session] = await handle.db.select().from(chargingSessions); + expect(session).toMatchObject({ source: 'rivian', energyKwh: 11.2, packKwh: 10.5 }); + expect(await handle.db.select().from(chargingCurvePoints)).toHaveLength(2); + + // Undo puts the charge back as Rivian had it. + await imports.undo(batch.id); + const [after] = await handle.db.select().from(chargingSessions); + expect(after).toEqual({ ...before!, id: after!.id }); + expect(await handle.db.select().from(chargingCurvePoints)).toHaveLength(0); + expect(await handle.db.select().from(importSessionFills)).toHaveLength(0); + }); + }); }); diff --git a/server/src/services/electrafi/imports.ts b/server/src/services/imports/imports.ts similarity index 74% rename from server/src/services/electrafi/imports.ts rename to server/src/services/imports/imports.ts index 229d514..25e81c8 100644 --- a/server/src/services/electrafi/imports.ts +++ b/server/src/services/imports/imports.ts @@ -1,16 +1,16 @@ import { and, asc, desc, eq, inArray, ne, sql } from 'drizzle-orm'; import { randomUUID } from 'node:crypto'; -import { createReadStream, createWriteStream } from 'node:fs'; -import { mkdtemp, open, readdir, rm } from 'node:fs/promises'; +import { createWriteStream } from 'node:fs'; +import { mkdtemp, readdir, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { basename, join } from 'node:path'; -import { Transform, type Readable } from 'node:stream'; +import type { Readable } from 'node:stream'; import { pipeline } from 'node:stream/promises'; -import { createGunzip } from 'node:zlib'; import type { ImportBatchDto, ImportCounts, ImportDraftDto, + ImportFormat, ImportPreviewDto, ImportSessionAddition, ImportSelection, @@ -20,6 +20,7 @@ import type { Db } from '../../db/client.js'; import { chargingCurvePoints, chargingSessions, + drivers, drives, importBatches, importSessionFills, @@ -33,6 +34,7 @@ import { type Logger, silentLogger } from '../../logger.js'; import { isTeslaMateVehicle } from '../../teslamate/ids.js'; import type { VehicleState } from '../../rivian/types.js'; import { isListedDrive } from '../drive-metrics.js'; +import { electraFiFormat } from '../electrafi/format.js'; import type { HistoryKeys } from '../history-keys.js'; import { MIN_MOVE_KM } from '../location-writer.js'; import { @@ -44,82 +46,72 @@ import { changesBetween, latestChanges, } from '../state-history.js'; +import { roamerFormat } from '../roamer/format.js'; import { haversineKm } from '../state-utils.js'; import { - type ElectraFiRow, - ROW_STATE_FIELDS, - firstDataId, - isTimeZone, - readRows, - rowState, -} from './parse.js'; + type ChargeSegment, + type DriveSegment, + type ExportFormat, + type Fix, + ImportError, + type ReadOptions, + type Reading, + type Scan, + capped, +} from './format.js'; import { type Existing, type Plan, type PlannedCharge, - type Scan, buildPlan, previewOf, rangeOf, - scanFiles, } from './plan.js'; -import type { ChargeSegment } from './segment.js'; +import { isTimeZone } from './time-zone.js'; + +export { ImportError } from './format.js'; const TEMP_PREFIX = 'rivianmate-import-'; -/** Decompressed, per file; a year of minute-by-minute rows is about 180 MB. */ +/** + * Per file, as stored: a year of ElectraFi's minute-by-minute rows is about + * 180 MB decompressed, and a Roamer archive stays compressed. + */ export const MAX_FILE_BYTES = 1024 ** 3; const MAX_FILES = 240; const INSERT_BATCH = 1000; const RESTARTED = 'RivianMate restarted during the import.'; -/** Something the user can act on; the route answers with its message. */ -export class ImportError extends Error { - constructor( - message: string, - readonly status = 400, - ) { - super(message); - } -} +const FORMATS: Readonly> = { + electrafi: electraFiFormat, + roamer: roamerFormat, +}; interface Draft { id: string; dir: string; + format: ExportFormat; files: { name: string; path: string }[]; - /** The files as last read: only they and the time zone change it, not the dates. */ - scan?: { timeZone: string; scan: Scan }; -} - -/** Stops a stream past `limit` bytes, so a small gzip can't fill the disk. */ -function capped(limit: number): Transform { - let seen = 0; - return new Transform({ - transform(chunk: Buffer, _encoding, done) { - seen += chunk.length; - if (seen > limit) done(new ImportError('This file is too large to import.', 413)); - else done(null, chunk); - }, - }); -} - -async function isGzip(path: string): Promise { - const file = await open(path, 'r'); - try { - const { bytesRead, buffer } = await file.read(Buffer.alloc(2), 0, 2, 0); - return bytesRead === 2 && buffer[0] === 0x1f && buffer[1] === 0x8b; - } finally { - await file.close(); - } + /** The VINs the files have data for; null when they don't say. */ + vins: string[] | null; + /** The files as last read: only they and the choices in its key change it, not the dates. */ + scan?: { key: string; scan: Scan }; } function emptyCounts(): ImportCounts { - return { drives: 0, sessions: 0, sessionsEnriched: 0, locationPoints: 0, stateChanges: 0 }; + return { + drives: 0, + sessions: 0, + sessionsEnriched: 0, + locationPoints: 0, + stateChanges: 0, + }; } function toDto(row: typeof importBatches.$inferSelect, progress: number | null): ImportBatchDto { return { id: row.id, vehicleId: row.vehicleId, + format: row.format, files: row.files, timeZone: row.timeZone, from: row.fromDate, @@ -134,16 +126,16 @@ function toDto(row: typeof importBatches.$inferSelect, progress: number | null): } /** - * Imports ElectraFi exports (one account per car, so the owner says which - * car). Files wait in a temp folder between upload and import, and go when - * it's done; a restart loses them along with any import under way. + * Imports other apps' exports, one kind per upload. Files wait in a temp + * folder between upload and import, and go when it's done; a restart loses + * them along with any import under way. * * Writes go straight to the tables, not through the monitor: what happened * months ago mustn't send notifications or look like live state. What's * already stored always wins, and every added row carries the batch's id so * an import can be removed whole. */ -export class ElectraFiImports { +export class Imports { private draft: Draft | null = null; /** The import under way (one at a time), and how far through its rows it is. */ private running: { id: number; done: number; total: number } | null = null; @@ -156,6 +148,7 @@ export class ElectraFiImports { private readonly onDrivesAdded: () => void = () => {}, private readonly now: () => number = Date.now, private readonly maxFileBytes = MAX_FILE_BYTES, + private readonly formats = FORMATS, ) {} /** After a restart: an import that was running is gone, and so are its temp files. */ @@ -175,52 +168,51 @@ export class ElectraFiImports { ); } - /** Stores an uploaded file, as sent or gzipped, adding it to the waiting draft. */ + /** Stores an uploaded file, adding it to the waiting draft of the same kind of export. */ async addFile( draftId: string | undefined, name: string, body: Readable, + formatId: ImportFormat = 'electrafi', ): Promise { + const format = this.formats[formatId]; if (draftId && this.draft?.id !== draftId) { throw new ImportError('That upload has gone. Add the files again.', 404); } + if (draftId && this.draft!.format !== format) { + throw new ImportError('Add files from one kind of export at a time.'); + } + if (draftId && format.singleFile) { + throw new ImportError('This export is a single file. Choose it again to replace it.'); + } // A new upload replaces whatever was left waiting. if (!draftId) await this.discardDraft(); const draft = (this.draft ??= { id: randomUUID(), dir: await mkdtemp(join(tmpdir(), TEMP_PREFIX)), + format, files: [], + vins: null, }); if (draft.files.length >= MAX_FILES) throw new ImportError('Too many files for one import.'); const stem = join(draft.dir, String(draft.files.length)); - const path = `${stem}.csv`; + let stored: { path: string; vins: string[] | null }; try { await pipeline(body, capped(this.maxFileBytes), createWriteStream(`${stem}.upload`)); - const stored = createWriteStream(path); - const uploaded = createReadStream(`${stem}.upload`); - if (await isGzip(`${stem}.upload`)) { - await pipeline(uploaded, createGunzip(), capped(this.maxFileBytes), stored); - } else { - await pipeline(uploaded, stored); - } - // Reads the header and first row, so a wrong file is refused here. - await firstDataId(path); + stored = await format.store(`${stem}.upload`, stem, this.maxFileBytes); } catch (err) { - await rm(path, { force: true }); // Nothing to wait for if this was the first file. if (draft.files.length === 0) await this.discardDraft(); if (err instanceof ImportError) throw err; - throw new ImportError( - err instanceof Error && err.message.includes('ElectraFi') - ? err.message - : "Couldn't read this file as an ElectraFi export.", - ); + this.log.info({ err, format: format.id }, 'upload not readable'); + throw new ImportError(format.unreadable); } finally { await rm(`${stem}.upload`, { force: true }); } draft.scan = undefined; - draft.files.push({ name: basename(name).slice(0, 200) || 'export.csv', path }); + draft.vins = stored.vins; + draft.files.push({ name: basename(name).slice(0, 200) || 'export', path: stored.path }); return this.draftDto(draft); } @@ -231,11 +223,28 @@ export class ElectraFiImports { await rm(draft.dir, { recursive: true, force: true }); } - private draftDto(draft: Draft): ImportDraftDto { - return { id: draft.id, files: draft.files.map((f) => f.name) }; + private async draftDto(draft: Draft): Promise { + const vins = draft.vins; + const covered = + vins && + (vins.length === 0 + ? [] + : await this.db + .select({ id: vehicles.id }) + .from(vehicles) + .where(inArray(vehicles.vin, vins))); + return { + id: draft.id, + format: draft.format.id, + files: draft.files.map((f) => f.name), + vehicleIds: covered && covered.map((v) => v.id), + }; } - private async requireSelection(draftId: string, selection: ImportSelection): Promise { + private async requireSelection( + draftId: string, + selection: ImportSelection, + ): Promise<{ draft: Draft; options: ReadOptions }> { const draft = this.draft; if (!draft || draft.id !== draftId || draft.files.length === 0) { throw new ImportError('That upload has gone. Add the files again.', 404); @@ -243,17 +252,20 @@ export class ElectraFiImports { if (!isTimeZone(selection.timeZone)) throw new ImportError('Choose a valid time zone.'); if (selection.from > selection.to) throw new ImportError('The start date is after the end.'); const [vehicle] = await this.db - .select({ id: vehicles.id }) + .select({ id: vehicles.id, vin: vehicles.vin }) .from(vehicles) .where(eq(vehicles.id, selection.vehicleId)); if (!vehicle || isTeslaMateVehicle(selection.vehicleId)) { throw new ImportError('Choose one of your Rivian vehicles.'); } - return draft; + if (draft.vins && !draft.vins.includes(vehicle.vin)) { + throw new ImportError('This export has no data for that vehicle.'); + } + return { draft, options: { timeZone: selection.timeZone, vin: vehicle.vin } }; } private async existing(vehicleId: string, seq: number): Promise { - const [beats, driveRows, sessionRows, mileage] = await Promise.all([ + const [beats, driveRows, sessionRows, mileage, places] = await Promise.all([ this.db .select({ at: stateHeartbeats.at, importId: stateHeartbeats.importId }) .from(stateHeartbeats) @@ -273,6 +285,14 @@ export class ElectraFiImports { .from(chargingSessions) .where(eq(chargingSessions.vehicleId, vehicleId)), changesBetween(this.db, vehicleId, ['vehicleMileage'], new Date(0), END_OF_TIME), + this.db + .select({ + lat: savedLocations.lat, + lon: savedLocations.lon, + radiusM: savedLocations.radiusM, + isHome: savedLocations.isHome, + }) + .from(savedLocations), ]); const live = beats.filter((b) => b.importId == null).at(-1); return { @@ -291,6 +311,7 @@ export class ElectraFiImports { mileage: mileage.flatMap((c) => typeof c.value === 'number' ? [{ at: c.at, value: c.value }] : [], ), + places, }; } @@ -301,16 +322,18 @@ export class ElectraFiImports { } async preview(draftId: string, selection: ImportSelection): Promise { - const draft = await this.requireSelection(draftId, selection); - const scan = await this.scanOf(draft, selection.timeZone); + const { draft, options } = await this.requireSelection(draftId, selection); + const scan = await this.scanOf(draft, options); const seq = await this.vehicleSeq(selection.vehicleId); const existing = await this.existing(selection.vehicleId, seq); const range = rangeOf(selection.from, selection.to, selection.timeZone); const plan = buildPlan(scan, existing, range, this.now()); - const counted = { readingsNew: 0, readingsRecorded: 0 }; + const counted = { readingsNew: 0, readingsRecorded: 0, readingsNewer: 0 }; for (const at of scan.readingTimes) { if (plan.scope.includes(at)) counted.readingsNew += 1; - else if (at >= range.from && at < range.to) counted.readingsRecorded += 1; + else if (at < range.from || at >= range.to) continue; + else if (existing.lastLiveAt != null && at > existing.lastLiveAt) counted.readingsNewer += 1; + else counted.readingsRecorded += 1; } const bySession = new Map(); for (const { segment, sessionId } of plan.charges) { @@ -326,18 +349,20 @@ export class ElectraFiImports { return previewOf(scan, plan, existing, counted, { timeZone: selection.timeZone, now: this.now(), - places: await this.db.select().from(savedLocations), + // As drives will be named once the places it adds are there. + places: [...(await this.db.select().from(savedLocations)), ...plan.places], additions, }); } - private async scanOf(draft: Draft, timeZone: string): Promise { - if (draft.scan?.timeZone === timeZone) return draft.scan.scan; - const scan = await scanFiles( + private async scanOf(draft: Draft, options: ReadOptions): Promise { + const key = draft.format.scanKey(options); + if (draft.scan?.key === key) return draft.scan.scan; + const scan = await draft.format.scan( draft.files.map((f) => f.path), - timeZone, + options, ); - draft.scan = { timeZone, scan }; + draft.scan = { key, scan }; return scan; } @@ -345,12 +370,12 @@ export class ElectraFiImports { async start(draftId: string, selection: ImportSelection): Promise { if (this.running) throw new ImportError('An import is already running.', 409); if (this.undoing) throw new ImportError('An import is being undone. Try again shortly.', 409); - const draft = await this.requireSelection(draftId, selection); + const { draft, options } = await this.requireSelection(draftId, selection); const [batch] = await this.db .insert(importBatches) .values({ vehicleId: selection.vehicleId, - format: 'electrafi', + format: draft.format.id, files: draft.files.map((f) => f.name), timeZone: selection.timeZone, fromDate: selection.from, @@ -360,7 +385,7 @@ export class ElectraFiImports { this.running = { id: batch!.id, done: 0, total: 0 }; // The files now belong to the import. this.draft = null; - void this.run(batch!.id, draft, selection); + void this.run(batch!.id, draft, selection, options); return toDto(batch!, 0); } @@ -369,10 +394,15 @@ export class ElectraFiImports { while (this.running) await new Promise((r) => setTimeout(r, 10)); } - private async run(id: number, draft: Draft, selection: ImportSelection): Promise { + private async run( + id: number, + draft: Draft, + selection: ImportSelection, + options: ReadOptions, + ): Promise { + const format = draft.format.id; try { - const paths = draft.files.map((f) => f.path); - const scan = await this.scanOf(draft, selection.timeZone); + const scan = await this.scanOf(draft, options); if (this.running) this.running.total = scan.rows; const seq = await this.vehicleSeq(selection.vehicleId); const existing = await this.existing(selection.vehicleId, seq); @@ -382,15 +412,15 @@ export class ElectraFiImports { rangeOf(selection.from, selection.to, selection.timeZone), this.now(), ); - const counts = await this.write(id, seq, selection, paths, plan); + const counts = await this.write(id, seq, selection.vehicleId, draft, options, plan); await this.db .update(importBatches) .set({ status: 'done', counts, finishedAt: new Date(this.now()) }) .where(eq(importBatches.id, id)); - this.log.info({ batch: id, ...counts }, 'ElectraFi import finished'); + this.log.info({ batch: id, format, ...counts }, 'import finished'); this.onDrivesAdded(); } catch (err) { - this.log.error({ err, batch: id }, 'ElectraFi import failed'); + this.log.error({ err, batch: id, format }, 'import failed'); await this.fail(id, 'The import failed and nothing was kept. See the logs for why.').catch( (cleanup: unknown) => this.log.error({ err: cleanup, batch: id }, 'could not clean up a failed import'), @@ -414,27 +444,26 @@ export class ElectraFiImports { private async write( id: number, seq: number, - selection: ImportSelection, - paths: readonly string[], + vehicleId: string, + draft: Draft, + options: ReadOptions, plan: Plan, ): Promise { const counts = emptyCounts(); - const driveIds = await this.insertDrives(id, selection.vehicleId, plan); + const format = draft.format; + if (plan.places.length > 0) { + await this.db.insert(savedLocations).values(plan.places); + counts.places = plan.places.length; + } + const driveIds = await this.insertDrives(id, vehicleId, plan); counts.drives = plan.drives.filter((d) => isListedDrive(d.distanceKm)).length; - await this.writeCharges(id, selection.vehicleId, plan.charges, counts); - - const writer = new RowWriter( - this.db, - this.keys, - id, - seq, - selection.vehicleId, - driveIds, - counts, - ); - for await (const { row, at } of readRows(paths, selection.timeZone)) { + await this.writeCharges(id, vehicleId, format.id, plan.charges, counts); + + const writer = new RowWriter(this.db, this.keys, id, seq, vehicleId, driveIds, counts, format); + const paths = draft.files.map((f) => f.path); + for await (const reading of format.readings(paths, options)) { if (this.running) this.running.done += 1; - if (plan.scope.includes(at)) await writer.take(row, at); + if (plan.scope.includes(reading.at)) await writer.take(reading); else await writer.closeWindow(); } await writer.closeWindow(); @@ -456,6 +485,7 @@ export class ElectraFiImports { ): Promise<{ id: number; startedAt: number; endedAt: number }[]> { const out: { id: number; startedAt: number; endedAt: number }[] = []; const heldBack = new Date(this.now()); + const driverOf = await this.driversByName(vehicleId, plan.drives); for (let i = 0; i < plan.drives.length; i += INSERT_BATCH) { const rows = await this.db .insert(drives) @@ -480,6 +510,10 @@ export class ElectraFiImports { driveMode: d.driveMode, elevationGainM: d.elevationGainM, elevationLossM: d.elevationLossM, + destinationName: d.destination?.name ?? null, + destinationLat: d.destination?.lat ?? null, + destinationLon: d.destination?.lon ?? null, + ...driverOf(d.driverName), placesCheckedAt: heldBack, weatherCheckedAt: heldBack, })), @@ -492,9 +526,57 @@ export class ElectraFiImports { return out.sort((a, b) => a.startedAt - b.startedAt); } + /** + * The driver for each name the drives give: the vehicle's driver by that + * name (the user's or the one it reported), or else one made for it, with + * no profile until the vehicle reports one by that name. A drive with no + * name is left for the vehicle's profile, as a live one is. + */ + private async driversByName( + vehicleId: string, + planned: readonly DriveSegment[], + ): Promise<(name: string | null) => { driverId?: number; driverSource?: 'profile' }> { + const key = (name: string) => name.trim().toLowerCase(); + const names = new Map( + planned.flatMap((d) => (d.driverName ? [[key(d.driverName), d.driverName.trim()]] : [])), + ); + const ids = new Map(); + if (names.size > 0) { + const known = await this.db + .select({ + id: drivers.id, + name: drivers.name, + reportedName: drivers.reportedName, + profileId: drivers.profileId, + }) + .from(drivers) + .where(eq(drivers.vehicleId, vehicleId)) + // A profile's driver first, so it wins over one made from a name. + .orderBy(sql`${drivers.profileId} IS NULL`, asc(drivers.id)); + for (const driver of known) { + for (const name of [driver.name, driver.reportedName]) { + if (name && !ids.has(key(name))) ids.set(key(name), driver.id); + } + } + const missing = [...names].filter(([k]) => !ids.has(k)); + if (missing.length > 0) { + const made = await this.db + .insert(drivers) + .values(missing.map(([, name]) => ({ vehicleId, reportedName: name }))) + .returning({ id: drivers.id, reportedName: drivers.reportedName }); + for (const driver of made) ids.set(key(driver.reportedName!), driver.id); + } + } + return (name) => { + const driverId = name ? ids.get(key(name)) : undefined; + return driverId == null ? {} : { driverId, driverSource: 'profile' }; + }; + } + private async writeCharges( id: number, vehicleId: string, + source: ImportFormat, charges: readonly PlannedCharge[], counts: ImportCounts, ): Promise { @@ -508,13 +590,15 @@ export class ElectraFiImports { .values({ vehicleId, importId: id, - source: 'electrafi', + source, startedAt: new Date(segment.startedAt), endedAt: new Date(segment.endedAt), chargingSeconds: segment.chargingSeconds, startSoc: segment.startSoc, endSoc: segment.endSoc, energyKwh: segment.energyKwh, + packKwh: segment.packKwh, + thermalKwh: segment.thermalKwh, rangeAddedKm: segment.rangeAddedKm, avgPowerKw: segment.avgPowerKw, maxPowerKw: segment.maxPowerKw, @@ -529,7 +613,7 @@ export class ElectraFiImports { await tx.insert(importSessionFills).values({ importId: id, sessionId: row!.id, fields }); return row!; }); - await this.insertCurve(id, created.id, segment); + await this.insertCurve(id, created.id, source, segment); counts.sessions += 1; } else { bySession.set(sessionId, [...(bySession.get(sessionId) ?? []), segment]); @@ -542,7 +626,7 @@ export class ElectraFiImports { if (hasFill) await this.fillSession(id, sessionId, fill); if (addCurve) { for (const segment of bySession.get(sessionId)!) { - await this.insertCurve(id, sessionId, segment); + await this.insertCurve(id, sessionId, source, segment); } } if (hasFill || addCurve) counts.sessionsEnriched += 1; @@ -593,7 +677,12 @@ export class ElectraFiImports { return out; } - private async insertCurve(id: number, sessionId: number, segment: ChargeSegment): Promise { + private async insertCurve( + id: number, + sessionId: number, + source: ImportFormat, + segment: ChargeSegment, + ): Promise { for (let i = 0; i < segment.curve.length; i += INSERT_BATCH) { await this.db .insert(chargingCurvePoints) @@ -601,7 +690,7 @@ export class ElectraFiImports { segment.curve.slice(i, i + INSERT_BATCH).map((p) => ({ sessionId, importId: id, - source: 'electrafi' as const, + source, ts: new Date(p.at), powerKw: p.powerKw, soc: p.soc, @@ -632,7 +721,7 @@ export class ElectraFiImports { .from(importBatches) .orderBy(desc(importBatches.createdAt), desc(importBatches.id)); return { - draft: this.draft && this.draftDto(this.draft), + draft: this.draft && (await this.draftDto(this.draft)), batches: rows.map((row) => { const running = this.running?.id === row.id ? this.running : null; return toDto( @@ -652,6 +741,8 @@ const SESSION_FIGURES = [ 'startSoc', 'endSoc', 'energyKwh', + 'packKwh', + 'thermalKwh', 'rangeAddedKm', 'avgPowerKw', 'maxPowerKw', @@ -672,6 +763,7 @@ function additionsOf(fill: SessionFill, addCurve: boolean): ImportSessionAdditio if (fill.chargingSeconds != null) adds.push('chargingTime'); if (fill.startSoc != null || fill.endSoc != null) adds.push('batteryLevels'); if (fill.energyKwh != null) adds.push('energy'); + if (fill.packKwh != null || fill.thermalKwh != null) adds.push('breakdown'); if (fill.rangeAddedKm != null) adds.push('range'); if (fill.avgPowerKw != null || fill.maxPowerKw != null) adds.push('power'); if (fill.lat != null) adds.push('location'); @@ -696,6 +788,8 @@ function fillFor( startSoc: row.startSoc == null ? segments[0]!.startSoc : null, endSoc: row.endSoc == null ? segments.at(-1)!.endSoc : null, energyKwh: row.energyKwh == null ? sum((s) => s.energyKwh) : null, + packKwh: row.packKwh == null ? sum((s) => s.packKwh) : null, + thermalKwh: row.thermalKwh == null ? sum((s) => s.thermalKwh) : null, rangeAddedKm: row.rangeAddedKm == null ? sum((s) => s.rangeAddedKm) : null, avgPowerKw: row.avgPowerKw == null && poweredSeconds > 0 @@ -789,7 +883,7 @@ async function removeImportedRows(db: Db, id: number): Promise { } /** - * Writes the rows an import takes. A window is a run of them with nothing + * Writes the readings an import takes. A window is a run of them with nothing * skipped between; state in it records only real changes from what was held * when it opened, and when it closes, whatever RivianMate last recorded is * put back, so the import doesn't change what history shows afterwards. @@ -811,11 +905,13 @@ class RowWriter { private readonly vehicleId: string, private readonly driveList: readonly { id: number; startedAt: number; endedAt: number }[], private readonly counts: ImportCounts, + private readonly format: ExportFormat, ) {} - async take(row: ElectraFiRow, at: number): Promise { + async take({ at, state, reporting, fix }: Reading): Promise { if (!this.window) { - const held = await latestChanges(this.db, this.vehicleId, ROW_STATE_FIELDS, new Date(at)); + const fields = this.format.stateFields; + const held = await latestChanges(this.db, this.vehicleId, fields, new Date(at)); this.window = { diff: new StateDiff(new Map(held.map((c) => [c.field, c.value]))), wrote: false, @@ -824,17 +920,16 @@ class RowWriter { } const window = this.window; window.lastAt = at; - const changes = window.diff.changes(rowState(row, at), at); + const changes = window.diff.changes(state, at); if (changes.length > 0) { window.wrote = true; this.changes.push(...changes); } - // Asleep rows have no state in them, and a sleeping car sends no heartbeats live either. - if (row.mileageM != null && at - this.lastBeatAt >= HEARTBEAT_MS) { + if (reporting && at - this.lastBeatAt >= HEARTBEAT_MS) { this.lastBeatAt = at; this.beats.push({ vehicleSeq: this.seq, at: new Date(at), importId: this.id }); } - if (row.lat != null && row.lon != null) this.takePoint(row, at); + if (fix) this.takePoint(fix, at); if ( this.changes.length >= INSERT_BATCH || this.points.length >= INSERT_BATCH || @@ -844,20 +939,26 @@ class RowWriter { } } - private takePoint(row: ElectraFiRow, at: number): void { - const lat = row.lat!; - const lon = row.lon!; + private takePoint(fix: Fix, at: number): void { + const { lat, lon } = fix; const last = this.lastPoint; - // Exports repeat the last fix between new ones, speed included. - if (last && haversineKm(last.lat, last.lon, lat, lon) < MIN_MOVE_KM) return; + // As live: a fix that hasn't moved is kept only while moving, unless the export repeats them. + if ( + last && + haversineKm(last.lat, last.lon, lat, lon) < MIN_MOVE_KM && + (this.format.repeatsFixes || (fix.speedKmh ?? 0) <= 0) + ) { + return; + } this.lastPoint = { lat, lon }; this.points.push({ vehicleId: this.vehicleId, ts: new Date(at), lat, lon, - speedKmh: row.speedKmh, - altitude: row.altitudeM, + speedKmh: fix.speedKmh, + bearing: fix.bearing, + altitude: fix.altitudeM, driveId: this.driveAt(at), importId: this.id, }); @@ -882,7 +983,7 @@ class RowWriter { await this.flush(); if (!window.wrote) return; const until = new Date(window.lastAt); - const held = await liveValues(this.db, this.seq, until); + const held = await liveValues(this.db, this.seq, until, this.format.stateFields); // A moment after the last row, so it isn't taken for that row's reading. const closeAt = window.lastAt + 1000; const state: VehicleState = {}; @@ -927,6 +1028,7 @@ async function liveValues( db: Db, seq: number, at: Date, + fields: readonly string[], ): Promise> { const rows = await db.execute<{ field: string; num: number | null; value: string | null }>(sql` SELECT f.name AS field, c.num, c.value @@ -938,7 +1040,7 @@ async function liveValues( ORDER BY at DESC LIMIT 1 ) c WHERE f.name IN (${sql.join( - ROW_STATE_FIELDS.map((f) => sql`${f}`), + fields.map((f) => sql`${f}`), sql`, `, )}) `); diff --git a/server/src/services/electrafi/plan.ts b/server/src/services/imports/plan.ts similarity index 86% rename from server/src/services/electrafi/plan.ts rename to server/src/services/imports/plan.ts index 138fcb0..f65e637 100644 --- a/server/src/services/electrafi/plan.ts +++ b/server/src/services/imports/plan.ts @@ -5,14 +5,10 @@ import type { ImportSessionAddition, } from '../../api-types.js'; import { isListedDrive } from '../drive-metrics.js'; -import { type Spot, nearbySpot } from '../locations.js'; -import { - type ChargeSegment, - ChargeSegmenter, - type DriveSegment, - DriveSegmenter, -} from './segment.js'; -import { type ReadStats, TimeZoneClock, readRows } from './parse.js'; +import { DEFAULT_RADIUS_M, type Spot, nearbySpot } from '../locations.js'; +import { haversineKm } from '../state-utils.js'; +import type { ChargeSegment, DriveSegment, ImportPlace, Scan } from './format.js'; +import { TimeZoneClock } from './time-zone.js'; /** * A row this close to a heartbeat is time RivianMate was already recording. @@ -24,18 +20,6 @@ const ODOMETER_SLACK_M = 1000; /** A row later than this past now has the wrong time zone. */ const FUTURE_SLACK_MS = 5 * 60_000; -export interface Scan extends ReadStats { - rows: number; - firstAt: number | null; - lastAt: number | null; - firstMileage: { at: number; value: number } | null; - lastMileage: { at: number; value: number } | null; - drives: DriveSegment[]; - charges: ChargeSegment[]; - /** When each row with readings was taken, so a new date range needn't read the files again. */ - readingTimes: number[]; -} - /** What's already stored for the vehicle. */ export interface Existing { /** Live and imported, in order. */ @@ -50,6 +34,7 @@ export interface Existing { imported: boolean; }[]; mileage: readonly { at: number; value: number }[]; + places: readonly (Spot & { isHome: boolean })[]; } export interface PlannedCharge { @@ -67,6 +52,9 @@ export interface Plan { /** Each drive and charge starting in the range, and what happens to it. */ driveDecisions: { segment: DriveSegment; status: ImportItemStatus }[]; chargeDecisions: { segment: ChargeSegment; status: ImportItemStatus; sessionId: number | null }[]; + /** Saved places to add; one where a place is already saved is left out. */ + places: ImportPlace[]; + placesRecorded: number; } type Span = readonly [number, number]; @@ -134,7 +122,7 @@ export class RowScope { /** Decides what an import of the range adds, leaving what's already stored alone. */ export function buildPlan( - scan: Pick, + scan: Pick & Partial>, existing: Existing, range: { from: number; to: number }, now: number, @@ -188,6 +176,7 @@ export function buildPlan( charges.push({ segment, sessionId: session?.id ?? null }); } + const places = newPlaces(scan.places ?? [], existing.places); return { from: range.from, to: range.to, @@ -195,6 +184,8 @@ export function buildPlan( charges, driveDecisions, chargeDecisions, + places, + placesRecorded: (scan.places?.length ?? 0) - places.length, scope: new RowScope( { ...range, lastLiveAt: existing.lastLiveAt }, existing.heartbeats, @@ -206,6 +197,26 @@ export function buildPlan( }; } +/** + * The places not already saved: one whose circle overlaps a saved one's is + * the same place. Home stays with a home already set. + */ +function newPlaces(places: readonly ImportPlace[], saved: Existing['places']): ImportPlace[] { + const out: ImportPlace[] = []; + let hasHome = saved.some((p) => p.isHome); + for (const place of places) { + const overlaps = [...saved, ...out].some( + (p) => + haversineKm(p.lat, p.lon, place.lat, place.lon) * 1000 <= + Math.max(p.radiusM ?? DEFAULT_RADIUS_M, place.radiusM), + ); + if (overlaps) continue; + out.push({ ...place, isHome: place.isHome && !hasHome }); + hasHome ||= place.isHome; + } + return out; +} + function closestOverlap( charge: ChargeSegment, sessions: Existing['sessions'], @@ -239,38 +250,6 @@ export function odometerFits( return true; } -export async function scanFiles(paths: readonly string[], timeZone: string): Promise { - const scan: Scan = { - rows: 0, - unreadable: 0, - firstAt: null, - lastAt: null, - firstMileage: null, - lastMileage: null, - drives: [], - charges: [], - readingTimes: [], - }; - const driveFinder = new DriveSegmenter(); - const chargeFinder = new ChargeSegmenter(); - for await (const { row, at } of readRows(paths, timeZone, scan)) { - scan.rows += 1; - scan.firstAt = Math.min(scan.firstAt ?? at, at); - scan.lastAt = Math.max(scan.lastAt ?? at, at); - if (row.mileageM != null) { - const reading = { at, value: row.mileageM }; - scan.firstMileage ??= reading; - scan.lastMileage = reading; - scan.readingTimes.push(at); - } - scan.drives.push(...driveFinder.push(row, at)); - scan.charges.push(...chargeFinder.push(row, at)); - } - scan.drives.push(...driveFinder.flush()); - scan.charges.push(...chargeFinder.flush()); - return scan; -} - /** The instants a selection's dates start and end (exclusive), in its time zone. */ export function rangeOf(from: string, to: string, timeZone: string): { from: number; to: number } { const clock = new TimeZoneClock(timeZone); @@ -298,7 +277,7 @@ export function previewOf( scan: Scan, plan: Plan, existing: Existing, - counted: { readingsNew: number; readingsRecorded: number }, + counted: { readingsNew: number; readingsRecorded: number; readingsNewer: number }, context: { timeZone: string; now: number; @@ -348,6 +327,9 @@ export function previewOf( sessionsEnriched: charges.filter((c) => c.status === 'enrich').length, readingsNew: counted.readingsNew, readingsRecorded: counted.readingsRecorded, + readingsNewer: counted.readingsNewer, + placesNew: plan.places.map((p) => p.name), + placesRecorded: plan.placesRecorded, warnings: [ ...(fits(scan.firstMileage) && fits(scan.lastMileage) ? [] : (['odometer'] as const)), ...(scan.lastAt != null && scan.lastAt > context.now + FUTURE_SLACK_MS @@ -363,6 +345,7 @@ export function previewOf( endBattery: d.end.battery, startPlace: placeName(d.start.lat, d.start.lon, context.places), endPlace: placeName(d.end.lat, d.end.lon, context.places), + driver: d.driverName, })), charges, }; diff --git a/server/src/services/imports/time-zone.ts b/server/src/services/imports/time-zone.ts new file mode 100644 index 0000000..8d5f8e7 --- /dev/null +++ b/server/src/services/imports/time-zone.ts @@ -0,0 +1,109 @@ +const OFFSET_SLOT_MS = 15 * 60_000; + +export interface LocalTime { + year: number; + month: number; + day: number; + hour: number; + minute: number; + second: number; +} + +/** + * Local times to instants. An hour that clocks repeat is ambiguous, so a row + * takes the first reading that isn't before the row ahead of it; rows are in + * order, so this walks through the repeat as the export did. + */ +export class TimeZoneClock { + private readonly format: Intl.DateTimeFormat; + private previous = -Infinity; + private readonly offsets = new Map(); + + constructor(timeZone: string) { + this.format = new Intl.DateTimeFormat('en-US', { + timeZone, + hourCycle: 'h23', + year: 'numeric', + month: 'numeric', + day: 'numeric', + hour: 'numeric', + minute: 'numeric', + second: 'numeric', + }); + } + + /** The zone's offset from UTC at an instant (ms). */ + private offsetAt(at: number): number { + const parts = Object.fromEntries( + this.format.formatToParts(at).map((p) => [p.type, Number(p.value)]), + ) as Record; + const wall = Date.UTC( + parts['year']!, + parts['month']! - 1, + parts['day']!, + parts['hour']!, + parts['minute']!, + parts['second']!, + ); + return wall - Math.floor(at / 1000) * 1000; + } + + /** The instant of a local time, keeping rows in order. */ + next(local: LocalTime): number { + const at = this.candidates(local).find((t) => t >= this.previous) ?? this.first(local); + this.previous = at; + return at; + } + + /** The instant of a local time, on its own (the earlier one if clocks repeat it). */ + first(local: LocalTime): number { + return this.candidates(local)[0]!; + } + + private candidates(local: LocalTime): number[] { + const wall = Date.UTC( + local.year, + local.month - 1, + local.day, + local.hour, + local.minute, + local.second, + ); + return this.offsetsNear(wall).map((offset) => wall - offset); + } + + /** + * The offsets a local time can have. Read for every row of an export and + * slow to work out, so kept per 15 minutes, the finest clocks change on. + */ + private offsetsNear(wall: number): number[] { + const slot = Math.floor(wall / OFFSET_SLOT_MS); + const known = this.offsets.get(slot); + if (known) return known; + const before = this.offsetAt(wall - 86_400_000); + const found = [...new Set([before, this.offsetAt(wall + 86_400_000)])] + .filter((offset) => this.offsetAt(wall - offset) === offset) + .sort((a, b) => b - a); + // A time clocks skip doesn't exist; it reads as the offset before the skip. + const offsets = found.length > 0 ? found : [before]; + this.offsets.set(slot, offsets); + return offsets; + } + + /** An instant's calendar date in this zone, as YYYY-MM-DD. */ + dateOf(at: number): string { + const parts = Object.fromEntries( + this.format.formatToParts(at).map((p) => [p.type, p.value]), + ) as Record; + return `${parts['year']}-${parts['month']!.padStart(2, '0')}-${parts['day']!.padStart(2, '0')}`; + } +} + +export function isTimeZone(value: string): boolean { + try { + new Intl.DateTimeFormat('en-US', { timeZone: value }); + return true; + } catch { + return false; + } +} diff --git a/server/src/services/roamer/archive.ts b/server/src/services/roamer/archive.ts new file mode 100644 index 0000000..00a1ec1 --- /dev/null +++ b/server/src/services/roamer/archive.ts @@ -0,0 +1,134 @@ +import { createInterface } from 'node:readline'; +import { type Entry, type ZipFile, openPromise } from 'yauzl'; + +/** The archive layout this reads; a later one may move or rename things. */ +const FORMAT_VERSION = 1; +const MANIFEST = 'manifest.json'; +/** The manifest is a listing; anything near this is something else. */ +const MAX_MANIFEST_BYTES = 64 * 1024 ** 2; + +export class NotARoamerExportError extends Error { + constructor(message = "This doesn't look like a Rivian Roamer export: it has no manifest.") { + super(message); + } +} + +interface ManifestFile { + dataset: string; + rows?: number; + /** The span of time a telemetry file holds. */ + from?: string; + until?: string; +} + +interface Manifest { + format_version: number; + files: Record; +} + +export interface ReadStats { + unreadable: number; +} + +/** + * A Roamer export, read in place: telemetry stays compressed in the zip and + * is streamed out file by file, a line (one JSON object) at a time. + */ +export class RoamerArchive { + private constructor( + private readonly zip: ZipFile, + private readonly entries: ReadonlyMap, + private readonly manifest: Manifest, + ) {} + + /** Raises NotARoamerExportError unless the zip has a manifest this can read. */ + static async open(path: string): Promise { + const zip = await openPromise(path, { lazyEntries: true, autoClose: false }); + try { + const entries = new Map(); + for await (const entry of zip.eachEntry()) entries.set(entry.fileName, entry); + const listing = entries.get(MANIFEST); + if (!listing || listing.uncompressedSize > MAX_MANIFEST_BYTES) { + throw new NotARoamerExportError(); + } + const chunks: Buffer[] = []; + for await (const chunk of await zip.openReadStreamPromise(listing)) { + chunks.push(chunk as Buffer); + } + let manifest: Manifest; + try { + manifest = JSON.parse(Buffer.concat(chunks).toString('utf8')) as Manifest; + } catch { + throw new NotARoamerExportError(); + } + if (manifest.format_version !== FORMAT_VERSION || typeof manifest.files !== 'object') { + throw new NotARoamerExportError( + 'This Rivian Roamer export is in a format RivianMate doesn’t know yet.', + ); + } + return new RoamerArchive(zip, entries, manifest); + } catch (err) { + zip.close(); + throw err; + } + } + + close(): void { + this.zip.close(); + } + + /** + * A dataset's files, in time order: the account-wide file at the top, or a + * vehicle's, which telemetry splits into one per stretch of time. + */ + files(dataset: string, vehicleId?: string): string[] { + const prefix = vehicleId == null ? null : `vehicles/${vehicleId}/`; + return Object.entries(this.manifest.files) + .filter( + ([name, file]) => + file.dataset === dataset && + this.entries.has(name) && + (prefix == null ? !name.includes('/') : name.startsWith(prefix)), + ) + .sort(([a, x], [b, y]) => (x.from ?? '').localeCompare(y.from ?? '') || a.localeCompare(b)) + .map(([name]) => name); + } + + /** How many records the manifest says the files hold. */ + rowsIn(names: readonly string[]): number { + return names.reduce((sum, name) => sum + (this.manifest.files[name]?.rows ?? 0), 0); + } + + /** Each record of the files in turn; a line that isn't a JSON object is counted and skipped. */ + async *records( + names: readonly string[], + stats: ReadStats = { unreadable: 0 }, + ): AsyncGenerator> { + for (const name of names) { + const entry = this.entries.get(name); + if (!entry) continue; + const input = await this.zip.openReadStreamPromise(entry); + const lines = createInterface({ input, crlfDelay: Infinity }); + try { + for await (const line of lines) { + if (line.trim() === '') continue; + let record: unknown; + try { + record = JSON.parse(line); + } catch { + record = null; + } + if (record && typeof record === 'object' && !Array.isArray(record)) { + yield record as Record; + } else { + stats.unreadable += 1; + } + } + } finally { + lines.close(); + // A reader that stops early leaves the rest of the file unread. + input.destroy(); + } + } + } +} diff --git a/server/src/services/roamer/format.ts b/server/src/services/roamer/format.ts new file mode 100644 index 0000000..b24a46f --- /dev/null +++ b/server/src/services/roamer/format.ts @@ -0,0 +1,272 @@ +import { rename, rm } from 'node:fs/promises'; +import { + type ChargeSegment, + type Destination, + type ExportFormat, + type ImportPlace, + ImportError, + type Reading, + type Scan, +} from '../imports/format.js'; +import { NotARoamerExportError, RoamerArchive } from './archive.js'; +import { + METRIC_STATE_FIELDS, + chargeOf, + chargingPower, + destinationsFor, + driveOf, + fixOf, + instant, + metricState, + navigationOf, + num, + placeOf, +} from './parse.js'; + +/** The export's vehicles, by Roamer's id for each. */ +async function vehiclesOf(archive: RoamerArchive): Promise<{ id: string; vin: string }[]> { + const out: { id: string; vin: string }[] = []; + for await (const record of archive.records(archive.files('vehicles'))) { + const { id, vin } = record; + if (typeof id === 'string' && typeof vin === 'string') out.push({ id, vin }); + } + return out; +} + +/** The account's saved places, Home marked by Roamer's favourite for it. */ +async function placesOf(archive: RoamerArchive): Promise { + const homes = new Set(); + for await (const record of archive.records(archive.files('rivian_favorites'))) { + if (record['key'] === 'home') homes.add(record['tagged_location_id']); + } + const places: ImportPlace[] = []; + for await (const record of archive.records(archive.files('saved_places'))) { + const place = placeOf(record, homes.has(record['id'])); + if (place) places.push(place); + } + return places; +} + +/** Opens the archive at the folder of the vehicle with this VIN. */ +async function openFor( + paths: readonly string[], + vin: string, +): Promise<{ archive: RoamerArchive; vehicleId: string }> { + const archive = await RoamerArchive.open(paths[0]!); + const vehicle = (await vehiclesOf(archive)).find((v) => v.vin === vin); + if (!vehicle) { + archive.close(); + throw new ImportError('This export has no data for that vehicle.'); + } + return { archive, vehicleId: vehicle.id }; +} + +/** Walks forward through charges in time order, for readings that come in time order too. */ +class ChargeCursor { + private index = 0; + + constructor(private readonly charges: readonly ChargeSegment[]) {} + + at(at: number): ChargeSegment | null { + while (this.index < this.charges.length && this.charges[this.index]!.endedAt < at) { + this.index += 1; + } + const charge = this.charges[this.index]; + return charge && charge.startedAt <= at ? charge : null; + } +} + +const byStart = (a: { startedAt: number }, b: { startedAt: number }) => a.startedAt - b.startedAt; + +/** + * Drives and charges are Roamer's own sessions. The metrics fill in what + * they don't keep: the capacity a drive set off with, where its navigation + * was headed, and the battery level through each charge for its curve. + */ +async function scanArchive(archive: RoamerArchive, vehicleId: string): Promise { + const scan: Scan = { + rows: 0, + unreadable: 0, + firstAt: null, + lastAt: null, + firstMileage: null, + lastMileage: null, + drives: [], + charges: [], + places: [], + readingTimes: [], + }; + const seen = (at: number) => { + scan.firstAt = Math.min(scan.firstAt ?? at, at); + scan.lastAt = Math.max(scan.lastAt ?? at, at); + }; + for await (const record of archive.records(archive.files('drive_sessions', vehicleId), scan)) { + const drive = driveOf(record); + if (drive) scan.drives.push(drive); + } + for await (const record of archive.records(archive.files('charging_sessions', vehicleId), scan)) { + const charge = chargeOf(record); + if (charge) scan.charges.push(charge); + } + scan.drives.sort(byStart); + scan.charges.sort(byStart); + const navigation: { at: number; to: Destination | null }[] = []; + for await (const record of archive.records(archive.files('navigation_metrics', vehicleId))) { + const reading = navigationOf(record); + if (reading) navigation.push(reading); + } + navigation.sort((a, b) => a.at - b.at); + destinationsFor(scan.drives, navigation).forEach((to, i) => { + scan.drives[i]!.destination = to; + }); + scan.places = await placesOf(archive); + for (const s of [...scan.drives, ...scan.charges]) { + seen(s.startedAt); + seen(s.endedAt); + } + + // The battery level through each charge, in time order. + const levels = new Map(); + const charging = new ChargeCursor(scan.charges); + let capacity: number | null = null; + let nextDrive = 0; + for await (const record of archive.records(archive.files('vehicle_metrics', vehicleId), scan)) { + const at = instant(record['created_at']); + if (at == null) { + scan.unreadable += 1; + continue; + } + scan.rows += 1; + seen(at); + while (nextDrive < scan.drives.length && scan.drives[nextDrive]!.startedAt <= at) { + scan.drives[nextDrive]!.batteryCapacityKwh = capacity; + nextDrive += 1; + } + if (Object.keys(metricState(record, at)).length > 0) scan.readingTimes.push(at); + const mileage = num(record['vehicle_mileage']); + if (mileage != null) { + scan.firstMileage ??= { at, value: mileage }; + scan.lastMileage = { at, value: mileage }; + } + capacity = num(record['battery_capacity']) ?? capacity; + const level = num(record['battery_level']); + const charge = level == null ? null : charging.at(at); + if (charge) { + const through = levels.get(charge) ?? []; + through.push({ at, value: level! }); + levels.set(charge, through); + } + } + for (const drive of scan.drives.slice(nextDrive)) drive.batteryCapacityKwh = capacity; + + for await (const record of archive.records(archive.files('location_metrics', vehicleId), scan)) { + const at = instant(record['created_at']); + if (at == null) { + scan.unreadable += 1; + continue; + } + scan.rows += 1; + seen(at); + } + + const curving = new ChargeCursor(scan.charges); + const levelIndex = new Map(); + for await (const record of archive.records(archive.files('charging_metrics', vehicleId), scan)) { + const at = instant(record['created_at']); + const powerKw = chargingPower(record); + const charge = at == null || powerKw == null ? null : curving.at(at); + if (!charge) continue; + // The latest level at or before the sample; before the first, where the charge began. + const known = levels.get(charge) ?? []; + let i = levelIndex.get(charge) ?? -1; + while (i + 1 < known.length && known[i + 1]!.at <= at!) i += 1; + levelIndex.set(charge, i); + charge.curve.push({ at: at!, powerKw, soc: known[i]?.value ?? charge.startSoc }); + } + return scan; +} + +/** The vehicle's metrics and fixes as readings, merged into time order. */ +async function* readingsOf(archive: RoamerArchive, vehicleId: string): AsyncGenerator { + async function* states(): AsyncGenerator { + for await (const record of archive.records(archive.files('vehicle_metrics', vehicleId))) { + const at = instant(record['created_at']); + if (at != null) yield { at, state: metricState(record, at), reporting: true, fix: null }; + } + } + async function* fixes(): AsyncGenerator { + for await (const record of archive.records(archive.files('location_metrics', vehicleId))) { + const at = instant(record['created_at']); + if (at != null) yield { at, state: {}, reporting: false, fix: fixOf(record) }; + } + } + yield* byTime(states(), fixes()); +} + +/** Merges streams that are each in time order. */ +async function* byTime(...sources: AsyncGenerator[]): AsyncGenerator { + const heads = await Promise.all(sources.map((s) => s.next())); + try { + for (;;) { + let pick = -1; + heads.forEach((head, i) => { + if (!head.done && (pick < 0 || head.value.at < (heads[pick]!.value as Reading).at)) { + pick = i; + } + }); + if (pick < 0) return; + yield heads[pick]!.value as Reading; + heads[pick] = await sources[pick]!.next(); + } + } finally { + await Promise.all(sources.map((s) => s.return(undefined))); + } +} + +/** Roamer's data export: one zip for the account, with exact times and each vehicle's VIN. */ +export const roamerFormat: ExportFormat = { + id: 'roamer', + singleFile: true, + unreadable: "Couldn't read this file as a Rivian Roamer export.", + + async store(upload, stem) { + const path = `${stem}.zip`; + await rename(upload, path); + try { + const archive = await RoamerArchive.open(path); + try { + return { path, vins: (await vehiclesOf(archive)).map((v) => v.vin) }; + } finally { + archive.close(); + } + } catch (err) { + await rm(path, { force: true }); + if (err instanceof NotARoamerExportError) throw new ImportError(err.message); + throw err; + } + }, + + // Times are exact, so the zone only sets where the chosen dates fall. + scanKey: ({ vin }) => vin, + + async scan(paths, { vin }) { + const { archive, vehicleId } = await openFor(paths, vin); + try { + return await scanArchive(archive, vehicleId); + } finally { + archive.close(); + } + }, + + async *readings(paths, { vin }) { + const { archive, vehicleId } = await openFor(paths, vin); + try { + yield* readingsOf(archive, vehicleId); + } finally { + archive.close(); + } + }, + + stateFields: METRIC_STATE_FIELDS, + repeatsFixes: false, +}; diff --git a/server/src/services/roamer/parse.ts b/server/src/services/roamer/parse.ts new file mode 100644 index 0000000..9bf72df --- /dev/null +++ b/server/src/services/roamer/parse.ts @@ -0,0 +1,257 @@ +import type { VehicleState } from '../../rivian/types.js'; +import { DEFAULT_RADIUS_M, MAX_RADIUS_M, MIN_RADIUS_M } from '../locations.js'; +import type { + ChargeSegment, + Destination, + DriveSegment, + Fix, + ImportPlace, +} from '../imports/format.js'; + +type RoamerRecord = Record; + +/** Location metrics keep Rivian's GNSS speed, in m/s. */ +const MPS_TO_KMH = 3.6; + +/** + * Vehicle metrics Roamer keeps as Rivian sent them, under Rivian's names in + * snake case, with the values live state has. Ones it has recoded (seat + * heat as 0–3, service mode and cold-battery limits as booleans, range + * thresholds with a prefix) are left out rather than guessed back. + */ +const NUMBER_METRICS = [ + 'vehicle_mileage', + 'battery_level', + 'distance_to_empty', + 'battery_capacity', + 'battery_limit', + 'cabin_climate_interior_temperature', + 'cabin_climate_driver_temperature', + 'tire_pressure_front_left', + 'tire_pressure_front_right', + 'tire_pressure_rear_left', + 'tire_pressure_rear_right', +]; +const CLOSURES = ['frunk', 'liftgate', 'tailgate', 'tonneau', 'side_bin_left', 'side_bin_right']; +const SIDES = ['front_left', 'front_right', 'rear_left', 'rear_right']; +const TEXT_METRICS = [ + 'power_state', + 'gear_status', + 'drive_mode', + 'charger_state', + 'charger_status', + 'charger_derate_status', + 'charge_port_state', + 'ota_current_version', + 'ota_available_version', + 'ota_status', + 'ota_current_status', + 'ota_install_ready', + ...SIDES.flatMap((side) => [`door_${side}_locked`, `door_${side}_closed`]), + ...CLOSURES.flatMap((closure) => [`closure_${closure}_locked`, `closure_${closure}_closed`]), + ...SIDES.map((side) => `window_${side}_closed`), + ...SIDES.map((side) => `tire_pressure_status_${side}`), + 'gear_guard_locked', + 'gear_guard_video_status', + 'pet_mode_status', + 'pet_mode_temperature_status', + 'cabin_preconditioning_status', + 'cabin_preconditioning_type', + 'defrost_defog_status', + 'twelve_volt_battery_health', + 'battery_hv_thermal_event', +]; + +const camel = (snake: string) => snake.replace(/_([a-z])/g, (_, c: string) => c.toUpperCase()); + +const METRICS: readonly { column: string; field: string; kind: 'number' | 'text' }[] = [ + ...NUMBER_METRICS.map((column) => ({ column, field: camel(column), kind: 'number' as const })), + ...TEXT_METRICS.map((column) => ({ column, field: camel(column), kind: 'text' as const })), +]; + +/** Every field `metricState` can write. */ +export const METRIC_STATE_FIELDS: readonly string[] = METRICS.map((m) => m.field); + +/** Charging metrics' state while power is flowing (Parallax's `charging_active`). */ +const CHARGING_ACTIVE = 3; + +/** A number; Postgres numerics can come as strings. Anything else is no reading. */ +export function num(value: unknown): number | null { + const n = typeof value === 'string' && value.trim() !== '' ? Number(value) : value; + return typeof n === 'number' && Number.isFinite(n) ? n : null; +} + +const text = (value: unknown): string | null => + typeof value === 'string' && value.trim() !== '' ? value : null; +const round = (n: number | null, places: number): number | null => + n == null ? null : Math.round(n * 10 ** places) / 10 ** places; + +/** An instant: Roamer's times are all UTC, with or without a suffix. */ +export function instant(value: unknown): number | null { + if (typeof value !== 'string') return null; + const at = Date.parse(/(Z|[+-]\d{2}:?\d{2})$/i.test(value) ? value : `${value}Z`); + return Number.isFinite(at) ? at : null; +} + +/** A vehicle metric's readings as vehicle state, stamped at `at`. */ +export function metricState(record: RoamerRecord, at: number): VehicleState { + const timeStamp = new Date(at).toISOString(); + const state: VehicleState = {}; + for (const { column, field, kind } of METRICS) { + const value = kind === 'number' ? num(record[column]) : text(record[column]); + if (value != null) state[field] = { value, timeStamp }; + } + return state; +} + +/** A location metric's fix, or null without a usable position. */ +export function fixOf(record: RoamerRecord): Fix | null { + const lat = num(record['latitude']); + const lon = num(record['longitude']); + if (lat == null || lon == null || Math.abs(lat) > 90 || Math.abs(lon) > 180) return null; + const speed = num(record['speed']); + return { + lat, + lon, + speedKmh: round(speed == null ? null : speed * MPS_TO_KMH, 2), + altitudeM: num(record['altitude']), + bearing: num(record['bearing']), + }; +} + +/** A drive session; null for one Roamer flagged as corrupted, or without a start and end. */ +export function driveOf(record: RoamerRecord): DriveSegment | null { + const startedAt = instant(record['started_at']); + const endedAt = instant(record['ended_at']); + if (record['is_corrupted'] === true || startedAt == null || endedAt == null) return null; + const odometerM = (column: string) => { + const km = num(record[column]); + return km == null ? null : Math.round(km * 1000); + }; + return { + startedAt, + endedAt, + start: { + mileageM: odometerM('start_odometer'), + battery: num(record['start_battery_level']), + rangeKm: num(record['start_range']), + lat: num(record['start_latitude']), + lon: num(record['start_longitude']), + }, + end: { + mileageM: odometerM('end_odometer'), + battery: num(record['end_battery_level']), + rangeKm: num(record['end_range']), + lat: num(record['end_latitude']), + lon: num(record['end_longitude']), + }, + // Not kept with the session: read from the metrics at its start. + batteryCapacityKwh: null, + driveMode: text(record['drive_mode_override']) ?? text(record['primary_drive_mode']), + distanceKm: round(num(record['total_distance']), 3), + elevationGainM: round(num(record['elevation_gain']), 1), + elevationLossM: round(num(record['elevation_loss']), 1), + driverName: text(record['primary_driver_profile'])?.trim() ?? null, + // Not kept with the session: read from the navigation metrics. + destination: null, + }; +} + +/** The time a session's periods spent charging, in seconds. */ +function periodSeconds(periods: unknown): number | null { + if (!Array.isArray(periods) || periods.length === 0) return null; + let total = 0; + for (const period of periods as RoamerRecord[]) { + const from = instant(period?.['started_at']); + const to = instant(period?.['ended_at']); + if (from == null || to == null) return null; + total += Math.max(0, to - from) / 1000; + } + return total; +} + +/** A charging session, with no curve yet; null for a corrupted one, or one still open. */ +export function chargeOf(record: RoamerRecord): ChargeSegment | null { + const startedAt = instant(record['started_at']); + const endedAt = instant(record['ended_at']); + if (record['is_corrupted'] === true || startedAt == null || endedAt == null) return null; + const minutes = num(record['duration_minutes']); + return { + startedAt, + endedAt, + chargingSeconds: Math.round( + periodSeconds(record['periods']) ?? (minutes == null ? 0 : minutes * 60), + ), + startSoc: round(num(record['start_battery_level']), 1), + endSoc: round(num(record['end_battery_level']), 1), + energyKwh: round(num(record['total_energy_kwh']), 3), + packKwh: round(num(record['pack_kwh']), 3), + thermalKwh: round(num(record['thermal_kwh']), 3), + rangeAddedKm: round(num(record['kilometers_added']), 3), + avgPowerKw: round(num(record['average_power_kw']), 2), + maxPowerKw: round(num(record['max_power_kw']), 2), + lat: num(record['latitude']), + lon: num(record['longitude']), + curve: [], + }; +} + +/** A charging metric's power, if power was flowing. */ +export function chargingPower(record: RoamerRecord): number | null { + return num(record['state']) === CHARGING_ACTIVE ? round(num(record['power_kw']), 2) : null; +} + +/** A saved place; `home` is whether Roamer's Home favourite points at it. */ +export function placeOf(record: RoamerRecord, home: boolean): ImportPlace | null { + const name = text(record['name'])?.trim(); + const lat = num(record['latitude']); + const lon = num(record['longitude']); + if (!name || lat == null || lon == null || Math.abs(lat) > 90 || Math.abs(lon) > 180) { + return null; + } + const radius = num(record['radius_meters']) ?? DEFAULT_RADIUS_M; + const cents = num(record['price_per_kwh_minor']); + return { + name, + lat, + lon, + radiusM: Math.round(Math.min(MAX_RADIUS_M, Math.max(MIN_RADIUS_M, radius))), + isHome: home, + ratePerKwh: cents == null ? null : cents / 100, + }; +} + +/** Where the navigation was headed as of a navigation metric; null when it wasn't navigating. */ +export function navigationOf(record: RoamerRecord): { at: number; to: Destination | null } | null { + const at = instant(record['observed_at']); + if (at == null) return null; + const name = text(record['destination_name']); + const lat = num(record['destination_latitude']); + const lon = num(record['destination_longitude']); + const navigating = + record['navigation_active'] === true && name != null && lat != null && lon != null; + return { at, to: navigating ? { name: name!, lat: lat!, lon: lon! } : null }; +} + +/** + * Each drive's destination, as live recording keeps it: where the + * navigation was headed as the drive set off, or later during it, the last + * one it heard of winning. Both lists are in time order. + */ +export function destinationsFor( + drives: readonly DriveSegment[], + navigation: readonly { at: number; to: Destination | null }[], +): (Destination | null)[] { + let i = 0; + let current: Destination | null = null; + return drives.map((drive) => { + while (i < navigation.length && navigation[i]!.at <= drive.startedAt) + current = navigation[i++]!.to; + let found = current; + while (i < navigation.length && navigation[i]!.at <= drive.endedAt) { + current = navigation[i++]!.to; + found = current ?? found; + } + return found; + }); +} diff --git a/server/src/services/roamer/roamer.test.ts b/server/src/services/roamer/roamer.test.ts new file mode 100644 index 0000000..e01d3b5 --- /dev/null +++ b/server/src/services/roamer/roamer.test.ts @@ -0,0 +1,327 @@ +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { ImportError, type Reading } from '../imports/format.js'; +import { ROAMER_VIN, roamerExport, zipOf } from '../../testing/roamer-archive.js'; +import { roamerFormat } from './format.js'; +import { + chargeOf, + destinationsFor, + driveOf, + fixOf, + instant, + metricState, + navigationOf, + placeOf, +} from './parse.js'; + +const at = (iso: string) => Date.parse(iso); + +describe('Roamer records', () => { + it('reads times as UTC, with or without a suffix', () => { + expect(instant('2026-11-01T18:51:53.863')).toBe(at('2026-11-01T18:51:53.863Z')); + expect(instant('2026-11-04T00:00:03.167+00:00')).toBe(at('2026-11-04T00:00:03.167Z')); + expect(instant('2026-11-01T02:43:37.256703')).toBe(at('2026-11-01T02:43:37.256Z')); + expect(instant('soon')).toBeNull(); + expect(instant(null)).toBeNull(); + }); + + it('keeps the metrics live state has, under its names', () => { + const state = metricState( + { + vehicle_mileage: 8481219, + battery_level: 55.4, + distance_to_empty: 297, + gear_status: 'park', + door_front_left_locked: 'locked', + closure_side_bin_left_closed: 'closed', + // Recoded by Roamer, so left out. + seat_front_left_heat: 2, + service_mode: false, + range_threshold: 'vehicle_range_normal', + cellular_carrier: 'AT&T', + battery_capacity: null, + }, + at('2026-11-04T00:00:00Z'), + ); + expect(Object.fromEntries(Object.entries(state).map(([k, v]) => [k, v.value]))).toEqual({ + vehicleMileage: 8481219, + batteryLevel: 55.4, + distanceToEmpty: 297, + gearStatus: 'park', + doorFrontLeftLocked: 'locked', + closureSideBinLeftClosed: 'closed', + }); + expect(state['gearStatus']!.timeStamp).toBe('2026-11-04T00:00:00.000Z'); + }); + + it('reads fixes, with speed from m/s', () => { + expect( + fixOf({ latitude: 33.1, longitude: -117.2, speed: 10, altitude: 50, bearing: 90 }), + ).toEqual({ lat: 33.1, lon: -117.2, speedKmh: 36, altitudeM: 50, bearing: 90 }); + expect(fixOf({ latitude: 133, longitude: -117.2 })).toBeNull(); + expect(fixOf({ speed: 3 })).toBeNull(); + }); + + it('takes drives in RivianMate’s units, leaving out corrupted ones', () => { + const drive = driveOf({ + started_at: '2026-11-01T18:51:53.863', + ended_at: '2026-11-01T19:02:19.053', + start_odometer: 8481.219, + end_odometer: 8488.769, + start_battery_level: 70.1, + end_battery_level: 69, + start_range: 376, + end_range: 371, + total_distance: 7.55, + primary_drive_mode: 'everyday', + drive_mode_override: 'sport', + elevation_gain: 85.355, + }); + expect(drive).toMatchObject({ + startedAt: at('2026-11-01T18:51:53.863Z'), + start: { mileageM: 8481219, battery: 70.1, rangeKm: 376 }, + end: { mileageM: 8488769 }, + distanceKm: 7.55, + driveMode: 'sport', + elevationGainM: 85.4, + }); + expect( + driveOf({ + started_at: '2026-11-01T18:51:53', + ended_at: '2026-11-01T19:02:19', + is_corrupted: true, + }), + ).toBeNull(); + expect(driveOf({ started_at: '2026-11-01T18:51:53' })).toBeNull(); + }); + + it('times a charge by its periods, with the breakdown', () => { + const charge = chargeOf({ + started_at: '2026-11-01T06:59:38.043', + ended_at: '2026-11-01T12:28:13.045', + duration_minutes: 329, + periods: [ + { started_at: '2026-11-01T06:59:38Z', ended_at: '2026-11-01T08:59:38Z' }, + { started_at: '2026-11-01T10:00:00Z', ended_at: '2026-11-01T10:30:00Z' }, + ], + total_energy_kwh: 37.70000076293945, + pack_kwh: '36.300', + thermal_kwh: 0, + start_battery_level: 37.600002, + }); + expect(charge).toMatchObject({ + chargingSeconds: 2.5 * 3600, + energyKwh: 37.7, + packKwh: 36.3, + thermalKwh: 0, + startSoc: 37.6, + }); + expect( + chargeOf({ + started_at: '2026-11-01T06:59:38', + ended_at: '2026-11-01T07:59:38', + duration_minutes: 60, + })!.chargingSeconds, + ).toBe(3600); + }); +}); + +describe('Roamer places and navigation', () => { + it('reads saved places, keeping the radius in range and the rate in currency units', () => { + expect( + placeOf( + { + name: ' Home ', + latitude: 33.07, + longitude: -117.25, + radius_meters: 5, + price_per_kwh_minor: 14.5, + }, + true, + ), + ).toEqual({ + name: 'Home', + lat: 33.07, + lon: -117.25, + radiusM: 25, + isHome: true, + ratePerKwh: 0.145, + }); + expect(placeOf({ name: 'Work', latitude: 32.88, longitude: -117.22 }, false)).toMatchObject({ + radiusM: 150, + ratePerKwh: null, + }); + expect(placeOf({ name: '', latitude: 1, longitude: 1 }, false)).toBeNull(); + }); + + it('reads a navigation metric as where it was headed, or nowhere', () => { + expect( + navigationOf({ + observed_at: '2026-11-01T18:53:43+00:00', + navigation_active: true, + destination_name: 'Beach', + destination_latitude: 32.7, + destination_longitude: -117.1, + }), + ).toEqual({ at: at('2026-11-01T18:53:43Z'), to: { name: 'Beach', lat: 32.7, lon: -117.1 } }); + expect( + navigationOf({ observed_at: '2026-11-01T18:53:43+00:00', navigation_active: false })!.to, + ).toBeNull(); + }); + + it('gives a drive where its navigation was headed, as live recording would', () => { + const drive = (from: string, to: string) => driveOf({ started_at: from, ended_at: to })!; + const beach = { name: 'Beach', lat: 1, lon: 1 }; + const store = { name: 'Store', lat: 2, lon: 2 }; + const drives = [ + // Set before setting off. + drive('2026-01-01T10:00:00', '2026-01-01T10:30:00'), + // Navigation stopped in between; this one set off without it. + drive('2026-01-01T12:00:00', '2026-01-01T12:30:00'), + // Set during the drive, and kept after navigation ends. + drive('2026-01-01T14:00:00', '2026-01-01T14:30:00'), + ]; + expect( + destinationsFor(drives, [ + { at: at('2026-01-01T09:58:00Z'), to: beach }, + { at: at('2026-01-01T10:31:00Z'), to: null }, + { at: at('2026-01-01T14:05:00Z'), to: store }, + { at: at('2026-01-01T14:20:00Z'), to: null }, + ]), + ).toEqual([beach, null, store]); + }); +}); + +describe('Roamer exports', () => { + let dir: string; + beforeAll(async () => { + dir = await mkdtemp(join(tmpdir(), 'roamer-test-')); + }); + afterAll(async () => { + await rm(dir, { recursive: true, force: true }); + }); + + let count = 0; + /** Stores a file as an upload would be, returning the kept path and VINs. */ + const store = async (bytes: Buffer) => { + count += 1; + const upload = join(dir, `${count}.upload`); + await writeFile(upload, bytes); + return roamerFormat.store(upload, join(dir, String(count)), 1024 ** 2); + }; + const options = { timeZone: 'UTC', vin: ROAMER_VIN }; + + const exported = roamerExport({ + drive_sessions: [ + { + started_at: '2026-01-01T10:00:00', + ended_at: '2026-01-01T10:30:00', + start_odometer: 100, + end_odometer: 120, + total_distance: 20, + primary_driver_profile: 'Laura V.', + }, + ], + navigation_metrics: [ + { + observed_at: '2026-01-01T10:02:00+00:00', + navigation_active: true, + destination_name: 'Beach', + destination_latitude: 32.7, + destination_longitude: -117.1, + }, + ], + saved_places: [ + { id: 'p1', name: 'Home', latitude: 33.07, longitude: -117.25, radius_meters: 100 }, + { id: 'p2', name: 'Work', latitude: 32.88, longitude: -117.22, radius_meters: 100 }, + ], + rivian_favorites: [{ key: 'home', tagged_location_id: 'p1' }], + charging_sessions: [ + { + started_at: '2026-01-01T12:00:00', + ended_at: '2026-01-01T13:00:00', + start_battery_level: 50, + end_battery_level: 60, + total_energy_kwh: 11, + pack_kwh: 10.5, + thermal_kwh: 0.2, + }, + ], + vehicle_metrics: [ + { created_at: '2026-01-01T09:00:00+00:00', battery_capacity: 110, vehicle_mileage: 100000 }, + { created_at: '2026-01-01T10:10:00+00:00', gear_status: 'drive', battery_capacity: 111 }, + { created_at: '2026-01-01T12:10:00+00:00', battery_level: 52 }, + { created_at: '2026-01-01T12:20:00+00:00', cellular_carrier: 'AT&T' }, + { created_at: '2026-01-01T13:30:00+00:00', vehicle_mileage: 120000 }, + ], + location_metrics: [ + { created_at: '2026-01-01T10:05:00+00:00', latitude: 33.1, longitude: -117.2, speed: 10 }, + { created_at: '2026-01-01T10:15:00+00:00', latitude: 33.2, longitude: -117.2, speed: 12 }, + ], + charging_metrics: [ + { created_at: '2026-01-01T12:05:00+00:00', state: 3, power_kw: 7.4 }, + { created_at: '2026-01-01T12:15:00+00:00', state: 3, power_kw: 7.5 }, + { created_at: '2026-01-01T12:59:00+00:00', state: 4, power_kw: 0 }, + ], + }); + + it('refuses a zip that isn’t an export, and a file that isn’t a zip', async () => { + await expect(store(zipOf({ 'notes.txt': 'hello' }))).rejects.toBeInstanceOf(ImportError); + await expect(store(Buffer.from('data_id,Date\n1,2026-01-01'))).rejects.toThrow(); + await expect( + store(zipOf({ 'manifest.json': JSON.stringify({ format_version: 2, files: {} }) })), + ).rejects.toThrow('format RivianMate doesn’t know'); + }); + + it('says which vehicles it has data for', async () => { + expect((await store(exported)).vins).toEqual([ROAMER_VIN]); + }); + + it('reads sessions, with what the metrics add to them', async () => { + const { path } = await store(exported); + const scan = await roamerFormat.scan([path], options); + expect(scan.drives).toHaveLength(1); + // The capacity known when the drive set off, not one read during it. + expect(scan.drives[0]!.batteryCapacityKwh).toBe(110); + expect(scan.drives[0]).toMatchObject({ + driverName: 'Laura V.', + destination: { name: 'Beach', lat: 32.7, lon: -117.1 }, + }); + expect(scan.places.map((p) => [p.name, p.isHome])).toEqual([ + ['Home', true], + ['Work', false], + ]); + expect(scan.charges).toHaveLength(1); + expect(scan.charges[0]!.curve).toEqual([ + { at: at('2026-01-01T12:05:00Z'), powerKw: 7.4, soc: 50 }, + { at: at('2026-01-01T12:15:00Z'), powerKw: 7.5, soc: 52 }, + ]); + expect(scan.rows).toBe(7); + // The carrier alone isn't a reading RivianMate keeps. + expect(scan.readingTimes).toHaveLength(4); + expect(scan.firstMileage?.value).toBe(100000); + expect(scan.lastMileage?.value).toBe(120000); + await expect(roamerFormat.scan([path], { ...options, vin: 'OTHER' })).rejects.toThrow( + 'no data for that vehicle', + ); + }); + + it('gives metrics and fixes in time order', async () => { + const { path } = await store(exported); + const readings: Reading[] = []; + for await (const r of roamerFormat.readings([path], options)) readings.push(r); + expect(readings.map((r) => new Date(r.at).toISOString().slice(11, 16))).toEqual([ + '09:00', + '10:05', + '10:10', + '10:15', + '12:10', + '12:20', + '13:30', + ]); + expect(readings[1]).toMatchObject({ reporting: false, fix: { speedKmh: 36 } }); + expect(readings[2]).toMatchObject({ reporting: true, fix: null }); + }); +}); diff --git a/server/src/testing/electrafi-rows.ts b/server/src/testing/electrafi-rows.ts index 4168e54..e2a4569 100644 --- a/server/src/testing/electrafi-rows.ts +++ b/server/src/testing/electrafi-rows.ts @@ -17,6 +17,7 @@ const COLUMNS = [ 'charge_limit_soc', 'car_version', 'spoiler_type', + 'autopark_state_v2', 'charging_state', 'chargeNumber', 'charger_power', diff --git a/server/src/testing/roamer-archive.ts b/server/src/testing/roamer-archive.ts new file mode 100644 index 0000000..c9f2013 --- /dev/null +++ b/server/src/testing/roamer-archive.ts @@ -0,0 +1,80 @@ +import { crc32 } from 'node:zlib'; + +/** A zip of the files, stored uncompressed: enough for the reader under test. */ +export function zipOf(files: Record): Buffer { + const locals: Buffer[] = []; + const central: Buffer[] = []; + let offset = 0; + for (const [name, content] of Object.entries(files)) { + const nameBytes = Buffer.from(name, 'utf8'); + const data = Buffer.from(content, 'utf8'); + const crc = crc32(data); + const local = Buffer.alloc(30); + local.writeUInt32LE(0x04034b50, 0); + local.writeUInt16LE(20, 4); + local.writeUInt32LE(crc, 14); + local.writeUInt32LE(data.length, 18); + local.writeUInt32LE(data.length, 22); + local.writeUInt16LE(nameBytes.length, 26); + locals.push(local, nameBytes, data); + const entry = Buffer.alloc(46); + entry.writeUInt32LE(0x02014b50, 0); + entry.writeUInt16LE(20, 4); + entry.writeUInt16LE(20, 6); + entry.writeUInt32LE(crc, 16); + entry.writeUInt32LE(data.length, 20); + entry.writeUInt32LE(data.length, 24); + entry.writeUInt16LE(nameBytes.length, 28); + entry.writeUInt32LE(offset, 42); + central.push(entry, nameBytes); + offset += local.length + nameBytes.length + data.length; + } + const directory = Buffer.concat(central); + const end = Buffer.alloc(22); + end.writeUInt32LE(0x06054b50, 0); + end.writeUInt16LE(Object.keys(files).length, 8); + end.writeUInt16LE(Object.keys(files).length, 10); + end.writeUInt32LE(directory.length, 12); + end.writeUInt32LE(offset, 16); + return Buffer.concat([...locals, directory, end]); +} + +export const ROAMER_VEHICLE = 'rv1'; +export const ROAMER_VIN = 'TESTVIN0000000001'; + +const lines = (records: readonly object[]) => records.map((r) => JSON.stringify(r)).join('\n'); + +/** + * A Roamer export for one vehicle: the datasets given, by name, listed in a + * manifest as Roamer lists them. Telemetry goes in one dated file each, and + * the account's places at the top. + */ +export function roamerExport( + datasets: Partial>, + vin = ROAMER_VIN, +): Buffer { + const telemetry = [ + 'vehicle_metrics', + 'location_metrics', + 'charging_metrics', + 'navigation_metrics', + ]; + const account = ['saved_places', 'rivian_favorites']; + const files: Record = { + 'vehicles.jsonl': lines([{ id: ROAMER_VEHICLE, vin, name: 'Test', model: 'R1S' }]), + }; + const listed: Record = { + 'vehicles.jsonl': { dataset: 'vehicles', rows: 1 }, + }; + for (const [dataset, records] of Object.entries(datasets)) { + const name = account.includes(dataset) + ? `${dataset}.jsonl` + : telemetry.includes(dataset) + ? `vehicles/${ROAMER_VEHICLE}/${dataset}/20260101T000000.jsonl` + : `vehicles/${ROAMER_VEHICLE}/${dataset}.jsonl`; + files[name] = lines(records ?? []); + listed[name] = { dataset, rows: records?.length ?? 0 }; + } + files['manifest.json'] = JSON.stringify({ format_version: 1, files: listed }); + return zipOf(files); +} diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 764754f..4d15677 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -17,6 +17,7 @@ import type { ImportChargePreview, ImportDraftDto, ImportDrivePreview, + ImportFormat, ImportPreviewDto, ImportSelection, ImportSessionAddition, @@ -52,6 +53,7 @@ export type { ImportChargePreview, ImportDraftDto, ImportDrivePreview, + ImportFormat, ImportPreviewDto, ImportSelection, ImportSessionAddition, @@ -269,8 +271,8 @@ export const api = { ), imports: () => request('/api/imports'), /** Sent as is, gzipped or not: the server works out which. */ - uploadImportFile: (file: File, draftId: string | undefined) => { - const params = new URLSearchParams({ name: file.name }); + uploadImportFile: (file: File, draftId: string | undefined, format: ImportFormat) => { + const params = new URLSearchParams({ name: file.name, format }); if (draftId) params.set('draft', draftId); return request(`/api/imports/files?${params}`, { method: 'POST', diff --git a/web/src/components/AuthCard.test.tsx b/web/src/components/AuthCard.test.tsx new file mode 100644 index 0000000..0e2209b --- /dev/null +++ b/web/src/components/AuthCard.test.tsx @@ -0,0 +1,12 @@ +import { renderToStaticMarkup } from 'react-dom/server'; +import { describe, expect, it } from 'vitest'; +import { TextField } from './AuthCard'; + +describe('TextField', () => { + it('keeps the show-password toggle out of the tab order', () => { + const html = renderToStaticMarkup( + {}} />, + ); + expect(html).toMatch(/]*tabindex="-1"[^>]*aria-label="Show password"/); + }); +}); diff --git a/web/src/components/AuthCard.tsx b/web/src/components/AuthCard.tsx index 0421498..f84a421 100644 --- a/web/src/components/AuthCard.tsx +++ b/web/src/components/AuthCard.tsx @@ -145,6 +145,8 @@ export function TextField(props: { - )} )} @@ -376,9 +440,10 @@ export function ImportFlow(props: { {step === 'review' && (

- {preview.isPending ? 'Reading your files…' : (summary?.label ?? 'No preview yet')} + {preview.isPending ? 'Reading files…' : (summary?.label ?? 'No preview yet')}

-
+ {/* Room for the usual two lines, so the summary arriving doesn't push the buttons down. */} +
{preview.isPending ? 'Large exports can take a minute.' : summary?.details.map((line) =>
{line}
)} @@ -396,10 +461,10 @@ export function ImportFlow(props: { {/* Above the review's list, so its length never puts these out of reach. */}
-
+
{step !== 'source' && ( )} {draft && ( '); expect(html).toContain('>Discard'); @@ -128,11 +135,15 @@ describe('ImportPanel while another import runs', () => { }); describe('ImportFlow', () => { - const renderStep = (initialStep: ImportStep, draft: ImportsDto['draft'] = null) => + const renderStep = ( + initialStep: ImportStep, + draft: ImportsDto['draft'] = null, + vehicles = [rivian], + ) => renderToStaticMarkup( { /> , ); - const draft = { id: 'd', files: ['export.csv'] }; + const draft = { + id: 'd', + format: 'electrafi' as const, + files: ['export.csv'], + vehicleIds: null, + }; + const archive = { id: 'r', format: 'roamer' as const, files: ['roamer.zip'], vehicleIds: ['v1'] }; - it('starts by asking for the source, showing where it is', () => { + it('starts by asking for the source, with Roamer first', () => { const html = renderStep('source'); - expect(html).toContain('ElectraFi'); + expect(html.indexOf('Roamer')).toBeGreaterThan(-1); + expect(html.indexOf('Roamer')).toBeLessThan(html.indexOf('ElectraFi')); expect(html).toContain('type="radio"'); expect(html).toContain('Step 1 of 4'); expect(html).not.toContain('>Back'); @@ -154,11 +172,14 @@ describe('ImportFlow', () => { it('holds at the files step until there are some', () => { const html = renderStep('files'); - expect(html).toContain('Download All'); - expect(html).toContain('Choose files'); + expect(html).toContain('Download a copy'); + expect(html).toContain('Choose file<'); + expect(html).toContain('No file added yet.'); + expect(html).not.toContain('multiple'); expect(html).toMatch(/disabled=""[^>]*>Continue/); expect(html).not.toContain('Discard upload'); const withFiles = renderStep('files', draft); + expect(withFiles).toContain('Add your ElectraFi export'); expect(withFiles).toContain('export.csv'); expect(withFiles).toContain('Add more'); expect(withFiles).not.toMatch(/disabled=""[^>]*>Continue/); @@ -166,9 +187,39 @@ describe('ImportFlow', () => { expect(withFiles).toContain('Save for later'); }); + it('replaces a Roamer archive rather than adding to it', () => { + const html = renderStep('files', archive); + expect(html).toContain('roamer.zip'); + expect(html).toContain('Choose another'); + expect(html).not.toContain('Add more'); + // The instructions stay with the file added. + expect(html).toContain('Download a copy'); + }); + + it('shows the vehicle the export was matched to, without a choice', () => { + const red = { ...rivian, id: 'v2', name: 'Red' } as VehicleDto; + const html = renderStep('settings', archive, [rivian, red]); + expect(html).toContain('Blue'); + expect(html).not.toContain('Red'); + expect(html).toContain('by VIN'); + expect(html).not.toContain('id="import-vehicle"'); + const both = renderStep('settings', { ...archive, vehicleIds: ['v1', 'v2'] }, [rivian, red]); + expect(both).toContain('id="import-vehicle"'); + const none = renderStep('settings', { ...archive, vehicleIds: [] }, [rivian]); + expect(none).toContain('No vehicle in this export is connected.'); + expect(none).toMatch(/disabled=""[^>]*>Review/); + }); + it('asks for settings once there are files', () => { const html = renderStep('settings', draft); expect(html).toContain('Time zone'); + // The rule stays put while what the files cover fills in below it. + expect(html).toContain('imported whole.'); + expect(html).toMatch(/Choose the dates\./); + // Held unseen until the dates are narrowed, so its showing doesn't re-wrap the text. + expect(html).toMatch(/invisible">Use all datesReading files…/); expect(html).toContain('Blue'); expect(html).not.toContain('Model Y'); expect(html).toContain('>Review'); @@ -215,8 +266,15 @@ describe('importSummary', () => { importSummary({ ...preview, sessionsEnriched: 1, readingsNew: 786, readingsRecorded: 638 }) .details, ).toEqual([ - 'Also adds 786 history readings and detail on 1 existing charge.', - 'Skips 638 history readings from times RivianMate was already recording.', + 'Also adds 786 history readings and detail on 1 charge.', + 'Skips 638 readings RivianMate already has.', + ]); + }); + + it('tells readings it skips from ones it leaves for live tracking', () => { + expect(importSummary({ ...preview, readingsNewer: 40 }).details).toEqual([ + 'Also adds 500 history readings.', + 'Leaves 40 newer readings to live tracking.', ]); }); @@ -232,6 +290,21 @@ describe('importSummary', () => { expect(summary.canImport).toBe(false); }); + it('names the places it adds, and counts the ones already saved', () => { + const summary = importSummary({ + ...preview, + readingsNew: 0, + placesNew: ['Home', 'Work'], + placesRecorded: 1, + }); + expect(summary.details).toEqual([ + 'Also adds 2 places (Home, Work).', + 'Skips 1 place already saved.', + ]); + const only = { ...preview, drivesNew: 0, sessionsNew: 0, readingsNew: 0 }; + expect(importSummary({ ...only, placesNew: ['Home'] }).canImport).toBe(true); + }); + it('still imports when it only adds detail or history', () => { const none = { ...preview, drivesNew: 0, sessionsNew: 0, readingsNew: 0 }; expect(importSummary({ ...none, sessionsEnriched: 1 }).canImport).toBe(true); @@ -249,8 +322,8 @@ describe('PreviewRows', () => { it('warns about a time zone that looks wrong, leaving the car to the vehicle row', () => { expect(html({ warnings: ['future'] })).toContain('in the future'); - expect(html({ warnings: ['odometer'] })).not.toContain('right car'); - expect(renderToStaticMarkup()).toContain('right car'); + expect(html({ warnings: ['odometer'] })).not.toContain('Check the vehicle'); + expect(renderToStaticMarkup()).toContain('Check the vehicle'); expect(html({})).not.toContain('role="alert"'); }); }); diff --git a/web/src/features/settings/ImportPanel.tsx b/web/src/features/settings/ImportPanel.tsx index 28d77ed..51f3428 100644 --- a/web/src/features/settings/ImportPanel.tsx +++ b/web/src/features/settings/ImportPanel.tsx @@ -18,6 +18,7 @@ export function batchSummary(batch: ImportBatchDto): string { plural(c.sessions, 'charge'), plural(c.locationPoints, 'location point'), plural(c.stateChanges, 'state change'), + ...(c.places ? [plural(c.places, 'saved place')] : []), ].join(', '); return c.sessionsEnriched > 0 ? `${added}; added detail to ${plural(c.sessionsEnriched, 'charge')}` @@ -88,7 +89,7 @@ export function ImportPanel(props: { vehicles: readonly VehicleDto[] }) {