using AsbCloudDb.Model; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Hosting; using System; using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Tasks; namespace AsbCloudInfrastructure.Services.DetectOperations { public class OperationDetectionBackgroundService : BackgroundService { private readonly IEnumerable detectors = new List { new Detectors.DetectorSlipsTime(), // new Detectors.DetectorDrillingRotor(), // new Detectors.DetectorDrillingSlide(), }; private readonly int minStepLength; private readonly int minFragmentLength; private readonly string connectionString; private readonly TimeSpan period = TimeSpan.FromHours(1); public OperationDetectionBackgroundService(IConfiguration configuration) { minStepLength = detectors.Min(d => d.StepLength); minStepLength = minStepLength > 0 ? minStepLength : 3; minFragmentLength = detectors.Min(d => d.FragmentLength); minFragmentLength = minFragmentLength > 0 ? minFragmentLength : 6; connectionString = configuration.GetConnectionString("DefaultConnection"); } protected override async Task ExecuteAsync(CancellationToken token = default) { var timeToStartAnalysis = DateTime.Now; var options = new DbContextOptionsBuilder() .UseNpgsql(connectionString) .Options; while (!token.IsCancellationRequested) { if (DateTime.Now > timeToStartAnalysis) { timeToStartAnalysis = DateTime.Now + period; try { using var context = new AsbCloudDbContext(options); var added = await DetectedAllTelemetriesAsync(context, token); Trace.TraceInformation($"Total detection complete. Added {added} operations."); } catch (Exception ex) { Trace.TraceError(ex.Message); } GC.Collect(); } var ms = (int)(timeToStartAnalysis - DateTime.Now).TotalMilliseconds; ms = ms > 100 ? ms : 100; await Task.Delay(ms, token).ConfigureAwait(false); } } public override async Task StopAsync(CancellationToken token) { await base.StopAsync(token).ConfigureAwait(false); } private async Task DetectedAllTelemetriesAsync(IAsbCloudDbContext db, CancellationToken token) { var lastDetectedDates = await db.DetectedOperations .GroupBy(o => o.IdTelemetry) .Select(g => new { IdTelemetry = g.Key, LastDate = g.Max(o => o.DateEnd) }) .ToListAsync(token); var telemetryIds = await db.Telemetries .Where(t => t.Info != null && t.TimeZone != null) .Select(t => t.Id) .ToListAsync(token); var JounedlastDetectedDates = telemetryIds .GroupJoin(lastDetectedDates, t => t, o => o.IdTelemetry, (outer, inner) => new { IdTelemetry = outer, LastDate = inner.SingleOrDefault()?.LastDate, }); var affected = 0; foreach (var item in JounedlastDetectedDates) { var newOperations = await DetectOperationsAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token); if (newOperations.Any()) { db.DetectedOperations.AddRange(newOperations); affected += await db.SaveChangesAsync(token); } } return affected; } private async Task> DetectOperationsAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token) { var query = db.TelemetryDataSaub .AsNoTracking() .Where(d => d.IdTelemetry == idTelemetry) .Select(d => new DetectableTelemetry { DateTime = d.DateTime, IdUser = d.IdUser, WellDepth = d.WellDepth, Pressure = d.Pressure, HookWeight = d.HookWeight, BlockPosition = d.BlockPosition, BitDepth = d.BitDepth, RotorSpeed = d.RotorSpeed, }) .OrderBy(d => d.DateTime); var take = 4 * 86_400; var startDate = begin; var detectedOperations = new List(8); var dbRequests_ = 0; var dbTime_ = 0d; var sw_ = new Stopwatch(); var otherTime_ = 0d; while (true) { sw_.Restart(); var data = await query .Where(d => d.DateTime > startDate) .Take(take) .ToArrayAsync(token); sw_.Stop(); dbTime_ += sw_.ElapsedMilliseconds; dbRequests_++; sw_.Restart(); if (data.Length < minFragmentLength) break; var skip = 0; var isDetected = false; while (data.Length > skip + minFragmentLength) { var isDetected1 = false; foreach (var detector in detectors) { if (data.Length < skip + detector.StepLength + detector.FragmentLength) continue; var detectedOperation = detector.DetectOrDefault(data, ref skip); if (detectedOperation is not null) { isDetected1 = true; isDetected = true; detectedOperation.IdTelemetry = idTelemetry; detectedOperations.Add(detectedOperation); startDate = detectedOperation.DateEnd; break; } } if (!isDetected1) skip += minStepLength; } sw_.Stop(); otherTime_ += sw_.ElapsedMilliseconds; if (!isDetected) { if (data.Length < take) break; var lastPartDate = data.Last().DateTime; startDate = startDate + (0.75 * (lastPartDate - startDate)); } } return detectedOperations; } } }