333 lines
14 KiB
JavaScript
333 lines
14 KiB
JavaScript
"use strict";
|
|
var __decorate = (this && this.__decorate) || function (decorators, target, key, desc) {
|
|
var c = arguments.length, r = c < 3 ? target : desc === null ? desc = Object.getOwnPropertyDescriptor(target, key) : desc, d;
|
|
if (typeof Reflect === "object" && typeof Reflect.decorate === "function") r = Reflect.decorate(decorators, target, key, desc);
|
|
else for (var i = decorators.length - 1; i >= 0; i--) if (d = decorators[i]) r = (c < 3 ? d(r) : c > 3 ? d(target, key, r) : d(target, key)) || r;
|
|
return c > 3 && r && Object.defineProperty(target, key, r), r;
|
|
};
|
|
var __metadata = (this && this.__metadata) || function (k, v) {
|
|
if (typeof Reflect === "object" && typeof Reflect.metadata === "function") return Reflect.metadata(k, v);
|
|
};
|
|
var __param = (this && this.__param) || function (paramIndex, decorator) {
|
|
return function (target, key) { decorator(target, key, paramIndex); }
|
|
};
|
|
var TelemetryAnalyzerService_1;
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
|
exports.TelemetryAnalyzerService = void 0;
|
|
const common_1 = require("@nestjs/common");
|
|
const typeorm_1 = require("@nestjs/typeorm");
|
|
const typeorm_2 = require("typeorm");
|
|
const schedule_1 = require("@nestjs/schedule");
|
|
const telemetry_entity_1 = require("./telemetry.entity");
|
|
const road_stat_entity_1 = require("../maps/road-stat.entity");
|
|
const candidate_road_entity_1 = require("../maps/candidate-road.entity");
|
|
const redis_service_1 = require("../common/redis.service");
|
|
const external_telemetry_service_1 = require("./external-telemetry.service");
|
|
let TelemetryAnalyzerService = TelemetryAnalyzerService_1 = class TelemetryAnalyzerService {
|
|
telemetryRepo;
|
|
roadStatRepo;
|
|
candidateRoadRepo;
|
|
dataSource;
|
|
redisService;
|
|
externalTelemetry;
|
|
logger = new common_1.Logger(TelemetryAnalyzerService_1.name);
|
|
TRAFFIC_CACHE_KEY = 'traffic_snapshot';
|
|
constructor(telemetryRepo, roadStatRepo, candidateRoadRepo, dataSource, redisService, externalTelemetry) {
|
|
this.telemetryRepo = telemetryRepo;
|
|
this.roadStatRepo = roadStatRepo;
|
|
this.candidateRoadRepo = candidateRoadRepo;
|
|
this.dataSource = dataSource;
|
|
this.redisService = redisService;
|
|
this.externalTelemetry = externalTelemetry;
|
|
}
|
|
async handleNightlyIntelligence() {
|
|
this.logger.log('⏰ Starting automated 3 AM intelligence process...');
|
|
try {
|
|
await this.syncExternalData(1);
|
|
await this.analyzeRoadSpeeds(24);
|
|
await this.discoverNewRoads(168);
|
|
await this.refreshTrafficCache();
|
|
this.logger.log('✅ 3 AM intelligence process complete.');
|
|
}
|
|
catch (error) {
|
|
this.logger.error('❌ Nightly intelligence failed:', error.stack);
|
|
}
|
|
}
|
|
async syncExternalData(days = 1) {
|
|
this.logger.log(`📡 Starting batch sync (Window: ${days} days)...`);
|
|
const tracks = await this.externalTelemetry.fetchCarTracks(days);
|
|
if (!tracks || tracks.length === 0) {
|
|
this.logger.warn(`⚠️ No tracks found on external server for the last ${days} days.`);
|
|
return { imported: 0 };
|
|
}
|
|
this.logger.log(`✅ Received ${tracks.length} tracks. Commencing batch insertion...`);
|
|
this.logger.log(`📥 Saving ${tracks.length} points to database...`);
|
|
const entities = tracks.map(t => ({
|
|
driverId: t.driver_id,
|
|
latitude: t.latitude,
|
|
longitude: t.longitude,
|
|
speed: t.speed,
|
|
heading: t.heading,
|
|
timestamp: new Date(t.created_at || t.timestamp),
|
|
location: {
|
|
type: 'Point',
|
|
coordinates: [t.longitude, t.latitude],
|
|
},
|
|
}));
|
|
const CHUNK_SIZE = 1000;
|
|
for (let i = 0; i < entities.length; i += CHUNK_SIZE) {
|
|
const chunk = entities.slice(i, i + CHUNK_SIZE);
|
|
const logs = this.telemetryRepo.create(chunk);
|
|
await this.telemetryRepo.save(logs);
|
|
}
|
|
return { imported: tracks.length };
|
|
}
|
|
async refreshTrafficCache() {
|
|
this.logger.log('🚀 Refreshing Redis traffic snapshot...');
|
|
const congested = await this.roadStatRepo.query(`
|
|
SELECT
|
|
"segmentId",
|
|
"congestionFactor",
|
|
ST_AsGeoJSON(geometry) as geojson
|
|
FROM road_segment_stats
|
|
WHERE "congestionFactor" > 1.1
|
|
`);
|
|
if (congested.length === 0) {
|
|
await this.redisService.del(this.TRAFFIC_CACHE_KEY);
|
|
return { cachedCount: 0 };
|
|
}
|
|
await this.redisService.set(this.TRAFFIC_CACHE_KEY, congested);
|
|
this.logger.log(`✅ Cached ${congested.length} congested segments in Redis.`);
|
|
return { cachedCount: congested.length };
|
|
}
|
|
async analyzeRoadSpeeds(sinceHours = 24) {
|
|
this.logger.log(`🔍 Starting road speed analysis for last ${sinceHours}h...`);
|
|
const matchedData = await this.dataSource.query(`
|
|
WITH matched_points AS (
|
|
SELECT
|
|
t.id AS telemetry_id,
|
|
t.speed,
|
|
t."driverId",
|
|
l.osm_id,
|
|
l.name,
|
|
l.highway,
|
|
l.way_4326,
|
|
ST_Distance(t.location::geography, l.way_4326::geography) AS distance_m
|
|
FROM telemetry_logs t
|
|
CROSS JOIN LATERAL (
|
|
SELECT osm_id, name, highway, ST_Transform(way, 4326) AS way_4326
|
|
FROM planet_osm_line
|
|
WHERE highway IS NOT NULL
|
|
ORDER BY way <-> ST_Transform(t.location::geometry, 3857)
|
|
LIMIT 1
|
|
) l
|
|
WHERE t.timestamp >= NOW() - INTERVAL '${sinceHours} hours'
|
|
AND t.speed > 2 -- Ignore stationary points / تجاهل النقاط الثابتة
|
|
AND ST_Distance(t.location::geography, l.way_4326::geography) < 15 -- 15m snap threshold
|
|
)
|
|
SELECT
|
|
osm_id::text AS segment_id,
|
|
name,
|
|
highway,
|
|
AVG(speed) AS avg_speed,
|
|
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY speed) AS median_speed,
|
|
COUNT(*) AS sample_count,
|
|
COUNT(DISTINCT "driverId") AS unique_drivers,
|
|
ST_AsGeoJSON(MIN(way_4326)) AS geojson
|
|
FROM matched_points
|
|
GROUP BY osm_id, name, highway
|
|
HAVING COUNT(*) >= 3 -- Minimum 3 samples for reliability / 3 عينات على الأقل
|
|
ORDER BY sample_count DESC
|
|
`);
|
|
let segmentsUpdated = 0;
|
|
let totalPoints = 0;
|
|
for (const row of matchedData) {
|
|
const medianSpeed = parseFloat(row.median_speed);
|
|
const sampleCount = parseInt(row.sample_count);
|
|
const geojson = JSON.parse(row.geojson);
|
|
totalPoints += sampleCount;
|
|
let congestionFactor = 1.0;
|
|
const majorRoadTypes = ['primary', 'secondary', 'trunk', 'motorway'];
|
|
if (majorRoadTypes.includes(row.highway) && medianSpeed < 30) {
|
|
congestionFactor = 30 / Math.max(medianSpeed, 1);
|
|
}
|
|
await this.roadStatRepo.upsert({
|
|
segmentId: row.segment_id,
|
|
averageSpeed: medianSpeed,
|
|
congestionFactor,
|
|
sampleCount,
|
|
lastUpdated: new Date(),
|
|
geometry: geojson,
|
|
}, ['segmentId']);
|
|
segmentsUpdated++;
|
|
}
|
|
return {
|
|
segmentsUpdated,
|
|
totalPointsProcessed: totalPoints,
|
|
topAdjustments: matchedData.slice(0, 10).map(d => ({
|
|
segmentId: d.segment_id,
|
|
name: d.name || 'Unnamed Road',
|
|
speed: Math.round(parseFloat(d.median_speed)),
|
|
samples: parseInt(d.sample_count)
|
|
}))
|
|
};
|
|
}
|
|
async discoverNewRoads(sinceHours = 168) {
|
|
this.logger.log(`🛣️ Starting road discovery for last ${sinceHours}h (${sinceHours / 24}d)...`);
|
|
const candidates = await this.dataSource.query(`
|
|
WITH off_road_points AS (
|
|
SELECT
|
|
t.id,
|
|
t."driverId",
|
|
t.speed,
|
|
t.heading,
|
|
t.location,
|
|
t.timestamp
|
|
FROM telemetry_logs t
|
|
WHERE t.timestamp >= NOW() - INTERVAL '${sinceHours} hours'
|
|
AND t.speed > 5 -- Moving, not parked / متحرك وليس متوقف
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM planet_osm_line l
|
|
WHERE l.highway IS NOT NULL
|
|
AND ST_DWithin(t.location::geography, ST_Transform(l.way, 4326)::geography, 15)
|
|
)
|
|
),
|
|
-- Step 2: Cluster nearby off-road points using DBSCAN
|
|
-- الخطوة 2: تجميع النقاط القريبة باستخدام DBSCAN
|
|
clustered AS (
|
|
SELECT
|
|
*,
|
|
ST_ClusterDBSCAN(location::geometry, eps := 0.0003, minpoints := 5)
|
|
OVER () AS cluster_id
|
|
FROM off_road_points
|
|
)
|
|
-- Step 3: Aggregate clusters into candidate road lines
|
|
-- الخطوة 3: تحويل التجمعات إلى خطوط طرق مرشحة
|
|
SELECT
|
|
cluster_id,
|
|
COUNT(*) AS total_points,
|
|
COUNT(DISTINCT "driverId") AS unique_drivers,
|
|
AVG(speed) AS avg_speed,
|
|
ST_AsGeoJSON(ST_MakeLine(location::geometry ORDER BY timestamp)) AS geojson_line,
|
|
ST_Length(ST_MakeLine(location::geometry ORDER BY timestamp)::geography) AS length_m
|
|
FROM clustered
|
|
WHERE cluster_id IS NOT NULL
|
|
GROUP BY cluster_id
|
|
HAVING COUNT(DISTINCT "driverId") >= 2 -- At least 2 drivers / سائقان على الأقل
|
|
AND COUNT(*) >= 10 -- At least 10 points / 10 نقاط على الأقل
|
|
ORDER BY unique_drivers DESC, total_points DESC
|
|
`);
|
|
const savedCandidates = [];
|
|
for (const c of candidates) {
|
|
const geojson = JSON.parse(c.geojson_line);
|
|
const confidence = this.calculateConfidence(parseInt(c.unique_drivers), parseInt(c.total_points), parseFloat(c.length_m));
|
|
const candidate = this.candidateRoadRepo.create({
|
|
geometry: geojson,
|
|
uniqueDriverCount: parseInt(c.unique_drivers),
|
|
totalPoints: parseInt(c.total_points),
|
|
averageSpeed: parseFloat(c.avg_speed),
|
|
lengthMeters: parseFloat(c.length_m),
|
|
confidence,
|
|
status: 'pending',
|
|
});
|
|
const saved = await this.candidateRoadRepo.save(candidate);
|
|
savedCandidates.push({
|
|
id: saved.id,
|
|
uniqueDrivers: saved.uniqueDriverCount,
|
|
totalPoints: saved.totalPoints,
|
|
averageSpeed: Math.round(saved.averageSpeed * 10) / 10,
|
|
lengthMeters: Math.round(saved.lengthMeters),
|
|
confidence: Math.round(saved.confidence * 100) / 100,
|
|
geojson,
|
|
});
|
|
}
|
|
this.logger.log(`✅ Road discovery complete: ${savedCandidates.length} candidates found`);
|
|
return {
|
|
candidatesFound: savedCandidates.length,
|
|
candidates: savedCandidates,
|
|
};
|
|
}
|
|
async getCongestionData(bounds) {
|
|
return this.dataSource.query(`
|
|
SELECT
|
|
rs."segmentId" AS segment_id,
|
|
rs."averageSpeed" AS avg_speed,
|
|
rs."congestionFactor" AS congestion_factor,
|
|
rs."sampleCount" AS sample_count,
|
|
ST_AsGeoJSON(rs.geometry) AS geojson
|
|
FROM road_segment_stats rs
|
|
WHERE rs.geometry IS NOT NULL
|
|
AND ST_Intersects(
|
|
rs.geometry::geometry,
|
|
ST_MakeEnvelope($1, $2, $3, $4, 4326)
|
|
)
|
|
AND rs."sampleCount" >= 3
|
|
ORDER BY rs."congestionFactor" DESC
|
|
`, [bounds.west, bounds.south, bounds.east, bounds.north]);
|
|
}
|
|
async getAnalysisSummary() {
|
|
const [telemetryCount] = await this.dataSource.query(`SELECT COUNT(*) as count FROM telemetry_logs`);
|
|
const [roadStatCount] = await this.dataSource.query(`SELECT COUNT(*) as count FROM road_segment_stats`);
|
|
const [candidateCount] = await this.dataSource.query(`SELECT COUNT(*) as count,
|
|
COUNT(*) FILTER (WHERE status = 'pending') as pending,
|
|
COUNT(*) FILTER (WHERE status = 'approved') as approved,
|
|
COUNT(*) FILTER (WHERE status = 'rejected') as rejected
|
|
FROM candidate_roads`);
|
|
const [recentActivity] = await this.dataSource.query(`
|
|
SELECT
|
|
COUNT(*) as points_last_24h,
|
|
COUNT(DISTINCT "driverId") as active_drivers_24h
|
|
FROM telemetry_logs
|
|
WHERE timestamp >= NOW() - INTERVAL '24 hours'
|
|
`);
|
|
return {
|
|
telemetry: {
|
|
totalPoints: parseInt(telemetryCount?.count || '0'),
|
|
last24h: parseInt(recentActivity?.points_last_24h || '0'),
|
|
activeDrivers24h: parseInt(recentActivity?.active_drivers_24h || '0'),
|
|
},
|
|
roadSegments: {
|
|
analyzed: parseInt(roadStatCount?.count || '0'),
|
|
},
|
|
candidateRoads: {
|
|
total: parseInt(candidateCount?.count || '0'),
|
|
pending: parseInt(candidateCount?.pending || '0'),
|
|
approved: parseInt(candidateCount?.approved || '0'),
|
|
rejected: parseInt(candidateCount?.rejected || '0'),
|
|
},
|
|
};
|
|
}
|
|
calculateConfidence(uniqueDrivers, totalPoints, lengthMeters) {
|
|
const driverScore = Math.min(uniqueDrivers / 5, 1.0) * 0.4;
|
|
const pointScore = Math.min(totalPoints / 50, 1.0) * 0.3;
|
|
let lengthScore = 0;
|
|
if (lengthMeters >= 50 && lengthMeters <= 2000) {
|
|
lengthScore = 0.3;
|
|
}
|
|
else if (lengthMeters > 2000) {
|
|
lengthScore = 0.2;
|
|
}
|
|
return Math.round((driverScore + pointScore + lengthScore) * 100) / 100;
|
|
}
|
|
};
|
|
exports.TelemetryAnalyzerService = TelemetryAnalyzerService;
|
|
__decorate([
|
|
(0, schedule_1.Cron)('0 3 * * *'),
|
|
__metadata("design:type", Function),
|
|
__metadata("design:paramtypes", []),
|
|
__metadata("design:returntype", Promise)
|
|
], TelemetryAnalyzerService.prototype, "handleNightlyIntelligence", null);
|
|
exports.TelemetryAnalyzerService = TelemetryAnalyzerService = TelemetryAnalyzerService_1 = __decorate([
|
|
(0, common_1.Injectable)(),
|
|
__param(0, (0, typeorm_1.InjectRepository)(telemetry_entity_1.TelemetryLog)),
|
|
__param(1, (0, typeorm_1.InjectRepository)(road_stat_entity_1.RoadSegmentStat)),
|
|
__param(2, (0, typeorm_1.InjectRepository)(candidate_road_entity_1.CandidateRoad)),
|
|
__metadata("design:paramtypes", [typeorm_2.Repository,
|
|
typeorm_2.Repository,
|
|
typeorm_2.Repository,
|
|
typeorm_2.DataSource,
|
|
redis_service_1.RedisService,
|
|
external_telemetry_service_1.ExternalTelemetryService])
|
|
], TelemetryAnalyzerService);
|
|
//# sourceMappingURL=telemetry-analyzer.service.js.map
|