forked from ddrilling/AsbCloudServer
284 lines
11 KiB
C#
284 lines
11 KiB
C#
using AsbCloudApp.Data.GTR;
|
|
using AsbCloudApp.Repositories;
|
|
using AsbCloudApp.Services;
|
|
using AsbCloudDb;
|
|
using AsbCloudDb.Model;
|
|
using AsbCloudDb.Model.GTR;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
|
|
namespace AsbCloudInfrastructure.Repository
|
|
{
|
|
public class GtrWitsRepository : IGtrRepository
|
|
{
|
|
private readonly IAsbCloudDbContext db;
|
|
private readonly ITelemetryService telemetryService;
|
|
private static ConcurrentDictionary<int, ConcurrentDictionary<(int, int), WitsItemRecordDto>> cache = new();
|
|
|
|
public GtrWitsRepository(
|
|
IAsbCloudDbContext db,
|
|
ITelemetryService telemetryService)
|
|
{
|
|
this.db = db;
|
|
this.telemetryService = telemetryService;
|
|
}
|
|
|
|
public async Task<IEnumerable<WitsRecordDto>> GetAsync(int idWell, DateTime? dateBegin, double intervalSec = 600, int approxPointsCount = 1024, CancellationToken token = default)
|
|
{
|
|
var telemetry = telemetryService.GetOrDefaultTelemetryByIdWell(idWell);
|
|
if (telemetry is null)
|
|
return Enumerable.Empty<WitsRecordDto>();
|
|
|
|
var timezone = telemetryService.GetTimezone(telemetry.Id);
|
|
|
|
DateTimeOffset? dateBeginUtc = dateBegin?.ToUtcDateTimeOffset(timezone.Hours);
|
|
var dateEnd = dateBeginUtc?.AddSeconds(intervalSec);
|
|
|
|
var witsRequest = new WitsRequest()
|
|
{
|
|
IdTelemetry = telemetry.Id,
|
|
DateBeginUtc = dateBeginUtc,
|
|
DateEnd = dateEnd,
|
|
ApproxPointsCount = approxPointsCount,
|
|
TimezoneHours = timezone.Hours
|
|
};
|
|
|
|
var recordAllInt = await GetItemsOrDefaultAsync<WitsItemInt, int>(witsRequest, token);
|
|
var recordAllFloat = await GetItemsOrDefaultAsync<WitsItemFloat, float>(witsRequest, token);
|
|
var recordAllString = await GetItemsOrDefaultAsync<WitsItemString, string>(witsRequest, token);
|
|
|
|
var dtos = (recordAllFloat.Union(recordAllInt)).Union(recordAllString)
|
|
.GroupBy(g => new
|
|
{
|
|
g.IdRecord,
|
|
g.Date
|
|
})
|
|
.Select(g => new WitsRecordDto
|
|
{
|
|
Id = g.Key.IdRecord,
|
|
Date = g.Key.Date,
|
|
Items = g.Select(r => new
|
|
{
|
|
Key = r.IdItem,
|
|
r.Value
|
|
}).ToDictionary(x => x.Key, x => x.Value)
|
|
});
|
|
return dtos;
|
|
}
|
|
|
|
public IEnumerable<WitsItemRecordDto> GetLastDataByRecordId(int idWell, int idRecord)
|
|
{
|
|
var result = GetLastData(idWell)
|
|
.Where(item => item.IdRecord == idRecord);
|
|
return result;
|
|
}
|
|
|
|
public IEnumerable<WitsItemRecordDto> GetLastData(int idWell)
|
|
{
|
|
var telemetry = telemetryService.GetOrDefaultTelemetryByIdWell(idWell);
|
|
if (telemetry is null)
|
|
return Enumerable.Empty<WitsItemRecordDto>();
|
|
|
|
var lastData = cache.GetValueOrDefault(telemetry.Id);
|
|
return lastData?.Values ?? Enumerable.Empty<WitsItemRecordDto>();
|
|
}
|
|
|
|
private async Task<IEnumerable<WitsItemRecordDto>> GetItemsOrDefaultAsync<TEntity, TValue>(
|
|
WitsRequest request,
|
|
CancellationToken token)
|
|
where TEntity : WitsItemBase<TValue>
|
|
where TValue : notnull
|
|
{
|
|
var query = BuildQuery<TEntity, TValue>(request);
|
|
|
|
var fullDataCount = await query.CountAsync(token);
|
|
if (fullDataCount == 0)
|
|
return Enumerable.Empty<WitsItemRecordDto>();
|
|
|
|
if (request.ApproxPointsCount is not null && fullDataCount > 1.75 * request.ApproxPointsCount)
|
|
{
|
|
var m = (int)Math.Round(1d * fullDataCount / request.ApproxPointsCount!.Value);
|
|
if (m > 1)
|
|
query = query.Where((d) => (((d.DateTime.DayOfYear * 24 + d.DateTime.Hour) * 60 + d.DateTime.Minute) * 60 + d.DateTime.Second) % m == 0);
|
|
}
|
|
|
|
var entities = await query
|
|
.OrderBy(d => d.DateTime)
|
|
.AsNoTracking()
|
|
.ToListAsync(token)
|
|
.ConfigureAwait(false);
|
|
|
|
var items = entities.Select(e => new WitsItemRecordDto
|
|
{
|
|
IdRecord = e.IdRecord,
|
|
Date = e.DateTime.ToRemoteDateTime(request.TimezoneHours),
|
|
IdItem = e.IdItem,
|
|
Value = new JsonValue(e.Value)
|
|
});
|
|
return items;
|
|
}
|
|
|
|
private IQueryable<TEntity> BuildQuery<TEntity, TValue>(WitsRequest request)
|
|
where TEntity : WitsItemBase<TValue>
|
|
where TValue : notnull
|
|
{
|
|
var query = db.Set<TEntity>().Where(i => i.IdTelemetry == request.IdTelemetry);
|
|
|
|
if (request.IdRecord is not null)
|
|
query = query
|
|
.Where(d => d.IdRecord == request.IdRecord);
|
|
|
|
if (request.DateBeginUtc is not null)
|
|
query = query
|
|
.Where(d => d.DateTime >= request.DateBeginUtc);
|
|
|
|
if (request.DateEnd is not null)
|
|
query = query
|
|
.Where(d => d.DateTime <= request.DateEnd);
|
|
|
|
return query;
|
|
}
|
|
|
|
public async Task SaveDataAsync(int idTelemetry, IEnumerable<WitsRecordDto> dtos, CancellationToken token)
|
|
{
|
|
var timezoneHours = telemetryService.GetTimezone(idTelemetry).Hours;
|
|
|
|
var cacheTelemetryItems = cache.GetValueOrDefault(idTelemetry);
|
|
|
|
var strings = new List<WitsItemString>(4);
|
|
var floats = new List<WitsItemFloat>(4);
|
|
var ints = new List<WitsItemInt>(4);
|
|
|
|
foreach (var record in dtos)
|
|
{
|
|
var dateTime = record.Date.ToUtcDateTimeOffset(timezoneHours);
|
|
foreach (var item in record.Items)
|
|
{
|
|
if (cacheTelemetryItems?.TryGetValue((record.Id, item.Key), out var cacheItem) == true)
|
|
if (Math.Abs((dateTime - cacheItem.Date).TotalSeconds) < 1)
|
|
continue;
|
|
|
|
if (item.Value.Value is string valueString)
|
|
{
|
|
var entity = MakeEntity<WitsItemString, string>(record.Id, item.Key, idTelemetry, dateTime, valueString);
|
|
strings.Add(entity);
|
|
}
|
|
if (item.Value.Value is float valueFloat)
|
|
{
|
|
var entity = MakeEntity<WitsItemFloat, float>(record.Id, item.Key, idTelemetry, dateTime, valueFloat);
|
|
floats.Add(entity);
|
|
}
|
|
if (item.Value.Value is int valueInt)
|
|
{
|
|
var entity = MakeEntity<WitsItemInt, int>(record.Id, item.Key, idTelemetry, dateTime, valueInt);
|
|
ints.Add(entity);
|
|
}
|
|
}
|
|
}
|
|
|
|
try
|
|
{
|
|
if (strings.Any())
|
|
await db.Database.ExecInsertOrIgnoreAsync(db.Set<WitsItemString>(), strings, token);
|
|
|
|
if (floats.Any())
|
|
await db.Database.ExecInsertOrIgnoreAsync(db.Set<WitsItemFloat>(), floats, token);
|
|
|
|
if (ints.Any())
|
|
await db.Database.ExecInsertOrIgnoreAsync(db.Set<WitsItemInt>(), ints, token);
|
|
}
|
|
catch(Exception ex)
|
|
{
|
|
Trace.TraceError("Exception while saving GTR Wits data", ex);
|
|
}
|
|
|
|
cache.AddOrUpdate(idTelemetry,
|
|
(_) => MakeNewCache(dtos),
|
|
(_, oldItemsDictionary) => {
|
|
foreach (var record in dtos)
|
|
foreach (var item in record.Items)
|
|
{
|
|
oldItemsDictionary.AddOrUpdate(
|
|
(record.Id, item.Key),
|
|
(_) => new WitsItemRecordDto
|
|
{
|
|
IdRecord = record.Id,
|
|
IdItem = item.Key,
|
|
Date = record.Date,
|
|
Value = item.Value
|
|
},
|
|
(_, _) => new WitsItemRecordDto
|
|
{
|
|
IdRecord = record.Id,
|
|
IdItem = item.Key,
|
|
Date = record.Date,
|
|
Value = item.Value
|
|
});
|
|
}
|
|
return oldItemsDictionary;
|
|
});
|
|
}
|
|
|
|
private static ConcurrentDictionary<(int, int), WitsItemRecordDto> MakeNewCache(IEnumerable<WitsRecordDto> dtos)
|
|
{
|
|
var items = dtos.SelectMany(record =>
|
|
record.Items.Select(
|
|
item => new WitsItemRecordDto {
|
|
IdItem = item.Key,
|
|
IdRecord = record.Id,
|
|
Date = record.Date,
|
|
Value = item.Value,
|
|
}));
|
|
|
|
var groups = items
|
|
.GroupBy(item => (item.IdRecord, item.IdItem));
|
|
|
|
var pairs = groups.Select(group => new KeyValuePair<(int, int), WitsItemRecordDto>(
|
|
group.Key,
|
|
group.OrderByDescending(item => item.Date).First()));
|
|
|
|
return new ConcurrentDictionary<(int, int), WitsItemRecordDto>(pairs);
|
|
}
|
|
|
|
private static TEntity MakeEntity<TEntity, TValue>(int idRecord, int idItem, int idTelemetry, DateTimeOffset dateTime, TValue value)
|
|
where TEntity : WitsItemBase<TValue>, new()
|
|
where TValue : notnull
|
|
=> new TEntity()
|
|
{
|
|
IdRecord = idRecord,
|
|
IdItem = idItem,
|
|
IdTelemetry = idTelemetry,
|
|
DateTime = dateTime,
|
|
Value = value,
|
|
};
|
|
|
|
private static TEntity MakeEntity<TEntity, TValue>(WitsItemRecordDto dto, int idTelemetry, DateTimeOffset dateTime)
|
|
where TEntity : WitsItemBase<TValue>, new()
|
|
where TValue : notnull
|
|
=> new TEntity()
|
|
{
|
|
IdRecord = dto.IdRecord,
|
|
IdItem = dto.IdItem,
|
|
IdTelemetry = idTelemetry,
|
|
DateTime = dateTime,
|
|
Value = (TValue)dto.Value.Value,
|
|
};
|
|
|
|
private class WitsRequest
|
|
{
|
|
public int IdTelemetry { get; set; }
|
|
public DateTimeOffset? DateBeginUtc { get; set; }
|
|
public DateTimeOffset? DateEnd { get; set; }
|
|
public int? ApproxPointsCount { get; set; }
|
|
public double TimezoneHours { get; set; }
|
|
public int? IdRecord { get; set; }
|
|
}
|
|
}
|
|
}
|