From f521d41b1a68ec9a87b3bb0133089db3f6b64403 Mon Sep 17 00:00:00 2001 From: Mitch Vitale Date: Sat, 7 Nov 2026 20:19:37 -0800 Subject: [PATCH 1/2] Import from Rivian Roamer, and bring drivers, places and destinations across MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rivian Roamer is now the first source under Settings → Import. Its data copy is one .zip, read in place, and matched to your Rivian by VIN. From a Rivian Roamer copy: - Drives and charges are Roamer's own sessions; ones it flagged as corrupted are left out. - Charges keep their energy breakdown and a curve from Roamer's charging readings. A charge you already have only gets what it's missing, and undo puts it back as it was. - Vehicle history comes from the metrics Roamer kept as Rivian sent them; ones it recoded (seat heat, service mode, range thresholds) are skipped. - Each drive gets the driver profile and navigation destination Roamer recorded, and the account's saved places are added unless one is already saved there. ElectraFi drives now get their driver too: ElectraFi keeps the active profile's name in Tesla's autopark_state_v2 column. Drivers are matched by name to ones RivianMate knows. A name it doesn't know becomes a driver without a profile, which the car's profile takes over once it reports that name. Undo keeps drivers and places. The import itself: - The shared steps (drafts, planning, writing, undo) move to services/imports, with each export as a format that reads its files. - The preview counts readings left for live tracking apart from ones already recorded, and lists the places it adds. - The flow's rows hold their size as files and dates load, and its text is shorter. --- README.md | 11 +- pnpm-lock.yaml | 26 ++ server/package.json | 2 + server/src/api-types.ts | 33 +- server/src/context.ts | 4 +- server/src/db/schema.ts | 25 +- server/src/index.ts | 4 +- server/src/routes/imports.ts | 9 +- server/src/services/charging-curve.ts | 13 +- server/src/services/drive-places.ts | 4 +- server/src/services/drivers.db.test.ts | 22 ++ server/src/services/drivers.ts | 49 ++- .../src/services/electrafi/electrafi.test.ts | 31 +- server/src/services/electrafi/format.ts | 119 ++++++ server/src/services/electrafi/parse.ts | 113 +----- server/src/services/electrafi/segment.ts | 52 +-- server/src/services/imports/format.ts | 160 ++++++++ .../{electrafi => imports}/imports.db.test.ts | 291 +++++++++++++- .../{electrafi => imports}/imports.ts | 370 +++++++++++------- .../services/{electrafi => imports}/plan.ts | 91 ++--- server/src/services/imports/time-zone.ts | 109 ++++++ server/src/services/roamer/archive.ts | 134 +++++++ server/src/services/roamer/format.ts | 272 +++++++++++++ server/src/services/roamer/parse.ts | 257 ++++++++++++ server/src/services/roamer/roamer.test.ts | 327 ++++++++++++++++ server/src/testing/electrafi-rows.ts | 1 + server/src/testing/roamer-archive.ts | 80 ++++ web/src/api/client.ts | 6 +- web/src/features/charging/EnergySplit.tsx | 4 +- web/src/features/charging/SessionCard.tsx | 3 + web/src/features/settings/ImportFlow.tsx | 247 ++++++++---- .../features/settings/ImportPanel.test.tsx | 97 ++++- web/src/features/settings/ImportPanel.tsx | 7 +- .../settings/ImportPreviewList.test.tsx | 5 + .../features/settings/ImportPreviewList.tsx | 2 + 35 files changed, 2500 insertions(+), 480 deletions(-) create mode 100644 server/src/services/electrafi/format.ts create mode 100644 server/src/services/imports/format.ts rename server/src/services/{electrafi => imports}/imports.db.test.ts (71%) rename server/src/services/{electrafi => imports}/imports.ts (74%) rename server/src/services/{electrafi => imports}/plan.ts (86%) create mode 100644 server/src/services/imports/time-zone.ts create mode 100644 server/src/services/roamer/archive.ts create mode 100644 server/src/services/roamer/format.ts create mode 100644 server/src/services/roamer/parse.ts create mode 100644 server/src/services/roamer/roamer.test.ts create mode 100644 server/src/testing/roamer-archive.ts 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/features/charging/EnergySplit.tsx b/web/src/features/charging/EnergySplit.tsx index bf1b797..679bcd6 100644 --- a/web/src/features/charging/EnergySplit.tsx +++ b/web/src/features/charging/EnergySplit.tsx @@ -44,7 +44,9 @@ export function EnergySplit(props: { : 'TeslaMate didn’t record the energy drawn for this session' : s.source === 'electrafi' ? 'ElectraFi exports don’t include a breakdown' - : 'Rivian didn’t report a breakdown for this session'; + : s.source === 'roamer' + ? 'Rivian Roamer’s export has no breakdown for this session' + : 'Rivian didn’t report a breakdown for this session'; return (
diff --git a/web/src/features/charging/SessionCard.tsx b/web/src/features/charging/SessionCard.tsx index 5f9b432..8232fe3 100644 --- a/web/src/features/charging/SessionCard.tsx +++ b/web/src/features/charging/SessionCard.tsx @@ -98,6 +98,9 @@ export function SessionCard(props: { {s.source === 'electrafi' && ( )} + {s.source === 'roamer' && ( + + )} {s.source === 'teslamate' && ( )} diff --git a/web/src/features/settings/ImportFlow.tsx b/web/src/features/settings/ImportFlow.tsx index d73edf8..4a3003e 100644 --- a/web/src/features/settings/ImportFlow.tsx +++ b/web/src/features/settings/ImportFlow.tsx @@ -1,16 +1,22 @@ import { useMutation } from '@tanstack/react-query'; -import { useEffect, useRef, useState } from 'react'; -import { type ImportDraftDto, type ImportPreviewDto, type VehicleDto, api } from '../../api/client'; +import { type ReactNode, useEffect, useRef, useState } from 'react'; +import { + type ImportDraftDto, + type ImportFormat, + type ImportPreviewDto, + type VehicleDto, + api, +} from '../../api/client'; import { StepDots } from '../../components/AuthCard'; import { ConfirmDialog } from '../../components/ConfirmDialog'; import { ImportPreviewList } from './ImportPreviewList'; import { Select } from '../../components/Select'; -import { SettingRow, secondaryButton } from './SettingsControls'; +import { SettingRow, dangerButton, secondaryButton } from './SettingsControls'; +const linkButton = + 'font-medium text-(--text-secondary) underline decoration-dotted underline-offset-2 hover:text-(--text-primary)'; const fieldClass = 'rounded-lg border border-(--border) bg-(--surface-2) px-3 py-1.5 text-sm'; const primaryButton = 'btn-primary rounded-md px-3 py-1.5 text-sm font-medium'; -const textButton = - 'font-medium text-(--text-secondary) underline decoration-dotted underline-offset-2 hover:text-(--text-primary) disabled:opacity-50'; /** Wider than any export, for reading what its files cover. */ const ALL_DATES = { from: '0001-01-01', to: '9999-12-31' }; @@ -23,31 +29,47 @@ export function timeZoneOptions(current: string): string[] { return known.includes(current) ? known : [current, ...known]; } +/** "a", "a and b", "a, b and c". */ +const listOf = (parts: readonly string[]) => + parts.length < 2 ? (parts[0] ?? '') : `${parts.slice(0, -1).join(', ')} and ${parts.at(-1)}`; + const plural = (n: number, one: string) => `${n.toLocaleString()} ${one}${n === 1 ? '' : 's'}`; export interface ImportSource { - id: string; + id: ImportFormat; label: string; description: string; /** Where to get the files. */ instructions: string; accept: string; + /** Whether an export comes as several files; otherwise another upload replaces it. */ + multiple: boolean; vehicleHelp: string; timeZoneHelp: string; } /** Each kind of export that can be imported. */ export const IMPORT_SOURCES: readonly ImportSource[] = [ + { + id: 'roamer', + label: 'Rivian Roamer', + description: 'Drives, charges, routes, history, drivers and places.', + instructions: + 'In Rivian Roamer: Account → Data → Download a copy. Request a copy, then Download ZIP once it’s ready.', + accept: '.zip,application/zip', + multiple: false, + vehicleHelp: 'Matched by VIN.', + timeZoneHelp: 'Sets where each day starts.', + }, { id: 'electrafi', label: 'ElectraFi', - description: 'Drives, charges, routes and vehicle history.', - instructions: - 'In ElectraFi, go to Settings → Account → Advanced → Download Data and choose Download All. Add the .csv or .csv.gz files it gives you.', + description: 'Drives, charges, routes, history and drivers.', + instructions: 'In ElectraFi: Settings → Account → Advanced → Download Data → Download All.', accept: '.csv,.gz,text/csv,application/gzip', - vehicleHelp: 'ElectraFi is one account per car, so choose which one this is.', - timeZoneHelp: - 'Exports have local times with no zone: use the one your ElectraFi account shows.', + multiple: true, + vehicleHelp: 'The car this export is from.', + timeZoneHelp: 'The one your ElectraFi account uses.', }, ]; @@ -138,8 +160,13 @@ export function ImportFlow(props: { const [saved] = useState(() => loadChoices(draft?.id)); const [requestedStep, setStep] = useState(props.initialStep); const step = reachableStep(requestedStep, !!draft); - const [sourceId, setSourceId] = useState(IMPORT_SOURCES[0]!.id); + const [sourceId, setSourceId] = useState(draft?.format ?? IMPORT_SOURCES[0]!.id); const source = IMPORT_SOURCES.find((s) => s.id === sourceId) ?? IMPORT_SOURCES[0]!; + // An upload from another source is replaced by the next one. + const sourceDraft = draft?.format === source.id ? draft : null; + const eligible = draft?.vehicleIds + ? props.vehicles.filter((v) => draft.vehicleIds!.includes(v.id)) + : props.vehicles; const [vehicleId, setVehicleId] = useState(saved?.vehicleId ?? ''); const [timeZone, setTimeZone] = useState(saved?.timeZone ?? browserTimeZone()); const [from, setFrom] = useState(saved?.from ?? ''); @@ -148,16 +175,14 @@ export function ImportFlow(props: { const [covered, setCovered] = useState(saved?.covered ?? null); const [discarding, setDiscarding] = useState(false); const [uploading, setUploading] = useState<{ at: number; of: number } | null>(null); - const chosen = props.vehicles.some((v) => v.id === vehicleId) - ? vehicleId - : (props.vehicles[0]?.id ?? ''); + const chosen = eligible.some((v) => v.id === vehicleId) ? vehicleId : (eligible[0]?.id ?? ''); const upload = useMutation({ mutationFn: async (files: File[]) => { - let id = draft?.id; + let id = source.multiple ? sourceDraft?.id : undefined; for (const [i, file] of files.entries()) { setUploading({ at: i + 1, of: files.length }); - id = (await api.uploadImportFile(file, id)).id; + id = (await api.uploadImportFile(file, id, source.id)).id; } }, onSettled: () => { @@ -222,10 +247,10 @@ export function ImportFlow(props: { const error = upload.error ?? preview.error ?? start.error; const stepNumber = IMPORT_STEPS.findIndex((s) => s.id === step) + 1; const title = { - source: 'Where is this import from?', - files: `Add your ${source.label} files`, + source: 'Choose a source', + files: `Add your ${source.label} export`, settings: 'Choose what to import', - review: 'Check and import', + review: 'Review', }[step]; return ( @@ -267,19 +292,41 @@ export function ImportFlow(props: { {step === 'files' && ( + {source.instructions} + + {sourceDraft + ? sourceDraft.files.join(', ') + : source.multiple + ? 'No files added yet.' + : 'No file added yet.'} + + + } >