Realtime DB 추가 및 Historical DB추가
This commit is contained in:
windpacer
@@ -14,6 +14,8 @@ public class ExperionDbContext : DbContext
|
||||
public DbSet<ExperionRecord> ExperionRecords => Set<ExperionRecord>();
|
||||
public DbSet<RawNodeMap> RawNodeMaps => Set<RawNodeMap>();
|
||||
public DbSet<NodeMapMaster> NodeMapMasters => Set<NodeMapMaster>();
|
||||
public DbSet<RealtimePoint> RealtimePoints => Set<RealtimePoint>();
|
||||
public DbSet<HistoryRecord> HistoryRecords => Set<HistoryRecord>();
|
||||
|
||||
protected override void OnModelCreating(ModelBuilder modelBuilder)
|
||||
{
|
||||
@@ -37,6 +39,20 @@ public class ExperionDbContext : DbContext
|
||||
e.HasIndex(x => x.NodeId);
|
||||
e.HasIndex(x => x.Level);
|
||||
});
|
||||
|
||||
modelBuilder.Entity<RealtimePoint>(e =>
|
||||
{
|
||||
e.HasKey(x => x.Id);
|
||||
e.HasIndex(x => x.NodeId).IsUnique();
|
||||
e.HasIndex(x => x.TagName);
|
||||
});
|
||||
|
||||
modelBuilder.Entity<HistoryRecord>(e =>
|
||||
{
|
||||
e.HasKey(x => x.Id);
|
||||
e.HasIndex(x => x.TagName);
|
||||
e.HasIndex(x => x.RecordedAt);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -83,6 +99,31 @@ public class ExperionDbService : IExperionDbService
|
||||
)
|
||||
""");
|
||||
|
||||
await _ctx.Database.ExecuteSqlRawAsync("""
|
||||
CREATE TABLE IF NOT EXISTS realtime_table (
|
||||
id SERIAL PRIMARY KEY,
|
||||
tagname TEXT NOT NULL,
|
||||
node_id TEXT NOT NULL UNIQUE,
|
||||
livevalue TEXT,
|
||||
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
)
|
||||
""");
|
||||
|
||||
await _ctx.Database.ExecuteSqlRawAsync("""
|
||||
CREATE TABLE IF NOT EXISTS history_table (
|
||||
id SERIAL PRIMARY KEY,
|
||||
tagname TEXT NOT NULL,
|
||||
node_id TEXT NOT NULL,
|
||||
value TEXT,
|
||||
recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
)
|
||||
""");
|
||||
|
||||
await _ctx.Database.ExecuteSqlRawAsync(
|
||||
"CREATE INDEX IF NOT EXISTS idx_history_tagname ON history_table(tagname)");
|
||||
await _ctx.Database.ExecuteSqlRawAsync(
|
||||
"CREATE INDEX IF NOT EXISTS idx_history_recorded_at ON history_table(recorded_at)");
|
||||
|
||||
_logger.LogInformation("[ExperionDb] 데이터베이스 초기화 완료");
|
||||
return true;
|
||||
}
|
||||
@@ -162,6 +203,155 @@ public class ExperionDbService : IExperionDbService
|
||||
return new NodeMapStats(total, objectCount, variableCount, maxLevel, dataTypes);
|
||||
}
|
||||
|
||||
// ── RealtimeTable ─────────────────────────────────────────────────────────
|
||||
|
||||
private static string ExtractTagName(string nodeId)
|
||||
{
|
||||
var idx = nodeId.LastIndexOf(':');
|
||||
return idx >= 0 ? nodeId[(idx + 1)..] : nodeId;
|
||||
}
|
||||
|
||||
public async Task<int> BuildRealtimeTableAsync(
|
||||
IEnumerable<string> names, IEnumerable<string> dataTypes)
|
||||
{
|
||||
var nameList = names.Where(n => !string.IsNullOrEmpty(n)).ToList();
|
||||
var dtList = dataTypes.Where(d => !string.IsNullOrEmpty(d)).ToList();
|
||||
|
||||
var q = _ctx.NodeMapMasters.AsQueryable();
|
||||
if (nameList.Count > 0) q = q.Where(x => nameList.Contains(x.Name));
|
||||
if (dtList.Count > 0) q = q.Where(x => dtList.Contains(x.DataType));
|
||||
|
||||
var sources = await q.ToListAsync();
|
||||
|
||||
await _ctx.Database.ExecuteSqlRawAsync(
|
||||
"TRUNCATE TABLE realtime_table RESTART IDENTITY");
|
||||
|
||||
var points = sources.Select(s => new RealtimePoint
|
||||
{
|
||||
TagName = ExtractTagName(s.NodeId),
|
||||
NodeId = s.NodeId,
|
||||
LiveValue = null,
|
||||
Timestamp = DateTime.UtcNow
|
||||
}).ToList();
|
||||
|
||||
await _ctx.RealtimePoints.AddRangeAsync(points);
|
||||
var saved = await _ctx.SaveChangesAsync();
|
||||
_logger.LogInformation("[ExperionDb] realtime_table 빌드: {Count}건", saved);
|
||||
return saved;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<RealtimePoint>> GetRealtimePointsAsync()
|
||||
=> await _ctx.RealtimePoints.OrderBy(x => x.TagName).ToListAsync();
|
||||
|
||||
public async Task<RealtimePoint> AddRealtimePointAsync(string nodeId)
|
||||
{
|
||||
var existing = await _ctx.RealtimePoints.FirstOrDefaultAsync(x => x.NodeId == nodeId);
|
||||
if (existing != null) return existing;
|
||||
|
||||
var point = new RealtimePoint
|
||||
{
|
||||
TagName = ExtractTagName(nodeId),
|
||||
NodeId = nodeId,
|
||||
LiveValue = null,
|
||||
Timestamp = DateTime.UtcNow
|
||||
};
|
||||
_ctx.RealtimePoints.Add(point);
|
||||
await _ctx.SaveChangesAsync();
|
||||
_logger.LogInformation("[ExperionDb] 수동 추가: {NodeId}", nodeId);
|
||||
return point;
|
||||
}
|
||||
|
||||
public async Task<bool> DeleteRealtimePointAsync(int id)
|
||||
{
|
||||
var point = await _ctx.RealtimePoints.FindAsync(id);
|
||||
if (point == null) return false;
|
||||
_ctx.RealtimePoints.Remove(point);
|
||||
await _ctx.SaveChangesAsync();
|
||||
return true;
|
||||
}
|
||||
|
||||
public async Task<int> UpdateLiveValueAsync(string nodeId, string? value, DateTime timestamp)
|
||||
{
|
||||
return await _ctx.RealtimePoints
|
||||
.Where(x => x.NodeId == nodeId)
|
||||
.ExecuteUpdateAsync(s => s
|
||||
.SetProperty(x => x.LiveValue, value)
|
||||
.SetProperty(x => x.Timestamp, timestamp));
|
||||
}
|
||||
|
||||
public async Task<int> BatchUpdateLiveValuesAsync(IEnumerable<LiveValueUpdate> updates)
|
||||
{
|
||||
var list = updates.ToList();
|
||||
if (list.Count == 0) return 0;
|
||||
|
||||
// 단일 DbContext(단일 DB 커넥션)에서 순차 업데이트 — 커넥션 폭발 방지
|
||||
int total = 0;
|
||||
foreach (var u in list)
|
||||
{
|
||||
total += await _ctx.RealtimePoints
|
||||
.Where(x => x.NodeId == u.NodeId)
|
||||
.ExecuteUpdateAsync(s => s
|
||||
.SetProperty(x => x.LiveValue, u.Value)
|
||||
.SetProperty(x => x.Timestamp, u.Timestamp));
|
||||
}
|
||||
return total;
|
||||
}
|
||||
|
||||
// ── HistoryTable ──────────────────────────────────────────────────────────
|
||||
|
||||
public async Task<int> SnapshotToHistoryAsync()
|
||||
{
|
||||
var now = DateTime.UtcNow;
|
||||
var points = await _ctx.RealtimePoints.ToListAsync();
|
||||
if (points.Count == 0) return 0;
|
||||
|
||||
var rows = points.Select(p => new HistoryRecord
|
||||
{
|
||||
TagName = p.TagName,
|
||||
NodeId = p.NodeId,
|
||||
Value = p.LiveValue,
|
||||
RecordedAt = now
|
||||
}).ToList();
|
||||
|
||||
await _ctx.HistoryRecords.AddRangeAsync(rows);
|
||||
var saved = await _ctx.SaveChangesAsync();
|
||||
_logger.LogInformation("[ExperionDb] history 스냅샷: {Count}건 @ {Time:HH:mm:ss}", saved, now);
|
||||
return saved;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<string>> GetTagNamesAsync()
|
||||
=> await _ctx.RealtimePoints.Select(x => x.TagName).OrderBy(x => x).ToListAsync();
|
||||
|
||||
public async Task<HistoryQueryResult> QueryHistoryAsync(
|
||||
IEnumerable<string> tagNames, DateTime? from, DateTime? to, int limit)
|
||||
{
|
||||
var tags = tagNames.Where(t => !string.IsNullOrEmpty(t)).ToList();
|
||||
|
||||
var q = _ctx.HistoryRecords.AsQueryable();
|
||||
if (tags.Count > 0) q = q.Where(x => tags.Contains(x.TagName));
|
||||
if (from.HasValue) q = q.Where(x => x.RecordedAt >= from.Value);
|
||||
if (to.HasValue) q = q.Where(x => x.RecordedAt <= to.Value);
|
||||
|
||||
var rows = await q.OrderBy(x => x.RecordedAt)
|
||||
.Take(Math.Min(limit, 5000))
|
||||
.ToListAsync();
|
||||
|
||||
// recorded_at 기준으로 행을 묶어서 pivot 구성
|
||||
var grouped = rows
|
||||
.GroupBy(x => x.RecordedAt)
|
||||
.Select(g => new HistoryRow(
|
||||
g.Key,
|
||||
g.ToDictionary(r => r.TagName, r => r.Value)
|
||||
as IReadOnlyDictionary<string, string?>))
|
||||
.ToList();
|
||||
|
||||
var usedTags = tags.Count > 0
|
||||
? tags
|
||||
: rows.Select(x => x.TagName).Distinct().OrderBy(x => x).ToList();
|
||||
|
||||
return new HistoryQueryResult(usedTags, grouped);
|
||||
}
|
||||
|
||||
public async Task<NodeMapQueryResult> QueryMasterAsync(
|
||||
int? minLevel, int? maxLevel, string? nodeClass,
|
||||
IEnumerable<string>? names, string? nodeId, string? dataType,
|
||||
|
||||
Reference in New Issue
Block a user