523 lines
20 KiB
JavaScript
523 lines
20 KiB
JavaScript
/* DEPRECATED — use scripts/migrate_applications.js instead.
|
||
* This script is kept for reference only.
|
||
*/
|
||
'use strict';
|
||
|
||
/**
|
||
* Application Aggregates Migration Script
|
||
*
|
||
* Backfills BOTH avgSpraySpeed AND totalSprLength/totalFlightLength on Application
|
||
* (and totalSprLength/totalFlightLength on AppFile)
|
||
* in a SINGLE traversal per application — no double-scanning.
|
||
*
|
||
* Fields updated:
|
||
* Application.avgSpraySpeed — weighted avg ground speed (m/s) across all spray-on AppDetail records
|
||
* Application.totalSprLength — cumulative geodesic spray distance (m) across all AppFile records
|
||
* Application.totalFlightLength — cumulative flight distance (m) across all AppFile records
|
||
* Application.avgXtError — weighted avg abs cross-track error (m) across spray-on records
|
||
* Application.avgHdop — weighted avg HDOP across spray-on records (lower = better)
|
||
* Application.flowAccuracyPct — (totalSprayMat/totalSprayed / appRate) × 100; backfilled in a second pass
|
||
* AppFile.totalSprLength — per-file spray distance (m)
|
||
* AppFile.totalFlightLength — per-file flight distance (m)
|
||
*
|
||
* Performance characteristics:
|
||
* - Applications processed most-recent-first (ObjectId descending = newest first)
|
||
* - AppDetail streamed via Mongoose cursor (never materialises all records for a file in memory)
|
||
* - Bulk writes to Application and AppFile flushed every --batch-size apps
|
||
* - Haversine (pure JS) replaces turf point objects in the inner loop — ~5× faster per record
|
||
* - `.lean()` on every Mongoose query
|
||
* - Tiered execution: --tier-days or --from-date limits the ObjectId range so you can target
|
||
* the most recent data first, then re-run without the flag for a full backfill
|
||
*
|
||
* Distance algorithm (matches readSatLogAsc in job_worker.js, corrected prevLonLat bug):
|
||
* For consecutive AppDetail rows sorted by gpsTime ASC:
|
||
* if ( (prevStat > 0 && curStat > 0) || curStat != prevStat )
|
||
* d = haversine(prevLon, prevLat, curLon, curLat)
|
||
* if d <= 1000m → totalSprLength += d
|
||
*
|
||
* Usage:
|
||
* node scripts/migrate_app_aggregates.js [options]
|
||
*
|
||
* # Tier 1 — most recent 90 days first (fastest to see results)
|
||
* DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js --tier-days=90
|
||
*
|
||
* # Tier 2 — everything from 2023 onward
|
||
* DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js --from-date=2023-01-01
|
||
*
|
||
* # Full backfill (all-time)
|
||
* DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js
|
||
*
|
||
* # Dry-run (logs what would change, no writes)
|
||
* DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js --dry-run --tier-days=7
|
||
*
|
||
* # Recompute already-set fields (e.g. after algorithm fix)
|
||
* DEBUG=agm:migrate-app-aggregates node scripts/migrate_app_aggregates.js --force --tier-days=30
|
||
*
|
||
* # Custom env file
|
||
* node scripts/migrate_app_aggregates.js --env ./environment_prod.env --tier-days=90
|
||
*
|
||
* Options:
|
||
* --env <path> Environment file (default: ./environment.env)
|
||
* --dry-run Report only; make no DB writes
|
||
* --force Recompute even when both fields are already set
|
||
* --batch-size=N Applications per bulkWrite flush (default: 50)
|
||
* --tier-days=N Only process apps created in the last N days
|
||
* --from-date=YYYY-MM-DD Only process apps created on/after this date (UTC)
|
||
* --concurrency=N Parallel file-level cursors per application (default: 3)
|
||
*/
|
||
|
||
// ─── Environment bootstrap ────────────────────────────────────────────────────
|
||
const path = require('path');
|
||
const args = process.argv.slice(2);
|
||
|
||
let envFile = './environment.env';
|
||
for (let i = 0; i < args.length; i++) {
|
||
if (args[i] === '--env' && args[i + 1]) { envFile = args[i + 1]; i++; }
|
||
else if (args[i].startsWith('--env=')) { envFile = args[i].split('=')[1]; }
|
||
}
|
||
require('dotenv').config({ path: path.resolve(process.cwd(), envFile) });
|
||
|
||
// ─── Imports (after env is loaded) ───────────────────────────────────────────
|
||
const debug = require('debug')('agm:migrate-app-aggregates');
|
||
const mongoose = require('mongoose');
|
||
const { DBConnection } = require('../helpers/db/connect');
|
||
const Application = require('../model/application');
|
||
const AppFile = require('../model/application_file');
|
||
const AppDetail = require('../model/application_detail');
|
||
|
||
// ─── Argument parsing ─────────────────────────────────────────────────────────
|
||
const cfg = {
|
||
dryRun: false,
|
||
force: false,
|
||
batchSize: 50,
|
||
tierDays: null, // Number – only apps newer than N days
|
||
fromDate: null, // ISO string – only apps on/after this date
|
||
concurrency: 3, // parallel file cursors per app
|
||
};
|
||
|
||
for (let i = 0; i < args.length; i++) {
|
||
const a = args[i];
|
||
if (a === '--dry-run') cfg.dryRun = true;
|
||
else if (a === '--force') cfg.force = true;
|
||
else if (a.startsWith('--batch-size=')) cfg.batchSize = parseInt(a.split('=')[1], 10) || 50;
|
||
else if (a.startsWith('--tier-days=')) cfg.tierDays = parseInt(a.split('=')[1], 10) || null;
|
||
else if (a.startsWith('--from-date=')) cfg.fromDate = a.split('=')[1] || null;
|
||
else if (a.startsWith('--concurrency=')) cfg.concurrency = parseInt(a.split('=')[1], 10) || 3;
|
||
// --env already consumed above
|
||
}
|
||
|
||
// ─── Constants ────────────────────────────────────────────────────────────────
|
||
const MAX_SEGMENT_METERS = 1000; // Sanity cap matching job_worker readSatLogAsc
|
||
const PROGRESS_LOG_INTERVAL = 100; // Log a line every N apps
|
||
|
||
// ─── Helpers ──────────────────────────────────────────────────────────────────
|
||
|
||
/**
|
||
* Fast Haversine distance in metres between two WGS-84 points.
|
||
* Avoids turf object allocation in tight loops.
|
||
*/
|
||
function haversineMeters(lon1, lat1, lon2, lat2) {
|
||
const R = 6371000;
|
||
const φ1 = lat1 * Math.PI / 180;
|
||
const φ2 = lat2 * Math.PI / 180;
|
||
const Δφ = (lat2 - lat1) * Math.PI / 180;
|
||
const Δλ = (lon2 - lon1) * Math.PI / 180;
|
||
const a = Math.sin(Δφ / 2) ** 2 + Math.cos(φ1) * Math.cos(φ2) * Math.sin(Δλ / 2) ** 2;
|
||
return R * 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));
|
||
}
|
||
|
||
/**
|
||
* Create a minimal ObjectId whose timestamp equals `date`.
|
||
*/
|
||
function objectIdFromDate(date) {
|
||
const ts = Math.floor(new Date(date).getTime() / 1000);
|
||
const hex = ts.toString(16).padStart(8, '0') + '0000000000000000';
|
||
return new mongoose.Types.ObjectId(hex);
|
||
}
|
||
|
||
/**
|
||
* Process all AppDetail records for one AppFile in a single streaming pass.
|
||
* Returns { sprLength, flightLength, speedSum, speedCount, xtSum, xtCount, hdopSum, hdopCount }.
|
||
*
|
||
* Validity gates (matching job_worker conventions):
|
||
* sprLength — spray-ON segment (prevStat > 0 OR curStat > 0) AND distance ≤ 1000 m AND time gap ≤ 120 s
|
||
* flightLength — distance ≤ 1000 m AND time gap ≤ 120 s (valid GPS segment, all movements)
|
||
* avgXtError — spray-on (sprayStat 1 or 3) records within valid (both-gate) segments, xTrack ≠ 0
|
||
* avgHdop — spray-on (sprayStat > 0) records within valid (both-gate) segments, stdHdop > 0
|
||
*/
|
||
async function processFile(fileId) {
|
||
let sprLength = 0;
|
||
let flightLength = 0;
|
||
let speedSum = 0;
|
||
let speedCount = 0;
|
||
let xtSum = 0;
|
||
let xtCount = 0;
|
||
let hdopSum = 0;
|
||
let hdopCount = 0;
|
||
|
||
let prevLon = null;
|
||
let prevLat = null;
|
||
let prevGpsTime = null;
|
||
let prevStat = -999;
|
||
|
||
// Sort by gpsTime for correct consecutive-point distance calculation.
|
||
// Tie-break by _id for determinism when gpsTime == 0 (legacy).
|
||
const cursor = AppDetail
|
||
.find({ fileId })
|
||
.select('lat lon gpsTime sprayStat grSpeed xTrack stdHdop')
|
||
.sort({ gpsTime: 1, _id: 1 })
|
||
.lean()
|
||
.cursor();
|
||
|
||
for await (const rec of cursor) {
|
||
const curLon = rec.lon;
|
||
const curLat = rec.lat;
|
||
const curGpsTime = rec.gpsTime;
|
||
const curStat = rec.sprayStat || 0;
|
||
|
||
if (prevStat !== -999 &&
|
||
(typeof curLon === 'number') && (typeof curLat === 'number') &&
|
||
(typeof prevLon === 'number') && (typeof prevLat === 'number')) {
|
||
|
||
const d = haversineMeters(prevLon, prevLat, curLon, curLat);
|
||
|
||
// Time gap with midnight-rollover handling (gpsTime is seconds-of-day)
|
||
let timeDif = (typeof curGpsTime === 'number' && typeof prevGpsTime === 'number')
|
||
? curGpsTime - prevGpsTime
|
||
: Infinity;
|
||
if (timeDif < 0 && Math.abs(timeDif) >= 80000) timeDif = 86400 - prevGpsTime + curGpsTime;
|
||
|
||
// ── Valid segment: both distance AND time gate ───────────────────────────
|
||
const segValid = d <= MAX_SEGMENT_METERS && timeDif > 0 && timeDif <= 120;
|
||
|
||
// ── Spray length: spray-ON segment within valid (both-gate) window ────────
|
||
// Matches job_worker _computeSprLength which applies both distance AND time gates.
|
||
if (segValid && (prevStat > 0 || curStat > 0)) {
|
||
sprLength += d;
|
||
}
|
||
|
||
// ── Flight length: all GPS movements within valid segments ───────────────
|
||
if (segValid) {
|
||
flightLength += d;
|
||
}
|
||
|
||
// ── XT error: spray-on records (stat 1 = on-swath, 3 = swath entry) ──────
|
||
// Only within valid segments; exclude zero (no reading) and non-numeric values.
|
||
if (segValid && (curStat === 1 || curStat === 3) &&
|
||
typeof rec.xTrack === 'number' && rec.xTrack !== 0) {
|
||
xtSum += Math.abs(rec.xTrack);
|
||
xtCount += 1;
|
||
}
|
||
|
||
// ── HDOP: spray-on records within valid segments, stdHdop > 0 ────────────
|
||
if (segValid && curStat > 0 && typeof rec.stdHdop === 'number' && rec.stdHdop > 0) {
|
||
hdopSum += rec.stdHdop;
|
||
hdopCount += 1;
|
||
}
|
||
}
|
||
|
||
// ── Speed accumulation (all spray-on, no segment gate needed) ────────────
|
||
if (curStat > 0 && typeof rec.grSpeed === 'number' && rec.grSpeed > 0) {
|
||
speedSum += rec.grSpeed;
|
||
speedCount += 1;
|
||
}
|
||
|
||
prevLon = curLon;
|
||
prevLat = curLat;
|
||
prevGpsTime = curGpsTime;
|
||
prevStat = curStat;
|
||
}
|
||
|
||
return { sprLength, flightLength, speedSum, speedCount, xtSum, xtCount, hdopSum, hdopCount };
|
||
}
|
||
|
||
/**
|
||
* Process all files for one Application with bounded parallelism.
|
||
* Returns:
|
||
* appSprLength – total spray length across all files (m)
|
||
* appFlightLength – total flight length across all files (m)
|
||
* avgSpraySpeed – weighted average ground speed across all files (m/s, null if no data)
|
||
* avgXtError – weighted average abs cross-track error (m, null if no data)
|
||
* avgHdop – weighted average HDOP over spray-on records (null if no data)
|
||
* fileOps – array of { fileId, sprLength, flightLength } for per-file bulkWrite
|
||
*/
|
||
async function processApplication(app, appFiles) {
|
||
if (!appFiles.length) {
|
||
return { appSprLength: 0, appFlightLength: 0, avgSpraySpeed: null, avgXtError: null, avgHdop: null, fileOps: [] };
|
||
}
|
||
|
||
// Process files in small parallel batches (bounded concurrency)
|
||
const fileResults = [];
|
||
for (let i = 0; i < appFiles.length; i += cfg.concurrency) {
|
||
const slice = appFiles.slice(i, i + cfg.concurrency);
|
||
const results = await Promise.all(slice.map(f => processFile(f._id)));
|
||
for (let j = 0; j < slice.length; j++) {
|
||
fileResults.push({ file: slice[j], result: results[j] });
|
||
}
|
||
}
|
||
|
||
let appSprLength = 0;
|
||
let appFlightLength = 0;
|
||
let totalSpeedSum = 0;
|
||
let totalSpeedCnt = 0;
|
||
let totalXtSum = 0;
|
||
let totalXtCnt = 0;
|
||
let totalHdopSum = 0;
|
||
let totalHdopCnt = 0;
|
||
const fileOps = [];
|
||
|
||
for (const { file, result } of fileResults) {
|
||
appSprLength += result.sprLength;
|
||
appFlightLength += result.flightLength;
|
||
totalSpeedSum += result.speedSum;
|
||
totalSpeedCnt += result.speedCount;
|
||
totalXtSum += result.xtSum;
|
||
totalXtCnt += result.xtCount;
|
||
totalHdopSum += result.hdopSum;
|
||
totalHdopCnt += result.hdopCount;
|
||
|
||
fileOps.push({ fileId: file._id, sprLength: result.sprLength, flightLength: result.flightLength });
|
||
}
|
||
|
||
const avgSpraySpeed = totalSpeedCnt > 0 ? totalSpeedSum / totalSpeedCnt : null;
|
||
const avgXtError = totalXtCnt > 0 ? totalXtSum / totalXtCnt : null;
|
||
const avgHdop = totalHdopCnt > 0 ? totalHdopSum / totalHdopCnt : null;
|
||
|
||
return { appSprLength, appFlightLength, avgSpraySpeed, avgXtError, avgHdop, fileOps };
|
||
}
|
||
|
||
// ─── Flow accuracy backfill (Application-level only) ─────────────────────────
|
||
|
||
/**
|
||
* Backfill Application.flowAccuracyPct for documents that already have
|
||
* totalSprayMat > 0, totalSprayed > 0, and appRate > 0 but no flowAccuracyPct.
|
||
* No AppDetail queries needed — all three source fields live on Application.
|
||
*/
|
||
async function backfillFlowAccuracy(dryRun, tierDays, fromDate) {
|
||
debug('─'.repeat(60));
|
||
debug('Second pass: backfill flowAccuracyPct …');
|
||
|
||
const filter = {
|
||
markedDelete: { $ne: true },
|
||
flowAccuracyPct: { $exists: false },
|
||
totalSprayed: { $gt: 0 },
|
||
totalSprayMat: { $gt: 0 },
|
||
appRate: { $gt: 0 },
|
||
};
|
||
|
||
if (tierDays) {
|
||
const cutoff = new Date(Date.now() - tierDays * 86400 * 1000);
|
||
filter._id = { $gte: objectIdFromDate(cutoff) };
|
||
} else if (fromDate) {
|
||
filter._id = { $gte: objectIdFromDate(fromDate + 'T00:00:00Z') };
|
||
}
|
||
|
||
debug('flowAccuracy filter: %o', filter);
|
||
|
||
if (dryRun) {
|
||
const count = await Application.countDocuments(filter);
|
||
debug(`[dry-run] Would update ${count} Application docs with flowAccuracyPct`);
|
||
return;
|
||
}
|
||
|
||
// MongoDB 4.2+ aggregation pipeline update — computes the field server-side
|
||
const result = await Application.updateMany(filter, [
|
||
{
|
||
$set: {
|
||
flowAccuracyPct: {
|
||
$round: [
|
||
{ $multiply: [{ $divide: [{ $divide: ['$totalSprayMat', '$totalSprayed'] }, '$appRate'] }, 100] },
|
||
2]
|
||
}
|
||
}
|
||
}
|
||
]);
|
||
|
||
debug(`flowAccuracyPct backfill: matched=${result.matchedCount} modified=${result.modifiedCount}`);
|
||
debug('─'.repeat(60));
|
||
}
|
||
|
||
// ─── Main migration ───────────────────────────────────────────────────────────
|
||
async function migrate() {
|
||
// Build Application filter
|
||
const appFilter = { markedDelete: { $ne: true } };
|
||
|
||
// Tier by recency
|
||
if (cfg.tierDays) {
|
||
const cutoff = new Date(Date.now() - cfg.tierDays * 86400 * 1000);
|
||
appFilter._id = { $gte: objectIdFromDate(cutoff) };
|
||
debug(`Tier: apps from last ${cfg.tierDays} days (>= ${cutoff.toISOString()})`);
|
||
} else if (cfg.fromDate) {
|
||
appFilter._id = { $gte: objectIdFromDate(cfg.fromDate + 'T00:00:00Z') };
|
||
debug(`Tier: apps from ${cfg.fromDate} onward`);
|
||
}
|
||
|
||
// Unless --force, revisit apps with missing/null fields and explicit zero values that may
|
||
// have been written by earlier buggy imports.
|
||
if (!cfg.force) {
|
||
appFilter.$or = [
|
||
{ avgSpraySpeed: { $exists: false } },
|
||
{ avgSpraySpeed: null },
|
||
{ totalSprLength: { $exists: false } },
|
||
{ totalSprLength: null },
|
||
{ totalSprLength: 0 },
|
||
{ totalFlightLength: { $exists: false } },
|
||
{ totalFlightLength: null },
|
||
{ totalFlightLength: 0 },
|
||
{ avgXtError: { $exists: false } },
|
||
{ avgXtError: null },
|
||
{ avgHdop: { $exists: false } },
|
||
{ avgHdop: null },
|
||
];
|
||
}
|
||
|
||
debug('Config: %o', { ...cfg, envFile });
|
||
debug('App filter: %o', appFilter);
|
||
|
||
// Stats
|
||
const stats = {
|
||
examined: 0,
|
||
updated: 0,
|
||
skipped: 0, // no AppFile / no AppDetail data
|
||
errors: 0,
|
||
startedAt: Date.now(),
|
||
};
|
||
|
||
// Pending bulk-write buffers
|
||
let appBulk = []; // Application updateOne ops
|
||
let fileBulk = []; // AppFile updateOne ops
|
||
|
||
async function flushBulk() {
|
||
if (cfg.dryRun) {
|
||
debug(`[dry-run] Would write ${appBulk.length} Application + ${fileBulk.length} AppFile ops`);
|
||
appBulk = [];
|
||
fileBulk = [];
|
||
return;
|
||
}
|
||
|
||
const writes = [];
|
||
if (appBulk.length) writes.push(Application.bulkWrite(appBulk, { ordered: false }));
|
||
if (fileBulk.length) writes.push(AppFile.bulkWrite(fileBulk, { ordered: false }));
|
||
await Promise.all(writes);
|
||
|
||
appBulk = [];
|
||
fileBulk = [];
|
||
}
|
||
|
||
// Stream Applications — newest first
|
||
const appCursor = Application
|
||
.find(appFilter)
|
||
.select('_id avgSpraySpeed totalSprLength')
|
||
.sort({ _id: -1 })
|
||
.lean()
|
||
.cursor();
|
||
|
||
for await (const app of appCursor) {
|
||
stats.examined++;
|
||
|
||
try {
|
||
// Load AppFile docs for this application (not deleted)
|
||
const appFiles = await AppFile
|
||
.find({ appId: app._id, markedDelete: { $ne: true } })
|
||
.select('_id')
|
||
.lean();
|
||
|
||
if (!appFiles.length) {
|
||
stats.skipped++;
|
||
} else {
|
||
const { appSprLength, appFlightLength, avgSpraySpeed, avgXtError, avgHdop, fileOps } = await processApplication(app, appFiles);
|
||
|
||
// Queue Application update
|
||
const $set = {
|
||
totalSprLength: appSprLength,
|
||
totalFlightLength: appFlightLength,
|
||
};
|
||
if (avgSpraySpeed !== null) $set.avgSpraySpeed = avgSpraySpeed;
|
||
if (avgXtError !== null) $set.avgXtError = avgXtError;
|
||
if (avgHdop !== null) $set.avgHdop = avgHdop;
|
||
|
||
appBulk.push({
|
||
updateOne: {
|
||
filter: { _id: app._id },
|
||
update: { $set },
|
||
},
|
||
});
|
||
|
||
// Queue AppFile updates
|
||
for (const op of fileOps) {
|
||
fileBulk.push({
|
||
updateOne: {
|
||
filter: { _id: op.fileId },
|
||
update: { $set: { totalSprLength: op.sprLength, totalFlightLength: op.flightLength } },
|
||
},
|
||
});
|
||
}
|
||
|
||
stats.updated++;
|
||
}
|
||
} catch (err) {
|
||
debug(`Error processing app ${app._id}: ${err.message}`);
|
||
stats.errors++;
|
||
}
|
||
|
||
// Flush when batch is full
|
||
if (appBulk.length >= cfg.batchSize) {
|
||
await flushBulk();
|
||
}
|
||
|
||
// Progress log
|
||
if (stats.examined % PROGRESS_LOG_INTERVAL === 0) {
|
||
const elapsed = ((Date.now() - stats.startedAt) / 1000).toFixed(1);
|
||
const rate = (stats.examined / parseFloat(elapsed)).toFixed(1);
|
||
debug(
|
||
`Progress: examined=${stats.examined} updated=${stats.updated} ` +
|
||
`skipped=${stats.skipped} errors=${stats.errors} ` +
|
||
`elapsed=${elapsed}s rate=${rate} apps/s`
|
||
);
|
||
}
|
||
}
|
||
|
||
// Final flush
|
||
await flushBulk();
|
||
|
||
const elapsed = ((Date.now() - stats.startedAt) / 1000).toFixed(1);
|
||
debug('─'.repeat(60));
|
||
debug(`Migration complete.`);
|
||
debug(` Examined : ${stats.examined}`);
|
||
debug(` Updated : ${stats.updated}`);
|
||
debug(` Skipped : ${stats.skipped} (no files or no detail data)`);
|
||
debug(` Errors : ${stats.errors}`);
|
||
debug(` Duration : ${elapsed}s`);
|
||
if (!cfg.dryRun && stats.updated > 0) {
|
||
debug(` Fields set: Application.avgSpraySpeed, Application.totalSprLength, Application.totalFlightLength, Application.avgXtError, Application.avgHdop`);
|
||
debug(` AppFile.totalSprLength, AppFile.totalFlightLength`);
|
||
}
|
||
debug('─'.repeat(60));
|
||
|
||
// ── Second pass: backfill flowAccuracyPct (Application-level only, no AppDetail needed) ──
|
||
await backfillFlowAccuracy(cfg.dryRun, cfg.tierDays, cfg.fromDate);
|
||
|
||
return stats;
|
||
}
|
||
|
||
// ─── Entry point ──────────────────────────────────────────────────────────────
|
||
process
|
||
.on('uncaughtException', err => { debug('Uncaught:', err); process.exit(1); })
|
||
.on('unhandledRejection', err => { debug('Unhandled rejection:', err); process.exit(1); });
|
||
|
||
async function main() {
|
||
const dbConn = new DBConnection('migrate-app-aggregates');
|
||
try {
|
||
await dbConn.initialize({ setupExitHandlers: false });
|
||
debug('DB connected');
|
||
await migrate();
|
||
} catch (err) {
|
||
debug('Fatal error:', err);
|
||
} finally {
|
||
await dbConn.close();
|
||
process.exit(0);
|
||
}
|
||
}
|
||
|
||
main();
|