DD.WellWorkover.Cloud/AsbCloudInfrastructure/Services/Subsystems/SubsystemOperationTimeBackgroundService.cs

using AsbCloudDb.Model;
using AsbCloudDb.Model.Subsystems;
using AsbCloudInfrastructure.Services.Subsystems.Utils;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using System;
using System.Collections.Generic;
using System.Data;
using System.Data.Common;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace AsbCloudInfrastructure.Services.Subsystems
{
#nullable enable    
    internal class SubsystemOperationTimeBackgroundService : BackgroundService
    {
        private readonly string connectionString;
        private readonly TimeSpan period = TimeSpan.FromHours(1);
        private const int idSubsytemTorqueMaster = 65537;
        private const int idSubsytemSpinMaster = 65536;
        private const int idSubsytemAkb = 1;
        private const int idSubsytemMse = 2;

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

        protected override async Task ExecuteAsync(CancellationToken token)
        {
            var timeToStart = DateTime.Now;
            var options = new DbContextOptionsBuilder<AsbCloudDbContext>()
                .UseNpgsql(connectionString)
                .Options;
            while (!token.IsCancellationRequested)
            {
                if (DateTime.Now > timeToStart)
                {
                    timeToStart = DateTime.Now + period;
                    try
                    {
                        using var context = new AsbCloudDbContext(options);
                        var added = await OperationTimeAllTelemetriesAsync(context, token);
                        Trace.TraceInformation($"Total subsystem operation time complete. Added {added} operations time.");
                    }
                    catch (Exception ex)
                    {
                        Trace.TraceError(ex.Message);
                    }
                    GC.Collect();
                }
                var ms = (int)(timeToStart - 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 static async Task<int> OperationTimeAllTelemetriesAsync(IAsbCloudDbContext db, CancellationToken token)
        {
            var lastDetectedDates = await db.SubsystemOperationTimes
                .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 telemetryLastDetectedDates = telemetryIds
                .GroupJoin(lastDetectedDates,
                    t => t,
                    o => o.IdTelemetry,
                    (outer, inner) => new
                    {
                        IdTelemetry = outer,
                        inner.SingleOrDefault()?.LastDate,
                    });

            var affected = 0;
            foreach (var item in telemetryLastDetectedDates)
            {
                var newOperationsSaub = await OperationTimeSaubAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
                if (newOperationsSaub?.Any() == true)
                {
                    db.SubsystemOperationTimes.AddRange(newOperationsSaub);
                    affected += await db.SaveChangesAsync(token);
                }
                var newOperationsSpin = await OperationTimeSpinAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
                if (newOperationsSpin?.Any() == true)
                {
                    db.SubsystemOperationTimes.AddRange(newOperationsSpin);
                    affected += await db.SaveChangesAsync(token);
                }
            }
            return affected;
        }

        private static async Task<DbDataReader> ExecuteReaderAsync(IAsbCloudDbContext db, string query, CancellationToken token)
        {
            var connection = db.Database.GetDbConnection();
            if (
                connection?.State is null || 
                connection.State == ConnectionState.Broken || 
                connection.State == ConnectionState.Closed)
            {
                await db.Database.OpenConnectionAsync(token);
                connection = db.Database.GetDbConnection();
            }
            using var command = connection.CreateCommand();
            command.CommandText = query;

            var result = await command.ExecuteReaderAsync(token);
            return result;
        }

        private static async Task<IEnumerable<SubsystemOperationTime>> OperationTimeSaubAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token)
        {
            static bool isSubsytemAkb(short? mode)
            {
                if (mode is null)
                    return false;
                if (mode == 1 | mode == 3)
                    return true;
                return false;
            }

            static bool IsSubsystemMse(short? state)
            {
                if (state is null)
                    return false;
                if ((state & 1) > 0)
                    return true;
                return false;
            }

            var query =
                $"select tt.date, tt.mode, tt.well_depth, tt.mse_state " +
                $"from ( " +
                $"  select " +
                $"      date, " +
                $"      mode, " +
                $"      mse_state, " +
                $"      well_depth, " +
                $"      lag(mode,1) over (order by date) as mode_lag, " +
                $"      lead(mode,1) over (order by date) as mode_lead " +
                $"  from t_telemetry_data_saub " +
                $"  where id_telemetry = {idTelemetry} and well_depth is not null and well_depth > 0" +
                $"  order by date ) as tt " +
                $"where (tt.mode_lag is null or (tt.mode != tt.mode_lag and tt.mode_lead != tt.mode_lag)) and tt.date >= '{begin:u}' " +
                $"order by tt.date;";

            using var result = await ExecuteReaderAsync(db, query, token);

            var subsystemsOperationTimes = new List<SubsystemOperationTime>();
            
            (bool isEnable, DateTimeOffset date, float depth) akbPre = default;
            (bool isEnable, DateTimeOffset date, float depth) msePre = default;

            while (result.Read())
            {
                var mode = result.GetFieldValue<short?>(1);
                var state = result.GetFieldValue<short?>(3);

                var isAkbEnable = isSubsytemAkb(mode);
                var isMseEnable = IsSubsystemMse(state);
                var date = result.GetFieldValue<DateTimeOffset>(0);
                var depth = result.GetFieldValue<float>(2);                

                if (!akbPre.isEnable && isAkbEnable)
                {
                    akbPre = (true, date, depth);
                }
                else if (akbPre.isEnable && !isAkbEnable) 
                {
                    var subsystemOperationTime = new SubsystemOperationTime
                    {
                        IdTelemetry = idTelemetry,
                        IdSubsystem = idSubsytemAkb,
                        DateStart = akbPre.date,
                        DateEnd = date,
                        DepthStart = akbPre.depth,
                        DepthEnd = depth,
                    };
                    if (IsValid(subsystemOperationTime))
                        subsystemsOperationTimes.Add(subsystemOperationTime);
                    akbPre.isEnable = false;
                }

                if (!msePre.isEnable && isMseEnable)
                {
                    msePre = (true, date, depth);
                }
                else if (msePre.isEnable && !isMseEnable)
                {
                    var subsystemOperationTime = new SubsystemOperationTime
                    {
                        IdTelemetry = idTelemetry,
                        IdSubsystem = idSubsytemMse,
                        DateStart = akbPre.date,
                        DateEnd = date,
                        DepthStart = akbPre.depth,
                        DepthEnd = depth,
                    };
                    if (IsValid(subsystemOperationTime))
                        subsystemsOperationTimes.Add(subsystemOperationTime);
                    
                    msePre.isEnable = false;
                }
            }
            
            return subsystemsOperationTimes;
        }

        private static async Task<IEnumerable<SubsystemOperationTime>> OperationTimeSpinAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token)
        {
            static int? GetSubsytemId(short? mode, int? state)
            {
                // При изменении следующего кода сообщи в Vladimir.Sobolev@nedra.digital
                if (state == 7 && (mode & 2) > 0)
                    return idSubsytemTorqueMaster;// демпфер

                if (state != 0 & state != 6 & state != 7)
                    return idSubsytemSpinMaster;// осцилляция

                return null;
            }

            var querySpin =
                $"select " +
                $"  tspin.date, " +
                $"  tspin.mode, " +
                $"  tspin.state " +                
                $"from ( " +
                $"  select " +
                $"      date, " +
                $"      mode, " +
                $"      lag(mode, 1) over (order by date) as mode_lag, " +
                $"      lead(mode, 1) over (order by date) as mode_lead, " +
                $"      state, " +
                $"      lag(state, 1) over (order by date) as state_lag " +
                $"  from t_telemetry_data_spin " +
                $"  where id_telemetry = {idTelemetry} and date >= '{begin:u}'" +
                $"  order by date ) as tspin " +                
                $"where mode_lag is null or state_lag is null or (mode != mode_lag and mode_lead != mode_lag) or state != state_lag " +
                $"order by date;";

            var rows = new List<(int? IdSubsystem, DateTimeOffset Date)>(32);
            {
                using var resultSpin = await ExecuteReaderAsync(db, querySpin, token);
                int? idSubsystemLast = null;
                while (resultSpin.Read())
                {
                    var mode = resultSpin.GetFieldValue<short?>(1);
                    var state = resultSpin.GetFieldValue<short?>(2);
                    var idSubsystem = GetSubsytemId(mode, state);
                    if(idSubsystemLast != idSubsystem)
                    {
                        idSubsystemLast = idSubsystem;
                        var date = resultSpin.GetFieldValue<DateTimeOffset>(0);
                        rows.Add((idSubsystem, date));
                    }                    
                }
                await resultSpin.DisposeAsync();
            }

            if (rows.Count < 2)
                return Enumerable.Empty<SubsystemOperationTime>();

            var minSpinDate = rows.Min(i => i.Date);
            var maxSpinDate = rows.Max(i => i.Date);
            var depthInterpolation = await GetInterpolation(db, idTelemetry, minSpinDate, maxSpinDate, token);

            if (depthInterpolation is null)
                return Enumerable.Empty<SubsystemOperationTime>();

            var subsystemsOperationTimes = new List<SubsystemOperationTime>(32);
            
            for (int i = 1; i < rows.Count; i++)
            {
                var r0 = rows[i - 1];
                var r1 = rows[i];
                if (r0.IdSubsystem is not null && r0.IdSubsystem != r1.IdSubsystem)
                {
                    var subsystemOperationTime = new SubsystemOperationTime()
                    {
                        IdTelemetry = idTelemetry,
                        IdSubsystem = r0.IdSubsystem.Value,
                        DateStart = r0.Date,
                        DateEnd = r1.Date,
                        DepthStart = depthInterpolation.GetDepth(r0.Date),
                        DepthEnd = depthInterpolation.GetDepth(r1.Date),
                    };
           
                    if (IsValid(subsystemOperationTime))
                        subsystemsOperationTimes.Add(subsystemOperationTime);
                }
            }
            
            return subsystemsOperationTimes;
        }

        private static bool IsValid(SubsystemOperationTime item)
        {
            var validateCode = GetValidateErrorCode(item);
            if (validateCode != 0)
            {
                var str = System.Text.Json.JsonSerializer.Serialize(item);
                Trace.TraceWarning($"Wrong({validateCode}) SubsystemOperationTime: {str}");
            }
            return validateCode == 0;
        }

        private static int GetValidateErrorCode(SubsystemOperationTime item)
        {
            if (item.DateStart > item.DateEnd)
                return -1;
            if ((item.DateEnd - item.DateStart).TotalHours > 48)
                return -2;
            if (item.DepthEnd < item.DepthStart)
                return -3;
            if (item.DepthEnd - item.DepthStart > 2000d)
                return -4;
            if (item.DepthEnd < 0d)
                return -5;
            if (item.DepthStart < 0d)
                return -6;
            if (item.DepthEnd > 24_0000d)
                return -7;
            if (item.DepthStart > 24_0000d)
                return -8;
            return 0;
        }

        private static async Task<DepthInterpolation?> GetInterpolation(IAsbCloudDbContext db, int idTelemetry, DateTimeOffset dateBegin, DateTimeOffset dateEnd, CancellationToken token)
        {
            var dataDepthFromSaub = await db.TelemetryDataSaub
                .Where(d => d.IdTelemetry == idTelemetry)
                .Where(d => d.DateTime >= dateBegin)
                .Where(d => d.DateTime <= dateEnd)
                .Where(d => d.WellDepth != null)
                .Where(d => d.WellDepth > 0)
                .GroupBy(d => Math.Ceiling(d.WellDepth ?? 0 * 10))
                .Select(g => new {
                    DateMin = g.Min(d => d.DateTime),
                    DepthMin = g.Min(d => d.WellDepth) ?? 0,
                })
                .OrderBy(i => i.DateMin)
                .ToArrayAsync(token);

            if (!dataDepthFromSaub.Any())
                return null;

            var depthInterpolation = new DepthInterpolation(dataDepthFromSaub.Select(i=>(i.DateMin, i.DepthMin)));
            return depthInterpolation;
        }
    }
#nullable disable
}