419 lines
19 KiB
JavaScript
419 lines
19 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 schedule_1 = require("@nestjs/schedule");
|
|
const typeorm_1 = require("@nestjs/typeorm");
|
|
const typeorm_2 = require("typeorm");
|
|
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 external_telemetry_service_1 = require("./external-telemetry.service");
|
|
const redis_service_1 = require("../common/redis.service");
|
|
const telegram_service_1 = require("../common/telegram.service");
|
|
const traffic_grid_service_1 = require("../maps/traffic-grid.service");
|
|
let TelemetryAnalyzerService = TelemetryAnalyzerService_1 = class TelemetryAnalyzerService {
|
|
telemetryRepo;
|
|
roadStatRepo;
|
|
candidateRoadRepo;
|
|
externalTelemetry;
|
|
redisService;
|
|
telegramService;
|
|
trafficGrid;
|
|
dataSource;
|
|
logger = new common_1.Logger(TelemetryAnalyzerService_1.name);
|
|
TRAFFIC_CACHE_KEY = 'live_traffic_congested';
|
|
constructor(telemetryRepo, roadStatRepo, candidateRoadRepo, externalTelemetry, redisService, telegramService, trafficGrid, dataSource) {
|
|
this.telemetryRepo = telemetryRepo;
|
|
this.roadStatRepo = roadStatRepo;
|
|
this.candidateRoadRepo = candidateRoadRepo;
|
|
this.externalTelemetry = externalTelemetry;
|
|
this.redisService = redisService;
|
|
this.telegramService = telegramService;
|
|
this.trafficGrid = trafficGrid;
|
|
this.dataSource = dataSource;
|
|
}
|
|
async runDeepIntelligence(hours = 48) {
|
|
this.logger.log(`🚀 Starting Full Intelligence Pipeline (Window: ${hours}h)...`);
|
|
await this.syncExternalData(Math.ceil(hours / 24));
|
|
const speedResult = await this.analyzeRoadSpeeds(hours);
|
|
const temporalResult = await this.analyzeTimeProfiles(hours);
|
|
const discoveryResult = await this.discoverNewRoads(hours);
|
|
await this.refreshTrafficCache();
|
|
await this.trafficGrid.refreshGrid();
|
|
const summary = await this.getAnalysisSummary();
|
|
await this.telegramService.sendIntelligenceReport({
|
|
syncResult: speedResult.totalPointsProcessed || 0,
|
|
updatedSegments: speedResult.segmentsUpdated || 0,
|
|
discoveredRoads: discoveryResult.candidatesFound || 0,
|
|
timeProfiles: temporalResult.bucketsUpdated || 0,
|
|
totalPoints: summary.telemetry.total,
|
|
days: Math.ceil(hours / 24),
|
|
});
|
|
this.logger.log('🏁 Intelligence Pipeline Finished Successfully.');
|
|
return {
|
|
success: true,
|
|
timestamp: new Date().toISOString(),
|
|
speedAnalysis: speedResult,
|
|
temporalAnalysis: temporalResult,
|
|
roadDiscovery: discoveryResult,
|
|
};
|
|
}
|
|
handleDeepIntelligence() {
|
|
this.runDeepIntelligence(240);
|
|
}
|
|
async syncExternalData(days = 7) {
|
|
this.logger.log(`📡 Starting batch sync (Window: ${days} days)...`);
|
|
try {
|
|
const tracks = await this.externalTelemetry.fetchCarTracks(days);
|
|
if (tracks.length === 0)
|
|
return;
|
|
this.logger.log(`📥 Saving ${tracks.length} points to database...`);
|
|
const values = tracks
|
|
.filter(t => t.driver_id && t.longitude && t.latitude)
|
|
.map(t => `('${t.driver_id}', ST_SetSRID(ST_Point(${t.longitude}, ${t.latitude}), 4326), ${t.latitude}, ${t.longitude}, ${t.speed || 0}, ${t.heading || 0}, '${t.created_at}')`).join(',');
|
|
await this.dataSource.query(`
|
|
INSERT INTO telemetry_logs ("driverId", location, latitude, longitude, speed, heading, timestamp)
|
|
VALUES ${values}
|
|
ON CONFLICT DO NOTHING
|
|
`);
|
|
this.logger.log(`✅ Batch sync complete.`);
|
|
}
|
|
catch (error) {
|
|
this.logger.error(`❌ Sync failed: ${error.message}`);
|
|
throw error;
|
|
}
|
|
}
|
|
async refreshTrafficCache() {
|
|
this.logger.log('🚀 Refreshing Redis traffic snapshot (v2.5.2 Optimized)...');
|
|
const congested = await this.roadStatRepo.query(`
|
|
SELECT "segmentId", "congestionFactor", ST_AsGeoJSON(geometry, 5) as geojson
|
|
FROM road_segment_stats
|
|
WHERE "congestionFactor" > 1.1
|
|
ORDER BY "congestionFactor" DESC
|
|
LIMIT 2000
|
|
`);
|
|
if (congested.length === 0) {
|
|
await this.redisService.del(this.TRAFFIC_CACHE_KEY);
|
|
return { cachedCount: 0 };
|
|
}
|
|
const snapshot = congested.map(row => ({
|
|
sid: row.segmentId,
|
|
cf: parseFloat(row.congestionFactor),
|
|
geo: JSON.parse(row.geojson)
|
|
}));
|
|
const sampleSize = Math.min(congested.length, 3);
|
|
const topSegments = congested.slice(0, sampleSize).map(s => `${s.segmentId} (F: ${s.congestionFactor})`).join(', ');
|
|
const totalSizeKB = Math.round(JSON.stringify(snapshot).length / 1024);
|
|
this.logger.log(`📊 Traffic Snapshot Sample (Top ${sampleSize}): ${topSegments}`);
|
|
this.logger.log(`📦 Redis Payload Size: ~${totalSizeKB} KB`);
|
|
await this.redisService.set(this.TRAFFIC_CACHE_KEY, snapshot);
|
|
this.logger.log(`✅ Redis traffic snapshot updated with ${congested.length} segments.`);
|
|
return { cachedCount: congested.length };
|
|
}
|
|
async analyzeRoadSpeeds(sinceHours = 24) {
|
|
this.logger.log(`🔍 Starting road speed analysis v2.5 for last ${sinceHours}h...`);
|
|
const query = `
|
|
WITH grid_points AS (
|
|
-- Group by 5m grid cell first to reduce spatial join volume (v2.5.1 WoW Performance)
|
|
SELECT ST_SnapToGrid(ST_Transform(location::geometry, 3857), 5) AS loc,
|
|
AVG(speed) as speed
|
|
FROM telemetry_logs
|
|
WHERE timestamp >= NOW() - INTERVAL '${sinceHours} hours' AND speed > 2
|
|
GROUP BY loc
|
|
),
|
|
matches AS (
|
|
-- Bulk Spatial Join (20m radius) with GIST optimization
|
|
SELECT l.osm_id::text as id, l.name, l.highway, g.speed, l.way
|
|
FROM grid_points g
|
|
INNER JOIN planet_osm_line l ON
|
|
l.highway IS NOT NULL AND
|
|
l.way && ST_Expand(g.loc, 20) AND
|
|
ST_DWithin(l.way, g.loc, 20)
|
|
),
|
|
stats AS (
|
|
-- Calculate median speed and aggregate samples
|
|
SELECT id, name, highway,
|
|
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY speed) as avg_speed,
|
|
COUNT(*) as samples,
|
|
ST_Transform(MIN(way), 4326) as way4326
|
|
FROM matches
|
|
GROUP BY id, name, highway
|
|
HAVING COUNT(*) >= 3
|
|
)
|
|
INSERT INTO road_segment_stats ("segmentId", "averageSpeed", "sampleCount", "lastUpdated", geometry, "congestionFactor")
|
|
SELECT id, avg_speed, samples, NOW(), way4326,
|
|
CASE WHEN highway IN ('primary','secondary','trunk','motorway') AND avg_speed < 30
|
|
THEN 30/GREATEST(avg_speed,1) ELSE 1.0 END
|
|
FROM stats
|
|
ON CONFLICT ("segmentId") DO UPDATE SET
|
|
"averageSpeed"=EXCLUDED."averageSpeed",
|
|
"sampleCount"=EXCLUDED."sampleCount",
|
|
"lastUpdated"=NOW(),
|
|
"geometry"=EXCLUDED."geometry",
|
|
"congestionFactor"=EXCLUDED."congestionFactor"
|
|
RETURNING "segmentId";
|
|
`;
|
|
const result = await this.dataSource.query(query);
|
|
return { segmentsUpdated: result.length };
|
|
}
|
|
async analyzeTimeProfiles(sinceHours = 720) {
|
|
this.logger.log(`🕒 Starting temporal profiling (Phase 2) for last ${sinceHours}h...`);
|
|
const query = `
|
|
WITH grid_points AS (
|
|
-- Bucket by 10m grid and Time (Hour + DOW)
|
|
SELECT
|
|
ST_SnapToGrid(ST_Transform(location::geometry, 3857), 10) AS loc,
|
|
EXTRACT(HOUR FROM timestamp)::int as hr,
|
|
EXTRACT(DOW FROM timestamp)::int as dow,
|
|
AVG(speed) as speed
|
|
FROM telemetry_logs
|
|
WHERE timestamp >= NOW() - INTERVAL '${sinceHours} hours' AND speed > 2
|
|
GROUP BY loc, hr, dow
|
|
),
|
|
matches AS (
|
|
-- Map to OSM segments
|
|
SELECT l.osm_id::text as sid, g.hr, g.dow, g.speed
|
|
FROM grid_points g
|
|
INNER JOIN planet_osm_line l ON
|
|
l.highway IS NOT NULL AND
|
|
l.way && ST_Expand(g.loc, 20) AND
|
|
ST_DWithin(l.way, g.loc, 20)
|
|
),
|
|
temporal_stats AS (
|
|
-- Aggregate by segment + time bucket
|
|
SELECT sid, hr, dow,
|
|
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY speed) as avg_spd,
|
|
COUNT(*) as smp
|
|
FROM matches
|
|
GROUP BY sid, hr, dow
|
|
HAVING COUNT(*) >= 2 -- Require minimum samples per bucket
|
|
)
|
|
INSERT INTO road_speed_profiles ("segmentId", "hourOfDay", "dayOfWeek", "averageSpeed", "sampleCount", "lastUpdated")
|
|
SELECT sid, hr, dow, avg_spd, smp, NOW()
|
|
FROM temporal_stats
|
|
ON CONFLICT ("segmentId", "hourOfDay", "dayOfWeek") DO UPDATE SET
|
|
"averageSpeed" = EXCLUDED."averageSpeed",
|
|
"sampleCount" = EXCLUDED."sampleCount",
|
|
"lastUpdated" = NOW()
|
|
RETURNING "segmentId";
|
|
`;
|
|
const result = await this.dataSource.query(query);
|
|
this.logger.log(`📊 Time-aware profiling complete: ${result.length} buckets updated.`);
|
|
return { bucketsUpdated: result.length };
|
|
}
|
|
async discoverNewRoads(sinceHours = 168) {
|
|
this.logger.log(`🛣️ Starting road discovery v2.5 for last ${sinceHours}h...`);
|
|
const query = `
|
|
WITH grid_points AS (
|
|
-- Group by cell and driver to reduce volume for clustering and anti-join (v2.5.1 WoW Performance)
|
|
SELECT "driverId", ST_SnapToGrid(ST_Transform(location::geometry, 3857), 10) as loc,
|
|
AVG(speed) as speed,
|
|
MIN(timestamp) as timestamp
|
|
FROM telemetry_logs
|
|
WHERE timestamp >= NOW() - INTERVAL '${sinceHours} hours' AND speed > 5
|
|
GROUP BY loc, "driverId"
|
|
),
|
|
off_road AS (
|
|
-- Spatial Anti-Join: Only points further than 35m from any existing highway
|
|
SELECT g.* FROM grid_points g
|
|
LEFT JOIN planet_osm_line l ON
|
|
l.highway IS NOT NULL AND
|
|
l.way && ST_Expand(g.loc, 35) AND
|
|
ST_DWithin(l.way, g.loc, 35)
|
|
WHERE l.osm_id IS NULL
|
|
),
|
|
clustered AS (
|
|
-- Density-based clustering to find linear paths
|
|
SELECT *, ST_ClusterDBSCAN(loc, eps := 30, minpoints := 3) OVER () as cid FROM off_road
|
|
),
|
|
cluster_stats AS (
|
|
-- 1. Calculate cluster-wide quality metrics
|
|
SELECT cid,
|
|
COUNT(DISTINCT "driverId") as drv_count,
|
|
COUNT(*) as pt_count,
|
|
AVG(speed) as avg_spd
|
|
FROM clustered
|
|
WHERE cid IS NOT NULL
|
|
GROUP BY cid
|
|
HAVING COUNT(DISTINCT "driverId") >= 3 AND COUNT(*) >= 15
|
|
),
|
|
cluster_path AS (
|
|
-- 2. Extract unique grid cells per cluster in chronological order
|
|
SELECT cid, loc, MIN(timestamp) as ts
|
|
FROM clustered
|
|
WHERE cid IN (SELECT cid FROM cluster_stats)
|
|
GROUP BY cid, loc
|
|
),
|
|
final_candidates AS (
|
|
-- 3. Build geometry and calculate final scoring
|
|
SELECT
|
|
s.cid,
|
|
ST_Transform(ST_MakeLine(p.loc ORDER BY p.ts), 4326) as geom,
|
|
s.drv_count,
|
|
s.pt_count,
|
|
s.avg_spd,
|
|
ST_Length(ST_Transform(ST_MakeLine(p.loc ORDER BY p.ts), 4326)::geography) as len
|
|
FROM cluster_stats s
|
|
JOIN cluster_path p ON s.cid = p.cid
|
|
GROUP BY s.cid, s.drv_count, s.pt_count, s.avg_spd
|
|
),
|
|
oneway_detect AS (
|
|
-- 4. Detect one-way direction from telemetry heading data
|
|
SELECT
|
|
fc.cid,
|
|
DEGREES(ST_Azimuth(
|
|
ST_StartPoint(fc.geom::geometry),
|
|
ST_EndPoint(fc.geom::geometry)
|
|
)) AS geom_bearing,
|
|
AVG(tl.heading) AS avg_heading,
|
|
COUNT(*) AS heading_samples
|
|
FROM final_candidates fc
|
|
JOIN telemetry_logs tl ON ST_DWithin(
|
|
tl.location::geometry,
|
|
fc.geom::geometry,
|
|
30
|
|
)
|
|
WHERE tl.speed > 3
|
|
GROUP BY fc.cid, fc.geom
|
|
HAVING COUNT(*) >= 5
|
|
),
|
|
oneway_scored AS (
|
|
SELECT
|
|
od.cid,
|
|
CASE
|
|
WHEN LEAST(ABS(od.avg_heading - od.geom_bearing), 360 - ABS(od.avg_heading - od.geom_bearing)) < 45
|
|
AND od.heading_samples >= 10 THEN 1
|
|
WHEN LEAST(ABS(od.avg_heading - (od.geom_bearing + 180) % 360), 360 - ABS(od.avg_heading - (od.geom_bearing + 180) % 360)) < 45
|
|
AND od.heading_samples >= 10 THEN -1
|
|
ELSE 0
|
|
END AS oneway_val
|
|
FROM oneway_detect od
|
|
)
|
|
INSERT INTO candidate_roads (geometry, "uniqueDriverCount", "totalPoints", "averageSpeed", "lengthMeters", confidence, status, oneway)
|
|
SELECT fc.geom, fc.drv_count, fc.pt_count, fc.avg_spd, fc.len,
|
|
ROUND((
|
|
LEAST(fc.drv_count::float / 5, 1.0) * 0.4 +
|
|
LEAST(fc.pt_count::float / 50, 1.0) * 0.3 +
|
|
CASE WHEN fc.len BETWEEN 50 AND 2000 THEN 0.3 ELSE 0.1 END
|
|
)::numeric, 2) as conf,
|
|
'pending',
|
|
COALESCE(os.oneway_val, 0)
|
|
FROM final_candidates fc
|
|
LEFT JOIN oneway_scored os ON os.cid = fc.cid
|
|
RETURNING id;
|
|
`;
|
|
const result = await this.dataSource.query(query);
|
|
return { candidatesFound: result.length };
|
|
}
|
|
async getCongestionData(bounds) {
|
|
return this.dataSource.query(`
|
|
SELECT rs."segmentId", rs."averageSpeed", rs."congestionFactor", rs."sampleCount", 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))
|
|
ORDER BY rs."congestionFactor" DESC
|
|
`, [bounds.west, bounds.south, bounds.east, bounds.north]);
|
|
}
|
|
async getAnalysisSummary() {
|
|
const [tCount] = await this.dataSource.query('SELECT COUNT(*) as count FROM telemetry_logs');
|
|
const [rCount] = await this.dataSource.query('SELECT COUNT(*) as count FROM road_segment_stats');
|
|
const [cCount] = await this.dataSource.query('SELECT COUNT(*) as count, COUNT(*) FILTER (WHERE status=\'pending\') as p FROM candidate_roads');
|
|
const [clCount] = await this.dataSource.query('SELECT COUNT(*) as count FROM road_segment_stats WHERE "isClosed" = true');
|
|
return {
|
|
telemetry: { total: tCount.count },
|
|
roads: { analyzed: rCount.count, closed: clCount.count },
|
|
candidates: { total: cCount.count, pending: cCount.p }
|
|
};
|
|
}
|
|
async getCandidates(status = 'pending', limit = 50) {
|
|
return this.candidateRoadRepo.find({
|
|
where: { status },
|
|
order: { confidence: 'DESC' },
|
|
take: limit
|
|
});
|
|
}
|
|
async updateCandidateStatus(id, status) {
|
|
await this.candidateRoadRepo.update(id, {
|
|
status,
|
|
reviewedAt: new Date()
|
|
});
|
|
return { success: true, id, status };
|
|
}
|
|
async discoverRoadClosures(sinceHours = 48) {
|
|
this.logger.log(`🚧 Analyzing road closures for last ${sinceHours}h...`);
|
|
await this.roadStatRepo.update({ isClosed: true }, { isClosed: false });
|
|
const query = `
|
|
WITH active_area AS (
|
|
-- Bounding box of some recent activity to prove drivers are on the map
|
|
SELECT ST_Expand(ST_Extent(location::geometry), 0.01) as bbox
|
|
FROM telemetry_logs
|
|
WHERE timestamp >= NOW() - INTERVAL '${sinceHours} hours'
|
|
),
|
|
possible_closures AS (
|
|
SELECT rs."segmentId"
|
|
FROM road_segment_stats rs
|
|
WHERE rs."sampleCount" > 50
|
|
AND rs.geometry && (SELECT bbox FROM active_area)
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM telemetry_logs t
|
|
WHERE t.timestamp >= NOW() - INTERVAL '${sinceHours} hours'
|
|
AND ST_DWithin(rs.geometry::geometry, t.location::geometry, 35)
|
|
)
|
|
)
|
|
UPDATE road_segment_stats
|
|
SET "isClosed" = true
|
|
WHERE "segmentId" IN (SELECT "segmentId" FROM possible_closures)
|
|
RETURNING "segmentId";
|
|
`;
|
|
const result = await this.dataSource.query(query);
|
|
this.logger.log(`✅ Road closure detection complete: ${result.length} roads flagged as closed.`);
|
|
return { roadsClosed: result.length };
|
|
}
|
|
async getClosures() {
|
|
return this.roadStatRepo.find({
|
|
where: { isClosed: true },
|
|
order: { sampleCount: 'DESC' }
|
|
});
|
|
}
|
|
calculateConfidence(drivers, points, len) {
|
|
const dScore = Math.min(drivers / 5, 1.0) * 0.4;
|
|
const pScore = Math.min(points / 50, 1.0) * 0.3;
|
|
const lScore = (len >= 50 && len <= 2000) ? 0.3 : 0.1;
|
|
return Math.round((dScore + pScore + lScore) * 100) / 100;
|
|
}
|
|
};
|
|
exports.TelemetryAnalyzerService = TelemetryAnalyzerService;
|
|
__decorate([
|
|
(0, schedule_1.Cron)('0 4 */10 * *'),
|
|
__metadata("design:type", Function),
|
|
__metadata("design:paramtypes", []),
|
|
__metadata("design:returntype", void 0)
|
|
], TelemetryAnalyzerService.prototype, "handleDeepIntelligence", 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,
|
|
external_telemetry_service_1.ExternalTelemetryService,
|
|
redis_service_1.RedisService,
|
|
telegram_service_1.TelegramService,
|
|
traffic_grid_service_1.TrafficGridService,
|
|
typeorm_2.DataSource])
|
|
], TelemetryAnalyzerService);
|
|
//# sourceMappingURL=telemetry-analyzer.service.js.map
|