1998 lines
78 KiB
JavaScript
1998 lines
78 KiB
JavaScript
'use strict';
|
||
|
||
const
|
||
util = require('util'),
|
||
async = require('async'),
|
||
amqp = require('amqplib/callback_api'),
|
||
{ DBConnection } = require('../helpers/db/connect.js'),
|
||
path = require('path'),
|
||
debug = require('debug')('agm:worker'),
|
||
glob = require('glob'),
|
||
fs = require('fs-extra'),
|
||
csv = require('fast-csv'),
|
||
unzip = util.callbackify(require('extract-zip')),
|
||
moment = require('moment'),
|
||
Redis = require('ioredis'),
|
||
{ globOps, sprayItemsOps, dataFilesPattern, areaFilesExtendedPattern } = require('../helpers/glob_options'),
|
||
turf = require('@turf/turf'),
|
||
Job = require('../model/job'),
|
||
App = require('../model/application'),
|
||
AppFile = require('../model/application_file'),
|
||
AppDetail = require('../model/application_detail'),
|
||
User = require('../model/user'),
|
||
Pilot = require('../model/pilot'),
|
||
ObjectId = require('mongodb').ObjectId,
|
||
utils = require('../helpers/utils'),
|
||
FILE = require('../helpers/file_constants'),
|
||
fileHelper = require('../helpers/file_helper'),
|
||
fileAgNav = require('../helpers/file_storage'),
|
||
fileKML = require('../helpers/file_kml'),
|
||
fileShp = require('../helpers/file_shp'),
|
||
fileSatLog = require('../helpers/file_satlog'),
|
||
jobUtil = require('../helpers/job_util'),
|
||
appDateTime = require('../helpers/application_datetime'),
|
||
geoUtil = require('../helpers/geo_util'),
|
||
subUtil = require('../helpers/subscription_util'),
|
||
{ DataUtil, WorkRecord } = require('../helpers/work_record'),
|
||
{ JobUpdateOp, JobStatus } = require('../helpers/job_constants'),
|
||
CVCST = require('../helpers/convert_constants'),
|
||
{ Errors, RecTypes, UserTypes, AppStatus, AppProStatus, Fields, DEL_APP_IDS, DEFAULT_LANG, RateUnits } = require('../helpers/constants'),
|
||
{ AppError, AppMembershipError, AppAuthError } = require('../helpers/app_error.js'),
|
||
{ SubFields } = require('../model/subscription.js'),
|
||
TaskTracker = require('../model/task_tracker'),
|
||
{ TaskTrackerStatus, ErrorCategory } = require('../model/task_tracker'),
|
||
{ generateTaskId, generateExecutionId } = require('../services/task_id_generator'),
|
||
env = require('../helpers/env'),
|
||
errorCommon = require('error-handler').common;
|
||
|
||
require('../model/crop');
|
||
|
||
// Initialize database connection
|
||
const workerDB = new DBConnection('Job Worker');
|
||
|
||
process.setMaxListeners(0);
|
||
// Avoid error-handler's key-file-storage corruption pattern; use atomic fatal reporter instead.
|
||
const { registerFatalHandlers } = require('../helpers/process_fatal_handlers');
|
||
registerFatalHandlers(process, {
|
||
env,
|
||
debug,
|
||
kindPrefix: 'job_worker',
|
||
reportFilePath: path.join(__dirname, 'job_worker.rlog'),
|
||
});
|
||
|
||
const WkrStatus = Object.freeze({
|
||
BROKE: 'broke',
|
||
CANCELLED: 'cancelled',
|
||
SKIPPED: 'skipped',
|
||
DONE: 'done'
|
||
});
|
||
|
||
const redis = new Redis({ password: env.REDIS_PWD, showFriendlyErrorStack: true });
|
||
|
||
const uploadPath = env.UPLOAD_DIR;
|
||
const unzipPath = env.UNZIP_DIR;
|
||
const jobQueue = env.PRODUCTION ? env.QUEUE_NAME_JOBS : 'dev_jobs';
|
||
|
||
let amqpConn = null;
|
||
let mqClosed = false;
|
||
|
||
let jobMainFile, jobUnZipPath, isKmlOrKmz = false;
|
||
|
||
// Initialize the database connection
|
||
workerDB.initialize({
|
||
setupExitHandlers: false,
|
||
onReady: () => {
|
||
// Start RabbitMQ connection after MongoDB is connected
|
||
if (!amqpConn) {
|
||
start();
|
||
}
|
||
}
|
||
});
|
||
|
||
// Rabbitmq connection. If the connection is closed or fails to be established at all, we will reconnect
|
||
function start() {
|
||
const conOps = {
|
||
protocol: 'amqp',
|
||
hostname: env.QUEUE_HOST || 'localhost',
|
||
port: env.QUEUE_PORT || 5672,
|
||
username: env.QUEUE_USR || 'agmuser',
|
||
password: env.QUEUE_PWD,
|
||
vhost: env.QUEUE_VHOST || '/',
|
||
heartbeat: env.QUEUE_HEARTBEAT || 0, // Will depends on server settings or whichever lower value, 0 will use server default 60s
|
||
frameMax: 0
|
||
};
|
||
|
||
amqp.connect(conOps, function (err, conn) {
|
||
|
||
if (err) {
|
||
return process.exit();
|
||
}
|
||
conn.on("error", function (err) {
|
||
debug(err);
|
||
return process.exit();
|
||
});
|
||
conn.on("close", function () {
|
||
mqClosed = true;
|
||
return process.exit();
|
||
});
|
||
|
||
amqpConn = conn;
|
||
process.once('SIGINT', () => {
|
||
closeOnErr('force_close_SIGINT');
|
||
});
|
||
whenConnected();
|
||
});
|
||
}
|
||
|
||
function whenConnected() {
|
||
debug("[AMQP] Job Worker connected");
|
||
mqClosed = false;
|
||
startWorker();
|
||
}
|
||
|
||
// A worker that acks messages only if processed succesfully
|
||
function startWorker() {
|
||
debug("Worker started !");
|
||
if (!amqpConn) {
|
||
debug("Ampq connection is null. Please check !");
|
||
return;
|
||
}
|
||
amqpConn.createChannel(function (err, ch) {
|
||
if (closeOnErr(err)) return;
|
||
|
||
ch.on("error", function (err) {
|
||
debug("[AMQP] channel error", err.message);
|
||
});
|
||
|
||
ch.on("close", function () {
|
||
debug("[AMQP] channel closed");
|
||
});
|
||
|
||
ch.prefetch(1); // Tell RabbitMQ not to give more than one message to a worker at a time
|
||
|
||
ch.assertQueue(jobQueue, { durable: true }, (err) => {
|
||
if (closeOnErr(err)) return;
|
||
ch.consume(jobQueue, processMsg, { noAck: false });
|
||
debug("Worker has started consuming messages...");
|
||
});
|
||
|
||
function processMsg(msg) {
|
||
if (!msg) return; // Ignore empty message
|
||
|
||
let impMsg;
|
||
try {
|
||
impMsg = JSON.parse(msg.content);
|
||
} catch (err) {
|
||
debug(err);
|
||
}
|
||
if (!impMsg) return;
|
||
|
||
// TaskTracker disabled
|
||
const msgJobId = impMsg.jobId;
|
||
|
||
work(impMsg, (msg.fields && msg.fields.redelivered), async (err, result) => {
|
||
if (mqClosed) {
|
||
debug("MQ conn already closed -- Skipping...");
|
||
return;
|
||
}
|
||
let confirmOk = true;
|
||
try {
|
||
if (!err) {
|
||
// Acknowledgement must be sent on the same channel the delivery it is for was received on.
|
||
if (result.appError || result.wasCancelled)
|
||
ch.reject(msg, false);
|
||
else
|
||
ch.ack(msg);
|
||
}
|
||
}
|
||
catch (rErr) {
|
||
confirmOk = false;
|
||
debug('[Work] channel ack/reject error', rErr);
|
||
closeOnErr(rErr);
|
||
}
|
||
finally {
|
||
const logDone = () => { debug("[x] Done for: ", msg.content.toString()); };
|
||
if (result.wasCancelled) {
|
||
const cleanJobFn = async (impMsg) => {
|
||
const app = await App.findOne({ _id: ObjectId(impMsg.appId) });
|
||
if (!app) return;
|
||
// The case of import file to create a job => remove that job fully
|
||
if (app.byImport && undefined === msgJobId && app.jobId) {
|
||
const job = await Job.findById(app.jobId);
|
||
if (job) await job.removeFull();
|
||
} else {
|
||
// Only need to remove the application
|
||
await jobUtil.deleteAppById(result.appId);
|
||
}
|
||
await redis.srem(DEL_APP_IDS, result.appId);
|
||
};
|
||
|
||
cleanJobFn(impMsg).catch(err => {
|
||
if (err) debug(err);
|
||
}).finally(() => {
|
||
debug(`Task with appId:${impMsg.appId} was cancelled!`);
|
||
});
|
||
} else if (result.wasSkipped) {
|
||
// Do nothing yet. Normally for marked as deleted applications. They going to be cleaned by a Cleanup Worker
|
||
logDone();
|
||
} else {
|
||
_cleanup(confirmOk, result.appError);
|
||
logDone();
|
||
}
|
||
}
|
||
});
|
||
}
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Clean up files or folders depends on the task processing conditions
|
||
* @param {boolean} taskAcked Whether the task was acknowledged (done either w/ success or error, and ready to take the next task).
|
||
* @param {boolean} hasAppError Where there was any application cope errors happened (crashes, unhandled promise exception occured)
|
||
*/
|
||
function _cleanup(taskAcked = true, hasAppError = false) {
|
||
let removeList = [];
|
||
if (taskAcked && hasAppError)
|
||
removeList = isKmlOrKmz ? [jobMainFile, path.join(jobUnZipPath, jobMainFile)] : [jobMainFile, jobUnZipPath];
|
||
else
|
||
removeList = isKmlOrKmz ? [path.join(jobUnZipPath, jobMainFile)] : [jobUnZipPath];
|
||
|
||
if (!utils.isEmptyArray(removeList))
|
||
fileHelper.removeFiles(removeList, err => { if (err) debug(err) });
|
||
}
|
||
|
||
/**
|
||
* Compute the difference (in hours) between the HH:MM encoded in the AgNav filename and
|
||
* the HH:MM of the GPS record's UTC time-of-day.
|
||
*
|
||
* AgNav filename encoding (from datasaveworker.cpp):
|
||
* - Date (YMMDD): device's LOCAL calendar date (QDate::currentDate())
|
||
* - Time (HHmm): GPS UTC time-of-day when the file was created
|
||
* (AppInfo::gpsTime ← gpsd._UTC from NMEA $GPGGA sentence)
|
||
*
|
||
* Because both the filename time and the GPS record time are UTC, this difference is
|
||
* near-zero for the first record (file is created at flight start; first record is from
|
||
* flight start). The result is used as a correction offset in computeStartEndDate().
|
||
*
|
||
* If GPS was not yet locked when the file was created (unusual edge case), the device
|
||
* falls back to QTime::currentTime() (local clock), and this function would then measure
|
||
* the device's UTC offset — but this is not the normal operational path.
|
||
*
|
||
* @param {number} gpsTimeSeconds GPS seconds since midnight UTC for the first data record
|
||
* @param {string} fileNameAgNavDT AgNav filename datetime code (format: YMMDDHHmm, 9+ digits)
|
||
* @returns {number} Difference in fractional hours between filename HH:MM and record HH:MM.
|
||
* Near 0 under normal operation (both are GPS UTC time).
|
||
*/
|
||
function computeGPSTimeOffsetLocalTime(gpsTimeSeconds, fileNameAgNavDT) {
|
||
const gpsDuration = moment.duration(new Date(1000 * gpsTimeSeconds).toISOString().substring(11, 16));
|
||
const localDuration = moment.duration(utils.dateTimePartsFromAgNav(fileNameAgNavDT).time);
|
||
const diff = localDuration.subtract(gpsDuration);
|
||
|
||
return Number(diff.asHours().toFixed(2));
|
||
}
|
||
|
||
/**
|
||
* Derive Application start/end datetimes from AgNav filename codes and GPS record times.
|
||
*
|
||
* AgNav filename encoding (from datasaveworker.cpp source):
|
||
* - Date (YMMDD): device's LOCAL calendar date (QDate::currentDate())
|
||
* - Time (HHmm): GPS UTC time-of-day (AppInfo::gpsTime ← gpsd._UTC)
|
||
*
|
||
* Result format stored in Application.startDateTime / endDateTime:
|
||
* "YYYYMMDDTHHmmss" where YYYYMMDD is the LOCAL date from the filename and
|
||
* HHmmss is the GPS UTC time from the data record. This is a HYBRID value:
|
||
* local calendar date + UTC time-of-day. It is neither pure UTC nor pure local time.
|
||
*
|
||
* Algorithm:
|
||
* 1. offsetHrs = filename_UTC_HH:MM − record_UTC_HH:MM
|
||
* Under normal operation (GPS locked before file creation) this is ≈ 0.
|
||
* 2. Build moment.utc(localDate + GPS_UTC_HHmmss) — moment.utc used to suppress
|
||
* server system timezone from interfering in step 3.
|
||
* 3. add(offsetHrs) — applies the near-zero correction; effectively a no-op in practice.
|
||
* 4. If endDate < startDate: add 1 day (handles GPS-UTC midnight crossover mid-flight).
|
||
*
|
||
* Why this hybrid format:
|
||
* - The legacy AgNav filename encoding anchors to the pilot's LOCAL CALENDAR DATE, which is what matters for correctly grouping records by day and for pilot-facing date displays.
|
||
* - Consumers (toRecordTimeUtc in api_pub.js / api_export.js) call startOf('day') on
|
||
* appStartDateTime to extract the pilot's LOCAL calendar date, then add GPS UTC seconds
|
||
* to reconstruct per-record UTC timestamps. The local date part is what matters for
|
||
* anchoring to the correct calendar day.
|
||
*
|
||
* Known edge case: for pilots in large east-of-UTC timezones whose UTC time crosses midnight
|
||
* BEFORE their local time does, the local date (YYYYMMDD part) will be one day ahead of
|
||
* the UTC date. toRecordTimeUtc() anchors to the local date, so per-record UTC timestamps
|
||
* will be off by one day during those early-UTC-morning hours.
|
||
*
|
||
* @param {number} startGPSTimeSecs GPS seconds since midnight UTC for the first data record
|
||
* @param {number} endGPSTimeSecs GPS seconds since midnight UTC for the last data record
|
||
* @param {string} firstFileAgNavDT AgNav filename datetime code of the first data file (YMMDDHHmm)
|
||
* @param {string} endFileAgNavDT AgNav filename datetime code of the last data file (YMMDDHHmm)
|
||
* @returns {{ start: moment.Moment, end: moment.Moment }}
|
||
* Moment objects (in UTC mode) representing: local calendar date + GPS UTC time-of-day.
|
||
* Use .format('YYYYMMDDTHHmmss') to store in Application.startDateTime / endDateTime.
|
||
*/
|
||
function computeStartEndDate(startGPSTimeSecs, endGPSTimeSecs, firstFileAgNavDT, endFileAgNavDT) {
|
||
const offsetHrs = computeGPSTimeOffsetLocalTime(startGPSTimeSecs, firstFileAgNavDT);
|
||
const sdate = utils.dateTimePartsFromAgNav(firstFileAgNavDT, 2);
|
||
const startDate = utils.toUTCDateTime(startGPSTimeSecs, sdate.date);
|
||
const edate = utils.dateTimePartsFromAgNav(endFileAgNavDT, 2);
|
||
const endDate = utils.toUTCDateTime(endGPSTimeSecs, edate.date);
|
||
|
||
if (offsetHrs) {
|
||
startDate.add(offsetHrs, 'hours');
|
||
endDate.add(offsetHrs, 'hours');
|
||
}
|
||
if (endDate < startDate) endDate.add(1, 'days');
|
||
|
||
return { start: startDate, end: endDate };
|
||
}
|
||
|
||
function removeAppData(appId, cb) {
|
||
AppFile.find({ appId: appId }, (err, appFiles) => {
|
||
if (err) {
|
||
return cb(err);
|
||
}
|
||
async.eachSeries(appFiles, (appFile, cb) => {
|
||
appFile.remove(cb);
|
||
}, cb);
|
||
});
|
||
}
|
||
|
||
function wasCancelled(appId, cb) {
|
||
let ret;
|
||
canAppProceed(appId)
|
||
.then(res => ret = res)
|
||
.catch(err => ret = err)
|
||
.finally(() => cb && cb(ret));
|
||
}
|
||
|
||
async function canAppProceed(appId) {
|
||
const wasCancelled = await redis.sismember(DEL_APP_IDS, appId);
|
||
if (wasCancelled)
|
||
return WkrStatus.CANCELLED;
|
||
|
||
const app = await App.findById(appId, { lean: true });
|
||
if (!app || app[Fields.MARKED_DELETE] === true)
|
||
return WkrStatus.SKIPPED;
|
||
else if (app && app.status === AppStatus.WAS_CANCELLED)
|
||
return WkrStatus.CANCELLED;
|
||
|
||
return null;
|
||
}
|
||
|
||
/**
|
||
* Import data or job items of the job into db.
|
||
*
|
||
* STEPS:
|
||
* 1. Unzip the arvchive into unzip folder
|
||
* 2. Create a new Application with status 1
|
||
* 3. Read area files and override the job's items (from NO1, PRJ)
|
||
* 4. Read data files and import to App-Detail collections
|
||
* 5. When finished importing, update status as done (3 - done, 0 - Error),
|
||
*
|
||
* Notes: Now import job items(spray, xcl, waypoint) found within .zip, (kml/kmz): buffer, placemark as well
|
||
* @param {any} impMsg the queued import message taken from the queue
|
||
* @param {boolean} redelivered whether the message was redelivered (failed to processed before)
|
||
* @param {any} cb callback method
|
||
*/
|
||
function work(impMsg, redelivered, cb) {
|
||
debug("Got msg ", impMsg);
|
||
|
||
let areaTypeAndFiles = [], areaFiles = [];
|
||
let appl = null, appData, hasData = false;
|
||
let _job, proStatus = AppProStatus.ERROR, hasAppError = false;
|
||
|
||
jobMainFile = path.join(uploadPath, impMsg.file.name);
|
||
isKmlOrKmz = /.*(.kml|.kmz)+$/i.test(jobMainFile);
|
||
|
||
jobUnZipPath = isKmlOrKmz ? uploadPath : path.join(unzipPath, impMsg.file.name.replace(path.extname(impMsg.file.name), ''));
|
||
jobUnZipPath = path.join(jobUnZipPath, '/');
|
||
|
||
async.series([
|
||
function (callback) {
|
||
return wasCancelled(ObjectId(impMsg.appId), callback);
|
||
},
|
||
function (callback) {
|
||
App.findById(ObjectId(impMsg.appId), (err, data) => {
|
||
if (err)
|
||
return callback(err);
|
||
appl = data;
|
||
if (!appl)
|
||
return callback(AppError.create(WkrStatus.BROKE));
|
||
else if (appl.status === AppStatus.DONE)
|
||
return callback(AppError.create(WkrStatus.DONE));
|
||
else if (appl[Fields.MARKED_DELETE] === true)
|
||
return callback(AppError.create(WkrStatus.SKIPPED));
|
||
|
||
callback();
|
||
});
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appl._id, callback);
|
||
},
|
||
function (callback) {
|
||
if (impMsg.jobId !== undefined) {
|
||
Job.findById(impMsg.jobId)
|
||
.populate({ path: 'crop', select: '_id name color', skipInvalidIds: true })
|
||
.then(job => {
|
||
if (job) {
|
||
_job = job;
|
||
return callback();
|
||
} else {
|
||
AppError.throw(Errors.JOB_NOT_FOUND);
|
||
}
|
||
})
|
||
.catch(err => callback(err));
|
||
}
|
||
else callback();
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appl._id, callback);
|
||
},
|
||
function (callback) {
|
||
// Handle redelivered (in case of crashed during processing the task) message to avoid duplicated data
|
||
if (redelivered) {
|
||
if (!utils.isBlank(appl.errorMsg) && errorCommon.isAppError(appl.errorMsg)) {
|
||
return callback(AppError.create(WkrStatus.SKIPPED));
|
||
} else if (appl.byImport && !appl.jobId) {
|
||
return appl.status === AppStatus.CREATED ? callback() : callback(AppError.create());
|
||
} else {
|
||
if (impMsg.updateOp !== JobUpdateOp.XCLS)
|
||
removeAppData(appl._id, callback); // remove previous fragmented imported data
|
||
else
|
||
callback(); // re-process the task
|
||
}
|
||
} else callback();
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appl._id, callback);
|
||
},
|
||
function (callback) {
|
||
if (appl.status !== AppStatus.IN_PROGRESS) {
|
||
appl.status = AppStatus.IN_PROGRESS;
|
||
return appl.save(callback);
|
||
}
|
||
else callback();
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appl._id, callback);
|
||
},
|
||
function (callback) {
|
||
if (isKmlOrKmz) {
|
||
areaFiles.push(jobMainFile);
|
||
return callback();
|
||
}
|
||
else {
|
||
async.series([
|
||
function (callback) {
|
||
unzip(jobMainFile, { dir: jobUnZipPath }, (err) => {
|
||
if (err) {
|
||
debug(err);
|
||
return callback(AppError.create(Errors.CORRUPTED_ZIP));
|
||
}
|
||
callback();
|
||
});
|
||
},
|
||
function (callback) {
|
||
if (impMsg.updateOp === JobUpdateOp.DATA_ONLY)
|
||
return callback();
|
||
// Find the files which might contain job items
|
||
glob(path.join(jobUnZipPath, areaFilesExtendedPattern), sprayItemsOps, (err, files) => {
|
||
if (err)
|
||
return callback(err);
|
||
|
||
areaFiles = files;
|
||
if (areaFiles.length === 0)
|
||
return callback(impMsg.updateOp === JobUpdateOp.OVERWRITE ? null : AppError.create(Errors.ITEMS_NOT_FOUND));
|
||
callback();
|
||
});
|
||
},
|
||
], callback);
|
||
}
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appl._id, callback);
|
||
},
|
||
function (callback) {
|
||
if (impMsg.updateOp !== JobUpdateOp.DATA_ONLY)
|
||
areaTypeAndFiles = fileHelper.areasFromList(jobUnZipPath, areaFiles);
|
||
callback();
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appl._id, callback);
|
||
},
|
||
function (callback) {
|
||
if (impMsg.updateOp > JobUpdateOp.DATA_ONLY) {
|
||
if (areaTypeAndFiles.length) {
|
||
updateJobItems(_job || {}, impMsg, jobUnZipPath, areaTypeAndFiles, impMsg.uid, (err, data) => {
|
||
if (err)
|
||
return callback(err);
|
||
|
||
_job = data.job; // Why _job passed by ref does not work any longer ????
|
||
|
||
if (data.updated) {
|
||
proStatus = AppProStatus.ERROR + 10;
|
||
if (!appl.jobId) appl.jobId = impMsg.jobId;
|
||
if (data.dup) {
|
||
if (appl.warnMsg) // Update numbers of duplicated areas
|
||
appl.warnMsg.set('dup', data.dup);
|
||
else appl.warnMsg = { dup: data.dup };
|
||
}
|
||
}
|
||
callback();
|
||
});
|
||
}
|
||
else
|
||
callback(!_job ? AppError.create(Errors.ITEMS_NOT_FOUND) : null);
|
||
}
|
||
else callback();
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appl._id, callback);
|
||
},
|
||
function (callback) {
|
||
if (isKmlOrKmz || impMsg.updateOp === JobUpdateOp.XCLS)
|
||
return callback(); // Not import data for KML/KMZ, XCL files
|
||
|
||
importData(jobUnZipPath, appl._id, _job.toObject(), (err, data) => {
|
||
if (err)
|
||
return callback(err);
|
||
|
||
appData = data;
|
||
hasData = !!(data);
|
||
callback();
|
||
});
|
||
}
|
||
], function (err) {
|
||
proStatus += hasData ? AppProStatus.WITH_DATA : AppProStatus.NO_DATA;
|
||
|
||
if (err) {
|
||
if ((hasAppError = (err instanceof AppError))) {
|
||
if (WkrStatus.BROKE === err.message)
|
||
return cb(null, { appError: true });
|
||
else if (WkrStatus.DONE === err.message)
|
||
return cb(null, { appError: false });
|
||
else if (WkrStatus.SKIPPED === err.message)
|
||
return cb(null, { wasSkipped: true });
|
||
if (WkrStatus.CANCELLED === err.message) {
|
||
return cb(null, { appError: hasAppError, wasCancelled: true, appId: impMsg.appId });
|
||
}
|
||
}
|
||
|
||
if (appl !== null) {
|
||
appl.errorMsg = hasAppError ? err.message : utils.trimStrTo(err.toString(), 100);
|
||
appl.status = AppStatus.ERROR;
|
||
}
|
||
else {
|
||
return cb(null, { appError: hasAppError });
|
||
}
|
||
}
|
||
else {
|
||
if (appl !== null) {
|
||
appl.status = AppStatus.DONE;
|
||
if (hasData) {
|
||
if (utils.isNumber(appData.appRate)) appl.appRate = utils.roundTo(appData.appRate);
|
||
if (utils.isNumber(appData.totalSprayed)) appl.totalSprayed = appData.totalSprayed * 1E-4; // update and convert from square meters to ha
|
||
if (utils.isNumber(appData.totalSprLength)) appl.totalSprLength = appData.totalSprLength; // meters
|
||
if (utils.isNumber(appData.totalFlightLength)) appl.totalFlightLength = appData.totalFlightLength; // meters
|
||
if (utils.isNumber(appData.totalTurnTime)) appl.totalTurnTime = appData.totalTurnTime;
|
||
if (utils.isNumber(appData.totalSprayTime)) appl.totalSprayTime = appData.totalSprayTime;
|
||
if (utils.isNumber(appData.totalFlightTime)) appl.totalFlightTime = appData.totalFlightTime;
|
||
if (utils.isNumber(appData.totalSprayMat)) {
|
||
appl.totalSprayMat = appData.totalSprayMat;
|
||
appl.totalSprayMatUnit = appData.totalSprayMatUnit;
|
||
}
|
||
if (utils.isNumber(appData.avgSpraySpeed)) appl.avgSpraySpeed = appData.avgSpraySpeed; // m/s, average ground speed during spray-on periods
|
||
if (utils.isNumber(appData.avgHdop)) appl.avgHdop = appData.avgHdop;
|
||
appl.avgXtError = appData.avgXtError ?? null;
|
||
// Flow accuracy: (actual L/ha or Kg/ha) / prescribed appRate × 100
|
||
if (utils.isNumber(appl.totalSprayed) && appl.totalSprayed > 0 &&
|
||
utils.isNumber(appl.totalSprayMat) && appl.totalSprayMat > 0 &&
|
||
utils.isNumber(appl.appRate) && appl.appRate > 0) {
|
||
const _actualRate = appl.totalSprayMat / appl.totalSprayed;
|
||
appl.flowAccuracyPct = Math.round((_actualRate / appl.appRate) * 10000) / 100;
|
||
}
|
||
if (utils.isNumber(appData.utcOffset)) appl.utcOffset = appData.utcOffset;
|
||
if (appData.startDateTimeUTC) appl.startDateTimeUTC = appData.startDateTimeUTC;
|
||
if (appData.endDateTimeUTC) appl.endDateTimeUTC = appData.endDateTimeUTC;
|
||
appl.startDateTime = appData.startDateTime.format('YYYYMMDDTHHmmss');
|
||
appl.endDateTime = appData.endDateTime.format('YYYYMMDDTHHmmss');
|
||
}
|
||
}
|
||
}
|
||
if (appl !== null) {
|
||
appl.proStatus = proStatus;
|
||
appl.updateDate = Date.now();
|
||
appl.save()
|
||
.then(() => {
|
||
if (hasData && impMsg.jobId)
|
||
return Job.updateOne({ _id: impMsg.jobId }, { $set: { status: JobStatus.SPRAYED } }); // Update Job Status as Downloaded
|
||
})
|
||
.then(() => cb(null, { appError: hasAppError }))
|
||
.catch(err => {
|
||
debug(err);
|
||
cb(err);
|
||
})
|
||
}
|
||
});
|
||
}
|
||
|
||
function updateJobItems(job, impMsg, filePath, areaTypeAndFiles, uid, cb) {
|
||
let jobItems = [], anyUpdated = false;
|
||
let updateMeta = false; // Determine whether to update new job meta from the first found file
|
||
let userWSettings, meta, dup = 0;
|
||
let usageLimits = { maxAcres: 0, totalSprAcres: 0 };
|
||
let sprZoneColor = 'blue';
|
||
|
||
const createJob = utils.isEmptyObj(job);
|
||
|
||
async.series([
|
||
function (callback) {
|
||
return wasCancelled(impMsg.appId, callback);
|
||
},
|
||
function (callback) {
|
||
if (!uid) return callback();
|
||
|
||
return User.aggregate([
|
||
{ $match: { _id: ObjectId(uid) } },
|
||
{ $project: { kind: 1, parent: 1, membership: 1, lang: { $ifNull: ["$lang", DEFAULT_LANG] } } },
|
||
{
|
||
$lookup:
|
||
{
|
||
from: "settings",
|
||
localField: "_id",
|
||
foreignField: "userId",
|
||
as: "cfg"
|
||
}
|
||
},
|
||
{ $unwind: { path: "$cfg", "preserveNullAndEmptyArrays": true } },
|
||
])
|
||
.exec((err, result) => {
|
||
if (err)
|
||
return callback(err);
|
||
|
||
if (result && result.length) {
|
||
userWSettings = result[0];
|
||
if (userWSettings.cfg && userWSettings.cfg.colors && userWSettings.cfg.colors["sprayZone"])
|
||
sprZoneColor = userWSettings.cfg.colors.sprayZone;
|
||
}
|
||
else return callback(AppError.create(Errors.USER_NOT_FOUND));
|
||
|
||
callback();
|
||
})
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(impMsg.appId, callback);
|
||
},
|
||
function (callback) {
|
||
// Get membership subscription info (only if it is about job modification) and check for the 1st round
|
||
if (impMsg.updateOp === JobUpdateOp.APPEND || impMsg.updateOp === JobUpdateOp.OVERWRITE) {
|
||
getUsageLimits(userWSettings)
|
||
.then(theUsageLimits => {
|
||
theUsageLimits && (usageLimits = theUsageLimits);
|
||
callback((usageLimits.maxAcres && (usageLimits.totalSprAcres) >= usageLimits.maxAcres) ? Errors.REACHED_AREA_LIMIT : null);
|
||
})
|
||
.catch(err => {
|
||
callback(err);
|
||
})
|
||
} else callback();
|
||
},
|
||
function (callback) {
|
||
if (createJob) { // create new job for the case of not existed
|
||
updateMeta = true;
|
||
let isUs = (userWSettings.lang === 'pt') ? false : true;
|
||
job = new Job({
|
||
client: ObjectId(impMsg.clientId),
|
||
name: utils.normalizeName(impMsg.file.originalName.replace(path.extname(impMsg.file.originalName), '')),
|
||
measureUnit: isUs,
|
||
swathWidth: isUs ? 30 : 10,
|
||
appRate: 10,
|
||
appRateUnit: isUs ? 1 : 3,
|
||
});
|
||
|
||
job.byPuid = userWSettings.kind === UserTypes.APP ? userWSettings._id : userWSettings.parent;
|
||
|
||
new Promise(resolve => resolve({ isPilot: userWSettings.kind === UserTypes.OPERATOR }))
|
||
.then(value => {
|
||
if (value.isPilot)
|
||
return Pilot.findById(ObjectId(uid), '-password', { lean: true });
|
||
else
|
||
return null;
|
||
})
|
||
.then(pilot => {
|
||
if (pilot)
|
||
job.operator = pilot._id;
|
||
// Try reading the first non-empty Qfile for aditional meta-data
|
||
return utils.execAsync(`find -P "${filePath}" -name "q*.t*" -type f -not -empty | head -n 1`);
|
||
})
|
||
.then(qfile => {
|
||
if (qfile) return fileAgNav.readQFile_Async(qfile.replace(/\r?\n|\r/g, ''));
|
||
else return null;
|
||
})
|
||
.then(data => {
|
||
if (data) {
|
||
job.flightNumber = data.flightNumber;
|
||
job.remark = data.remark;
|
||
// job.crop = data.crob; // May be try to lookup w entities first with case-insensitive ?
|
||
job.appRate = utils.toNumber(data.appRate, job.appRate);
|
||
job.appRateUnit = utils.rateStringToCode(data.appRateUnitStr);
|
||
job.measureUnit = (job.appRateUnit > 2) ? false : true;
|
||
}
|
||
callback();
|
||
})
|
||
.catch(err => callback(err));
|
||
}
|
||
else callback();
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(impMsg.appId, callback);
|
||
},
|
||
function (outterCB) {
|
||
const ops = { sprZoneColor: sprZoneColor, isUS: job.measureUnit, appRate: job.appRate, crop: job.crop };
|
||
|
||
async.eachSeries(areaTypeAndFiles, (areaFile, esCB) => {
|
||
// debug("Reading file: ", areaFile);
|
||
async.series([
|
||
function (callback) {
|
||
return wasCancelled(impMsg.appId, callback);
|
||
},
|
||
function (callback) {
|
||
switch (areaFile.type) {
|
||
case FILE.FILE_NO1:
|
||
case FILE.FILE_PRJ:
|
||
fileAgNav.readAGN_PRJ(filePath, areaFile, ops, (err, data) => {
|
||
if (err) {
|
||
debug(err.stack);
|
||
if (areaTypeAndFiles.length === 1)
|
||
return callback(AppError.create(Errors.INVALID_JOB_FILE));
|
||
//else Just ignore to move to the next file but log
|
||
}
|
||
if (data) {
|
||
jobItems.push({ type: FILE.FILE_NO1, items: data });
|
||
if (updateMeta && !meta && data.meta) {
|
||
meta = data.meta;
|
||
}
|
||
}
|
||
callback();
|
||
});
|
||
break;
|
||
case FILE.FILE_KMZ:
|
||
case FILE.FILE_KML:
|
||
fileKML.readKmlKmzToGeoItems(filePath, areaFile, ops, (err, data) => {
|
||
if (err) {
|
||
debug(err);
|
||
if (areaTypeAndFiles.length === 1)
|
||
return callback(AppError.create(Errors.INVALID_JOB_FILE));
|
||
//else Just ignore to move to the next file but log
|
||
}
|
||
if (data)
|
||
jobItems.push({ type: FILE.FILE_KML, items: data });
|
||
callback();
|
||
});
|
||
break;
|
||
case FILE.FILE_SHP:
|
||
fileShp.readSHPToGeoItems(filePath, areaFile, ops)
|
||
.then((data) => {
|
||
if (data)
|
||
jobItems.push({ type: FILE.FILE_SHP, items: data });
|
||
callback();
|
||
})
|
||
.catch(err => {
|
||
if (err) {
|
||
if (err instanceof AppError)
|
||
return callback(err);
|
||
if (areaTypeAndFiles.length === 1)
|
||
callback(AppError.create(Errors.INVALID_JOB_FILE));
|
||
}
|
||
else
|
||
callback(); // Ignore error in the case the zip file contains multiple area files
|
||
})
|
||
break;
|
||
case FILE.FILE_SATLOG_JOB:
|
||
fileSatLog.readSatLogJob(filePath, areaFile, ops, (err, data) => {
|
||
if (err) {
|
||
debug(err);
|
||
if (areaTypeAndFiles.length === 1)
|
||
return callback(AppError.create(Errors.INVALID_JOB_FILE));
|
||
//else Just ignore to move to the next file but log
|
||
}
|
||
if (data) {
|
||
jobItems.push({ type: FILE.FILE_SATLOG_JOB, items: data });
|
||
if (updateMeta && !meta && data.meta) {
|
||
meta = data.meta;
|
||
}
|
||
}
|
||
callback();
|
||
});
|
||
break;
|
||
default:
|
||
return callback();
|
||
}
|
||
},
|
||
function (callback) {
|
||
if (updateMeta && !meta && areaFile.agn && (areaFile.type === FILE.FILE_SHP || areaFile.type === FILE.FILE_KML || areaFile.type === FILE.FILE_KMZ)) {
|
||
fileAgNav.readAGN_PRJ(filePath, { area: areaFile.agn, type: FILE.FILE_AGN }, ops, (err, data) => {
|
||
if (data && data.meta) {
|
||
meta = data.meta;
|
||
}
|
||
callback();
|
||
});
|
||
}
|
||
else callback();
|
||
}
|
||
], (err) => {
|
||
esCB(err);
|
||
});
|
||
}, (err) => {
|
||
outterCB(err);
|
||
});
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(impMsg.appId, callback);
|
||
},
|
||
function (callback) {
|
||
if (!jobItems.length)
|
||
return callback(impMsg.updateOp === JobUpdateOp.OVERWRITE ? null : AppError.create(Errors.ITEMS_NOT_FOUND));
|
||
|
||
try {
|
||
|
||
if (updateMeta && meta) { // Update the new job with read measureUnit and swath
|
||
job.measureUnit = (meta.measureUnit == 1);
|
||
if (meta.swath > 0) job.swathWidth = meta.swath;
|
||
}
|
||
|
||
// Collect all items from area files here and do the Job update
|
||
let sprayAreas = [], xclAreas = [], waypoints = [], bufs = [], places = [];
|
||
for (let i = 0; i < jobItems.length; i++) {
|
||
const item = jobItems[i];
|
||
if (item.items) {
|
||
if (!utils.isEmptyArray(item.items.sprayAreas))
|
||
sprayAreas = sprayAreas.concat(item.items.sprayAreas);
|
||
|
||
if (!utils.isEmptyArray(item.items.xclAreas))
|
||
xclAreas = xclAreas.concat(item.items.xclAreas);
|
||
|
||
if (!utils.isEmptyArray(item.items.waypoints))
|
||
waypoints = waypoints.concat(item.items.waypoints);
|
||
|
||
if (item.type === FILE.FILE_KML) {
|
||
if (!utils.isEmptyArray(item.items.bufs))
|
||
bufs = bufs.concat(item.items.bufs);
|
||
|
||
if (!utils.isEmptyArray(item.items.places))
|
||
places = places.concat(item.items.places);
|
||
}
|
||
}
|
||
}
|
||
|
||
let oldAreaIds = [], checkRes;
|
||
//Check for the import item modification options to decide how to update the job's items
|
||
sprayAreas = jobUtil.cleanAreas(sprayAreas);
|
||
xclAreas = jobUtil.cleanAreas(xclAreas);
|
||
const shouldCheckLimits = !!(usageLimits.maxAcres);
|
||
|
||
if (impMsg.updateOp === JobUpdateOp.APPEND) {
|
||
const allAreas = utils.appendArray(job.sprayAreas, job.excludedAreas);
|
||
if (!utils.isEmptyArray(sprayAreas)) {
|
||
checkRes = jobUtil.checkDupAreas(allAreas, sprayAreas);
|
||
dup += checkRes.dup;
|
||
job.sprayAreas = utils.appendArray(job.sprayAreas, checkRes.areas);
|
||
}
|
||
if (!utils.isEmptyArray(xclAreas)) {
|
||
checkRes = jobUtil.checkDupAreas(allAreas, xclAreas);
|
||
dup += checkRes.dup;
|
||
job.excludedAreas = utils.appendArray(job.excludedAreas, checkRes.areas);
|
||
}
|
||
|
||
job.waypoints = jobUtil.cleanGeoPoints(utils.appendArray(job.waypoints, waypoints));
|
||
job.bufs = utils.appendArray(job.bufs, bufs);
|
||
job.places = jobUtil.cleanGeoPoints(utils.appendArray(job.places, places));
|
||
|
||
if (shouldCheckLimits && job.ttSprArea) usageLimits.totalSprAcres -= job.ttSprArea * CVCST.SM2ACR;
|
||
}
|
||
else if (impMsg.updateOp === JobUpdateOp.OVERWRITE) {
|
||
if (job.sprayAreas.length)
|
||
oldAreaIds = job.sprayAreas.map(a => a._id);
|
||
|
||
job.sprayAreas = sprayAreas;
|
||
job.excludedAreas = xclAreas;
|
||
job.waypoints = jobUtil.cleanGeoPoints(waypoints);
|
||
job.bufs = bufs;
|
||
job.places = jobUtil.cleanGeoPoints(places);
|
||
|
||
if (shouldCheckLimits && job.ttSprArea) usageLimits.totalSprAcres -= job.ttSprArea * CVCST.SM2ACR;
|
||
}
|
||
else if (impMsg.updateOp === JobUpdateOp.XCLS) {
|
||
let xcl;
|
||
for (let i = 0; i < sprayAreas.length; i++) { // Convert all found sprayAreas to xcls
|
||
xcl = jobUtil.sprayToXCL(sprayAreas[i]);
|
||
if (xcl) xclAreas.push(xcl);
|
||
}
|
||
checkRes = jobUtil.checkDupAreas(job.excludedAreas, xclAreas);
|
||
dup += checkRes.dup;
|
||
job.excludedAreas = jobUtil.cleanAreas(utils.appendArray(job.excludedAreas, checkRes.areas));
|
||
}
|
||
|
||
// Ensure Maximum of Items
|
||
if (!utils.isEmptyArray(job.sprayAreas) && job.sprayAreas.length >= FILE.MAX_ITEM)
|
||
job.sprayAreas = job.sprayAreas.slice(0, FILE.MAX_ITEM);
|
||
|
||
if (!utils.isEmptyArray(job.excludedAreas) && job.excludedAreas.length >= FILE.MAX_ITEM)
|
||
job.excludedAreas = job.excludedAreas.slice(0, FILE.MAX_ITEM);
|
||
|
||
if (!utils.isEmptyArray(job.waypoints) && job.waypoints.length >= FILE.MAX_ITEM)
|
||
job.waypoints = job.waypoints.slice(0, FILE.MAX_ITEM);
|
||
|
||
if (!utils.isEmptyArray(job.bufs) && job.bufs.length >= FILE.MAX_ITEM)
|
||
job.bufs = job.bufs.slice(0, FILE.MAX_ITEM);
|
||
|
||
if (!utils.isEmptyArray(job.places) && job.places.length >= FILE.MAX_ITEM)
|
||
job.places = job.places.slice(0, FILE.MAX_ITEM);
|
||
|
||
const newTTSprSqrMeters = jobUtil.calcTTSprayAreas(job.sprayAreas, job.excludedAreas);
|
||
|
||
if (shouldCheckLimits) {
|
||
if (usageLimits.totalSprAcres + (newTTSprSqrMeters * CVCST.SM2ACR) >= usageLimits.maxAcres)
|
||
AppMembershipError.throw(Errors.REACHED_AREA_LIMIT);
|
||
}
|
||
|
||
job.ttSprArea = newTTSprSqrMeters * CVCST.SM2HA;
|
||
|
||
const updateFn = async (job) => {
|
||
if (undefined === job._id) {
|
||
// Use mongoose directly for transactions
|
||
const session = await workerDB.getConnection().startSession();
|
||
try {
|
||
await session.withTransaction(async () => {
|
||
await job.save({ session });
|
||
await App.updateOne({ _id: ObjectId(impMsg.appId) }, { $set: { jobId: job._id } }).session(session);
|
||
});
|
||
} finally {
|
||
await session.endSession();
|
||
}
|
||
} else {
|
||
await job.save({ validateModifiedOnly: true });
|
||
}
|
||
};
|
||
|
||
updateFn(job)
|
||
.then(() => {
|
||
if (!impMsg.jobId)
|
||
impMsg.jobId = job._id;
|
||
|
||
if (impMsg.updateOp === JobUpdateOp.OVERWRITE)
|
||
return jobUtil.deleteAreaLines(oldAreaIds); // Remove grid lines of previous areas (before sprayAreas were overriten)
|
||
})
|
||
.then(() => {
|
||
anyUpdated = true;
|
||
callback();
|
||
})
|
||
.catch(err => {
|
||
debug(err.stack);
|
||
return callback(err);
|
||
});
|
||
} catch (err) {
|
||
callback(err);
|
||
}
|
||
}
|
||
], function (err) {
|
||
jobItems = null;
|
||
|
||
if (err) {
|
||
debug('Error while updating Job Item', err);
|
||
cb(err);
|
||
} else {
|
||
cb(null, { job: job, updated: anyUpdated, dup: dup });
|
||
}
|
||
});
|
||
}
|
||
|
||
async function getUsageLimits(user) {
|
||
if (!user || (user.kind !== UserTypes.APP) && !user.parent) AppAuthError.throw(Errors.INVALID_ACCOUNT);
|
||
|
||
let memUser = user, usageLimits = { maxAcres: 0, totalSprAcres: 0 };
|
||
|
||
if (UserTypes.APP !== memUser.kind && memUser.parent) {
|
||
memUser = await User.findById(memUser.parent, 'membership', { lean: true });
|
||
}
|
||
if (!memUser) AppAuthError.throw(Errors.APPLICATOR_NOT_FOUND);
|
||
|
||
const pkgSub = subUtil.getPkgSubfromUserInfo(memUser);
|
||
// Check for customer-specific override first, fallback to package limit
|
||
usageLimits.maxAcres = memUser.membership?.customLimits?.maxAcres
|
||
?? subUtil.getSubMetaField(pkgSub, SubFields.MAX_ACRES)
|
||
?? 0;
|
||
if (usageLimits.maxAcres) {
|
||
usageLimits.totalSprAcres = await subUtil.calcTotalAreaByUser(memUser._id, pkgSub.periodStart, pkgSub.periodEnd) * CVCST.HA2ACR;
|
||
}
|
||
return usageLimits;
|
||
}
|
||
|
||
function importData(dataPath, appId, job, cb) {
|
||
let appData, totalSprays = 0, totalSprLength = 0, totalFlightLength = 0, avgRates = [], totalTurnTime = 0, totalSprayTime = 0, totalFlightTime = 0, totalSprMats = 0, dataFiles = [], sprMatsUnit;
|
||
let totalSpeedAcc = 0, totalSpeedCount = 0; // for avgSpraySpeed
|
||
let totalHdopAcc = 0, totalHdopCount = 0; // for avgHdop
|
||
let totalXtAcc = 0, totalXtCount = 0; // for avgXtError
|
||
const importInfo = [];
|
||
const begin = Date.now(); // DEBUG - Measering total import data time
|
||
|
||
async.series([
|
||
function (callback) {
|
||
return wasCancelled(appId, callback);
|
||
},
|
||
function (callback) {
|
||
glob(path.join(dataPath, dataFilesPattern), globOps, (err, files) => {
|
||
if (err || files.length === 0) {
|
||
return callback();
|
||
}
|
||
// Classify found data files into known data format file entries
|
||
let match, file, typedFiles = [];
|
||
for (let i = 0; i < files.length; i++) {
|
||
file = files[i];
|
||
let basename = path.basename(file), agn;
|
||
|
||
// AG-NAV data nt files
|
||
if ((match = basename.match(/^n(\d{7})(-\d+)?.*\.t(\d{1,2})$/i))) {
|
||
agn = match.length >= 3 ? match[1] + match[3] + (match[2] || "") : match[1] + match[2];
|
||
typedFiles.push({ type: FILE.DATA_AGNAV, agn: agn, file: file });
|
||
// ESRI Shape spray on/off files
|
||
} else if ((match = basename.match(/^n(\d{9})(-\d+)?.*spr(?:on|off).*\.dbf$/i))) {
|
||
agn = match[1];
|
||
if (match[2] && match[2].startsWith('-'))
|
||
agn += match[2];
|
||
typedFiles.push({ type: FILE.DATA_SHAPE, agn: agn, file: file });
|
||
// SATLOG exported ascii data files
|
||
} else if ((match = basename.match(/^.*.asc$/i))) {
|
||
const stats = fs.statSync(file);
|
||
const m = stats && stats.mtime ? moment.utc(stats.mtime) : moment.utc();
|
||
// Note: might think about updating this later from the data if needed
|
||
agn = m.format('YYMMDDHHmm').substring(1);
|
||
typedFiles.push({ type: FILE.DATA_SALOG, agn: agn, file: file });
|
||
}
|
||
}
|
||
// fs.writeFileSync('./data.json', JSON.stringify(typedFiles) , 'utf-8');
|
||
|
||
if (typedFiles.length) {
|
||
typedFiles.sort(utils.dynamicSort('agn')); // Sort them in ascending datetime (from file names) order
|
||
|
||
// Pick non-SHAPE data file items
|
||
dataFiles = utils.appendArray(dataFiles, typedFiles.filter(it => it.type !== FILE.DATA_SHAPE).map(it => [it]));
|
||
|
||
// Pick SHAPE spray-on and off pairs or spray-on data file items only
|
||
let i = 0, type2Files = typedFiles.filter(it => it.type === FILE.DATA_SHAPE);
|
||
while (type2Files.length > 0) {
|
||
let item = type2Files[i];
|
||
type2Files.splice(i, 1);
|
||
|
||
let m = item.file.match(/^(.*n(?:\d{9}).*spr)(on|off).*\.dbf$/i);
|
||
item['sprayOn'] = (m[2].toLocaleLowerCase() === 'on');
|
||
let find = item['sprayOn'] ? 'off' : 'on';
|
||
|
||
let j = type2Files.length - 1;
|
||
let foundPair = false;
|
||
while (type2Files.length && j >= 0) {
|
||
let regex = new RegExp(`^${m[1]}${find}.*.dbf$`, 'i');
|
||
if (regex.test(type2Files[j].file) && item.agn === type2Files[j].agn) {
|
||
type2Files[j]['sprayOn'] = !item['sprayOn'];
|
||
dataFiles.push([item, type2Files[j]]);
|
||
type2Files.splice(j, 1);
|
||
foundPair = true;
|
||
break;
|
||
}
|
||
j--;
|
||
}
|
||
if (!foundPair) {
|
||
if (/^.*n\d{9}.*spron.*\.dbf$/i.test(item.file))
|
||
dataFiles.push([item]);
|
||
}
|
||
}
|
||
}
|
||
return callback();
|
||
});
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appId, callback);
|
||
},
|
||
function (callback) {
|
||
if (!dataFiles.length) return callback();
|
||
// debug("Files: ", dataFiles);
|
||
async.eachSeries(dataFiles,
|
||
function (fileItems, callback) {
|
||
async.series([
|
||
function (callback) {
|
||
wasCancelled(appId, callback)
|
||
},
|
||
function (callback) {
|
||
importDataFiles(fileItems, appId, job, (err, data) => {
|
||
// debug("Done read for: ", fileItems);
|
||
if (data) {
|
||
importInfo.push({ agn: fileItems[0].agn, info: data });
|
||
|
||
if (utils.isNumber(data.totalSprayed)) totalSprays += data.totalSprayed;
|
||
if (utils.isNumber(data.totalSprLength)) totalSprLength += data.totalSprLength;
|
||
if (utils.isNumber(data.totalFlightLength)) totalFlightLength += data.totalFlightLength;
|
||
if (utils.isNumber(data.turnTime)) totalTurnTime += data.turnTime;
|
||
if (utils.isNumber(data.sprayTime)) totalSprayTime += data.sprayTime;
|
||
if (utils.isNumber(data.totalTime)) totalFlightTime += data.totalTime;
|
||
|
||
if (utils.isNumber(data.spraySpeedCount) && data.spraySpeedCount > 0 && utils.isNumber(data.avgSpraySpeed)) {
|
||
totalSpeedAcc += data.avgSpraySpeed * data.spraySpeedCount;
|
||
totalSpeedCount += data.spraySpeedCount;
|
||
}
|
||
if (utils.isNumber(data.hdopCount) && data.hdopCount > 0 && utils.isNumber(data.hdopSum)) {
|
||
totalHdopAcc += data.hdopSum;
|
||
totalHdopCount += data.hdopCount;
|
||
}
|
||
if (utils.isNumber(data.xtCount) && data.xtCount > 0 && utils.isNumber(data.xtSum)) {
|
||
totalXtAcc += data.xtSum;
|
||
totalXtCount += data.xtCount;
|
||
}
|
||
|
||
if (data.avgRate)
|
||
avgRates.push(data.avgRate);
|
||
|
||
if (utils.isNumber(data.totalSprayMat)) {
|
||
totalSprMats += data.totalSprayMat;
|
||
if (data.totalSprayMat > 0 && data.totalSprayMatUnit) {
|
||
sprMatsUnit = data.totalSprayMatUnit;
|
||
}
|
||
}
|
||
}
|
||
callback();
|
||
});
|
||
}
|
||
], callback);
|
||
}, err => { // If any of the saves produced an error, err would equal that error
|
||
debug(`All ${dataFiles.length} files Imported.`);
|
||
if (err) callback(err);
|
||
else callback();
|
||
});
|
||
},
|
||
function (callback) {
|
||
return wasCancelled(appId, callback);
|
||
},
|
||
function (callback) {
|
||
if (!importInfo.length) return callback();
|
||
|
||
let first = importInfo[0];
|
||
let last = importInfo[importInfo.length - 1];
|
||
// Compute application start and end time
|
||
const startendDate = computeStartEndDate(first.info.firstTime, last.info.lastTime, first.agn, last.agn);
|
||
// Coordinates are captured into firstLat/firstLon before records are GC'd inside importDataFiles
|
||
let firstLat = null, firstLon = null;
|
||
for (const entry of importInfo) {
|
||
if (entry.info && utils.isNumber(entry.info.firstLat) && utils.isNumber(entry.info.firstLon)) {
|
||
firstLat = entry.info.firstLat;
|
||
firstLon = entry.info.firstLon;
|
||
break;
|
||
}
|
||
}
|
||
const dateFields = appDateTime.buildApplicationDateFields({
|
||
startDateTime: startendDate.start.format('YYYYMMDDTHHmmss'),
|
||
endDateTime: startendDate.end.format('YYYYMMDDTHHmmss'),
|
||
latitude: firstLat,
|
||
longitude: firstLon
|
||
});
|
||
appData = {
|
||
startDateTime: startendDate.start,
|
||
endDateTime: startendDate.end,
|
||
utcOffset: dateFields.utcOffset,
|
||
startDateTimeUTC: dateFields.startDateTimeUTC,
|
||
endDateTimeUTC: dateFields.endDateTimeUTC,
|
||
appRate: avgRates.length ? avgRates.reduce(function (a, b) { return Number(a) + Number(b); }) / avgRates.length : 0,
|
||
totalSprayed: totalSprays,
|
||
totalSprLength: totalSprLength,
|
||
totalFlightLength: totalFlightLength,
|
||
totalSprayTime: totalSprayTime,
|
||
totalTurnTime: totalTurnTime,
|
||
totalFlightTime: totalFlightTime,
|
||
totalSprayMat: totalSprMats,
|
||
totalSprayMatUnit: sprMatsUnit,
|
||
avgSpraySpeed: totalSpeedCount > 0 ? totalSpeedAcc / totalSpeedCount : null,
|
||
avgHdop: totalHdopCount > 0 ? totalHdopAcc / totalHdopCount : null,
|
||
avgXtError: totalXtCount > 0 ? totalXtAcc / totalXtCount : null
|
||
}
|
||
|
||
const duration = Date.now() - begin;
|
||
debug('Total (data): %dms', duration);
|
||
callback();
|
||
}
|
||
], (err) => {
|
||
if (err) {
|
||
return cb(err);
|
||
}
|
||
cb(null, appData);
|
||
});
|
||
}
|
||
|
||
function importDataFiles(fileItems, appId, job, cb) {
|
||
let appFile, firstTime, lastTime, totalSprMats = 0, sprMatsUnit;
|
||
let importInfo = { firstTime: 0, lastTime: 0, turnTime: 0, sprayTime: 0, totalTime: 0 };
|
||
let fileName = path.basename(fileItems[0].file), fileMeta, hasData = false;
|
||
|
||
const dataType = fileItems[0].type;
|
||
|
||
if (fileItems.length > 1 && dataType === FILE.DATA_SHAPE) {
|
||
fileName = fileName.replace(new RegExp("(on|off)", "i"), "onoff");
|
||
fileItems.sort(utils.dynamicSort('-sprayOn')); // Make spray-on file first in the list
|
||
}
|
||
|
||
async.series([
|
||
function (callback) {
|
||
// Read q file if exists for AGNAV binary data
|
||
const file = fileItems[0];
|
||
if (file.type === FILE.DATA_AGNAV || file.type === FILE.DATA_SHAPE) {
|
||
let qfilePath = path.join(path.dirname(file.file), 'q' + (file.type === FILE.DATA_AGNAV ? path.basename(file.file).slice(1) : file.agn));
|
||
fileAgNav.readQFile(qfilePath, (err, meta) => {
|
||
if (!utils.isEmptyObj(meta)) {
|
||
meta.hasQfile = true; // Mark that Q file was successfully read
|
||
fileMeta = meta;
|
||
}
|
||
callback();
|
||
});
|
||
} else {
|
||
callback();
|
||
}
|
||
},
|
||
function (callback) {
|
||
const fileObj = { appId: appId, name: fileName, agn: fileItems[0].agn };
|
||
|
||
// Normalize metadata according to DATA_FORMAT_NOTES.md
|
||
const { DataTypes, MatTypes } = require('../helpers/constants');
|
||
let normalizedMeta;
|
||
// Determine default material type from job's application rate unit
|
||
// Dry units: LBS_PER_ACRE, KG_PER_HA
|
||
// Wet units: OZ_PER_ACRE, GAL_PER_ACRE, LIT_PER_HA
|
||
const defaultMatType = (job.appRateUnit === RateUnits.LBS_PER_ACRE || job.appRateUnit === RateUnits.KG_PER_HA)
|
||
? MatTypes.DRY
|
||
: MatTypes.WET;
|
||
|
||
if (utils.isEmptyObj(fileMeta)) {
|
||
fileObj.note = "NO_QFILE";
|
||
// No Q file - use job defaults
|
||
// Create fileMeta for use in rateInfoFromFileMeta()
|
||
fileMeta = {
|
||
fcType: 'none',
|
||
appRate: job.appRate,
|
||
rateUnit: job.appRateUnit,
|
||
hasQfile: false
|
||
};
|
||
normalizedMeta = {
|
||
type: DataTypes.AGNAV,
|
||
matType: defaultMatType,
|
||
operator: null,
|
||
fcName: null,
|
||
// Original fields for backward compatibility
|
||
...fileMeta
|
||
};
|
||
} else {
|
||
// Normalize Q file metadata
|
||
// Determine material type from qfile's appRateUnitStr or fcType
|
||
let matType = null;
|
||
if (fileMeta.appRateUnitStr) {
|
||
const unitStr = fileMeta.appRateUnitStr.toLowerCase();
|
||
// Liquid units: gal/ac, L/ha, oz/ac. Dry units: lbs/ac, Kg/ha
|
||
matType = (unitStr.includes('gal') || unitStr.includes('l/') || unitStr.includes('oz')) ? MatTypes.WET : MatTypes.DRY;
|
||
}
|
||
if (!matType) {
|
||
// Fallback to AgNav fcType (Flow Controller name) mapping if any
|
||
const matTypeFromFCType = utils.matTypeFromFCType(fileMeta.fcType);
|
||
if (matTypeFromFCType !== 'none') {
|
||
matType = matTypeFromFCType === 'dry' ? MatTypes.DRY : MatTypes.WET;
|
||
}
|
||
}
|
||
if (!matType) {
|
||
// Fallback to job default inferred from job application rate unit
|
||
matType = defaultMatType;
|
||
}
|
||
|
||
normalizedMeta = {
|
||
// Normalized fields (common format for both AgNav and SatLoc)
|
||
type: DataTypes.AGNAV,
|
||
matType: matType,
|
||
operator: fileMeta.operator || null, // From OPERATOR field in Q file
|
||
fcName: fileMeta.fcType || null, // From FC TYPE field in Q file
|
||
|
||
// All original Q file fields (preserved completely)
|
||
...fileMeta,
|
||
|
||
// Additional backward compatibility flags
|
||
hasQfile: true
|
||
};
|
||
}
|
||
|
||
fileObj.meta = normalizedMeta;
|
||
|
||
appFile = new AppFile(fileObj);
|
||
|
||
appFile.save(err => {
|
||
if (err) return callback(err);
|
||
callback();
|
||
});
|
||
},
|
||
function (callback) {
|
||
async.eachSeries(fileItems,
|
||
function (dataFile, callback) {
|
||
switch (dataFile.type) {
|
||
case FILE.DATA_AGNAV:
|
||
readNTFile(dataFile.file, fileMeta, appFile._id, (err, data) => {
|
||
if (err) return callback(err);
|
||
importInfo = data;
|
||
if (importInfo && !utils.isEmptyArray(importInfo.records)) {
|
||
firstTime = importInfo.records[0].gpsTime;
|
||
lastTime = importInfo.records[importInfo.records.length - 1].gpsTime;
|
||
if (utils.isNumber(data.totalSprayMat)) {
|
||
totalSprMats += data.totalSprayMat;
|
||
if (data.totalSprayMat > 0 && data.totalSprayMatUnit) {
|
||
sprMatsUnit = data.totalSprayMatUnit;
|
||
}
|
||
}
|
||
}
|
||
callback();
|
||
});
|
||
break;
|
||
case FILE.DATA_SHAPE:
|
||
readShapeDataFile(dataFile, fileMeta, appFile._id, (err, data) => {
|
||
if (err) return callback(err);
|
||
if (data) {
|
||
if (dataFile['sprayOn']) {
|
||
importInfo.totalSprayed = data.totalSprayed;
|
||
importInfo.avgRate = data.avgRate;
|
||
if (utils.isNumber(data.totalSprayMat)) {
|
||
totalSprMats += data.totalSprayMat;
|
||
if (data.totalSprayMat > 0 && data.totalSprayMatUnit) {
|
||
sprMatsUnit = data.totalSprayMatUnit;
|
||
}
|
||
}
|
||
}
|
||
if (!utils.isEmptyArray(data.records)) {
|
||
if (utils.isEmptyArray(importInfo.records)) {
|
||
firstTime = data.records[0].gpsTime;
|
||
lastTime = data.records[data.records.length - 1].gpsTime;
|
||
} else {
|
||
if (data.records[0].gpsTime < firstTime)
|
||
firstTime = data.records[0].gpsTime;
|
||
if (data.records[data.records.length - 1].gpsTime > lastTime)
|
||
lastTime = data.records[data.records.length - 1].gpsTime;
|
||
}
|
||
}
|
||
// Aggregate per-file totals to avoid a full rescanning pass
|
||
importInfo.totalSprLength = (importInfo.totalSprLength || 0) + (data.totalSprLength || 0);
|
||
importInfo.totalFlightLength = (importInfo.totalFlightLength || 0) + (data.totalFlightLength || 0);
|
||
|
||
// Account for cross-file boundary segment (last of existing -> first of new)
|
||
if (!utils.isEmptyArray(importInfo.records) && !utils.isEmptyArray(data.records)) {
|
||
const prev = importInfo.records[importInfo.records.length - 1];
|
||
const curr = data.records[0];
|
||
if (prev && curr && utils.isNumber(prev.utmX) && utils.isNumber(prev.utmY) && utils.isNumber(curr.utmX) && utils.isNumber(curr.utmY) && utils.isNumber(prev.gpsTime) && utils.isNumber(curr.gpsTime)) {
|
||
let dt = curr.gpsTime - prev.gpsTime;
|
||
if (dt < 0 && Math.abs(dt) >= 80000) dt = (86400 - prev.gpsTime) + curr.gpsTime;
|
||
if (dt > 0 && dt <= 120) {
|
||
const segDist = Math.hypot(curr.utmX - prev.utmX, curr.utmY - prev.utmY);
|
||
if (segDist <= 1000) {
|
||
importInfo.totalFlightLength += segDist;
|
||
if ((prev.sprayStat && prev.sprayStat > 0) || (curr.sprayStat && curr.sprayStat > 0)) {
|
||
importInfo.totalSprLength += segDist;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
importInfo.records = utils.appendArray(importInfo.records, data.records);
|
||
}
|
||
// else if (!dataFile['sprayOn']) {
|
||
// // Skip the spray-off if no sprayon file
|
||
// return callback(new Error(Errors.DATA_NO_SPRAYON));
|
||
// }
|
||
callback();
|
||
});
|
||
break;
|
||
case FILE.DATA_SALOC: // Removed support for SATLOG exported ascii data file
|
||
break;
|
||
default:
|
||
return callback();
|
||
}
|
||
}, err => {
|
||
// Ensure skipping the loop normally incase of no spray-on data
|
||
if (err && err.message == Errors.DATA_NO_SPRAYON)
|
||
callback();
|
||
else
|
||
callback(err);
|
||
}) // eachSeries
|
||
},
|
||
// Empty file ? => remove the created appFile entry
|
||
function (callback) {
|
||
hasData = (importInfo && !utils.isEmptyArray(importInfo.records));
|
||
if (appFile && !hasData) {
|
||
appFile.remove(callback);
|
||
}
|
||
else callback();
|
||
},
|
||
// Save records of the file to db
|
||
function (callback) {
|
||
if (!hasData) return callback();
|
||
|
||
// Sort spray-on and off data by gspTime (provided data in the two files has been already compensated a day after passing midnight)
|
||
if (fileItems.length > 1 && dataType === FILE.DATA_SHAPE)
|
||
importInfo.records.sort(utils.dynamicSort('gpsTime', true));
|
||
|
||
const chunks = utils.chunkArray(importInfo.records, 1000);
|
||
async.eachSeries(chunks, (chunk, cb) => { // Use series to ensure correct insert order of the records
|
||
AppDetail.insertMany(chunk, { rawResult: true, ordered: true }, (err) => {
|
||
if (err) return cb(err);
|
||
cb();
|
||
});
|
||
}, (err) => {
|
||
// err & console.log(err);
|
||
callback(err);
|
||
});
|
||
},
|
||
function (callback) {
|
||
if (!hasData) return callback();
|
||
|
||
/* Calculate turn time for each file or each pair of shape on/off files
|
||
Turn Time: total turn times after each spray line. Counting start from spray OFF the previous line to spray ON on the next line
|
||
(skip the segments within the same line)
|
||
*/
|
||
let turnTime = { line: null, at: null, nextOff: false, total: 0 }, timeDif = 0, totalSprTime = 0, totalTime = 0;
|
||
let totalSpeedAcc = 0, spraySpeedCount = 0; // for avgSpraySpeed
|
||
let hdopAcc = 0, hdopCount = 0; // for avgHdop
|
||
let xtAcc = 0, xtCount = 0; // for avgXtError
|
||
let prevTime = -999, prevSprTime = -999;
|
||
let record;
|
||
for (let i = 0; i < importInfo.records.length; i++) {
|
||
record = importInfo.records[i];
|
||
|
||
// Calculate total flight time
|
||
if (prevTime != -999 && prevTime !== record.gpsTime) {
|
||
timeDif = record.gpsTime - prevTime;
|
||
if (timeDif < 0) {
|
||
if (Math.abs(timeDif) >= 80000)
|
||
timeDif = (86400 - prevTime) + record.gpsTime;
|
||
}
|
||
if (timeDif > 0 && timeDif <= 120)
|
||
totalTime += timeDif;
|
||
}
|
||
prevTime = record.gpsTime;
|
||
|
||
// Calculate spray time (secs)
|
||
if (record.sprayStat > 0) {
|
||
if (prevSprTime != -999 && record.sprayStat !== 3) {
|
||
timeDif = record.gpsTime - prevSprTime;
|
||
if (timeDif < 0) {
|
||
if (Math.abs(timeDif) >= 80000)
|
||
timeDif = (86400 - prevSprTime) + record.gpsTime;
|
||
}
|
||
if (timeDif > 0 && timeDif <= 120)
|
||
totalSprTime += timeDif;
|
||
}
|
||
if (utils.isNumber(record.grSpeed) && record.grSpeed !== 0) {
|
||
totalSpeedAcc += record.grSpeed;
|
||
spraySpeedCount++;
|
||
}
|
||
if (utils.isNumber(record.stdHdop) && record.stdHdop > 0) {
|
||
hdopAcc += record.stdHdop;
|
||
hdopCount++;
|
||
}
|
||
if ((record.sprayStat === 1 || record.sprayStat === 3) &&
|
||
utils.isNumber(record.xTrack) && record.xTrack !== 0) {
|
||
xtAcc += Math.abs(record.xTrack);
|
||
xtCount++;
|
||
}
|
||
prevSprTime = record.gpsTime;
|
||
}
|
||
|
||
if (fileItems[0].type !== FILE.DATA_SALOC) {
|
||
// Calculate turn time (secs)
|
||
if (null === turnTime.line) {
|
||
if (!record.sprayStat) {
|
||
turnTime.line = record.llnum;
|
||
turnTime.at = record.gpsTime;
|
||
}
|
||
} else {
|
||
if (turnTime.line != record.llnum) {
|
||
if (record.sprayStat) {
|
||
timeDif = record.gpsTime - turnTime.at;
|
||
if (timeDif < 0) {
|
||
if (Math.abs(timeDif) >= 80000)
|
||
timeDif = (86400 - turnTime.at) + record.gpsTime;
|
||
}
|
||
if (timeDif >= 5 && timeDif <= 120)
|
||
turnTime.total += timeDif;
|
||
|
||
turnTime.line = record.llnum;
|
||
turnTime.nextOff = true;
|
||
}
|
||
}
|
||
else {
|
||
if (!record.sprayStat && turnTime.nextOff) { // Mark start for the next turn
|
||
turnTime.at = record.gpsTime;
|
||
turnTime.nextOff = false;
|
||
} else if (record.sprayStat) {
|
||
turnTime.nextOff = true;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
importInfo.turnTime = turnTime.total;
|
||
importInfo.sprayTime = totalSprTime;
|
||
importInfo.totalTime = totalTime;
|
||
importInfo.avgSpraySpeed = spraySpeedCount > 0 ? totalSpeedAcc / spraySpeedCount : null; // m/s
|
||
importInfo.spraySpeedCount = spraySpeedCount;
|
||
importInfo.hdopSum = hdopAcc;
|
||
importInfo.hdopCount = hdopCount;
|
||
importInfo.xtSum = xtAcc;
|
||
importInfo.xtCount = xtCount;
|
||
|
||
callback();
|
||
},
|
||
function (callback) {
|
||
if (hasData && appFile) {
|
||
if (utils.isNumber(importInfo.totalSprayed)) appFile.totalSprayed = importInfo.totalSprayed * 1E-4; // m2 to ha
|
||
if (utils.isNumber(importInfo.turnTime)) appFile.totalTurnTime = importInfo.turnTime;
|
||
if (utils.isNumber(importInfo.sprayTime)) appFile.totalSprayTime = importInfo.sprayTime;
|
||
if (utils.isNumber(importInfo.totalTime)) appFile.totalFlightTime = importInfo.totalTime;
|
||
if (utils.isNumber(importInfo.totalSprLength)) appFile.totalSprLength = importInfo.totalSprLength;
|
||
if (utils.isNumber(importInfo.totalFlightLength)) appFile.totalFlightLength = importInfo.totalFlightLength;
|
||
if (utils.isNumber(totalSprMats) && sprMatsUnit !== undefined) {
|
||
appFile.totalSprayMat = totalSprMats;
|
||
appFile.totalSprayMatUnit = sprMatsUnit;
|
||
}
|
||
appFile.save(callback);
|
||
} else callback();
|
||
}
|
||
], (err) => {
|
||
if (err) return cb(err);
|
||
|
||
if (importInfo && !utils.isEmptyArray(importInfo.records)) {
|
||
importInfo.firstTime = firstTime;
|
||
importInfo.lastTime = lastTime;
|
||
|
||
// Use aggregated per-file totals (computed during file parsing). Fall back to
|
||
// full-scan only if these weren't computed for some reason.
|
||
importInfo.totalSprLength = (importInfo.totalSprLength !== undefined) ? importInfo.totalSprLength : _computeSprLength(importInfo.records);
|
||
importInfo.totalFlightLength = (importInfo.totalFlightLength !== undefined) ? importInfo.totalFlightLength : _computeFlightLength(importInfo.records);
|
||
|
||
importInfo.totalSprayMat = totalSprMats;
|
||
importInfo.totalSprayMatUnit = sprMatsUnit;
|
||
|
||
// Capture first valid coordinate before records are discarded (used for timezone lookup)
|
||
const firstWithCoords = importInfo.records.find(r => utils.isNumber(r.lat) && utils.isNumber(r.lon));
|
||
importInfo.firstLat = firstWithCoords ? firstWithCoords.lat : null;
|
||
importInfo.firstLon = firstWithCoords ? firstWithCoords.lon : null;
|
||
|
||
delete importInfo.records;
|
||
gc();
|
||
return cb(null, importInfo);
|
||
} else return cb();
|
||
});
|
||
}
|
||
|
||
function readNTFile(file, fileMeta, fileId, cb) {
|
||
let binBuf, records = [];
|
||
let sprayedSeg = 0, totalSprays = 0, totalSprMats = 0, sprMatsUnit, totalAppRates = 0, totalSprayRecs = 0;
|
||
let prevUTM_X, prevUTM_Y, prevSwath, prevLine, prevStat = 0;
|
||
let totalFlightLen = 0, totalSprLen = 0;
|
||
let prevRecUTM_X = null, prevRecUTM_Y = null, prevRecTime = null, prevRecSprStat = null;
|
||
let rateInfo = null;
|
||
|
||
async.series([
|
||
function (callback) {
|
||
fs.readFile(file, (err, data) => {
|
||
if (err) return callback(err);
|
||
binBuf = data;
|
||
callback();
|
||
});
|
||
},
|
||
function (callback) {
|
||
if (!binBuf || !Buffer.isBuffer(binBuf)) return callback();
|
||
|
||
let latlon, startIdx = 0, timeOffset = 0, appliedRate;
|
||
rateInfo = utils.rateInfoFromFileMeta(fileMeta, RecTypes.AGN_BIN_LQD);
|
||
const recType = rateInfo.recType;
|
||
|
||
while (binBuf.length - startIdx >= FILE.AGN_PACK_SIZE) {
|
||
if (DataUtil.isValidAgn(binBuf, startIdx)) {
|
||
const record = DataUtil.readAgnBinary(binBuf, startIdx, recType);
|
||
if (!record) continue;
|
||
|
||
if (record.type === RecTypes.AGN_AMS) {
|
||
if (records.length) {
|
||
// Copy AMS record fields to the last main record
|
||
records[records.length - 1] = DataUtil.mergeAgnAms(records[records.length - 1], record);
|
||
}
|
||
} else {
|
||
if (record.timeAdv > 0.0) {
|
||
// Shift the lat lon coordinate according to time Advance (system lag)
|
||
latlon = geoUtil.projectLatLong(record.lat, record.lon, record.grSpeed, record.head, record.timeAdv);
|
||
record.lat = latlon.lat; record.lon = latlon.lon;
|
||
}
|
||
record["fileId"] = fileId;
|
||
records.push(record);
|
||
|
||
// For compensation with a day when gpsTime passing mid-night (rolloff)
|
||
if (records.length > 1 && timeOffset === 0
|
||
&& ((records[records.length - 1].gpsTime - records[records.length - 2].gpsTime) < 0
|
||
&& Math.abs(records[records.length - 1].gpsTime - records[records.length - 2].gpsTime) >= 80000))
|
||
timeOffset = 86400;
|
||
record.adjustGpsTime(timeOffset);
|
||
|
||
// Incremental distance accumulation with time-gap and outlier guards
|
||
if (prevRecUTM_X !== null && prevRecUTM_Y !== null && utils.isNumber(record.utmX) && utils.isNumber(record.utmY) && prevRecTime !== null) {
|
||
let dt = record.gpsTime - prevRecTime;
|
||
if (dt < 0 && Math.abs(dt) >= 80000) dt = (86400 - prevRecTime) + record.gpsTime;
|
||
if (dt > 0 && dt <= 120) {
|
||
const segDist = Math.hypot(record.utmX - prevRecUTM_X, record.utmY - prevRecUTM_Y);
|
||
if (segDist <= 1000) {
|
||
totalFlightLen += segDist;
|
||
if ((prevRecSprStat && prevRecSprStat > 0) || (record.sprayStat && record.sprayStat > 0)) {
|
||
totalSprLen += segDist;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Update running previous-record markers for next iteration
|
||
prevRecUTM_X = record.utmX; prevRecUTM_Y = record.utmY; prevRecTime = record.gpsTime; prevRecSprStat = record.sprayStat;
|
||
|
||
if (record.sprayStat > 0) {
|
||
({ appliedRate, sprMatsUnit } = getAppliedRate(record, rateInfo, rateInfo.recType === RecTypes.AGN_BIN_LQD));
|
||
|
||
if (record.lhaReq > 0) {
|
||
totalSprayRecs++;
|
||
totalAppRates += record.lhaReq;
|
||
}
|
||
if (record.sprayStat === 3) {
|
||
prevUTM_X = record.utmX;
|
||
prevUTM_Y = record.utmY;
|
||
prevSwath = record.swath;
|
||
prevLine = record.llnum;
|
||
prevStat = record.sprayStat;
|
||
}
|
||
else {
|
||
if (prevStat > 0 && prevLine === record.llnum) {
|
||
sprayedSeg = Math.hypot(record.utmX - prevUTM_X, record.utmY - prevUTM_Y) * prevSwath;
|
||
if (sprayedSeg) {
|
||
totalSprays += sprayedSeg;
|
||
if (appliedRate > 0) {
|
||
totalSprMats += (sprayedSeg * CVCST.SM2HA) * appliedRate;
|
||
}
|
||
}
|
||
}
|
||
prevUTM_X = record.utmX;
|
||
prevUTM_Y = record.utmY;
|
||
prevSwath = record.swath;
|
||
prevLine = record.llnum;
|
||
prevStat = record.sprayStat;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
startIdx += FILE.AGN_PACK_SIZE;
|
||
}
|
||
binBuf = null;
|
||
callback();
|
||
}
|
||
], err => {
|
||
if (err) return cb(err);
|
||
|
||
// avgRate: prefer mean lhaReq from binary records; fall back to the Q-file /
|
||
// job planned rate (metric-converted) when no per-record lhaReq was captured.
|
||
let avgRate = 0;
|
||
if (totalSprayRecs > 0) {
|
||
avgRate = totalAppRates / totalSprayRecs;
|
||
} else if (rateInfo && rateInfo.appRate > 0) {
|
||
avgRate = utils.toMetricRate(rateInfo.appRate, rateInfo.rateUnit).value;
|
||
}
|
||
|
||
const fileDataInfo = {
|
||
records: records,
|
||
totalSprayMat: totalSprMats,
|
||
totalSprayMatUnit: sprMatsUnit,
|
||
totalSprayed: totalSprays,
|
||
totalSprLength: totalSprLen,
|
||
totalFlightLength: totalFlightLen,
|
||
avgRate: avgRate
|
||
}
|
||
// gc();
|
||
cb(null, fileDataInfo);
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Compute total flight path length in meters from an array of records with utmX/utmY.
|
||
* Used as a FALLBACK only — inline aggregation in readNTFile/readShapeDataFile is preferred.
|
||
* ALL segments are counted regardless of spray status (includes turns).
|
||
* Segments are skipped when:
|
||
* - consecutive time gap > 120 s (GPS dropout / instrument pause)
|
||
* - distance > 1000 m (GPS position outlier)
|
||
* Midnight-rollover of gpsTime (seconds-of-day) is handled automatically.
|
||
* @param {Array} records
|
||
* @returns {number} total travel distance in meters
|
||
*/
|
||
function _computeFlightLength(records) {
|
||
let total = 0;
|
||
let prev = null;
|
||
let prevTime = null;
|
||
for (let i = 0; i < records.length; i++) {
|
||
const curr = records[i];
|
||
if (!prev) {
|
||
prev = curr;
|
||
prevTime = prev && prev.gpsTime !== undefined ? prev.gpsTime : null;
|
||
continue;
|
||
}
|
||
if (prev.utmX && prev.utmY && curr.utmX && curr.utmY) {
|
||
const dist = Math.hypot(curr.utmX - prev.utmX, curr.utmY - prev.utmY);
|
||
// time gap check (handle midnight rollovers similar to other logic)
|
||
let timeOk = true;
|
||
if (prevTime !== null && curr.gpsTime !== undefined) {
|
||
let dt = curr.gpsTime - prevTime;
|
||
if (dt < 0 && Math.abs(dt) >= 80000) dt = (86400 - prevTime) + curr.gpsTime;
|
||
if (dt <= 0 || dt > 120) timeOk = false;
|
||
}
|
||
if (timeOk && dist <= 1000) total += dist;
|
||
}
|
||
prev = curr;
|
||
prevTime = curr && curr.gpsTime !== undefined ? curr.gpsTime : null;
|
||
}
|
||
return total;
|
||
}
|
||
|
||
/**
|
||
* Compute spray path length in meters from an array of records with utmX/utmY.
|
||
* Used as a FALLBACK only — inline aggregation in readNTFile/readShapeDataFile is preferred.
|
||
* Only segments where at least one endpoint has spray ON (sprayStat > 0) are counted,
|
||
* matching the behaviour of readSatLogAsc() which skips pure spray-off segments (turns).
|
||
* Segments are skipped when:
|
||
* - both endpoints have sprayStat == 0 (turn between spray lines)
|
||
* - consecutive time gap > 120 s (GPS dropout / instrument pause)
|
||
* - distance > 1000 m (GPS position outlier)
|
||
* Midnight-rollover of gpsTime (seconds-of-day) is handled automatically.
|
||
* @param {Array} records
|
||
* @returns {number} spray distance in meters
|
||
*/
|
||
function _computeSprLength(records) {
|
||
let total = 0;
|
||
let prev = null;
|
||
let prevTime = null;
|
||
for (let i = 0; i < records.length; i++) {
|
||
const curr = records[i];
|
||
if (!prev) {
|
||
prev = curr;
|
||
prevTime = prev && prev.gpsTime !== undefined ? prev.gpsTime : null;
|
||
continue;
|
||
}
|
||
// Skip pure spray-off segments (turns between spray lines)
|
||
if (!(prev.sprayStat > 0 || curr.sprayStat > 0)) {
|
||
prev = curr; prevTime = curr && curr.gpsTime !== undefined ? curr.gpsTime : null; continue;
|
||
}
|
||
if (prev.utmX && prev.utmY && curr.utmX && curr.utmY) {
|
||
const dist = Math.hypot(curr.utmX - prev.utmX, curr.utmY - prev.utmY);
|
||
// time gap check (handle midnight rollovers)
|
||
let timeOk = true;
|
||
if (prevTime !== null && curr.gpsTime !== undefined) {
|
||
let dt = curr.gpsTime - prevTime;
|
||
if (dt < 0 && Math.abs(dt) >= 80000) dt = (86400 - prevTime) + curr.gpsTime;
|
||
if (dt <= 0 || dt > 120) timeOk = false;
|
||
}
|
||
if (timeOk && dist <= 1000) total += dist;
|
||
}
|
||
prev = curr;
|
||
prevTime = curr && curr.gpsTime !== undefined ? curr.gpsTime : null;
|
||
}
|
||
return total;
|
||
}
|
||
|
||
/**
|
||
* Determine the applied rate based on metadata from file or from recorded data record
|
||
* @param {*} record
|
||
* @param {*} rateInfo
|
||
* @param {*} isLiquid
|
||
* @returns { appliedRate, sprMatsUnit } in Metric: L/Ha or Kg/Ha
|
||
*/
|
||
function getAppliedRate(record, rateInfo, isLiquid) {
|
||
let appliedRate, sprMatsUnit;
|
||
|
||
if (rateInfo && rateInfo.appRate && (!rateInfo.useFC || !record.lminApp)) {
|
||
appliedRate = rateInfo.appRate;
|
||
if (rateInfo.rateUnit === 0) {
|
||
appliedRate = appliedRate * CVCST.OZPA2LPHA;
|
||
sprMatsUnit = RateUnits.LIT_PER_HA;
|
||
} else {
|
||
sprMatsUnit = rateInfo.rateUnit;
|
||
// Convert to metric rate to store to db later
|
||
const metricRate = utils.toMetricRate(appliedRate, sprMatsUnit);
|
||
appliedRate = metricRate.value;
|
||
sprMatsUnit = metricRate.unit;
|
||
}
|
||
} else {
|
||
if (isLiquid) {
|
||
appliedRate = utils.appRateFromFlowRate(record.lminApp, record.swath, record.grSpeed);
|
||
sprMatsUnit = RateUnits.LIT_PER_HA;
|
||
} else {
|
||
appliedRate = record.lminApp;
|
||
sprMatsUnit = RateUnits.KG_PER_HA;
|
||
}
|
||
}
|
||
return { appliedRate, sprMatsUnit };
|
||
}
|
||
|
||
function readShapeDataFile(dataFile, fileMeta, fileId, cb) {
|
||
let sprayedSeg = 0, totalAppRates = 0, totalSprMats = 0, sprMatsUnit, totalSprays = 0, totalSprayRecs = 0;
|
||
let prevUTM_X, prevUTM_Y, prevSwath, prevLine;
|
||
let records = [], latlon;
|
||
let totalFlightLen = 0, totalSprLen = 0;
|
||
let prevRecUTM_X = null, prevRecUTM_Y = null, prevRecTime = null, prevRecSprStat = null;
|
||
|
||
fileShp.readDBF4Items(dataFile.file, ["GPSTIME", "LATITUDE", "LONGITUDE", "GRNDSPEED"], (err, items) => {
|
||
if (err) {
|
||
debug(`Error in readShapeDataFile(): ${dataFile}`, err)
|
||
cb(err);
|
||
return;
|
||
}
|
||
let timeOffset = 0, appliedRate;
|
||
const rateInfo = utils.rateInfoFromFileMeta(fileMeta, RecTypes.AGN_SHP);
|
||
|
||
for (let i = 0; i < items.length; i++) {
|
||
const record = DataUtil.readShpRecord(items[i]);
|
||
if (!record) continue;
|
||
|
||
if (record.timeAdv > 0.0) {
|
||
// Shift the lat lon coordinate according to time Advance (system lag)
|
||
latlon = geoUtil.projectLatLong(record.lat, record.lon, record.grSpeed, record.head, record.timeAdv);
|
||
record.lat = latlon.lat; record.lon = latlon.lon;
|
||
}
|
||
record["fileId"] = fileId;
|
||
records.push(record);
|
||
|
||
// For compensation with a day when gpsTime passing mid-night (rolloff)
|
||
if (i > 0 && records.length > 1 && timeOffset === 0
|
||
&& ((records[records.length - 1].gpsTime - records[records.length - 2].gpsTime) < 0
|
||
&& Math.abs(records[records.length - 1].gpsTime - records[records.length - 2].gpsTime) >= 80000))
|
||
record.adjustGpsTime(timeOffset);
|
||
|
||
// Incremental distance accumulation with time-gap and outlier guards
|
||
if (prevRecUTM_X !== null && prevRecUTM_Y !== null && utils.isNumber(record.utmX) && utils.isNumber(record.utmY) && prevRecTime !== null) {
|
||
let dt = record.gpsTime - prevRecTime;
|
||
if (dt < 0 && Math.abs(dt) >= 80000) dt = (86400 - prevRecTime) + record.gpsTime;
|
||
if (dt > 0 && dt <= 120) {
|
||
const segDist = Math.hypot(record.utmX - prevRecUTM_X, record.utmY - prevRecUTM_Y);
|
||
if (segDist <= 1000) {
|
||
totalFlightLen += segDist;
|
||
if ((prevRecSprStat && prevRecSprStat > 0) || (record.sprayStat && record.sprayStat > 0)) {
|
||
totalSprLen += segDist;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
prevRecUTM_X = record.utmX; prevRecUTM_Y = record.utmY; prevRecTime = record.gpsTime; prevRecSprStat = record.sprayStat;
|
||
|
||
// FOR DEDUG
|
||
// fs.appendFileSync(`./${fileName}.csv`, `${appDetail.gpsTime}, ${appDetail.sprayStat}, ${appDetail.llnum}, ${appDetail.lat}, ${appDetail.lon}` + endOfLine);
|
||
if (dataFile['sprayOn']) {
|
||
if (record.lhaReq > 0) {
|
||
totalSprayRecs++;
|
||
totalAppRates += record.lhaReq;
|
||
}
|
||
if (record.sprayStat === 3) {
|
||
prevUTM_X = record.utmX;
|
||
prevUTM_Y = record.utmY;
|
||
prevSwath = record.swath;
|
||
prevLine = record.llnum;
|
||
}
|
||
else {
|
||
({ appliedRate, sprMatsUnit } = getAppliedRate(record, rateInfo, rateInfo.recType === RecTypes.AGN_SHP));
|
||
|
||
if (prevLine === record.llnum) {
|
||
sprayedSeg = Math.hypot(record.utmX - prevUTM_X, record.utmY - prevUTM_Y) * prevSwath;
|
||
if (sprayedSeg) {
|
||
totalSprays += sprayedSeg;
|
||
if (appliedRate > 0) {
|
||
totalSprMats += (sprayedSeg * CVCST.SM2HA) * appliedRate;
|
||
}
|
||
}
|
||
}
|
||
prevUTM_X = record.utmX;
|
||
prevUTM_Y = record.utmY;
|
||
prevSwath = record.swath;
|
||
prevLine = record.llnum;
|
||
}
|
||
}
|
||
}
|
||
|
||
let fileDataInfo = null;
|
||
if (records.length) {
|
||
let avgRate = 0;
|
||
if (totalSprayRecs > 0) {
|
||
avgRate = totalAppRates / totalSprayRecs;
|
||
} else if (rateInfo && rateInfo.appRate > 0) {
|
||
avgRate = utils.toMetricRate(rateInfo.appRate, rateInfo.rateUnit).value;
|
||
}
|
||
fileDataInfo = {
|
||
records: records,
|
||
totalSprayMat: totalSprMats,
|
||
totalSprayMatUnit: sprMatsUnit,
|
||
totalSprayed: totalSprays,
|
||
totalSprLength: totalSprLen,
|
||
totalFlightLength: totalFlightLen,
|
||
avgRate: avgRate
|
||
}
|
||
}
|
||
items = null;
|
||
return cb(null, fileDataInfo);
|
||
});
|
||
}
|
||
|
||
function readSatLogAsc(dataFile, fileId, cb) {
|
||
/*
|
||
const hdrs = ['Time', undefined, undefined, 'Alt', undefined, undefined, undefined, undefined, undefined, 'Date', undefined, 'DOP', undefined, undefined,
|
||
undefined, 'Hdg', 'Lat', 'Lon', 'RHumi', 'Speed', 'Spray', undefined, 'SU', undefined, 'Temperature', undefined, undefined, undefined, undefined, 'X-Track'];
|
||
*/
|
||
let records = [], totalSprLength = 0, totalFlightLength = 0, currStat = -999, prevStat = -999, curLonLat = turf.point([0, 0]), prevLonLat = turf.point([0, 0]), segLength = 0;
|
||
let prevTime = null;
|
||
fs.createReadStream(dataFile)
|
||
.pipe(csv.parse({ headers: true, ignoreEmpty: true }))
|
||
.on('error', err => {
|
||
if (cb) cb(err);
|
||
})
|
||
.on('data', rowRec => {
|
||
const record = new WorkRecord();
|
||
record.readSLAscRecord(rowRec);
|
||
if (record.gpsTime && /(?:[01]\d|2[0123]):(?:[012345]\d):(?:[012345]\d)/.test(record.gpsTime)) {
|
||
record.decodeSL();
|
||
record["fileId"] = fileId;
|
||
records.push(record);
|
||
|
||
curLonLat.geometry.coordinates = [record.lon, record.lat];
|
||
currStat = record.sprayStat;
|
||
if (prevStat != -999 && prevTime !== null) {
|
||
// compute time gap (prevTime is in seconds)
|
||
let dt = record.gpsTime - prevTime;
|
||
if (dt < 0 && Math.abs(dt) >= 80000) dt = (86400 - prevTime) + record.gpsTime;
|
||
if (dt > 0 && dt <= 120) {
|
||
segLength = turf.distance(prevLonLat, curLonLat, { units: "meters" });
|
||
if (segLength <= 1000) {
|
||
totalFlightLength += segLength;
|
||
if (prevStat > 0 && currStat > 0 || record.sprayStat != prevStat) {
|
||
totalSprLength += segLength;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
prevStat = record.sprayStat;
|
||
prevTime = record.gpsTime;
|
||
prevLonLat.geometry.coordinates = [record.lon, record.lat];
|
||
}
|
||
})
|
||
.on('end', rowCount => {
|
||
curLonLat = prevLonLat = null;
|
||
|
||
const fileDataInfo = {
|
||
records: records,
|
||
totalSprayed: 0,
|
||
totalSprLength: totalSprLength,
|
||
totalFlightLength: totalFlightLength,
|
||
avgRate: 0
|
||
}
|
||
if (cb) cb(null, fileDataInfo);
|
||
});
|
||
}
|
||
|
||
function closeOnErr(err) {
|
||
if (!err) return false;
|
||
try {
|
||
debug("[AMQP] error:", err);
|
||
amqpConn.close();
|
||
} catch (error) {
|
||
}
|
||
return true;
|
||
}
|
||
|
||
function gc() {
|
||
if (global.gc) global.gc();
|
||
}
|