on.nemtina/DD.WellWorkover.Cloud
1
0
Fork 0
forked from ddrilling/AsbCloudServer
Code Pull Requests Activity

Add backgroundController for monitoring works states

Browse Source
This commit is contained in:
ngfrolov 2023-10-09 13:12:45 +05:00
parent 1560c6bf91
commit c28315b795
Signed by untrusted user who does not match committer: ng.frolov
GPG Key ID: E99907A0357B29A7
23 changed files with 867 additions and 848 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
AsbCloudApp/Data/BackgroudWorkDto.cs → AsbCloudApp/Data/BackgroundWorkDto.cs
View File

@ -6,7 +6,7 @@ namespace AsbCloudApp.Data
/// <summary>
/// Информация о фоновой работе
/// </summary>
public class BackgroudWorkDto
public class BackgroundWorkDto
{
/// <summary>
/// Идентификатор работы. Должен быть уникальным. Используется в логах и передается в колбэки.
@ -59,6 +59,11 @@ namespace AsbCloudApp.Data
/// </summary>
public class LastErrorInfo : LastCompleteInfo
{
/// <summary>
///
/// </summary>
/// <param name="state"></param>
/// <param name="errorText"></param>
public LastErrorInfo(CurrentStateInfo state, string errorText)
: base(state)
{
@ -96,6 +101,10 @@ namespace AsbCloudApp.Data
/// </summary>
public string State { get; init; }
/// <summary>
/// ctor
/// </summary>
/// <param name="state"></param>
public LastCompleteInfo(CurrentStateInfo state)
{
Start = state.Start;

5
AsbCloudApp/Data/WellboreDto.cs
View File

@ -7,7 +7,10 @@ namespace AsbCloudApp.Data;
/// </summary>
public class WellboreDto
{
public WellWithTimezoneDto Well { get; set; }
/// <summary>
/// Скважина
/// </summary>
public WellWithTimezoneDto Well { get; set; } = null!;
/// <summary>
/// Идентификатор

1
AsbCloudApp/Services/Notifications/NotificationService.cs
View File

@ -90,7 +90,6 @@ public class NotificationService
/// Отправка уведомлений, которые не были отправлены
/// </summary>
/// <param name="idUser"></param>
/// <param name="request"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async Task RenotifyAsync(int idUser, CancellationToken cancellationToken)

2
AsbCloudInfrastructure/Background/BackgroundWorker.cs
View File

@ -40,7 +40,7 @@ public class BackgroundWorker : BackgroundService
var result = await work.Start(scope.ServiceProvider, token);
if (!result)
WorkStore.Falled.Add(work);
WorkStore.Felled.Add(work);
CurrentWork = null;
await Task.Delay(minDelay, token);

160
AsbCloudInfrastructure/Background/Work.cs
View File

@ -3,86 +3,96 @@ using System;
using System.Threading;
using System.Threading.Tasks;
namespace AsbCloudInfrastructure.Background
namespace AsbCloudInfrastructure.Background;
/// <summary>
/// Класс разовой работы.
/// Разовая работа приоритетнее периодической.
/// </summary>
public abstract class Work : BackgroundWorkDto
{
/// <summary>
/// Класс разовой работы.
/// Разовая работа приоритетнее периодической.
/// </summary>
public class Work : BackgroudWorkDto
private sealed class WorkBase : Work
{
private Func<string, IServiceProvider, Action<string, double?>, CancellationToken, Task> ActionAsync { get; }
/// <summary>
/// Делегат обработки ошибки.
/// Не должен выполняться долго.
/// </summary>
public Func<string, Exception, CancellationToken, Task>? OnErrorAsync { get; set; }
public TimeSpan OnErrorHandlerTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Базовая работа
/// </summary>
/// <param name="id"></param>
/// <param name="actionAsync">
/// Делегат работы.
/// <para>
/// Параметры:
/// <list type="number">
/// <item>
/// <term>string</term>
/// <description>Id Идентификатор работы</description>
/// </item>
/// <item>
/// <term>IServiceProvider</term>
/// <description>Поставщик сервисов</description>
/// </item>
/// <item>
/// <term>Action&lt;string, double?&gt;</term>
/// <description>on progress callback. String - new state text. double? - optional progress 0-100%.</description>
/// </item>
/// <item>
/// <term>CancellationToken</term>
/// <description>Токен отмены задачи</description>
/// </item>
/// </list>
/// </para>
/// </param>
public Work(string id, Func<string, IServiceProvider, Action<string, double?>, CancellationToken, Task> actionAsync)
public WorkBase(string id, Func<string, IServiceProvider, Action<string, double?>, CancellationToken, Task> actionAsync)
: base(id)
{
Id = id;
ActionAsync = actionAsync;
}
/// <summary>
/// Запустить работу
/// </summary>
/// <param name="services"></param>
/// <param name="token"></param>
/// <returns>True - susess, False = Fail</returns>
public async Task<bool> Start(IServiceProvider services, CancellationToken token)
{
SetStatusStart();
try
{
var task = ActionAsync(Id, services, UpdateStatus, token);
await task.WaitAsync(Timeout, token);
SetStatusComplete();
return true;
}
catch (Exception exception)
{
SetLastError(exception.Message);
if (OnErrorAsync is not null)
{
var task = Task.Run(
async () => await OnErrorAsync(Id, exception, token),
token);
await task.WaitAsync(OnErrorHandlerTimeout, token);
}
}
return false;
}
protected override Task Action(string id, IServiceProvider services, Action<string, double?> onProgressCallback, CancellationToken token)
=> ActionAsync(id, services, onProgressCallback, token);
}
/// <summary>
/// Делегат обработки ошибки.
/// Не должен выполняться долго.
/// </summary>
public Func<string, Exception, CancellationToken, Task>? OnErrorAsync { get; set; }
/// <summary>
/// макс продолжительность обработки исключения
/// </summary>
public TimeSpan OnErrorHandlerTimeout { get; set; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Базовая работа
/// </summary>
/// <param name="id"></param>
public Work(string id)
{
Id = id;
}
/// <summary>
/// Создать работу на основе делегата
/// </summary>
/// <param name="id"></param>
/// <param name="actionAsync"></param>
/// <returns></returns>
[Obsolete("Use implement Work class")]
public static Work CreateByDelegate(string id, Func<string, IServiceProvider, Action<string, double?>, CancellationToken, Task> actionAsync)
{
return new WorkBase(id, actionAsync);
}
/// <summary>
/// Запустить работу
/// </summary>
/// <param name="services"></param>
/// <param name="token"></param>
/// <returns>True - success, False = fail</returns>
public async Task<bool> Start(IServiceProvider services, CancellationToken token)
{
SetStatusStart();
try
{
var task = Action(Id, services, UpdateStatus, token);
await task.WaitAsync(Timeout, token);
SetStatusComplete();
return true;
}
catch (Exception exception)
{
SetLastError(exception.Message);
if (OnErrorAsync is not null)
{
var task = Task.Run(
async () => await OnErrorAsync(Id, exception, token),
token);
await task.WaitAsync(OnErrorHandlerTimeout, token);
}
}
return false;
}
/// <summary>
/// делегат фоновой работы
/// </summary>
/// <param name="id">Идентификатор работы</param>
/// <param name="services">Поставщик сервисов</param>
/// <param name="onProgressCallback">on progress callback. String - new state text. double? - optional progress 0-100%</param>
/// <param name="token"></param>
/// <returns></returns>
protected abstract Task Action(string id, IServiceProvider services, Action<string, double?> onProgressCallback, CancellationToken token);
}

25
AsbCloudInfrastructure/Background/WorkStore.cs
View File

@ -18,16 +18,35 @@ public class WorkStore
/// Список периодических задач
/// </summary>
public IEnumerable<WorkPeriodic> Periodics => periodics;
/// <summary>
/// Работы выполняемые один раз
/// </summary>
public Queue<Work> RunOnceQueue { get; private set; } = new(8);
/// <summary>
/// Завершывшиеся с ошибкой
/// Завершившиеся с ошибкой
/// </summary>
public CyclycArray<Work> Falled { get; } = new(16);
public CyclycArray<Work> Felled { get; } = new(16);
/// <summary>
/// Добавить фоновую работу выполняющуюся с заданным периодом
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="period"></param>
public void AddPeriodic<T>(TimeSpan period)
where T : Work, new()
{
var work = new T();
var periodic = new WorkPeriodic(work, period);
periodics.Add(periodic);
}
/// <summary>
/// Добавить фоновую работу выполняющуюся с заданным периодом
/// </summary>
/// <param name="work"></param>
/// <param name="period"></param>
public void AddPeriodic(Work work, TimeSpan period)
{
var periodic = new WorkPeriodic(work, period);

2
AsbCloudInfrastructure/Services/AutoGeneratedDailyReports/AutoGeneratedDailyReportService.cs
View File

@ -226,7 +226,7 @@ public class AutoGeneratedDailyReportService : IAutoGeneratedDailyReportService
.OrderBy(w => w.DateStart);
}
private Task<IEnumerable<SubsystemStatDto>?> GetSubsystemStatsAsync(int idWell, DateTime startDate,
private Task<IEnumerable<SubsystemStatDto>> GetSubsystemStatsAsync(int idWell, DateTime startDate,
DateTime finishDate, CancellationToken cancellationToken)
{
var request = new SubsystemOperationTimeRequest

164
AsbCloudInfrastructure/Services/DetectOperations/OperationDetectionWorkFactory.cs
View File

@ -1,164 +0,0 @@
using AsbCloudDb.Model;
using Microsoft.EntityFrameworkCore;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AsbCloudInfrastructure.Services.DetectOperations.Detectors;
using AsbCloudInfrastructure.Background;
using Microsoft.Extensions.DependencyInjection;
namespace AsbCloudInfrastructure.Services.DetectOperations
{
public static class OperationDetectionWorkFactory
{
private const string workId = "Operation detection";
private static string progress = "no progress";
private static readonly DetectorAbstract[] detectors = new DetectorAbstract[]
{
new DetectorRotor(),
new DetectorSlide(),
//new DetectorDevelopment(),
//new DetectorTemplating(),
new DetectorSlipsTime(),
//new DetectorStaticSurveying(),
//new DetectorFlashingBeforeConnection(),
//new DetectorFlashing(),
//new DetectorTemplatingWhileDrilling(),
};
public static Work MakeWork() => new Work(workId, WorkAction)
{
Timeout = TimeSpan.FromMinutes(20),
OnErrorAsync = (id, exception, token) =>
{
var text = $"work {id}, when {progress}, throw error:{exception.Message}";
Trace.TraceWarning(text);
return Task.CompletedTask;
}
};
// TODO: Разделить этот акшн на более мелкие части И использовать telemetryServiceData<..> вместо прямого обращения к БД.
private static async Task WorkAction(string _, IServiceProvider serviceProvider, Action<string, double?> onProgress, CancellationToken token)
{
using var db = serviceProvider.GetRequiredService<IAsbCloudDbContext>();
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 joinedlastDetectedDates = telemetryIds
.GroupJoin(lastDetectedDates,
t => t,
o => o.IdTelemetry,
(outer, inner) => new
{
IdTelemetry = outer,
inner.SingleOrDefault()?.LastDate,
});
var affected = 0;
var count = joinedlastDetectedDates.Count();
var i = 0;
foreach (var item in joinedlastDetectedDates)
{
var stopwatch = Stopwatch.StartNew();
var startDate = item.LastDate ?? DateTimeOffset.MinValue;
onProgress($"start detecting telemetry: {item.IdTelemetry} from {startDate}", i++ / count);
var newOperations = await DetectOperationsAsync(item.IdTelemetry, startDate, db, token);
stopwatch.Stop();
if (newOperations.Any())
{
db.DetectedOperations.AddRange(newOperations);
affected += await db.SaveChangesAsync(token);
}
}
}
private static async Task<IEnumerable<DetectedOperation>> DetectOperationsAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token)
{
var query = db.TelemetryDataSaub
.AsNoTracking()
.Where(d => d.IdTelemetry == idTelemetry)
.Where(d => d.BlockPosition >= 0)
.Select(d => new DetectableTelemetry
{
DateTime = d.DateTime,
IdUser = d.IdUser,
WellDepth = d.WellDepth ?? float.NaN,
Pressure = d.Pressure ?? float.NaN,
HookWeight = d.HookWeight ?? float.NaN,
BlockPosition = d.BlockPosition ?? float.NaN,
BitDepth = d.BitDepth ?? float.NaN,
RotorSpeed = d.RotorSpeed ?? float.NaN,
})
.OrderBy(d => d.DateTime);
var take = 4 * 86_400; // 4 дня
var startDate = begin;
var detectedOperations = new List<DetectedOperation>(8);
DetectedOperation? lastDetectedOperation = null;
const int minOperationLength = 5;
const int maxDetectorsInterpolationFrameLength = 30;
const int gap = maxDetectorsInterpolationFrameLength + minOperationLength;
while (true)
{
var data = await query
.Where(d => d.DateTime > startDate)
.Take(take)
.ToArrayAsync(token);
if (data.Length < gap)
break;
var isDetected = false;
var positionBegin = 0;
var positionEnd = data.Length - gap;
var step = 10;
while (positionEnd > positionBegin)
{
step ++;
for (int i = 0; i < detectors.Length; i++)
{
progress = $"telemetry:{idTelemetry}, date:{startDate}, pos:{positionBegin}, detector:{detectors[i]}";
if (detectors[i].TryDetect(idTelemetry, data, positionBegin, positionEnd, lastDetectedOperation, out OperationDetectorResult? result))
{
detectedOperations.Add(result!.Operation);
lastDetectedOperation = result.Operation;
isDetected = true;
step = 1;
positionBegin = result.TelemetryEnd;
break;
}
}
if (step > 20)
step = 10;
positionBegin += step;
}
if (isDetected)
startDate = lastDetectedOperation!.DateEnd;
else
startDate = data[positionEnd].DateTime;
}
return detectedOperations;
}
}
}

157
AsbCloudInfrastructure/Services/DetectOperations/WorkOperationDetection.cs Normal file
View File

@ -0,0 +1,157 @@
using AsbCloudDb.Model;
using Microsoft.EntityFrameworkCore;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using AsbCloudInfrastructure.Services.DetectOperations.Detectors;
using AsbCloudInfrastructure.Background;
using Microsoft.Extensions.DependencyInjection;
namespace AsbCloudInfrastructure.Services.DetectOperations;
public class WorkOperationDetection: Work
{
private static readonly DetectorAbstract[] detectors = new DetectorAbstract[]
{
new DetectorRotor(),
new DetectorSlide(),
//new DetectorDevelopment(),
//new DetectorTemplating(),
new DetectorSlipsTime(),
//new DetectorStaticSurveying(),
//new DetectorFlashingBeforeConnection(),
//new DetectorFlashing(),
//new DetectorTemplatingWhileDrilling(),
};
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;
};
}
protected override async Task Action(string id, IServiceProvider services, Action<string, double?> onProgressCallback, CancellationToken token)
{
using var db = services.GetRequiredService<IAsbCloudDbContext>();
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 joinedlastDetectedDates = telemetryIds
.GroupJoin(lastDetectedDates,
t => t,
o => o.IdTelemetry,
(outer, inner) => new
{
IdTelemetry = outer,
inner.SingleOrDefault()?.LastDate,
});
var affected = 0;
var count = joinedlastDetectedDates.Count();
var i = 0d;
foreach (var item in joinedlastDetectedDates)
{
var stopwatch = Stopwatch.StartNew();
var startDate = item.LastDate ?? DateTimeOffset.MinValue;
onProgressCallback($"start detecting telemetry: {item.IdTelemetry} from {startDate}", i++ / count);
var newOperations = await DetectOperationsAsync(item.IdTelemetry, startDate, db, token);
stopwatch.Stop();
if (newOperations.Any())
{
db.DetectedOperations.AddRange(newOperations);
affected += await db.SaveChangesAsync(token);
}
}
}
private static async Task<IEnumerable<DetectedOperation>> DetectOperationsAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token)
{
var query = db.TelemetryDataSaub
.AsNoTracking()
.Where(d => d.IdTelemetry == idTelemetry)
.Where(d => d.BlockPosition >= 0)
.Select(d => new DetectableTelemetry
{
DateTime = d.DateTime,
IdUser = d.IdUser,
WellDepth = d.WellDepth ?? float.NaN,
Pressure = d.Pressure ?? float.NaN,
HookWeight = d.HookWeight ?? float.NaN,
BlockPosition = d.BlockPosition ?? float.NaN,
BitDepth = d.BitDepth ?? float.NaN,
RotorSpeed = d.RotorSpeed ?? float.NaN,
})
.OrderBy(d => d.DateTime);
var take = 4 * 86_400; // 4 дня
var startDate = begin;
var detectedOperations = new List<DetectedOperation>(8);
DetectedOperation? lastDetectedOperation = null;
const int minOperationLength = 5;
const int maxDetectorsInterpolationFrameLength = 30;
const int gap = maxDetectorsInterpolationFrameLength + minOperationLength;
while (true)
{
var data = await query
.Where(d => d.DateTime > startDate)
.Take(take)
.ToArrayAsync(token);
if (data.Length < gap)
break;
var isDetected = false;
var positionBegin = 0;
var positionEnd = data.Length - gap;
var step = 10;
while (positionEnd > positionBegin)
{
step ++;
for (int i = 0; i < detectors.Length; i++)
{
if (detectors[i].TryDetect(idTelemetry, data, positionBegin, positionEnd, lastDetectedOperation, out OperationDetectorResult? result))
{
detectedOperations.Add(result!.Operation);
lastDetectedOperation = result.Operation;
isDetected = true;
step = 1;
positionBegin = result.TelemetryEnd;
break;
}
}
if (step > 20)
step = 10;
positionBegin += step;
}
if (isDetected)
startDate = lastDetectedOperation!.DateEnd;
else
startDate = data[positionEnd].DateTime;
}
return detectedOperations;
}
}

7
AsbCloudInfrastructure/Services/DrillingProgram/DrillingProgramService.cs
View File

@ -540,11 +540,8 @@ namespace AsbCloudInfrastructure.Services.DrillingProgram
return Task.CompletedTask;
};
var work = new Work(workId, workAction)
{
OnErrorAsync = onErrorAction
};
var work = Work.CreateByDelegate(workId, workAction);
work.OnErrorAsync = onErrorAction;
backgroundWorker.WorkStore.RunOnceQueue.Enqueue(work);
}
}

2
AsbCloudInfrastructure/Services/Email/EmailNotificationTransportService.cs
View File

@ -56,7 +56,7 @@ namespace AsbCloudInfrastructure.Services.Email
{
var workAction = MakeEmailSendWorkAction(notification);
var work = new Work(workId, workAction);
var work = Work.CreateByDelegate(workId, workAction);
backgroundWorker.WorkStore.RunOnceQueue.Enqueue(work);
}

149
AsbCloudInfrastructure/Services/LimitingParameterBackgroundService.cs
View File

@ -1,149 +0,0 @@
using AsbCloudDb.Model;
using Microsoft.EntityFrameworkCore;
using System;
using System.Data.Common;
using System.Data;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;
using AsbCloudInfrastructure.Background;
using Microsoft.Extensions.DependencyInjection;
namespace AsbCloudInfrastructure.Services
{
internal static class LimitingParameterCalcWorkFactory
{
private const string workId = "Limiting parameter calc";
public static Work MakeWork() => new Work(workId, WorkAction)
{
Timeout = TimeSpan.FromMinutes(30)
};
// TODO: Разделить этот акшн на более мелкие части И использовать telemetryServiceData<..> вместо прямого обращения к БД.
private static async Task WorkAction(string _, IServiceProvider serviceProvider, Action<string, double?> onProgress, CancellationToken token)
{
using var db = serviceProvider.GetRequiredService<IAsbCloudDbContext>();
var lastDetectedDates = await db.LimitingParameter
.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 count = telemetryLastDetectedDates.Count();
var i = 0;
foreach (var item in telemetryLastDetectedDates)
{
onProgress($"Start hanling telemetry: {item.IdTelemetry} from {item.LastDate}", i++/count);
var newLimitingParameters = await GetLimitingParameterAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
if (newLimitingParameters?.Any() == true)
{
db.LimitingParameter.AddRange(newLimitingParameters);
await db.SaveChangesAsync(token);
}
}
}
private static async Task<IEnumerable<LimitingParameter>> GetLimitingParameterAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token)
{
var query =
$"select " +
$"limiting_parameters.date, limiting_parameters.id_feed_regulator, limiting_parameters.well_depth " +
$"from ( " +
$"select " +
$"date, id_feed_regulator, well_depth, " +
$"lag(id_feed_regulator, 1) over (order by date) as id_feed_regulator_lag, " +
$"lead(id_feed_regulator, 1) over (order by date) as id_feed_regulator_lead " +
$"from t_telemetry_data_saub " +
$"where id_feed_regulator is not null " +
$"and id_telemetry = {idTelemetry}" +
$"and date >= '{begin:u}'" +
$"order by date) as limiting_parameters " +
$"where id_feed_regulator_lag is null " +
$"or (id_feed_regulator != id_feed_regulator_lag and id_feed_regulator_lead != id_feed_regulator_lag) " +
$"order by date;";
var limitingParameters = new List<LimitingParameter>(32);
using (var result = await ExecuteReaderAsync(db, query, token))
{
LimitingParameter? limitingLast = null;
while (result.Read())
{
var date = result.GetFieldValue<DateTimeOffset>(0);
var idLimiting = result.GetFieldValue<short>(1);
var wellDepth = result.GetFieldValue<float>(2);
if (limitingLast is null)
{
limitingLast = new LimitingParameter
{
DateStart = date,
DepthStart = wellDepth,
IdFeedRegulator = idLimiting
};
}
if (limitingLast.IdFeedRegulator != idLimiting || limitingLast.DepthStart < wellDepth)
{
limitingParameters.Add(new LimitingParameter {
IdTelemetry = idTelemetry,
IdFeedRegulator = limitingLast.IdFeedRegulator,
DateStart = limitingLast.DateStart,
DateEnd = date,
DepthStart = limitingLast.DepthStart,
DepthEnd = wellDepth
});
limitingLast = new LimitingParameter
{
DateStart = date,
DepthStart = wellDepth,
IdFeedRegulator = idLimiting
};
}
}
}
return limitingParameters;
}
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;
}
}
}

2
AsbCloudInfrastructure/Services/ReportService.cs
View File

@ -94,7 +94,7 @@ namespace AsbCloudInfrastructure.Services
context.SaveChanges();
};
var work = new Work(workId, workAction);
var work = Work.CreateByDelegate(workId, workAction);
backgroundWorkerService.WorkStore.RunOnceQueue.Enqueue(work);
progressHandler.Invoke(new ReportProgressDto

17
AsbCloudInfrastructure/Services/SAUB/TelemetryDataCache.cs
View File

@ -48,7 +48,7 @@ namespace AsbCloudInfrastructure.Services.SAUB
instance = new TelemetryDataCache<TDto>();
var worker = provider.GetRequiredService<BackgroundWorker>();
var workId = $"Telemetry cache loading from DB {typeof(TEntity).Name}";
var work = new Work(workId, async (workId, provider, onProgress, token) => {
var work = Work.CreateByDelegate(workId, async (workId, provider, onProgress, token) => {
var db = provider.GetRequiredService<IAsbCloudDbContext>();
await instance.InitializeCacheFromDBAsync<TEntity>(db, onProgress, token);
});
@ -159,7 +159,6 @@ namespace AsbCloudInfrastructure.Services.SAUB
isLoading = true;
var defaultTimeout = db.Database.GetCommandTimeout();
System.Diagnostics.Trace.TraceInformation($"cache loading starting. Setting CommandTimeout 90s ({defaultTimeout})");
db.Database.SetCommandTimeout(TimeSpan.FromSeconds(90));
Well[] wells = await db.Set<Well>()
@ -169,7 +168,7 @@ namespace AsbCloudInfrastructure.Services.SAUB
.ToArrayAsync(token);
var count = wells.Length;
var i = 0;
var i = 0d;
foreach (Well well in wells)
{
var capacity = well.IdState == 1
@ -178,21 +177,13 @@ namespace AsbCloudInfrastructure.Services.SAUB
var idTelemetry = well.IdTelemetry!.Value;
var hoursOffset = well.Timezone.Hours;
// TODO: remove traces
System.Diagnostics.Trace.TraceInformation($"cache<{typeof(TDto).Name}>: Loading for well: {well.Cluster?.Caption}/{well.Caption} (capacity:{capacity}) idTelemetry:{idTelemetry}");
onProgress($"Loading for well: {well.Cluster?.Caption}/{well.Caption} (capacity:{capacity}) idTelemetry:{idTelemetry}", i++/count);
var cacheItem = await GetOrDefaultCacheDataFromDbAsync<TEntity>(db, idTelemetry, capacity, hoursOffset, token);
if(cacheItem is not null)
{
caches.TryAdd(idTelemetry, cacheItem);
System.Diagnostics.Trace.TraceInformation($"cache<{typeof(TDto).Name}> for well: {well.Cluster?.Caption}/{well.Caption} loaded");
}
else
{
System.Diagnostics.Trace.TraceInformation($"cache<{typeof(TDto).Name}> for well: {well.Cluster?.Caption}/{well.Caption} has no data");
}
}
System.Diagnostics.Trace.TraceInformation($"cache<{typeof(TDto).Name}> load complete");
isLoading = false;
db.Database.SetCommandTimeout(defaultTimeout);
}

298
AsbCloudInfrastructure/Services/Subsystems/SubsystemOperationTimeCalcWorkFactory.cs
View File

@ -1,298 +0,0 @@
using AsbCloudDb.Model;
using AsbCloudDb.Model.Subsystems;
using AsbCloudInfrastructure.Background;
using AsbCloudInfrastructure.Services.Subsystems.Utils;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
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
{
internal static class SubsystemOperationTimeCalcWorkFactory
{
private const string workId = "Subsystem operation time calc";
private const int idSubsytemTorqueMaster = 65537;
private const int idSubsytemSpinMaster = 65536;
private const int idSubsystemAPDRotor = 11;
private const int idSubsystemAPDSlide = 12;
private const int idSubsytemMse = 2;
public static Work MakeWork() => new Work(workId, WorkAction)
{
Timeout = TimeSpan.FromMinutes(20)
};
// TODO: Разделить этот акшн на более мелкие части И использовать telemetryServiceData<..> вместо прямого обращения к БД.
private static async Task WorkAction(string _, IServiceProvider serviceProvider, Action<string, double?> onProgress, CancellationToken token)
{
using var db = serviceProvider.GetRequiredService<IAsbCloudDbContext>();
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 count = telemetryLastDetectedDates.Count();
var i = 0;
foreach (var item in telemetryLastDetectedDates)
{
onProgress($"Start hanling telemetry: {item.IdTelemetry} from {item.LastDate}", i++ / count);
var newOperationsSaub = await OperationTimeSaubAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
if (newOperationsSaub?.Any() == true)
{
db.SubsystemOperationTimes.AddRange(newOperationsSaub);
await db.SaveChangesAsync(token);
}
var newOperationsSpin = await OperationTimeSpinAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
if (newOperationsSpin?.Any() == true)
{
db.SubsystemOperationTimes.AddRange(newOperationsSpin);
await db.SaveChangesAsync(token);
}
}
}
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 isSubsytemAkbRotor(short? mode) => mode == 1;
static bool isSubsytemAkbSlide(short? mode) => mode == 3;
static bool IsSubsystemMse(short? state) => (state & 1) > 0;
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>();
var detectorRotor = new SubsystemDetector(idTelemetry, idSubsystemAPDRotor, isSubsytemAkbRotor, IsValid);
var detectorSlide = new SubsystemDetector(idTelemetry, idSubsystemAPDSlide, isSubsytemAkbSlide, IsValid);
var detectorMse = new SubsystemDetector(idTelemetry, idSubsytemMse, IsSubsystemMse, IsValid);
while (result.Read())
{
var mode = result.GetFieldValue<short?>(1);
var state = result.GetFieldValue<short?>(3);
var isAkbRotorEnable = isSubsytemAkbRotor(mode);
var isAkbSlideEnable = isSubsytemAkbSlide(mode);
var isMseEnable = IsSubsystemMse(state);
var date = result.GetFieldValue<DateTimeOffset>(0);
var depth = result.GetFieldValue<float>(2);
if (detectorRotor.TryDetect(mode, date, depth, out var detectedRotor))
subsystemsOperationTimes.Add(detectedRotor!);
if (detectorSlide.TryDetect(mode, date, depth, out var detectedSlide))
subsystemsOperationTimes.Add(detectedSlide!);
if (detectorMse.TryDetect(mode, date, depth, out var detectedMse))
subsystemsOperationTimes.Add(detectedMse!);
}
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 != 5 && 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;
}
}
}

294
AsbCloudInfrastructure/Services/Subsystems/WorkSubsystemOperationTimeCalc.cs Normal file
View File

@ -0,0 +1,294 @@
using AsbCloudDb.Model;
using AsbCloudDb.Model.Subsystems;
using AsbCloudInfrastructure.Background;
using AsbCloudInfrastructure.Services.Subsystems.Utils;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
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;
public class WorkSubsystemOperationTimeCalc: Work
{
private const int idSubsytemTorqueMaster = 65537;
private const int idSubsytemSpinMaster = 65536;
private const int idSubsystemAPDRotor = 11;
private const int idSubsystemAPDSlide = 12;
private const int idSubsytemMse = 2;
public WorkSubsystemOperationTimeCalc()
: base("Subsystem operation time calc")
{
Timeout = TimeSpan.FromMinutes(20);
}
protected override async Task Action(string id, IServiceProvider services, Action<string, double?> onProgressCallback, CancellationToken token)
{
using var db = services.GetRequiredService<IAsbCloudDbContext>();
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 count = telemetryLastDetectedDates.Count();
var i = 0d;
foreach (var item in telemetryLastDetectedDates)
{
onProgressCallback($"Start hanling telemetry: {item.IdTelemetry} from {item.LastDate}", i++ / count);
var newOperationsSaub = await OperationTimeSaubAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
if (newOperationsSaub?.Any() == true)
{
db.SubsystemOperationTimes.AddRange(newOperationsSaub);
await db.SaveChangesAsync(token);
}
var newOperationsSpin = await OperationTimeSpinAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
if (newOperationsSpin?.Any() == true)
{
db.SubsystemOperationTimes.AddRange(newOperationsSpin);
await db.SaveChangesAsync(token);
}
}
}
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 isSubsytemAkbRotor(short? mode) => mode == 1;
static bool isSubsytemAkbSlide(short? mode) => mode == 3;
static bool IsSubsystemMse(short? state) => (state & 1) > 0;
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>();
var detectorRotor = new SubsystemDetector(idTelemetry, idSubsystemAPDRotor, isSubsytemAkbRotor, IsValid);
var detectorSlide = new SubsystemDetector(idTelemetry, idSubsystemAPDSlide, isSubsytemAkbSlide, IsValid);
var detectorMse = new SubsystemDetector(idTelemetry, idSubsytemMse, IsSubsystemMse, IsValid);
while (result.Read())
{
var mode = result.GetFieldValue<short?>(1);
var state = result.GetFieldValue<short?>(3);
var isAkbRotorEnable = isSubsytemAkbRotor(mode);
var isAkbSlideEnable = isSubsytemAkbSlide(mode);
var isMseEnable = IsSubsystemMse(state);
var date = result.GetFieldValue<DateTimeOffset>(0);
var depth = result.GetFieldValue<float>(2);
if (detectorRotor.TryDetect(mode, date, depth, out var detectedRotor))
subsystemsOperationTimes.Add(detectedRotor!);
if (detectorSlide.TryDetect(mode, date, depth, out var detectedSlide))
subsystemsOperationTimes.Add(detectedSlide!);
if (detectorMse.TryDetect(mode, date, depth, out var detectedMse))
subsystemsOperationTimes.Add(detectedMse!);
}
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 != 5 && 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;
}
}

173
AsbCloudInfrastructure/Services/WellInfoService.cs
View File

@ -18,53 +18,26 @@ using System.Threading.Tasks;
using AsbCloudApp.IntegrationEvents;
using AsbCloudApp.IntegrationEvents.Interfaces;
namespace AsbCloudInfrastructure.Services
namespace AsbCloudInfrastructure.Services;
public class WellInfoService
{
public class WellInfoService
public class WorkWellInfoUpdate : Work
{
class WellMapInfoWithComanies : WellMapInfoDto
public WorkWellInfoUpdate()
: base("Well statistics update")
{
public int? IdTelemetry { get; set; }
public IEnumerable<int> IdsCompanies { get; set; } = null!;
Timeout = TimeSpan.FromMinutes(20);
}
private const string workId = "Well statistics update";
private readonly TelemetryDataCache<TelemetryDataSaubDto> telemetryDataSaubCache;
private readonly TelemetryDataCache<TelemetryDataSpinDto> telemetryDataSpinCache;
private readonly IWitsRecordRepository<Record7Dto> witsRecord7Repository;
private readonly IWitsRecordRepository<Record1Dto> witsRecord1Repository;
private readonly IGtrRepository gtrRepository;
private static IEnumerable<WellMapInfoWithComanies> WellMapInfo = Enumerable.Empty<WellMapInfoWithComanies>();
public WellInfoService(
TelemetryDataCache<TelemetryDataSaubDto> telemetryDataSaubCache,
TelemetryDataCache<TelemetryDataSpinDto> telemetryDataSpinCache,
IWitsRecordRepository<Record7Dto> witsRecord7Repository,
IWitsRecordRepository<Record1Dto> witsRecord1Repository,
IGtrRepository gtrRepository)
protected override async Task Action(string id, IServiceProvider services, Action<string, double?> onProgressCallback, CancellationToken token)
{
this.telemetryDataSaubCache = telemetryDataSaubCache;
this.telemetryDataSpinCache = telemetryDataSpinCache;
this.witsRecord7Repository = witsRecord7Repository;
this.witsRecord1Repository = witsRecord1Repository;
this.gtrRepository = gtrRepository;
}
public static Work MakeWork() => new Work(workId, WorkAction)
{
Timeout = TimeSpan.FromMinutes(20)
};
private static async Task WorkAction(string workName, IServiceProvider serviceProvider, Action<string, double?> onProgress, CancellationToken token)
{
var wellService = serviceProvider.GetRequiredService<IWellService>();
var operationsStatService = serviceProvider.GetRequiredService<IOperationsStatService>();
var processMapRepository = serviceProvider.GetRequiredService<IProcessMapPlanRepository>();
var subsystemOperationTimeService = serviceProvider.GetRequiredService<ISubsystemOperationTimeService>();
var telemetryDataSaubCache = serviceProvider.GetRequiredService<TelemetryDataCache<TelemetryDataSaubDto>>();
var messageHub = serviceProvider.GetRequiredService<IIntegrationEventHandler<UpdateWellInfoEvent>>();
var wellService = services.GetRequiredService<IWellService>();
var operationsStatService = services.GetRequiredService<IOperationsStatService>();
var processMapRepository = services.GetRequiredService<IProcessMapPlanRepository>();
var subsystemOperationTimeService = services.GetRequiredService<ISubsystemOperationTimeService>();
var telemetryDataSaubCache = services.GetRequiredService<TelemetryDataCache<TelemetryDataSaubDto>>();
var messageHub = services.GetRequiredService<IIntegrationEventHandler<UpdateWellInfoEvent>>();
var wells = await wellService.GetAllAsync(token);
@ -82,30 +55,30 @@ namespace AsbCloudInfrastructure.Services
});
var operationsStat = await operationsStatService.GetWellsStatAsync(wellsIds, token);
var subsystemStat = await subsystemOperationTimeService
.GetStatByActiveWells(wellsIds, token);
var count = wells.Count();
var i = 0;
var i = 0d;
WellMapInfo = wells.Select(well => {
var wellMapInfo = well.Adapt<WellMapInfoWithComanies>();
wellMapInfo.IdState = well.IdState;
onProgress($"Start updating info by well({well.Id}): {well.Caption}", i++ / count);
onProgressCallback($"Start updating info by well({well.Id}): {well.Caption}", i++ / count);
double? currentDepth = null;
TelemetryDataSaubDto? lastSaubTelemetry = null;
if (well.IdTelemetry.HasValue)
{
wellMapInfo.IdTelemetry = well.IdTelemetry.Value;
lastSaubTelemetry = telemetryDataSaubCache.GetLastOrDefault(well.IdTelemetry.Value);
if(lastSaubTelemetry is not null)
if (lastSaubTelemetry is not null)
{
currentDepth = lastSaubTelemetry.WellDepth;
}
}
var wellOperationsStat = operationsStat.FirstOrDefault(s => s.Id == well.Id);
var wellOperationsStat = operationsStat.FirstOrDefault(s => s.Id == well.Id);
var wellLastFactSection = wellOperationsStat?.Sections.LastOrDefault(s => s.Fact is not null);
currentDepth ??= wellLastFactSection?.Fact?.WellDepthEnd;
@ -120,7 +93,7 @@ namespace AsbCloudInfrastructure.Services
{
wellProcessMap = wellProcessMaps.FirstOrDefault(p => p.IdWellSectionType == idSection);
}
else if(currentDepth.HasValue)
else if (currentDepth.HasValue)
{
wellProcessMap = wellProcessMaps.FirstOrDefault(p => p.DepthStart <= currentDepth.Value && p.DepthEnd >= currentDepth.Value);
}
@ -130,8 +103,8 @@ namespace AsbCloudInfrastructure.Services
planTotalDepth ??= wellOperationsStat?.Total.Plan?.WellDepthEnd;
wellMapInfo.Section = wellLastFactSection?.Caption;
wellMapInfo.FirstFactOperationDateStart = wellOperationsStat?.Total.Fact?.Start
wellMapInfo.FirstFactOperationDateStart = wellOperationsStat?.Total.Fact?.Start
?? wellOperationsStat?.Total.Plan?.Start;
wellMapInfo.LastPredictOperationDateEnd = wellOperationsStat?.Total.Plan?.End;
@ -159,7 +132,7 @@ namespace AsbCloudInfrastructure.Services
Plan = wellProcessMap?.Pressure.Plan,
Fact = lastSaubTelemetry?.Pressure
};
wellMapInfo.PressureSp = lastSaubTelemetry?.PressureSp;
wellMapInfo.WellDepth = new()
@ -191,51 +164,79 @@ namespace AsbCloudInfrastructure.Services
return wellMapInfo;
}).ToArray();
var updateWellInfoEventTasks = wellsIds.Select(idWell =>
var updateWellInfoEventTasks = wellsIds.Select(idWell =>
messageHub.HandleAsync(new UpdateWellInfoEvent(idWell), token));
await Task.WhenAll(updateWellInfoEventTasks);
}
}
private WellMapInfoWithTelemetryStat Convert(WellMapInfoWithComanies wellInfo)
class WellMapInfoWithComanies : WellMapInfoDto
{
public int? IdTelemetry { get; set; }
public IEnumerable<int> IdsCompanies { get; set; } = null!;
}
private readonly TelemetryDataCache<TelemetryDataSaubDto> telemetryDataSaubCache;
private readonly TelemetryDataCache<TelemetryDataSpinDto> telemetryDataSpinCache;
private readonly IWitsRecordRepository<Record7Dto> witsRecord7Repository;
private readonly IWitsRecordRepository<Record1Dto> witsRecord1Repository;
private readonly IGtrRepository gtrRepository;
private static IEnumerable<WellMapInfoWithComanies> WellMapInfo = Enumerable.Empty<WellMapInfoWithComanies>();
public WellInfoService(
TelemetryDataCache<TelemetryDataSaubDto> telemetryDataSaubCache,
TelemetryDataCache<TelemetryDataSpinDto> telemetryDataSpinCache,
IWitsRecordRepository<Record7Dto> witsRecord7Repository,
IWitsRecordRepository<Record1Dto> witsRecord1Repository,
IGtrRepository gtrRepository)
{
this.telemetryDataSaubCache = telemetryDataSaubCache;
this.telemetryDataSpinCache = telemetryDataSpinCache;
this.witsRecord7Repository = witsRecord7Repository;
this.witsRecord1Repository = witsRecord1Repository;
this.gtrRepository = gtrRepository;
}
private WellMapInfoWithTelemetryStat Convert(WellMapInfoWithComanies wellInfo)
{
var result = wellInfo.Adapt<WellMapInfoWithTelemetryStat>();
if (wellInfo.IdTelemetry.HasValue)
{
var result = wellInfo.Adapt<WellMapInfoWithTelemetryStat>();
if (wellInfo.IdTelemetry.HasValue)
{
var idTelemetry = wellInfo.IdTelemetry.Value;
result.LastDataSaub = telemetryDataSaubCache.GetLastOrDefault(idTelemetry);
result.LastDataSpin = telemetryDataSpinCache.GetLastOrDefault(idTelemetry);
result.LastDataDdsDate = GetLastOrDefaultDdsTelemetry(idTelemetry);
result.LastDataGtrDate = gtrRepository.GetLastData(wellInfo.Id)
.MaxOrDefault(item => item.Date);
result.LastDataDpcsDate = null;
result.LastDataDpcsDate = null;
}
return result;
var idTelemetry = wellInfo.IdTelemetry.Value;
result.LastDataSaub = telemetryDataSaubCache.GetLastOrDefault(idTelemetry);
result.LastDataSpin = telemetryDataSpinCache.GetLastOrDefault(idTelemetry);
result.LastDataDdsDate = GetLastOrDefaultDdsTelemetry(idTelemetry);
result.LastDataGtrDate = gtrRepository.GetLastData(wellInfo.Id)
.MaxOrDefault(item => item.Date);
result.LastDataDpcsDate = null;
result.LastDataDpcsDate = null;
}
private DateTime? GetLastOrDefaultDdsTelemetry(int idTelemetry)
{
var lastDdsRecord1Date = witsRecord1Repository.GetLastOrDefault(idTelemetry)?.DateTime;
var lastDdsRecord7Date = witsRecord7Repository.GetLastOrDefault(idTelemetry)?.DateTime;
return result;
}
if (lastDdsRecord1Date.HasValue && lastDdsRecord7Date.HasValue)
if (lastDdsRecord1Date.Value > lastDdsRecord7Date.Value)
return lastDdsRecord1Date.Value;
else
return lastDdsRecord7Date.Value;
private DateTime? GetLastOrDefaultDdsTelemetry(int idTelemetry)
{
var lastDdsRecord1Date = witsRecord1Repository.GetLastOrDefault(idTelemetry)?.DateTime;
var lastDdsRecord7Date = witsRecord7Repository.GetLastOrDefault(idTelemetry)?.DateTime;
return lastDdsRecord1Date ?? lastDdsRecord7Date;
}
if (lastDdsRecord1Date.HasValue && lastDdsRecord7Date.HasValue)
if (lastDdsRecord1Date.Value > lastDdsRecord7Date.Value)
return lastDdsRecord1Date.Value;
else
return lastDdsRecord7Date.Value;
public WellMapInfoWithTelemetryStat? FirstOrDefault(Func<WellMapInfoDto, bool> predicate)
{
var first = WellMapInfo.FirstOrDefault(predicate);
if (first is WellMapInfoWithComanies wellMapInfoWithComanies)
return Convert(wellMapInfoWithComanies);
return lastDdsRecord1Date ?? lastDdsRecord7Date;
}
return null;
}
public WellMapInfoWithTelemetryStat? FirstOrDefault(Func<WellMapInfoDto, bool> predicate)
{
var first = WellMapInfo.FirstOrDefault(predicate);
if (first is WellMapInfoWithComanies wellMapInfoWithComanies)
return Convert(wellMapInfoWithComanies);
return null;
}
}

144
AsbCloudInfrastructure/Services/WorkLimitingParameterCalc.cs Normal file
View File

@ -0,0 +1,144 @@
using AsbCloudDb.Model;
using Microsoft.EntityFrameworkCore;
using System;
using System.Data.Common;
using System.Data;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;
using AsbCloudInfrastructure.Background;
using Microsoft.Extensions.DependencyInjection;
namespace AsbCloudInfrastructure.Services;
public class WorkLimitingParameterCalc : Work
{
public WorkLimitingParameterCalc()
: base("Limiting parameter calc")
{
Timeout = TimeSpan.FromMinutes(30);
}
protected override async Task Action(string id, IServiceProvider services, Action<string, double?> onProgressCallback, CancellationToken token)
{
using var db = services.GetRequiredService<IAsbCloudDbContext>();
var lastDetectedDates = await db.LimitingParameter
.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 count = telemetryLastDetectedDates.Count();
var i = 0d;
foreach (var item in telemetryLastDetectedDates)
{
onProgressCallback($"Start hanling telemetry: {item.IdTelemetry} from {item.LastDate}", i++/count);
var newLimitingParameters = await GetLimitingParameterAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
if (newLimitingParameters?.Any() == true)
{
db.LimitingParameter.AddRange(newLimitingParameters);
await db.SaveChangesAsync(token);
}
}
}
private static async Task<IEnumerable<LimitingParameter>> GetLimitingParameterAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token)
{
var query =
$"select " +
$"limiting_parameters.date, limiting_parameters.id_feed_regulator, limiting_parameters.well_depth " +
$"from ( " +
$"select " +
$"date, id_feed_regulator, well_depth, " +
$"lag(id_feed_regulator, 1) over (order by date) as id_feed_regulator_lag, " +
$"lead(id_feed_regulator, 1) over (order by date) as id_feed_regulator_lead " +
$"from t_telemetry_data_saub " +
$"where id_feed_regulator is not null " +
$"and id_telemetry = {idTelemetry}" +
$"and date >= '{begin:u}'" +
$"order by date) as limiting_parameters " +
$"where id_feed_regulator_lag is null " +
$"or (id_feed_regulator != id_feed_regulator_lag and id_feed_regulator_lead != id_feed_regulator_lag) " +
$"order by date;";
var limitingParameters = new List<LimitingParameter>(32);
using (var result = await ExecuteReaderAsync(db, query, token))
{
LimitingParameter? limitingLast = null;
while (result.Read())
{
var date = result.GetFieldValue<DateTimeOffset>(0);
var idLimiting = result.GetFieldValue<short>(1);
var wellDepth = result.GetFieldValue<float>(2);
if (limitingLast is null)
{
limitingLast = new LimitingParameter
{
DateStart = date,
DepthStart = wellDepth,
IdFeedRegulator = idLimiting
};
}
if (limitingLast.IdFeedRegulator != idLimiting || limitingLast.DepthStart < wellDepth)
{
limitingParameters.Add(new LimitingParameter {
IdTelemetry = idTelemetry,
IdFeedRegulator = limitingLast.IdFeedRegulator,
DateStart = limitingLast.DateStart,
DateEnd = date,
DepthStart = limitingLast.DepthStart,
DepthEnd = wellDepth
});
limitingLast = new LimitingParameter
{
DateStart = date,
DepthStart = wellDepth,
IdFeedRegulator = idLimiting
};
}
}
}
return limitingParameters;
}
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;
}
}

10
AsbCloudInfrastructure/Startup.cs
View File

@ -31,10 +31,10 @@ namespace AsbCloudInfrastructure
_ = provider.GetRequiredService<TelemetryDataCache<TelemetryDataSpinDto>>();
var backgroundWorker = provider.GetRequiredService<BackgroundWorker>();
backgroundWorker.WorkStore.AddPeriodic(WellInfoService.MakeWork(), TimeSpan.FromMinutes(30));
backgroundWorker.WorkStore.AddPeriodic(OperationDetectionWorkFactory.MakeWork(), TimeSpan.FromMinutes(15));
backgroundWorker.WorkStore.AddPeriodic(SubsystemOperationTimeCalcWorkFactory.MakeWork(), TimeSpan.FromMinutes(30));
backgroundWorker.WorkStore.AddPeriodic(LimitingParameterCalcWorkFactory.MakeWork(), TimeSpan.FromMinutes(30));
backgroundWorker.WorkStore.AddPeriodic<WellInfoService.WorkWellInfoUpdate>(TimeSpan.FromMinutes(30));
backgroundWorker.WorkStore.AddPeriodic<WorkOperationDetection>(TimeSpan.FromMinutes(15));
backgroundWorker.WorkStore.AddPeriodic<WorkSubsystemOperationTimeCalc>(TimeSpan.FromMinutes(30));
backgroundWorker.WorkStore.AddPeriodic<WorkLimitingParameterCalc>(TimeSpan.FromMinutes(30));
backgroundWorker.WorkStore.AddPeriodic(MakeMemoryMonitoringWork(), TimeSpan.FromMinutes(1));
var notificationBackgroundWorker = provider.GetRequiredService<NotificationBackgroundWorker>();
@ -55,7 +55,7 @@ namespace AsbCloudInfrastructure
System.Diagnostics.Trace.TraceInformation($"Total memory allocated is {bytesString} bytes. DbContext count is:{AsbCloudDbContext.ReferenceCount}");
return Task.CompletedTask;
};
var work = new Work("Memory monitoring", workAction);
var work = Work.CreateByDelegate("Memory monitoring", workAction);
return work;
}

16
AsbCloudWebApi.Tests/Middlware/UserConnectionsLimitMiddlwareTest.cs
View File

@ -1,4 +1,6 @@
using AsbCloudApp.Data.SAUB;
using AsbCloudApp.Data;
using AsbCloudApp.Data.SAUB;
using AsbCloudApp.Requests;
using AsbCloudApp.Services;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
@ -35,12 +37,22 @@ namespace AsbCloudWebApi.Tests.Middlware
public class TelemetryDataSaubService : ITelemetryDataSaubService
{
public async Task<IEnumerable<TelemetryDataSaubDto>?> GetAsync(int idWell, DateTime dateBegin = default, double intervalSec = 600, int approxPointsCount = 1024, CancellationToken token = default)
public async Task<IEnumerable<TelemetryDataSaubDto>> GetAsync(int idWell, DateTime dateBegin = default, double intervalSec = 600, int approxPointsCount = 1024, CancellationToken token = default)
{
await Task.Delay(1000, token);
return Enumerable.Empty<TelemetryDataSaubDto>();
}
public Task<IEnumerable<TelemetryDataSaubDto>> GetAsync(int idWell, TelemetryDataRequest request, CancellationToken token)
{
throw new NotImplementedException();
}
public Task<DatesRangeDto?> GetRangeAsync(int idWell, DateTimeOffset start, DateTimeOffset end, CancellationToken token)
{
throw new NotImplementedException();
}
public Task<IEnumerable<TelemetryDataSaubStatDto>> GetTelemetryDataStatAsync(int idTelemetry, CancellationToken token) => throw new NotImplementedException();
public Task<Stream> GetZippedCsv(int idWell, DateTime beginDate, DateTime endDate, CancellationToken token)

34
AsbCloudWebApi/Controllers/BackgroundWork.cs Normal file
View File

@ -0,0 +1,34 @@
using AsbCloudApp.Data;
using AsbCloudInfrastructure.Background;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using System.Linq;
namespace AsbCloudWebApi.Controllers
{
[Route("api/[controller]")]
[Authorize]
[ApiController]
public class BackgroundWork : ControllerBase
{
private readonly BackgroundWorker backgroundWorker;
public BackgroundWork(BackgroundWorker backgroundWorker)
{
this.backgroundWorker = backgroundWorker;
}
[HttpGet]
//[ProducesResponseType(typeof(IEnumerable<DepositDto>), (int)System.Net.HttpStatusCode.OK)]
public IActionResult GetAll()
{
var result = new {
CurrentWork = (BackgroundWorkDto?)backgroundWorker.CurrentWork,
RunOnceQueue = backgroundWorker.WorkStore.RunOnceQueue.Select(work => (BackgroundWorkDto)work),
Periodics = backgroundWorker.WorkStore.Periodics.Select(work => (BackgroundWorkDto)work.Work),
Felled = backgroundWorker.WorkStore.Felled.Select(work => (BackgroundWorkDto)work),
};
return Ok(result);
}
}
}

2
AsbCloudWebApi/SignalR/Services/SignalRNotificationTransportService.cs
View File

@ -33,7 +33,7 @@ public class SignalRNotificationTransportService : INotificationTransportService
return Task.CompletedTask;
var workAction = MakeSignalRSendWorkAction(notifications);
var work = new Work(workId, workAction);
var work = Work.CreateByDelegate(workId, workAction);
backgroundWorker.WorkStore.RunOnceQueue.Enqueue(work);
return Task.CompletedTask;

40
ConsoleApp1/DebugWellOperationsStatService.cs
View File

@ -1,40 +0,0 @@
using AsbCloudApp.Data;
using System;
using System.Collections.Generic;
namespace ConsoleApp1
{
public static class DebugWellOperationsStatService
{
public static void Main(/*string[] args*/)
{
//var options = new DbContextOptionsBuilder<AsbCloudDbContext>()
// .UseNpgsql("Host=localhost;Database=postgres;Username=postgres;Password=q;Persist Security Info=True")
// .Options;
//using var db = new AsbCloudDbContext(options);
//var cacheDb = new CacheDb();
//var telemetryService = new TelemetryService(db, new TelemetryTracker(cacheDb), cacheDb);
//var wellService = new WellService(db, telemetryService, cacheDb);
//var wellOptsStat = new OperationsStatService(db, cacheDb, wellService);
//var tvd = wellOptsStat.GetTvdAsync(1, default).Result;
//Print(tvd);
}
private static void Print(IEnumerable<PlanFactPredictBase<WellOperationDto>> tvd)
{
Console.WriteLine("|\tplan\t|\tfact\t|\tprog\t|");
Console.WriteLine("|:-------------:|:-------------:|:-------------:|");
foreach (var item in tvd)
Print(item);
}
private static void Print(PlanFactPredictBase<WellOperationDto> item)
{
static string GetText(WellOperationDto item)
=> (item is null)
? " --------- "
: $"{item.IdCategory} d:{item.DepthStart} ";
Console.WriteLine($"|\t{GetText(item.Plan)}\t|\t{GetText(item.Fact)}\t|\t{GetText(item.Predict)}\t|");
}
}
}