DD.WellWorkover.Cloud/AsbCloudInfrastructure/Services/DetectOperations/OperationDetectionBackgroundService.cs

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;
using AsbCloudInfrastructure.Services.DetectOperations.Detectors;

namespace AsbCloudInfrastructure.Services.DetectOperations
{
    public class OperationDetectionBackgroundService : BackgroundService
    {
        private readonly int stepLength = 3;
        private readonly int fragmentLength = 6;
        private readonly string connectionString;
        private readonly TimeSpan period = TimeSpan.FromHours(1);

        private static readonly Func<DetectableTelemetry[], int, bool> isBitLt0_03mToBottom = FragmentDetector.MakeInstantDelegate(d => 
            (double)(d.WellDepth - d.BitDepth) < 0.03d);
        private static readonly Func<DetectableTelemetry[], int, bool> isBitLt2_5mToBottom = FragmentDetector.MakeInstantDelegate(d =>
            (double)(d.WellDepth - d.BitDepth) < 2.5d);
        private static readonly Func<DetectableTelemetry[], int, bool> isPressureGt25 = FragmentDetector.MakeInstantDelegate(d => 
            d.Pressure > 25);
        private static readonly Func<DetectableTelemetry[], int, bool> isPressureLt15 = FragmentDetector.MakeInstantDelegate(d => 
            d.Pressure < 15);
        private static readonly Func<DetectableTelemetry[], int, bool> isBlockPositionLt8 = FragmentDetector.MakeInstantDelegate(d => 
            d.BlockPosition < 8);
        private static readonly Func<DetectableTelemetry[], int, bool> isHookWeightLt20 = FragmentDetector.MakeInstantDelegate(d => 
            d.HookWeight < 20);
        private static readonly Func<DetectableTelemetry[], int, bool> isRotorSpeedAvgGt5 = FragmentDetector.MakeInterpolationDelegate(d => (double)d.RotorSpeed, line => 
            line.IsAverageYGreaterThan(5), 12);
        private static readonly Func<DetectableTelemetry[], int, bool> isRotorSpeedAvgLt5 = FragmentDetector.MakeInterpolationDelegate(d => (double)d.RotorSpeed, line => 
            line.IsAverageYLessThan(5), 12);

        private static readonly List<IDetector> detectors = new List<IDetector> 
        {
            new FragmentDetector("isBitLt2_5mToBottom", isBitLt2_5mToBottom,
                new FragmentDetector("isBitLt0_03mToBottom", isBitLt0_03mToBottom,
                    new FragmentDetector("isPressureGt25", isPressureGt25,
                        new FragmentDetector("isRotorSpeedAvgGt5", isRotorSpeedAvgGt5, new OperationDrilling(2)),
                        new FragmentDetector("isRotorSpeedAvgLt5", isRotorSpeedAvgLt5, new OperationDrilling(3)))),
                new FragmentDetector("isPressureLt15", isPressureLt15,
                    new FragmentDetector("isBlockPositionLt8", isBlockPositionLt8,
                        new FragmentDetector("isHookWeightLt20", isHookWeightLt20, new OperationSlipsTime())))),
        };

        public OperationDetectionBackgroundService(IConfiguration configuration)
        {
            connectionString = configuration.GetConnectionString("DefaultConnection");
        }

        protected override async Task ExecuteAsync(CancellationToken token = default)
        {
            var timeToStartAnalysis = DateTime.Now;
            var options = new DbContextOptionsBuilder<AsbCloudDbContext>()
                .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<int> 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<IEnumerable<DetectedOperation>> 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; // 4 дня
            var startDate = begin;
            var detectedOperations = new List<DetectedOperation>(8);

            while (true)
            {
                var data = await query
                    .Where(d => d.DateTime > startDate)
                    .Take(take)
                    .ToArrayAsync(token);

                if (data.Length < 2 * fragmentLength + stepLength)
                    break;

                var isDetected = false;
                var positionBegin = 0;
                var positionEnd = data.Length - fragmentLength;
                
                foreach (var detector in detectors)
                {
                    if (detector is FragmentDetector fragmentDetector)
                    {
                        var minLengthToDetect = fragmentDetector.StepLength + fragmentDetector.FragmentLength;
                        if (data.Length < positionBegin + minLengthToDetect)
                            continue;

                        if (fragmentDetector.TryDetect(idTelemetry, data, positionBegin, positionEnd, out IEnumerable<OperationDetectorResult> results))
                        {
                            isDetected = true;
                            var operations = results.Select(r => r.Operation);
                            detectedOperations.AddRange(operations);                                
                            positionBegin = results.Max(r=>r.TelemetryEnd) + 1;
                            break;
                        }
                    }
                }

                if (isDetected)
                {
                    startDate = detectedOperations.Max(o => o.DateEnd);
                }
                else
                {
                    int i = positionEnd - (int)(data.Length * 0.8);
                    if (i < 0)
                        break;

                    startDate = data[i].DateTime;
                }
            }

            return detectedOperations;
        }
    }
}