using DD.Persistence.Database.Entity; using DD.Persistence.Models; using DD.Persistence.Models.Common; using DD.Persistence.Repositories; using Microsoft.EntityFrameworkCore; namespace DD.Persistence.Database.Postgres.Repositories; public class TimestampedValuesRepository : ITimestampedValuesRepository { private readonly DbContext db; public TimestampedValuesRepository(DbContext db) { this.db = db; } protected virtual IQueryable GetQueryReadOnly() => this.db.Set(); public async virtual Task AddRange(Guid discriminatorId, IEnumerable dtos, CancellationToken token) { var timestampedValuesEntities = dtos.Select(dto => new TimestampedValues() { DiscriminatorId = discriminatorId, Timestamp = dto.Timestamp.ToUniversalTime(), Values = dto.Values.Values.ToArray() }); await db.Set().AddRangeAsync(timestampedValuesEntities, token); var result = await db.SaveChangesAsync(token); return result; } public async virtual Task>> Get(IEnumerable discriminatorIds, DateTimeOffset? timestampBegin, IEnumerable? columnNames, int skip, int take, CancellationToken token) { var query = GetQueryReadOnly() .Where(entity => discriminatorIds.Contains(entity.DiscriminatorId)); // Фильтрация по дате if (timestampBegin.HasValue) { query = ApplyGeTimestamp(query, timestampBegin.Value); } // Группировка отсортированных значений по DiscriminatorId var groupQuery = query .GroupBy(e => e.DiscriminatorId) .Select(g => KeyValuePair.Create(g.Key, g.OrderBy(i => i.Timestamp).Skip(skip).Take(take))); var entities = await groupQuery.ToArrayAsync(token); var result = entities.ToDictionary(k => k.Key, v => v.Value.Select(e => ( e.Timestamp, e.Values ))); return result; } public async virtual Task> GetFirst(Guid discriminatorId, int takeCount, CancellationToken token) { var query = GetQueryReadOnly() .OrderBy(e => e.Timestamp) .Take(takeCount); var entities = await query.ToArrayAsync(token); var result = entities.Select(e => ( e.Timestamp, e.Values )); return result; } public async virtual Task> GetLast(Guid discriminatorId, int takeCount, CancellationToken token) { var query = GetQueryReadOnly() .OrderByDescending(e => e.Timestamp) .Take(takeCount); var entities = await query.ToArrayAsync(token); var result = entities.Select(e => ( e.Timestamp, e.Values )); return result; } // ToDo: прореживание должно осуществляться до материализации public async virtual Task> GetResampledData( Guid discriminatorId, DateTimeOffset dateBegin, double intervalSec = 600d, int approxPointsCount = 1024, CancellationToken token = default) { var result = await GetGtDate(discriminatorId, dateBegin, token); var dateEnd = dateBegin.AddSeconds(intervalSec); result = result .Where(i => i.Item1 <= dateEnd); var ratio = result.Count() / approxPointsCount; if (ratio > 1) result = result .Where((_, index) => index % ratio == 0); return result; } public async virtual Task> GetGtDate(Guid discriminatorId, DateTimeOffset timestampBegin, CancellationToken token) { var query = GetQueryReadOnly() .Where(e => e.Timestamp > timestampBegin); var entities = await query.ToArrayAsync(token); var result = entities.Select(e => ( e.Timestamp, e.Values )); return result; } public async virtual Task GetDatesRange(Guid discriminatorId, CancellationToken token) { var query = GetQueryReadOnly() .GroupBy(entity => entity.DiscriminatorId) .Select(group => new { Min = group.Min(entity => entity.Timestamp), Max = group.Max(entity => entity.Timestamp), }); var item = await query.FirstOrDefaultAsync(token); if (item is null) { return null; } var dto = new DatesRangeDto { From = item.Min, To = item.Max, }; return dto; } public virtual Task Count(Guid discriminatorId, CancellationToken token) { var dbSet = db.Set(); var query = dbSet.Where(entity => entity.DiscriminatorId == discriminatorId); return query.CountAsync(token); } /// /// Применить фильтр по дате /// /// /// /// private IQueryable ApplyGeTimestamp(IQueryable query, DateTimeOffset timestampBegin) { var geTimestampUtc = timestampBegin.ToUniversalTime(); var result = query .Where(entity => entity.Timestamp >= geTimestampUtc); return result; } }