diff --git a/AsbCloudApp/Repositories/IDetectedOperationRepository.cs b/AsbCloudApp/Repositories/IDetectedOperationRepository.cs index c588a168..cda3379a 100644 --- a/AsbCloudApp/Repositories/IDetectedOperationRepository.cs +++ b/AsbCloudApp/Repositories/IDetectedOperationRepository.cs @@ -1,11 +1,9 @@ -using System; using AsbCloudApp.Data.DetectedOperation; using AsbCloudApp.Requests; using System.Collections.Generic; using System.Threading.Tasks; using System.Threading; using AsbCloudApp.Data; -using AsbCloudApp.Data.WellOperation; namespace AsbCloudApp.Repositories; @@ -53,14 +51,14 @@ public interface IDetectedOperationRepository /// /// Task> GetPageAsync(DetectedOperationByTelemetryRequest request, CancellationToken token); - + /// - /// Получение дат последних определённых операций + /// Получение последних автоопределённых операций /// /// /// - Task> GetLastDetectedDatesAsync(CancellationToken token); - + Task> GetLastDetectedOperationsAsync(CancellationToken token); + /// /// Удалить операции /// diff --git a/AsbCloudApp/Services/IDetectedOperationService.cs b/AsbCloudApp/Services/IDetectedOperationService.cs index ac7d650c..3b555f37 100644 --- a/AsbCloudApp/Services/IDetectedOperationService.cs +++ b/AsbCloudApp/Services/IDetectedOperationService.cs @@ -79,9 +79,13 @@ namespace AsbCloudApp.Services /// Определение операций /// /// - /// + /// + /// /// /// - Task> DetectOperationsAsync(int idTelemetry, DateTimeOffset? beginDate, CancellationToken token); + Task<(DateTimeOffset LastDate, IEnumerable Items)> DetectOperationsAsync(int idTelemetry, + TelemetryDataRequest request, + DetectedOperationDto? lastDetectedOperation, + CancellationToken token); } } diff --git a/AsbCloudApp/Services/ITelemetryDataSaubService.cs b/AsbCloudApp/Services/ITelemetryDataSaubService.cs index e9f2d1ac..f684b3d6 100644 --- a/AsbCloudApp/Services/ITelemetryDataSaubService.cs +++ b/AsbCloudApp/Services/ITelemetryDataSaubService.cs @@ -4,6 +4,7 @@ using System.Collections.Generic; using System.IO; using System.Threading; using System.Threading.Tasks; +using AsbCloudApp.Data; namespace AsbCloudApp.Services { @@ -44,5 +45,12 @@ namespace AsbCloudApp.Services /// /// Task GetZippedCsv(int idWell, DateTime beginDate, DateTime endDate, CancellationToken token); + + /// + /// Получение диапозона дат телеметрий + /// + /// + /// + Task> GetDateRangesAsync(CancellationToken token); } } \ No newline at end of file diff --git a/AsbCloudInfrastructure/Repository/DetectedOperationRepository.cs b/AsbCloudInfrastructure/Repository/DetectedOperationRepository.cs index b788dbda..a9123bda 100644 --- a/AsbCloudInfrastructure/Repository/DetectedOperationRepository.cs +++ b/AsbCloudInfrastructure/Repository/DetectedOperationRepository.cs @@ -25,7 +25,29 @@ public class DetectedOperationRepository : CrudRepositoryBase> GetLastDetectedOperationsAsync(CancellationToken token) + { + var entities = await dbContext.Set() + .GroupBy(o => o.IdTelemetry) + .Select(g => new + { + IdTelemetry = g.Key, + LastDetectedOperation = g.OrderBy(o => o.DateEnd).Last() + }) + .ToDictionaryAsync(x => x.IdTelemetry, x => x.LastDetectedOperation, token); + + var dtos = entities.ToDictionary(x => x.Key, x => + { + if (x.Value == null) + throw new ArgumentNullException(nameof(x.Value), "The value of the last detected operation cannot be null"); + + return Convert(x.Value); + }); + + return dtos; + } + public async Task DeleteAsync(DetectedOperationByTelemetryRequest request, CancellationToken token) { var query = BuildQuery(request); @@ -57,17 +79,7 @@ public class DetectedOperationRepository : CrudRepositoryBase> GetLastDetectedDatesAsync(CancellationToken token) => - await dbContext.Set() - .GroupBy(o => o.IdTelemetry) - .Select(g => new - { - IdTelemetry = g.Key, - LastDate = g.Max(o => o.DateEnd) - }) - .ToDictionaryAsync(x => x.IdTelemetry, x => x.LastDate, token); - + public async Task> Get(DetectedOperationByTelemetryRequest request, CancellationToken token) { var query = BuildQuery(request) diff --git a/AsbCloudInfrastructure/Services/DetectOperations/DetectedOperationService.cs b/AsbCloudInfrastructure/Services/DetectOperations/DetectedOperationService.cs index 299db028..db2ca4ea 100644 --- a/AsbCloudInfrastructure/Services/DetectOperations/DetectedOperationService.cs +++ b/AsbCloudInfrastructure/Services/DetectOperations/DetectedOperationService.cs @@ -32,7 +32,7 @@ public class DetectedOperationService : IDetectedOperationService new DetectorFlashing(), new DetectorConditioning(), }; - + public DetectedOperationService( IDetectedOperationRepository operationRepository, IWellOperationCategoryRepository wellOperationCategoryRepository, @@ -177,72 +177,70 @@ public class DetectedOperationService : IDetectedOperationService return dtos; } - public async Task> DetectOperationsAsync(int idTelemetry, DateTimeOffset? beginDate, CancellationToken token) + public async Task<(DateTimeOffset LastDate, IEnumerable Items)> DetectOperationsAsync(int idTelemetry, + TelemetryDataRequest request, + DetectedOperationDto? lastDetectedOperation, + CancellationToken token) { - const int take = 4 * 86_400; - - var detectedOperations = new List(); - DetectedOperationDto? lastDetectedOperation = null; const int minOperationLength = 5; const int maxDetectorsInterpolationFrameLength = 30; const int gap = maxDetectorsInterpolationFrameLength + minOperationLength; + + var telemetries = await telemetryDataSaubService.GetByTelemetryAsync(idTelemetry, request, token); + + var count = telemetries.Count(); + + if (count == 0) + throw new InvalidOperationException("InvalidOperation_EmptyTelemetries"); + var timezone = telemetryService.GetTimezone(idTelemetry); - while (true) + if (telemetries.Count() <= gap) { - var request = new TelemetryDataRequest - { - GeDate = beginDate, - Take = take, - Order = 0 - }; - - var dtos = await telemetryDataSaubService.GetByTelemetryAsync(idTelemetry, request, token); - var detectableTelemetries = dtos - .Where(t => t.BlockPosition >= 0) - .Select(t => new DetectableTelemetry - { - DateTime = new DateTimeOffset(t.DateTime, timezone.Offset), - IdUser = t.IdUser, - Mode = t.Mode, - WellDepth = t.WellDepth, - Pressure = t.Pressure, - HookWeight = t.HookWeight, - BlockPosition = t.BlockPosition, - BitDepth = t.BitDepth, - RotorSpeed = t.RotorSpeed, - AxialLoad = t.AxialLoad, - }).ToArray(); - - if (detectableTelemetries.Length <= gap) - break; - - var isDetected = false; - var positionBegin = 0; - var positionEnd = detectableTelemetries.Length - gap; - while (positionEnd > positionBegin) - { - foreach (var detector in detectors) - { - if (!detector.TryDetect(idTelemetry, detectableTelemetries, positionBegin, positionEnd, lastDetectedOperation, out var result)) - continue; - - detectedOperations.Add(result!.Operation); - lastDetectedOperation = result.Operation; - isDetected = true; - positionBegin = result.TelemetryEnd; - break; - } - - positionBegin += 1; - } - - beginDate = isDetected - ? lastDetectedOperation!.DateEnd - : detectableTelemetries[positionEnd].DateTime; + var lastTelemetry = telemetries.Last(); + var lastDateTelemetry = new DateTimeOffset(lastTelemetry.DateTime, timezone.Offset); + return (lastDateTelemetry, Enumerable.Empty()); } - return detectedOperations; + var detectedOperations = new List(); + + var detectableTelemetries = telemetries.Select(t => new DetectableTelemetry + { + DateTime = new DateTimeOffset(t.DateTime, timezone.Offset), + IdUser = t.IdUser, + Mode = t.Mode, + WellDepth = t.WellDepth, + Pressure = t.Pressure, + HookWeight = t.HookWeight, + BlockPosition = t.BlockPosition, + BitDepth = t.BitDepth, + RotorSpeed = t.RotorSpeed, + AxialLoad = t.AxialLoad, + }).ToArray(); + + var positionBegin = 0; + var positionEnd = detectableTelemetries.Length - gap; + + while (positionEnd > positionBegin) + { + foreach (var detector in detectors) + { + if (!detector.TryDetect(idTelemetry, detectableTelemetries, positionBegin, positionEnd, lastDetectedOperation, + out var result)) + continue; + + detectedOperations.Add(result!.Operation); + lastDetectedOperation = result.Operation; + positionBegin = result.TelemetryEnd; + break; + } + + positionBegin += 1; + } + + var lastDate = lastDetectedOperation?.DateEnd ?? detectableTelemetries[positionEnd].DateTime; + + return (lastDate, detectedOperations); } public async Task DeleteAsync(DetectedOperationByWellRequest request, CancellationToken token) diff --git a/AsbCloudInfrastructure/Services/DetectOperations/Detectors/DetectorAbstract.cs b/AsbCloudInfrastructure/Services/DetectOperations/Detectors/DetectorAbstract.cs index c751dfd5..a0adec19 100644 --- a/AsbCloudInfrastructure/Services/DetectOperations/Detectors/DetectorAbstract.cs +++ b/AsbCloudInfrastructure/Services/DetectOperations/Detectors/DetectorAbstract.cs @@ -48,6 +48,8 @@ namespace AsbCloudInfrastructure.Services.DetectOperations.Detectors while (positionEnd < end) { + //TODO: поиск провалов телеметрии + positionEnd += 1; if (positionEnd > end) break; diff --git a/AsbCloudInfrastructure/Services/DetectOperations/WorkOperationDetection.cs b/AsbCloudInfrastructure/Services/DetectOperations/WorkOperationDetection.cs index 9b50f67d..6ad52f4b 100644 --- a/AsbCloudInfrastructure/Services/DetectOperations/WorkOperationDetection.cs +++ b/AsbCloudInfrastructure/Services/DetectOperations/WorkOperationDetection.cs @@ -1,54 +1,93 @@ using System; +using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Tasks; using AsbCloudApp.Data; using AsbCloudApp.Repositories; +using AsbCloudApp.Requests; using AsbCloudApp.Services; using AsbCloudInfrastructure.Background; using Microsoft.Extensions.DependencyInjection; namespace AsbCloudInfrastructure.Services.DetectOperations; -public class WorkOperationDetection: Work +public class WorkOperationDetection : Work { + private static readonly IDictionary CacheOfStartDatesByTelemetryId = new Dictionary(); - public WorkOperationDetection() - :base("Operation detection") - { - Timeout = TimeSpan.FromMinutes(20); - OnErrorAsync = (id, exception, token) => - { - var text = $"work {id}, when {CurrentState?.State}, throw error:{exception.Message}"; - Trace.TraceWarning(text); - return Task.CompletedTask; - }; - } + public WorkOperationDetection() + : base("Operation detection") + { + Timeout = TimeSpan.FromMinutes(20); + OnErrorAsync = (id, exception, _) => + { + var text = $"work {id}, when {CurrentState?.State}, throw error:{exception.Message}"; + Trace.TraceWarning(text); + return Task.CompletedTask; + }; + } - protected override async Task Action(string id, IServiceProvider services, Action onProgressCallback, CancellationToken token) - { - var telemetryRepository = services.GetRequiredService>(); - var detectedOperationRepository = services.GetRequiredService(); - var detectedOperationService = services.GetRequiredService(); + protected override async Task Action(string id, + IServiceProvider services, + Action onProgressCallback, + CancellationToken token) + { + var telemetryRepository = services.GetRequiredService>(); + var detectedOperationRepository = services.GetRequiredService(); + var detectedOperationService = services.GetRequiredService(); + var telemetryDataSaubService = services.GetRequiredService(); - var telemetryIds = (await telemetryRepository.GetAllAsync(token)) - .Select(t => t.Id) - .ToArray(); + var idsTelemetry = (await telemetryRepository.GetAllAsync(token)) + .Select(t => t.Id) + .ToArray(); - var lastDetectedDates = await detectedOperationRepository.GetLastDetectedDatesAsync(token); + var telemetriesDateRanges = await telemetryDataSaubService.GetDateRangesAsync(token); - for (var i = 0; i < telemetryIds.Length; i++) - { - var telemetryId = telemetryIds[i]; - - var beginDate = lastDetectedDates.TryGetValue(telemetryId, out var date) ? date : (DateTimeOffset?)null; + var lastDetectedOperations = await detectedOperationRepository.GetLastDetectedOperationsAsync(token); - onProgressCallback($"Start detecting telemetry: {telemetryId} from {beginDate}", i / telemetryIds.Length); - var detectedOperations = await detectedOperationService.DetectOperationsAsync(telemetryId, beginDate, token); + for (int i = 0; i < idsTelemetry.Length; i++) + { + var idTelemetry = idsTelemetry[i]; + + if (!telemetriesDateRanges.TryGetValue(idTelemetry, out var telemetryDateRange)) + continue; - if (detectedOperations.Any()) - await detectedOperationRepository.InsertRangeAsync(detectedOperations, token); - } - } -} + var dateBegin = telemetryDateRange.From.DateTime; + var dateEnd = telemetryDateRange.To.DateTime; + + if (lastDetectedOperations.TryGetValue(idTelemetry, out var lastDetectedOperation)) + dateBegin = lastDetectedOperation.DateEnd.UtcDateTime; + + if (CacheOfStartDatesByTelemetryId.TryGetValue(idTelemetry, out var dateBeginFromCahce)) + dateBegin = dateBeginFromCahce; + + onProgressCallback.Invoke($"Start detecting telemetry: {idTelemetry} from {dateBegin}", i / idsTelemetry.Length); + + const int pointsCount = 4 * 86_400; + + while (dateBegin < dateEnd) + { + var request = new TelemetryDataRequest + { + GeDate = dateBegin, + Take = pointsCount, + Order = 0, + GeBlockPosition = 0 + }; + + var detectedOperations = + await detectedOperationService.DetectOperationsAsync(idTelemetry, request, lastDetectedOperation, token); + + await detectedOperationRepository.InsertRangeAsync(detectedOperations.Items, token); + + dateBegin = detectedOperations.LastDate.UtcDateTime; + + CacheOfStartDatesByTelemetryId[idTelemetry] = dateBegin; + + await detectedOperationRepository.InsertRangeAsync(detectedOperations.Items, token); + } + } + } +} \ No newline at end of file diff --git a/AsbCloudInfrastructure/Services/SAUB/TelemetryDataBaseService.cs b/AsbCloudInfrastructure/Services/SAUB/TelemetryDataBaseService.cs index 88b11941..ea2b7912 100644 --- a/AsbCloudInfrastructure/Services/SAUB/TelemetryDataBaseService.cs +++ b/AsbCloudInfrastructure/Services/SAUB/TelemetryDataBaseService.cs @@ -177,7 +177,7 @@ namespace AsbCloudInfrastructure.Services.SAUB return dtos; } - private IQueryable BuildQuery(int idTelemetry, TelemetryDataRequest request) + protected virtual IQueryable BuildQuery(int idTelemetry, TelemetryDataRequest request) { var dbSet = db.Set(); @@ -268,7 +268,7 @@ namespace AsbCloudInfrastructure.Services.SAUB }; return range; } - + public DatesRangeDto? GetRange(int idWell) { var telemetry = telemetryService.GetOrDefaultTelemetryByIdWell(idWell); diff --git a/AsbCloudInfrastructure/Services/SAUB/TelemetryDataSaubService.cs b/AsbCloudInfrastructure/Services/SAUB/TelemetryDataSaubService.cs index 78a63b2c..c66cd984 100644 --- a/AsbCloudInfrastructure/Services/SAUB/TelemetryDataSaubService.cs +++ b/AsbCloudInfrastructure/Services/SAUB/TelemetryDataSaubService.cs @@ -13,6 +13,8 @@ using System.Linq; using System.Text.Csv; using System.Threading; using System.Threading.Tasks; +using AsbCloudApp.Data; +using AsbCloudApp.Requests; namespace AsbCloudInfrastructure.Services.SAUB; @@ -174,4 +176,33 @@ public class TelemetryDataSaubService : TelemetryDataBaseService> GetDateRangesAsync(CancellationToken token) + { + return await db.Set().GroupBy(x => x.IdTelemetry) + .Select(g => new + { + IdTelemetry = g.Key, + From = g.Min(x => x.DateTime), + To = g.Max(x => x.DateTime) + }) + .ToDictionaryAsync(x => x.IdTelemetry, x => new DatesRangeDto + { + From = x.From, + To = x.To + }, token); + } + + protected override IQueryable BuildQuery(int idTelemetry, TelemetryDataRequest request) + { + var query = base.BuildQuery(idTelemetry, request); + + if (request.GeBlockPosition.HasValue) + query = query.Where(e => e.BlockPosition >= request.GeBlockPosition); + + if (request.LeBlockPosition.HasValue) + query = query.Where(e => e.BlockPosition <= request.LeBlockPosition); + + return query; + } }