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

using AsbCloudDb.Model;
using AsbCloudDb.Model.Subsystems;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Npgsql;
using System;
using System.Collections.Generic;
using System.Data;
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<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;
            }            

            static List<SubsystemOperationTime> GetSubsystemOperationTimes (List<DataRow> dataRows, Predicate<DataRow> satisfyCondition, int idSubsystem, int idTelemetry)
            {
                if (dataRows is null)
                    return new List<SubsystemOperationTime>();
                if (dataRows.Count < 2)
                    return new List<SubsystemOperationTime>();
                var listSubsystemOperationTime = new List<SubsystemOperationTime>();
                var foundSubsystem = satisfyCondition(dataRows[0]);
                var dateStart = dataRows[0].Date;
                var depthStart = dataRows[0].Depth;               

                for (int i = 1; i<dataRows.Count; i++)
                {
                    var dateEnd = dataRows[i].Date;
                    var depthEnd = dataRows[i].Depth;                    
                    var currentSatisfy = satisfyCondition(dataRows[i]);
                    var endSubsystem = !currentSatisfy;                    
                    if (foundSubsystem && endSubsystem)
                    {
                        var operationTimeItem = new SubsystemOperationTime()
                        {
                            IdTelemetry = idTelemetry,
                            IdSubsystem = idSubsystem,
                            DateStart = dateStart,
                            DateEnd = dateEnd,
                            DepthStart = depthStart,
                            DepthEnd = depthEnd
                        };
                        listSubsystemOperationTime.Add(operationTimeItem);
                        foundSubsystem = false;
                    }
                    if (currentSatisfy && !foundSubsystem)
                    {                        
                        dateStart = dateEnd;
                        foundSubsystem = true;
                        depthStart = depthEnd;
                    }                   
                }
                return listSubsystemOperationTime;
            }

            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_prev " +
                $"  from t_telemetry_data_saub " +
                $"  where id_telemetry = @idTelemetry " +
                $"  order by date ) as tt " +
                $"where (tt.mode_prev is null or tt.mode != tt.mode_prev) and tt.date >= @begin " +
                $"order by tt.date;";

            var idTelemetryParam = new NpgsqlParameter("@idTelemetry", idTelemetry);
            var beginParam = new NpgsqlParameter("@begin", begin);

            await db.Database.OpenConnectionAsync(token);
            using var command = db.Database.GetDbConnection().CreateCommand();
            command.CommandText = query;
            command.Parameters.Add(idTelemetryParam);
            command.Parameters.Add(beginParam);

            using var result = await command.ExecuteReaderAsync(token);

            var subsystemOperationTime = new List<SubsystemOperationTime>();
            var dataRowList = new List<DataRow>();
            while (result.Read())
            {
                var dateRowItem = new DataRow()
                {
                    Date = result.GetFieldValue<DateTimeOffset>(0),
                    Mode = result.GetFieldValue<short?>(1),
                    Depth = result.GetFieldValue<float?>(2),
                    State = result.GetFieldValue<short?>(3)
                };
                dataRowList.Add(dateRowItem);
            }
            var akbOperationTimes = GetSubsystemOperationTimes(dataRowList, d => isSubsytemAkb(d.Mode), idSubsytemAkb, idTelemetry);
            var mseOperationTimes = GetSubsystemOperationTimes(dataRowList, d => IsSubsystemMse(d.State), idSubsytemMse, idTelemetry);
            subsystemOperationTime.AddRange(akbOperationTimes);
            subsystemOperationTime.AddRange(mseOperationTimes);
            return subsystemOperationTime;
        }

        private static async Task<IEnumerable<SubsystemOperationTime>?> OperationTimeSpinAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token)
        {
            static int? GetSubsytemId(short mode, int state)
            {
                if (state == 7 && (mode & 2) > 0)
                    return idSubsytemTorqueMaster;

                if (state != 0 & state != 6 & state != 7)
                    return idSubsytemSpinMaster;

                return null;
            }

            var query =
                $"select " +
                $"  tspin.date, " +
                $"  tspin.mode, " +
                $"  tspin.state, " +
                $"  tsaub.well_depth " +
                $"from ( " +
                $"  select " +
                $"      date, " +
                $"      mode, " +
                $"      lag(mode, 1) over (order by date) as mode_pre, " +
                $"      state, " +
                $"      lag(state, 1) over (order by date) as state_pre " +
                $"  from t_telemetry_data_spin " +
                $"  where id_telemetry = @idTelemetry and date >= @begin" +
                $"  order by date ) as tspin " +
                $"left outer join ( " +
                $"	select " +
                $"    	date, " +
                $"    	well_depth " +
                $"	from t_telemetry_data_saub " +
                $"	where id_telemetry = @idTelemetry and date >= @begin) as tsaub " +
                $"on EXTRACT(EPOCH from tspin.date) = EXTRACT(EPOCH from tsaub.date) " +
                $"where mode_pre is null or state_pre is null or mode != mode_pre or state != state_pre " +
                $"order by date;";

            var idTelemetryParam = new NpgsqlParameter("@idTelemetry", idTelemetry);
            var beginParam = new NpgsqlParameter("@begin", begin);

            await db.Database.OpenConnectionAsync(token);
            using var command = db.Database.GetDbConnection().CreateCommand();
            command.CommandText = query;
            command.Parameters.Add(idTelemetryParam);
            command.Parameters.Add(beginParam);

            using var result = await command.ExecuteReaderAsync(token);

            var subsystemOperationTime = new List<SubsystemOperationTime>(32);

            if (result.Read())
            {
                var mode = result.GetFieldValue<short>(1);
                var state = result.GetFieldValue<int>(2);
                var idSubsystem = GetSubsytemId(mode, state);
                var dateStart = result.GetFieldValue<DateTimeOffset>(0);
                var depthStart = result.GetFieldValue<float?>(3);

                while (result.Read())
                {
                    var dateEnd = result.GetFieldValue<DateTimeOffset>(0);
                    var depthEnd = result.GetFieldValue<float?>(3);                    
                    if (idSubsystem.HasValue)
                    {
                        var operationTimeItem = new SubsystemOperationTime()
                        {
                            IdTelemetry = idTelemetry,
                            IdSubsystem = idSubsystem.Value,
                            DateStart = dateStart,
                            DateEnd = dateEnd,
                            DepthStart = depthStart,
                            DepthEnd = depthEnd
                        };
                        subsystemOperationTime.Add(operationTimeItem);
                    }
                    mode = result.GetFieldValue<short>(1);
                    state = result.GetFieldValue<int>(2);
                    idSubsystem = GetSubsytemId(mode, state);
                    dateStart = dateEnd;
                    depthStart = depthEnd;
                }
            }
            return subsystemOperationTime;
        }
    }

    internal class DataRow
    {        
        public DateTimeOffset Date { get; set; }
        public short? Mode { get; set; }
        public float? Depth { get; set; }
        public short? State { get; set; }
    }

#nullable disable
}