700 lines
28 KiB
JavaScript
700 lines
28 KiB
JavaScript
/**
|
||
* SatLoc Application Processor - Handles log grouping and application creation logic
|
||
* Similar to Job Worker pattern but designed for SatLoc log files
|
||
*/
|
||
|
||
const debug = require('debug')('agm:satloc-processor');
|
||
const { enhancedRunInTransaction, runWithSessionOrTransaction } = require('./mongo_enhanced');
|
||
const JobAssign = require('../model/job_assign');
|
||
const Application = require('../model/application');
|
||
const ApplicationFile = require('../model/application_file');
|
||
const ApplicationDetail = require('../model/application_detail');
|
||
const appDateTime = require('./application_datetime');
|
||
const { SatLocLogParser } = require('./satloc_log_parser');
|
||
const { AppStatus, AppProStatus, AssignStatus, UserTypes, FCTypes, RateUnits } = require('./constants');
|
||
const { JobUpdateOp } = require('./job_constants');
|
||
const { AppInputError } = require('./app_error');
|
||
const env = require('./env');
|
||
const path = require('path');
|
||
const fs = require('fs').promises;
|
||
const _ = require('lodash');
|
||
|
||
const MAX_TIME_DIFF = 120;
|
||
|
||
class SatLocApplicationProcessor {
|
||
constructor(options = {}) {
|
||
this.options = {
|
||
batchSize: options.batchSize || 1000,
|
||
enableRetryLogic: options.enableRetryLogic !== false,
|
||
...options
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Process a SatLoc log file and create/update Application records with proper grouping
|
||
* @param {Object} logFileData - Log file information
|
||
* @param {Object} contextData - Context data (jobId, userId, etc.)
|
||
* @returns {Promise<Object>} Processing results
|
||
*/
|
||
async processLogFile(logFileData, contextData = {}) {
|
||
debug(`Processing SatLoc log file: ${logFileData.filePath}`);
|
||
|
||
let result = {
|
||
success: false,
|
||
matchedJobs: [],
|
||
numberOfJobGroups: 0,
|
||
applications: [],
|
||
applicationFiles: [],
|
||
appFileId: null, // For backward compatibility - first application file ID
|
||
totalDetails: 0,
|
||
error: null
|
||
};
|
||
|
||
try {
|
||
// 1. Use SatLocLogParser directly with integrated calculations AND grouping
|
||
const parserOptions = {
|
||
skipUnknownRecords: true,
|
||
validateChecksums: true
|
||
};
|
||
|
||
// Only add maxPositionsPerJob if it's set in ENV
|
||
if (env.SATLOC_MAX_POSITIONS_PER_JOB !== undefined) {
|
||
parserOptions.maxPositionsPerJob = env.SATLOC_MAX_POSITIONS_PER_JOB;
|
||
}
|
||
|
||
const parser = new SatLocLogParser(parserOptions);
|
||
|
||
const parseResults = await parser.parseFile(logFileData.filePath, contextData);
|
||
if (!parseResults.success) {
|
||
throw new Error(`Failed to parse log file: ${parseResults.error}`);
|
||
}
|
||
|
||
// 2. Use pre-grouped job groups from parsing (no duplication)
|
||
const jobGroups = parseResults.jobGroups;
|
||
result.numberOfJobGroups = Object.keys(jobGroups).length;
|
||
result.statistics = parseResults.statistics;
|
||
debug(`Found ${result.numberOfJobGroups} job groups in log file`);
|
||
const assignedSet = _.groupBy(contextData.taskInfo.assignments, it => it.jobName || it.job._id);
|
||
let matchedAssign = null;
|
||
|
||
// 3. Process each job group separately by jobId and aircraftId
|
||
for (const [satlocJobId, applicationDetails] of Object.entries(jobGroups)) {
|
||
// Check if we have any assignments before processing
|
||
const assignments = contextData.taskInfo?.assignments;
|
||
if (!assignments || assignments.length === 0) {
|
||
debug(`No assignments available for job group: ${satlocJobId} => skip processing.`);
|
||
continue;
|
||
}
|
||
|
||
// Make it configurable via env param to toggle off later when needed
|
||
if (env.SATLOC_1ST_ASSIGNMENT_ALWAYS_MATCH) {
|
||
matchedAssign = assignments[0]; // TODO: To be removed later. For testing, assume single assignment
|
||
} else {
|
||
// Only process if the current job group has a matching, (.job (the or ._id) or jobName), assignment
|
||
if (!(matchedAssign = assignedSet[satlocJobId])) {
|
||
debug(`No assignment found for job group: ${satlocJobId} => skip processing !`);
|
||
continue;
|
||
}
|
||
}
|
||
|
||
const jobGroupResult = await enhancedRunInTransaction(async (session) => {
|
||
// Create consolidated job group context
|
||
const jobGroupContext = {
|
||
taskInfo: contextData.taskInfo,
|
||
matchedAssign,
|
||
applicationDetails,
|
||
metadata: { ...contextData.metadata, ...parseResults.metadata } || {},
|
||
geoInfo: { utmZone: parseResults.utmZone, boundingBox: parseResults.boundingBox },
|
||
parseResults
|
||
};
|
||
return await this.processJobGroup(jobGroupContext, session);
|
||
});
|
||
|
||
// Collect results from job group processing
|
||
if (jobGroupResult) {
|
||
if (jobGroupResult.application) {
|
||
result.applications.push(jobGroupResult.application);
|
||
result.matchedJobs.push({
|
||
assignId: matchedAssign._id, // Job assignment reference
|
||
jobId: jobGroupResult.application.jobId,
|
||
confidence: 1.0 // Full confidence for direct assignment match
|
||
});
|
||
}
|
||
if (jobGroupResult.applicationFile) {
|
||
result.applicationFiles.push(jobGroupResult.applicationFile);
|
||
// Set first appFileId for backward compatibility
|
||
if (!result.appFileId) {
|
||
result.appFileId = jobGroupResult.applicationFile._id;
|
||
}
|
||
}
|
||
result.totalDetails += jobGroupResult.detailsCount || 0;
|
||
}
|
||
}
|
||
|
||
result.success = true;
|
||
} catch (error) {
|
||
debug(`Error processing log file: ${error.message}`);
|
||
result.error = error.message;
|
||
throw error;
|
||
}
|
||
|
||
// Multiple jobs: return new format
|
||
return {
|
||
...result,
|
||
multiJob: result.numberOfJobGroups > 1,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Process a single job group with calculations and database operations
|
||
* @param {Object} groupContext - Consolidated context containing all job group data
|
||
* @param {Object} session - MongoDB session
|
||
* @returns {Promise<Object>} Processing results for this job group
|
||
*/
|
||
async processJobGroup(groupContext, session) {
|
||
const { matchedAssign, applicationDetails, taskInfo, metadata, geoInfo, parseResults } = groupContext;
|
||
debug(`Processing job group: ${matchedAssign?.job?._id} with ${applicationDetails.length} details`);
|
||
|
||
// Check the assignment still exists and valid for processing against the db
|
||
const freshAssign = await JobAssign.findById(matchedAssign._id, {}, { lean: true }).session(session);
|
||
if (!freshAssign) {
|
||
debug(`No valid assignment found for job group: ${matchedAssign.job._id} => skip processing.`);
|
||
return;
|
||
}
|
||
|
||
const finalJobId = matchedAssign.job._id;
|
||
|
||
// Extract user ID from assignment - this is the pilot/operator who will perform the application
|
||
const assignedUserId = matchedAssign.user || freshAssign.user;
|
||
|
||
// Create context for this job group
|
||
const appContext = {
|
||
jobId: finalJobId,
|
||
fileName: taskInfo.logFileName,
|
||
fileSize: metadata.fileSize,
|
||
userId: assignedUserId // Use the assigned user (pilot) from the job assignment
|
||
};
|
||
|
||
// Create new Application for this job group (1:1 mapping per job group)
|
||
const application = await this.createApplication(appContext, session);
|
||
|
||
// Process each application detail in this job group (with needed calculations: UTM coords, spray stats, etc.).
|
||
const processedDetails = [];
|
||
|
||
// Initialize aggregation variables for this job group
|
||
let totalSprayTime = 0, totalFlightTime = 0, totalSprayed = 0, totalSprayMat = 0;
|
||
let totalSprayLength = 0;
|
||
let rateSum = 0, rateCount = 0; // For computing avgAppRate (mean lhaApp/lhaReq across spray-on records)
|
||
let spraySegments = [];
|
||
let startDateTime = null, endDateTime = null;
|
||
|
||
// Tracking variables for calculations
|
||
let prevTime = -999, prevSprTime = -999, prevUTM_X, prevUTM_Y;
|
||
let prevSprayStat = -1; // Track previous spray status for transition detection
|
||
let currentSegment = null;
|
||
|
||
// Load UTM conversion library
|
||
const utmMod = await import('@mickeyjohn/geodesy/utm.js');
|
||
const LatLonUTM = utmMod.LatLon;
|
||
const refUTMZone = geoInfo.utmZone;
|
||
|
||
debug(`Processing ${applicationDetails.length} application details for job group ${finalJobId}`);
|
||
debug(`Using UTM zone: ${refUTMZone.zoneNumber}${refUTMZone.hemisphere} for coordinate conversion`);
|
||
|
||
// Create ApplicationFile with calculated data, and ref for application details for this job group
|
||
// Normalize metadata according to DATA_FORMAT_NOTES.md
|
||
const { DataTypes, MatTypes } = require('./constants');
|
||
|
||
const normalizedMeta = {
|
||
// Normalized fields (common format for both AgNav and SatLoc)
|
||
type: DataTypes.SATLOC, // Data source type
|
||
matType: metadata.fcType === FCTypes.DRY ? MatTypes.DRY : MatTypes.WET, // Material type: 'wet' or 'dry'
|
||
operator: metadata.pilotName || null, // Pilot name (from System Setup record Type 100)
|
||
fcName: metadata.fcName || null, // Flow controller name (from Controller Type record Type 46)
|
||
|
||
// All original metadata fields (preserved completely)
|
||
...metadata,
|
||
|
||
// Additional processing metadata
|
||
recordCount: applicationDetails.length,
|
||
parseStats: parseResults.statistics,
|
||
version: parseResults.headerInfo?.version,
|
||
utmZone: parseResults.utmZone,
|
||
bbox: parseResults.boundingBox
|
||
};
|
||
|
||
const appFile = await this.createApplicationFile(application._id, appContext.fileName, taskInfo.uploadedDate,
|
||
normalizedMeta,
|
||
session
|
||
);
|
||
|
||
// Process each detail with calculations
|
||
for (let i = 0; i < applicationDetails.length; i++) {
|
||
const record = applicationDetails[i];
|
||
|
||
// Convert lat/lon to UTM coordinates using reference zone
|
||
if (record.lat && record.lon && !isNaN(record.lat) && !isNaN(record.lon) && refUTMZone.zoneNumber) {
|
||
try {
|
||
const latLonObj = new LatLonUTM(record.lat, record.lon);
|
||
const utm = latLonObj.toUtm(refUTMZone.zoneNumber, refUTMZone.hemisphere);
|
||
record.utmX = utm.easting;
|
||
record.utmY = utm.northing;
|
||
} catch (error) {
|
||
debug(`UTM conversion error for record ${i}: ${error.message}`);
|
||
record.utmX = 0;
|
||
record.utmY = 0;
|
||
}
|
||
} else {
|
||
debug(`Skipping record ${i} due to invalid coordinates`);
|
||
continue; // Skip records without valid coordinates
|
||
}
|
||
|
||
// Track overall time range for this job group
|
||
if (!startDateTime || record.gpsTime < startDateTime) {
|
||
startDateTime = record.gpsTime;
|
||
}
|
||
if (!endDateTime || record.gpsTime > endDateTime) {
|
||
endDateTime = record.gpsTime;
|
||
}
|
||
|
||
// Calculate total flight time
|
||
if (prevTime !== -999 && prevTime !== record.gpsTime) {
|
||
let timeDif = record.gpsTime - prevTime;
|
||
// Only count reasonable time differences
|
||
if (timeDif > 0 && timeDif <= MAX_TIME_DIFF) {
|
||
totalFlightTime += timeDif;
|
||
}
|
||
}
|
||
prevTime = record.gpsTime;
|
||
|
||
// Detect spray segment transitions based on sprayStat changes
|
||
const curSprayStat = record.sprayStat || 0;
|
||
|
||
// Handle first record: if spray starts ON, create initial segment
|
||
if (prevSprayStat === -1) {
|
||
if (curSprayStat > 0) {
|
||
env && !env.PRODUCTION && debug(`Job group ${finalJobId}: Log starts with spray ON at time ${record.gpsTime}, creating initial segment`);
|
||
currentSegment = {
|
||
startTime: record.gpsTime,
|
||
endTime: record.gpsTime,
|
||
startLat: record.lat,
|
||
startLon: record.lon,
|
||
endLat: record.lat,
|
||
endLon: record.lon,
|
||
distance: 0,
|
||
area: 0,
|
||
points: []
|
||
};
|
||
}
|
||
} else {
|
||
// Check for spray transitions: OFF to ON (start segment) or ON to OFF (end segment)
|
||
// Spray turning ON (previous OFF, current ON) - start new segment
|
||
if (prevSprayStat === 0 && curSprayStat > 0) {
|
||
env && !env.PRODUCTION && debug(`Job group ${finalJobId}: Spray ON detected at time ${record.gpsTime}, starting new segment`);
|
||
currentSegment = {
|
||
startTime: record.gpsTime,
|
||
endTime: record.gpsTime,
|
||
startLat: record.lat,
|
||
startLon: record.lon,
|
||
endLat: record.lat,
|
||
endLon: record.lon,
|
||
distance: 0,
|
||
area: 0,
|
||
points: []
|
||
};
|
||
}
|
||
|
||
// Spray turning OFF (previous ON, current OFF) - close current segment
|
||
if (prevSprayStat > 0 && curSprayStat === 0) {
|
||
env && !env.PRODUCTION && debug(`Job group ${finalJobId}: Spray OFF detected at time ${record.gpsTime}, closing segment`);
|
||
if (currentSegment) {
|
||
spraySegments.push(currentSegment);
|
||
currentSegment = null;
|
||
}
|
||
}
|
||
}
|
||
|
||
// Calculate spray time and area when spray is active
|
||
if (curSprayStat > 0) {
|
||
// ALWAYS add spray ON points to current segment
|
||
if (currentSegment) {
|
||
currentSegment.endTime = record.gpsTime;
|
||
currentSegment.endLat = record.lat;
|
||
currentSegment.endLon = record.lon;
|
||
|
||
// Add EVERY spray ON point to the segment
|
||
currentSegment.points.push({
|
||
lat: record.lat,
|
||
lon: record.lon,
|
||
gpsTime: record.gpsTime,
|
||
sprayStat: record.sprayStat
|
||
});
|
||
}
|
||
|
||
// Calculate spray time and area calculations
|
||
if (prevUTM_X !== undefined && prevUTM_Y !== undefined && prevSprayStat > 0) {
|
||
const timeDif = record.gpsTime - prevSprTime;
|
||
if (timeDif > 0 && timeDif <= MAX_TIME_DIFF) {
|
||
totalSprayTime += timeDif;
|
||
}
|
||
|
||
const distance = Math.hypot(record.utmX - prevUTM_X, record.utmY - prevUTM_Y);
|
||
if (distance > 0 && record.swath > 0) {
|
||
const swathArea = distance * record.swath;
|
||
totalSprayed += swathArea;
|
||
totalSprayLength += distance;
|
||
|
||
// Track spray material usage and accumulate prescribed rate for appRate average
|
||
const recRate = record.lhaApp || record.lhaReq || 0;
|
||
if (recRate > 0) {
|
||
totalSprayMat += (swathArea * recRate) / 10000; // Convert m² × L/ha → L
|
||
rateSum += recRate;
|
||
rateCount += 1;
|
||
}
|
||
|
||
// Update segment distance and area
|
||
if (currentSegment) {
|
||
currentSegment.distance += distance;
|
||
currentSegment.area += swathArea;
|
||
}
|
||
}
|
||
}
|
||
|
||
prevSprTime = record.gpsTime;
|
||
// Update UTM position for next iteration (only when spray is ON)
|
||
prevUTM_X = record.utmX;
|
||
prevUTM_Y = record.utmY;
|
||
}
|
||
|
||
// Update previous spray status for next iteration (MOVED OUTSIDE spray condition)
|
||
// This ensures ALL transitions (ON→OFF, OFF→ON) are properly tracked
|
||
prevSprayStat = curSprayStat;
|
||
|
||
// Create processed detail with calculations (will be linked to file after ApplicationFile creation)
|
||
const processedDetail = {
|
||
...record,
|
||
fileId: appFile._id
|
||
};
|
||
processedDetails.push(processedDetail);
|
||
}
|
||
|
||
// Convert units (like job worker)
|
||
totalSprayed = totalSprayed * 1E-4; // Convert m² to hectares
|
||
|
||
// Compute average prescribed application rate from per-record lhaApp/lhaReq values.
|
||
// This mirrors job_worker.js readNTFile() which uses mean(lhaReq) across spray-ON records.
|
||
const avgAppRate = rateCount > 0 ? rateSum / rateCount : 0;
|
||
|
||
// Compute flow accuracy only when all three source fields are positive.
|
||
const flowAccuracyPct = avgAppRate > 0 && totalSprayed > 0 && totalSprayMat > 0
|
||
? Math.round((totalSprayMat / totalSprayed / avgAppRate) * 10000) / 100
|
||
: undefined;
|
||
|
||
debug(`Job group ${finalJobId} calculated totals - Flight: ${totalFlightTime} s, Spray: ${totalSprayTime} s, Area: ${totalSprayed} Ha, Material: ${totalSprayMat} L/Kg, AppRate: ${avgAppRate} L/ha, FlowAccuracy: ${flowAccuracyPct ?? 'n/a'}%, Segments: ${spraySegments.length}`);
|
||
|
||
const sprayMatUnit = metadata.fcType === FCTypes.LIQUID ? RateUnits.LIT_PER_HA : RateUnits.KG_PER_HA;
|
||
|
||
// Save ApplicationDetails in batches for efficiency
|
||
// await this.saveApplicationDetails(processedDetails, session);
|
||
// ==============================================================
|
||
// Prepare job group statistics for Application update
|
||
const jobGroupStats = {
|
||
totalSprayTime,
|
||
totalFlightTime,
|
||
totalSprayed,
|
||
totalSprayMat,
|
||
totalSprayMatUnit: sprayMatUnit,
|
||
totalSprayLength,
|
||
startDateTime,
|
||
endDateTime,
|
||
spraySegments
|
||
};
|
||
const referenceDetail = processedDetails.find(detail => typeof detail.lat === 'number' && typeof detail.lon === 'number');
|
||
const dateFields = appDateTime.buildApplicationDateFields({
|
||
startDateTime: jobGroupStats.startDateTime ? new Date(jobGroupStats.startDateTime * 1000).toISOString() : null,
|
||
endDateTime: jobGroupStats.endDateTime ? new Date(jobGroupStats.endDateTime * 1000).toISOString() : null,
|
||
latitude: referenceDetail ? referenceDetail.lat : null,
|
||
longitude: referenceDetail ? referenceDetail.lon : null
|
||
});
|
||
|
||
// Batch insert ApplicationDetails for this job group
|
||
if (processedDetails.length > 0) {
|
||
await this.saveApplicationDetails(processedDetails, session);
|
||
}
|
||
|
||
// Update Application with calculated totals for this job group
|
||
await ApplicationFile.updateOne({ _id: appFile._id },
|
||
{
|
||
$set: {
|
||
totalSprLength: jobGroupStats.totalSprayLength || 0,
|
||
totalSprayTime: jobGroupStats.totalSprayTime || 0,
|
||
totalFlightTime: jobGroupStats.totalFlightTime || 0,
|
||
totalSprayed: jobGroupStats.totalSprayed || 0,
|
||
totalSprayMat: jobGroupStats.totalSprayMat || 0,
|
||
totalSprayMatUnit: jobGroupStats.totalSprayMatUnit || sprayMatUnit,
|
||
}
|
||
},
|
||
{ session }
|
||
);
|
||
|
||
// Update Application with calculated totals for this job group
|
||
await Application.updateOne({ _id: application._id },
|
||
{
|
||
$set: {
|
||
status: AppStatus.DONE,
|
||
proStatus: processedDetails.length > 0 ? AppProStatus.WITH_DATA : AppProStatus.NO_DATA,
|
||
totalSprayTime: jobGroupStats.totalSprayTime || 0,
|
||
totalFlightTime: jobGroupStats.totalFlightTime || 0,
|
||
totalSprayed: jobGroupStats.totalSprayed || 0,
|
||
totalSprayMat: jobGroupStats.totalSprayMat || 0,
|
||
totalSprayMatUnit: jobGroupStats.totalSprayMatUnit || sprayMatUnit,
|
||
totalSprLength: jobGroupStats.totalSprayLength || 0,
|
||
appRate: avgAppRate,
|
||
...(flowAccuracyPct !== undefined ? { flowAccuracyPct } : {}),
|
||
startDateTime: jobGroupStats.startDateTime ? new Date(jobGroupStats.startDateTime * 1000).toISOString() : null,
|
||
endDateTime: jobGroupStats.endDateTime ? new Date(jobGroupStats.endDateTime * 1000).toISOString() : null,
|
||
startDateTimeUTC: dateFields.startDateTimeUTC,
|
||
endDateTimeUTC: dateFields.endDateTimeUTC,
|
||
utcOffset: dateFields.utcOffset,
|
||
updateDate: new Date()
|
||
}
|
||
},
|
||
{ session }
|
||
);
|
||
|
||
debug(`Completed job group ${finalJobId}: ${processedDetails.length} details processed, ${jobGroupStats.totalSprayed.toFixed(2)}ha sprayed, ${jobGroupStats.spraySegments.length} segments`);
|
||
|
||
// Finalize last segment for this job group
|
||
if (currentSegment) {
|
||
const lastRecord = applicationDetails[applicationDetails.length - 1];
|
||
|
||
// Only finalize if segment has meaningful data (consecutive spray ON points)
|
||
if (currentSegment.points.length > 0) {
|
||
spraySegments.push(currentSegment);
|
||
debug(`Job group ${finalJobId}: Finalized last spray segment with ${currentSegment.points.length} consecutive spray ON points`);
|
||
} else {
|
||
debug(`Job group ${finalJobId}: Discarded empty last segment - no spray ON points`);
|
||
}
|
||
}
|
||
|
||
// Return results for this job group including application and applicationFile for tracker update
|
||
return {
|
||
success: true,
|
||
application, // Application document for this job group
|
||
applicationFile: appFile, // ApplicationFile document for this job group
|
||
detailsCount: processedDetails.length,
|
||
skipped: false
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Find matching assignment where
|
||
* @param {string} aircraftId - Aircraft ID to match against user's partnerInfo.partnerAircraftId, if needed
|
||
* @param {string} satlocJobId - SatLoc job ID for additional matching, using now
|
||
* @param {Object} session - MongoDB session
|
||
* @returns {Promise<Object>} Assignment with populated job or null
|
||
*/
|
||
async findJobAssignment(aircraftId, satlocJobId, session = null) {
|
||
let recentAssignment = null;
|
||
try {
|
||
debug(`Looking for assignment with partnerAircraftId: ${aircraftId}, satlocJobId: ${satlocJobId}`);
|
||
|
||
if (!aircraftId || !satlocJobId) AppInputError.throw();
|
||
|
||
const validAssignments = await JobAssign.find({ status: { $in: [AssignStatus.UPLOADED] }, jobName: satlocJobId }, { lean: true })
|
||
.populate({ path: 'job', select: '_id status' })
|
||
.populate({ path: 'partnerInfo.partner', model: UserTypes.PARTNER, select: 'partnerAircraftId' })
|
||
.session(session);
|
||
|
||
if (validAssignments.length > 0) {
|
||
recentAssignment = validAssignments[0];
|
||
debug(`Found recent assignment: ${recentAssignment.job} (${recentAssignment.job.status}) for user: ${recentAssignment.userId.username}`);
|
||
if (recentAssignment.userId.partnerInfo?.partnerAircraftId !== aircraftId) {
|
||
debug(`Warning: partnerAircraftId mismatch: expected ${aircraftId}, found ${recentAssignment.userId.partnerInfo?.partnerAircraftId}`);
|
||
}
|
||
}
|
||
return recentAssignment;
|
||
} catch (error) {
|
||
debug(`Error finding assignment by partner aircraft ID: ${error.message}`);
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Handle retry logic for existing files - resets data and reprocesses
|
||
* @param {string} filePath - Path to the log file
|
||
* @param {Object} contextData - Context data
|
||
* @returns {Promise<Object>} Processing results
|
||
*/
|
||
async retryLogFile(filePath, contextData = {}) {
|
||
debug(`Retrying SatLoc log file: ${filePath}`);
|
||
|
||
try {
|
||
/* NOTE: Use markedDelete flag instead of deleting data for the case of accesive old application details
|
||
=> Cleanup worker will handle actual data deletion later
|
||
*/
|
||
await enhancedRunInTransaction(async (session) => {
|
||
// Find existing ApplicationFile by the log file name
|
||
const logFileName = contextData.taskInfo?.logFileName || path.basename(filePath);
|
||
await ApplicationFile.updateMany({ fileName: logFileName }, { $set: { markedDelete: true } }).session(session);
|
||
});
|
||
|
||
// Now process the file normally
|
||
return await this.processLogFile({ filePath }, contextData);
|
||
|
||
} catch (error) {
|
||
debug(`Error retrying log file: ${error.message}`);
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Find existing Application or create new one - simplified for 1:1 file-to-application mapping
|
||
* Each log file gets its own Application record, similar to Job Worker pattern
|
||
* @param {Object} ctxData - Context data (jobId, fileName, etc.)
|
||
* @param {Object} session - MongoDB session
|
||
* @returns {Promise<Object>} Application document
|
||
*/
|
||
async createApplication(ctxData, session) {
|
||
// Always create a new Application for each log file
|
||
const newApplication = new Application({
|
||
jobId: ctxData.jobId,
|
||
fileName: ctxData.fileName,
|
||
savedFilename: ctxData.fileName,
|
||
fileSize: ctxData.fileSize,
|
||
updateOp: JobUpdateOp.DATA_ONLY,
|
||
status: AppStatus.CREATED,
|
||
createdDate: new Date(),
|
||
updateDate: new Date(),
|
||
errorMsg: null,
|
||
cid: null,
|
||
byUser: ctxData.userId || null, // User ID from job assignment (vehicle/partner aircraft)
|
||
byImport: true // Mark as import since this comes from partner sync, not direct user upload, needs review later
|
||
});
|
||
|
||
const savedApplication = await newApplication.save({ session });
|
||
debug(`Created new application ${savedApplication._id} for SatLoc log file: ${ctxData.fileName}`);
|
||
|
||
return savedApplication;
|
||
}
|
||
|
||
/**
|
||
* Create ApplicationFile record for individual log file
|
||
* @param {ObjectId} applicationId - Application ID
|
||
* @param {string} fileName - Log file name
|
||
* @param {string} uploadedDate - Uploaded date as string (timezone unknown, stored as-is from partner API)
|
||
* @param {Object} logMetadata - Normalized metadata object with fields:
|
||
* - type: 'satloc' | 'agnav' - Data source type
|
||
* - matType: 'wet' | 'dry' - Material type (normalized from fcType)
|
||
* - operator: string - Pilot name (from pilotName)
|
||
* - fcName: string - Flow controller name
|
||
* - aircraftId: string - Aircraft identifier (SatLoc-specific)
|
||
* - jobId: string - Job identifier (SatLoc-specific)
|
||
* - fcType: FCTypes.LIQUID | FCTypes.DRY - Original flow controller type
|
||
* - recordCount: number - Number of application details
|
||
* - parseStats: Object - Parser statistics
|
||
* - version: string - Log file version
|
||
* - utmZone: Object - UTM zone information
|
||
* - bbox: Array - Bounding box [minLon, minLat, maxLon, maxLat]
|
||
* @param {Object} session - MongoDB session
|
||
* @returns {Promise<Object>} ApplicationFile document
|
||
*/
|
||
async createApplicationFile(applicationId, fileName, uploadedDate, logMetadata, session) {
|
||
debug(`Creating ApplicationFile for application ${applicationId}, file: ${fileName}`);
|
||
|
||
const applicationFile = new ApplicationFile({
|
||
appId: applicationId,
|
||
name: fileName,
|
||
agn: (uploadedDate && this.generateAGNFromISO(uploadedDate)),
|
||
meta: logMetadata || {}
|
||
});
|
||
|
||
const savedFile = await applicationFile.save({ session });
|
||
debug(`Created application file ${savedFile._id} for log ${fileName}`);
|
||
|
||
return savedFile;
|
||
}
|
||
|
||
/**
|
||
* Generate AGN (AgNav identifier) from timestamp
|
||
*/
|
||
generateAGN(timestamp) {
|
||
if (!timestamp) return '0000000000';
|
||
|
||
const date = new Date(timestamp);
|
||
const year = date.getFullYear().toString().substr(-2);
|
||
const month = String(date.getMonth() + 1).padStart(2, '0');
|
||
const day = String(date.getDate()).padStart(2, '0');
|
||
const hour = String(date.getHours()).padStart(2, '0');
|
||
const minute = String(date.getMinutes()).padStart(2, '0');
|
||
|
||
return `${year}${month}${day}${hour}${minute}`;
|
||
}
|
||
|
||
|
||
/**
|
||
* Generate AGN (AgNav File Group Name) from ISO 8601 timestamp
|
||
* @param {string} isoString - ISO 8601 timestamp, i.e. "2023-10-05T14:30:00Z"
|
||
* @returns {string} - AGN string
|
||
*/
|
||
generateAGNFromISO(isoString) {
|
||
if (!isoString) return '0000000000';
|
||
|
||
const date = new Date(isoString);
|
||
if (isNaN(date.getTime())) return '0000000000';
|
||
|
||
const year = date.getFullYear().toString().substr(-2);
|
||
const month = String(date.getMonth() + 1).padStart(2, '0');
|
||
const day = String(date.getDate()).padStart(2, '0');
|
||
const hour = String(date.getHours()).padStart(2, '0');
|
||
const minute = String(date.getMinutes()).padStart(2, '0');
|
||
|
||
return `${year}${month}${day}${hour}${minute}`;
|
||
}
|
||
|
||
/**
|
||
* Batch insert application details with proper error handling
|
||
*/
|
||
async saveApplicationDetails(applicationDetails, session) {
|
||
if (!applicationDetails || applicationDetails.length === 0) {
|
||
return { inserted: 0 };
|
||
}
|
||
|
||
const batchSize = this.options.batchSize;
|
||
let totalInserted = 0;
|
||
|
||
for (let i = 0; i < applicationDetails.length; i += batchSize) {
|
||
const batch = applicationDetails.slice(i, i + batchSize);
|
||
|
||
try {
|
||
const result = await ApplicationDetail.insertMany(batch, {
|
||
ordered: false,
|
||
lean: true,
|
||
session
|
||
});
|
||
totalInserted += result.length;
|
||
|
||
env && !env.PRODUCTION && debug(`Inserted batch ${Math.floor(i / batchSize) + 1}: ${result.length} records`);
|
||
|
||
} catch (error) {
|
||
debug(`Error inserting batch: ${error.message}`);
|
||
// Continue with next batch on error
|
||
}
|
||
}
|
||
|
||
return { inserted: totalInserted };
|
||
}
|
||
|
||
/**
|
||
* Get processing statistics
|
||
*/
|
||
getStatistics() {
|
||
return {
|
||
batchSize: this.options.batchSize,
|
||
enableRetryLogic: this.options.enableRetryLogic
|
||
};
|
||
}
|
||
}
|
||
|
||
module.exports = SatLocApplicationProcessor;
|