forked from ddrilling/AsbCloudServer
312 lines
12 KiB
C#
312 lines
12 KiB
C#
using AsbCloudApp.Data;
|
|
using AsbCloudApp.Exceptions;
|
|
using AsbCloudApp.Repositories;
|
|
using AsbCloudApp.Requests;
|
|
using AsbCloudApp.Services;
|
|
using AsbCloudDb;
|
|
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;
|
|
|
|
namespace AsbCloudInfrastructure.Services.SAUB
|
|
{
|
|
public abstract class TelemetryDataBaseService<TDto, TEntity> : ITelemetryDataService<TDto>
|
|
where TDto : AsbCloudApp.Data.ITelemetryData
|
|
where TEntity : class, AsbCloudDb.Model.ITelemetryData
|
|
{
|
|
protected readonly IAsbCloudDbContext db;
|
|
protected readonly ITelemetryService telemetryService;
|
|
protected readonly ITelemetryDataCache<TDto> telemetryDataCache;
|
|
|
|
protected TelemetryDataBaseService(
|
|
IAsbCloudDbContext db,
|
|
ITelemetryService telemetryService,
|
|
ITelemetryDataCache<TDto> telemetryDataCache)
|
|
{
|
|
this.db = db;
|
|
this.telemetryService = telemetryService;
|
|
this.telemetryDataCache = telemetryDataCache;
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public virtual async Task<int> UpdateDataAsync(string uid, IEnumerable<TDto> dtos, CancellationToken token = default)
|
|
{
|
|
if (dtos == default || !dtos.Any())
|
|
return 0;
|
|
|
|
var dtosList = dtos.OrderBy(d => d.DateTime).ToList();
|
|
|
|
var dtoMinDate = dtosList.First().DateTime;
|
|
var dtoMaxDate = dtosList.Last().DateTime;
|
|
|
|
if (dtosList.Count > 1)
|
|
{
|
|
var duplicates = new List<TDto>(8);
|
|
for (int i = 1; i < dtosList.Count; i++)
|
|
if (dtosList[i].DateTime - dtosList[i - 1].DateTime < TimeSpan.FromMilliseconds(100))
|
|
duplicates.Add(dtosList[i - 1]);
|
|
foreach (var duplicate in duplicates)
|
|
dtosList.Remove(duplicate);
|
|
}
|
|
|
|
var telemetry = telemetryService.GetOrCreateTelemetryByUid(uid);
|
|
var timezone = telemetryService.GetTimezone(telemetry.Id);
|
|
|
|
telemetryDataCache.AddRange(telemetry.Id, dtos);
|
|
|
|
var entities = dtosList.Select(dto =>
|
|
{
|
|
var entity = Convert(dto, timezone.Hours);
|
|
entity.IdTelemetry = telemetry.Id;
|
|
return entity;
|
|
});
|
|
|
|
var stopwatch = Stopwatch.StartNew();
|
|
var dbSet = db.Set<TEntity>();
|
|
try
|
|
{
|
|
return await db.Database.ExecInsertOrUpdateAsync(dbSet, entities, token).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
stopwatch.Stop();
|
|
Trace.WriteLine($"Fail to save data telemetry " +
|
|
$"uid: {uid}, " +
|
|
$"idTelemetry {telemetry.Id}, " +
|
|
$"count: {entities.Count()}, " +
|
|
$"dataDate: {entities.FirstOrDefault()?.DateTime}, " +
|
|
$"dbSaveDurationTime:{stopwatch.ElapsedMilliseconds}ms. " +
|
|
$"Message: {ex.Message}");
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public virtual async Task<IEnumerable<TDto>> GetByWellAsync(int idWell,
|
|
DateTime dateBegin = default, double intervalSec = 600d,
|
|
int approxPointsCount = 1024, CancellationToken token = default)
|
|
{
|
|
var telemetry = telemetryService.GetOrDefaultTelemetryByIdWell(idWell);
|
|
if (telemetry is null)
|
|
return Enumerable.Empty<TDto>();
|
|
|
|
var timezone = telemetryService.GetTimezone(telemetry.Id);
|
|
|
|
var filterByDateEnd = dateBegin != default;
|
|
DateTimeOffset dateBeginUtc;
|
|
if (dateBegin == default)
|
|
{
|
|
var dateRange = telemetryDataCache.GetOrDefaultWellDataDateRange(telemetry.Id);
|
|
dateBeginUtc = (dateRange?.To.ToUniversalTime() ?? DateTimeOffset.UtcNow)
|
|
.AddSeconds(-intervalSec);
|
|
}
|
|
else
|
|
{
|
|
dateBeginUtc = dateBegin.ToUtcDateTimeOffset(timezone.Hours);
|
|
}
|
|
|
|
var cacheData = telemetryDataCache.GetOrDefault(telemetry.Id, dateBeginUtc.ToRemoteDateTime(timezone.Hours), intervalSec, approxPointsCount);
|
|
if (cacheData is not null)
|
|
return cacheData;
|
|
|
|
var dateEnd = dateBeginUtc.AddSeconds(intervalSec);
|
|
var dbSet = db.Set<TEntity>();
|
|
|
|
var query = dbSet
|
|
.Where(d => d.IdTelemetry == telemetry.Id
|
|
&& d.DateTime >= dateBeginUtc);
|
|
|
|
if (filterByDateEnd)
|
|
query = query.Where(d => d.DateTime <= dateEnd);
|
|
|
|
var fullDataCount = await query.CountAsync(token)
|
|
.ConfigureAwait(false);
|
|
|
|
if (fullDataCount == 0)
|
|
return Enumerable.Empty<TDto>();
|
|
|
|
if (fullDataCount > 1.75 * approxPointsCount)
|
|
{
|
|
var m = (int)Math.Round(1d * fullDataCount / approxPointsCount);
|
|
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
|
|
.AsNoTracking()
|
|
.ToArrayAsync(token);
|
|
|
|
var dtos = entities.Select(e => Convert(e, timezone.Hours));
|
|
|
|
return dtos;
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public virtual async Task<IEnumerable<TDto>> GetByWellAsync(int idWell, TelemetryDataRequest request, CancellationToken token)
|
|
{
|
|
var telemetry = telemetryService.GetOrDefaultTelemetryByIdWell(idWell);
|
|
if (telemetry is null)
|
|
return Enumerable.Empty<TDto>();
|
|
|
|
return await GetByTelemetryAsync(telemetry.Id, request, token);
|
|
}
|
|
|
|
public async Task<IEnumerable<TDto>> GetByTelemetryAsync(int idTelemetry, TelemetryDataRequest request, CancellationToken token)
|
|
{
|
|
var timezone = telemetryService.GetTimezone(idTelemetry);
|
|
|
|
var cache = telemetryDataCache.GetOrDefault(idTelemetry, request);
|
|
|
|
if (cache is not null)
|
|
return cache;
|
|
|
|
var query = BuildQuery(idTelemetry, request);
|
|
|
|
var entities = await query
|
|
.AsNoTracking()
|
|
.ToArrayAsync(token);
|
|
|
|
var dtos = entities.Select(e => Convert(e, timezone.Hours));
|
|
|
|
return dtos;
|
|
}
|
|
|
|
protected virtual IQueryable<TEntity> BuildQuery(int idTelemetry, TelemetryDataRequest request)
|
|
{
|
|
var dbSet = db.Set<TEntity>();
|
|
|
|
var query = dbSet
|
|
.Where(d => d.IdTelemetry == idTelemetry);
|
|
|
|
if (request.GeDate.HasValue)
|
|
{
|
|
var geDate = request.GeDate.Value.UtcDateTime;
|
|
query = query.Where(d => d.DateTime >= geDate);
|
|
}
|
|
|
|
if (request.LeDate.HasValue)
|
|
{
|
|
var leDate = request.LeDate.Value.UtcDateTime;
|
|
query = query.Where(d => d.DateTime <= leDate);
|
|
}
|
|
|
|
if (request.Divider > 1)
|
|
query = query.Where((d) => (((d.DateTime.DayOfYear * 24 + d.DateTime.Hour) * 60 + d.DateTime.Minute) * 60 + d.DateTime.Second) % request.Divider == 0);
|
|
|
|
switch (request.Order)
|
|
{
|
|
case 1:// Поздние вперед
|
|
query = query
|
|
.OrderByDescending(d => d.DateTime)
|
|
.Skip(request.Skip)
|
|
.Take(request.Take)
|
|
.OrderBy(d => d.DateTime);
|
|
break;
|
|
default:// Ранние вперед
|
|
query = query
|
|
.OrderBy(d => d.DateTime)
|
|
.Skip(request.Skip)
|
|
.Take(request.Take);
|
|
break;
|
|
}
|
|
|
|
return query;
|
|
}
|
|
|
|
private IQueryable<TEntity> BuildQuery(TelemetryPartDeleteRequest request)
|
|
{
|
|
var query = db.Set<TEntity>()
|
|
.Where(o => o.IdTelemetry == request.IdTelemetry);
|
|
|
|
if (request.GeDate is not null)
|
|
{
|
|
var geDate = request.GeDate.Value.ToUniversalTime();
|
|
query = query.Where(o => o.DateTime <= geDate);
|
|
}
|
|
|
|
if (request.LeDate is not null)
|
|
{
|
|
var leDate = request.LeDate.Value.ToUniversalTime();
|
|
query = query.Where(o => o.DateTime >= leDate);
|
|
}
|
|
|
|
return query;
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async Task<DatesRangeDto?> GetRangeAsync(int idWell, DateTimeOffset geDate, DateTimeOffset? leDate, CancellationToken token)
|
|
{
|
|
var telemetry = telemetryService.GetOrDefaultTelemetryByIdWell(idWell)
|
|
?? throw new ArgumentInvalidException(nameof(idWell), $"По скважине id:{idWell} нет телеметрии");
|
|
|
|
if ((DateTimeOffset.UtcNow - geDate) < TimeSpan.FromHours(12))
|
|
{
|
|
// пробуем обойтись кэшем
|
|
var cachedRange = telemetryDataCache.GetOrDefaultCachedDataDateRange(telemetry.Id);
|
|
if (cachedRange is not null)
|
|
{
|
|
var datesRange = new DatesRangeDto { From = cachedRange.From, To = cachedRange.To };
|
|
if (geDate >= cachedRange.From)
|
|
datesRange.From = geDate.ToOffset(cachedRange.From.Offset);
|
|
|
|
if (leDate.HasValue && leDate <= cachedRange.To)
|
|
datesRange.To = leDate.Value.ToOffset(cachedRange.To.Offset);
|
|
|
|
return datesRange;
|
|
}
|
|
}
|
|
|
|
var query = db.Set<TEntity>()
|
|
.Where(entity => entity.IdTelemetry == telemetry.Id)
|
|
.Where(entity => entity.DateTime >= geDate.ToUniversalTime());
|
|
|
|
if (leDate.HasValue)
|
|
query = query.Where(entity => entity.DateTime <= leDate.Value.ToUniversalTime());
|
|
|
|
var groupQuery = query
|
|
.GroupBy(entity => entity.IdTelemetry)
|
|
.Select(group => new
|
|
{
|
|
MinDate = group.Min(entity => entity.DateTime),
|
|
MaxDate = group.Max(entity => entity.DateTime),
|
|
});
|
|
|
|
var result = await groupQuery.FirstOrDefaultAsync(token);
|
|
if (result is null)
|
|
return null;
|
|
|
|
var range = new DatesRangeDto
|
|
{
|
|
From = result.MinDate.ToOffset(telemetry.TimeZone!.Offset),
|
|
To = result.MaxDate.ToOffset(telemetry.TimeZone!.Offset),
|
|
};
|
|
return range;
|
|
}
|
|
|
|
public DatesRangeDto? GetRange(int idWell)
|
|
{
|
|
var telemetry = telemetryService.GetOrDefaultTelemetryByIdWell(idWell);
|
|
if (telemetry is null)
|
|
return default;
|
|
|
|
return telemetryDataCache.GetOrDefaultWellDataDateRange(telemetry.Id);
|
|
}
|
|
|
|
protected abstract TDto Convert(TEntity src, double timezoneOffset);
|
|
|
|
protected abstract TEntity Convert(TDto src, double timezoneOffset);
|
|
|
|
public async Task<int> DeleteAsync(TelemetryPartDeleteRequest request, CancellationToken token)
|
|
{
|
|
var query = BuildQuery(request);
|
|
db.Set<TEntity>().RemoveRange(query);
|
|
return await db.SaveChangesAsync(token);
|
|
}
|
|
}
|
|
}
|