julian
353 lines
19 KiB
C#
353 lines
19 KiB
C#
using System.Timers;
|
|
using adas_core.Domain.Enums;
|
|
using adas_core.Domain.Models;
|
|
using adas_core.Domain.Models.GroupedObservations;
|
|
using adas_core.Domain.Utils;
|
|
using MongoDB.Bson;
|
|
using Serilog;
|
|
using static adas_core.Domain.Models.GroupedObservation;
|
|
using Timer = System.Timers.Timer;
|
|
|
|
namespace adas_core.Application.Subscriptions;
|
|
|
|
/**
|
|
* This class is used to group the subscribers by the same group
|
|
*/
|
|
/// <summary>
|
|
/// Represents a grouped WebSocket subscriber that aggregates and manages multiple related subscription registrations.
|
|
/// </summary>
|
|
public class WsSubscriberGrouped
|
|
{
|
|
private readonly EventHandler<string> _sendEvent;
|
|
|
|
/// <summary>
|
|
/// Initializes a new instance of <see cref="WsSubscriberGrouped"/> that registers <paramref name="wsId"/> in the shared WsSubscriber collection and copies the grouping configuration from <paramref name="gf"/>. The constructor also seeds the last-observation state through <see cref="UpdateLastGo"/>, computes the identity hash via <see cref="CryptoAdas.CreateMd5GroupedObs"/>, and starts the internal <see cref="Timer"/> that drives periodic emission through <paramref name="sendEvent"/>.
|
|
/// </summary>
|
|
/// <param name="patientId">The <see cref="ObjectId"/> of the patient whose grouped observations this subscriber tracks.</param>
|
|
/// <param name="wsId">The web socket subscriber identifier added to the shared WsSubscriber list and used as the key in the per-instance group dictionary.</param>
|
|
/// <param name="timeZoneId">The Windows or IANA time zone identifier applied to the observations; when null, falls back to Romance Standard Time.</param>
|
|
/// <param name="gf">The <see cref="GroupedField"/> providing the grouping configuration consumed for names, max, since, start time shift, regularity, result, and group key.</param>
|
|
/// <param name="lastGroupedObservationObs">The most recent <see cref="GroupedObservation"/> passed to <see cref="UpdateLastGo"/> to initialize the last-observation state.</param>
|
|
/// <param name="sendEvent">The <see cref="EventHandler{String}"/> invoked by the timer to forward outgoing messages to the subscriber.</param>
|
|
/// <!-- aidoc:v1 sig=16f9dfc body=cc103b1 -->
|
|
public WsSubscriberGrouped(ObjectId patientId, string wsId, string? timeZoneId,
|
|
GroupedField gf, GroupedObservation lastGroupedObservationObs, EventHandler<string> sendEvent)
|
|
{
|
|
_sendEvent = sendEvent;
|
|
if (WsSubscriber.ToList().FirstOrDefault(s => s == wsId) == null) WsSubscriber.Add(wsId);
|
|
|
|
TimeZoneId = timeZoneId ?? "Romance Standard Time";
|
|
PatientId = patientId;
|
|
Names = CollectionsUtils.IfEmptyOrNull(gf.Names, [gf.Name ?? string.Empty]);
|
|
Max = gf.Max;
|
|
Since = gf.Since;
|
|
StartTimeShift = gf.StartTimeShift;
|
|
Regularity = gf.Regularity;
|
|
Result = gf.Result;
|
|
Group.Add(wsId, gf.Group ?? string.Empty);
|
|
UpdateLastGo(lastGroupedObservationObs);
|
|
HashCode = CryptoAdas.CreateMd5GroupedObs(gf, patientId);
|
|
|
|
Timer = new Timer
|
|
{
|
|
Interval = SetUpTimerInterval()
|
|
};
|
|
Timer.Elapsed += Timer_Elapsed;
|
|
if (Regularity != GroupedObservationEnum.Regularity.Times) Timer.Start();
|
|
}
|
|
|
|
public List<string> WsSubscriber { get; set; } = [];
|
|
public ObjectId PatientId { get; set; }
|
|
public List<string> Names { get; set; }
|
|
public int Max { get; set; }
|
|
public GroupedObservationEnum.Regularity? Regularity { get; set; }
|
|
public List<GroupedObservationEnum.Result> Result { get; set; }
|
|
public GroupedObservationEnum.Since Since { get; set; }
|
|
public List<string> StartTimeShift { get; set; }
|
|
public string HashCode { get; set; }
|
|
public string TimeZoneId { get; set; }
|
|
public List<GroupedObservationObs> LastGroupedObservationObs { get; set; } = [];
|
|
public Dictionary<string, string> Group { get; set; } = new();
|
|
public Timer Timer { get; set; }
|
|
|
|
/// <summary>
|
|
/// Refreshes the cached last observations by matching each known name against the supplied grouped observation, falling back to the most recent observation when no match is found.
|
|
/// </summary>
|
|
/// <param name="lastGroupedObservationObs">The source grouped observation whose observations are searched to populate the last-observation cache.</param>
|
|
public void UpdateLastGo(GroupedObservation lastGroupedObservationObs)
|
|
{
|
|
LastGroupedObservationObs.Clear();
|
|
foreach (var name in Names)
|
|
{
|
|
var lastObs = lastGroupedObservationObs.Observations.LastOrDefault(c => c.Name == name);
|
|
if (lastObs != null)
|
|
{
|
|
LastGroupedObservationObs.Add(lastObs);
|
|
}
|
|
else
|
|
{
|
|
lastObs = lastGroupedObservationObs.Observations.LastOrDefault();
|
|
if (lastObs != null) LastGroupedObservationObs.Add(lastObs);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles the timer elapsed event by stopping the timer, checking whether any cached grouped observation matches the current time at the configured <see cref="GroupedObservationEnum.Regularity"/> granularity (second, minute, day, or hour by default), invoking the send event when no match is found, and restarting the timer.
|
|
/// </summary>
|
|
/// <param name="sender">The source of the timer elapsed event.</param>
|
|
/// <param name="e">The <see cref="System.Timers.ElapsedEventArgs"/> instance containing the elapsed event data.</param>
|
|
/// <!-- aidoc:v1 sig=168f657 body=7634f8f -->
|
|
private void Timer_Elapsed(object? sender, ElapsedEventArgs e)
|
|
{
|
|
var currentDateTime = DateTime.Now;
|
|
Timer.Stop();
|
|
try
|
|
{
|
|
bool hasCurrentObs;
|
|
|
|
switch (Regularity)
|
|
{
|
|
case GroupedObservationEnum.Regularity.Second:
|
|
// Filtra observaciones con el mismo segundo actual
|
|
hasCurrentObs = LastGroupedObservationObs.Any(obs => obs.Time.Second == currentDateTime.Second
|
|
&& obs.Time.Minute == currentDateTime.Minute
|
|
&& obs.Time.Hour == currentDateTime.Hour
|
|
&& obs.Time.Date == currentDateTime.Date);
|
|
break;
|
|
|
|
case GroupedObservationEnum.Regularity.Minute:
|
|
// Filtra observaciones con el mismo minuto actual
|
|
hasCurrentObs = LastGroupedObservationObs.Any(obs => obs.Time.Minute == currentDateTime.Minute
|
|
&& obs.Time.Hour == currentDateTime.Hour
|
|
&& obs.Time.Date == currentDateTime.Date);
|
|
break;
|
|
|
|
case GroupedObservationEnum.Regularity.Day:
|
|
// Filtra observaciones con la misma fecha actual (día completo)
|
|
hasCurrentObs = LastGroupedObservationObs.Any(obs => obs.Time.Date == currentDateTime.Date);
|
|
break;
|
|
|
|
default:
|
|
// Filtra observaciones con la misma hora actual
|
|
hasCurrentObs = LastGroupedObservationObs.Any(obs => obs.Time.Hour == currentDateTime.Hour
|
|
&& obs.Time.Date == currentDateTime.Date);
|
|
break;
|
|
}
|
|
|
|
// Si no hay al menos una observación que coincide, invoca el evento
|
|
if (!hasCurrentObs) _sendEvent(this, HashCode);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Log.Warning(
|
|
$"Exception on WsSubsciptor for patient: {PatientId} message: {ex.Message} exception: {ex}");
|
|
}
|
|
|
|
TimerReestart();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Restarts the timer by recalculating its interval through <c>SetUpTimerInterval</c> and then starting it.
|
|
/// </summary>
|
|
private void TimerReestart()
|
|
{
|
|
Timer.Interval = SetUpTimerInterval();
|
|
Timer.Start();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Calculates the timer interval in milliseconds until the next scheduled occurrence based on the configured <see cref="Regularity"/>, supporting Second, Minute, Day, and a default (Hour) case, and adds a per-regularity grace time.
|
|
/// </summary>
|
|
/// <returns>The number of milliseconds to wait until the next interval, including the grace time.</returns>
|
|
private int SetUpTimerInterval()
|
|
{
|
|
var currentDateTime = DateTime.Now;
|
|
int graceTime;
|
|
DateTime nextIntervalTime;
|
|
|
|
switch (Regularity)
|
|
{
|
|
case GroupedObservationEnum.Regularity.Second:
|
|
nextIntervalTime = new DateTime(
|
|
currentDateTime.Year,
|
|
currentDateTime.Month,
|
|
currentDateTime.Day,
|
|
currentDateTime.Hour,
|
|
currentDateTime.Minute,
|
|
currentDateTime.Second
|
|
).AddSeconds(1);
|
|
graceTime = 500;
|
|
break;
|
|
|
|
case GroupedObservationEnum.Regularity.Minute:
|
|
nextIntervalTime = new DateTime(
|
|
currentDateTime.Year,
|
|
currentDateTime.Month,
|
|
currentDateTime.Day,
|
|
currentDateTime.Hour,
|
|
currentDateTime.Minute,
|
|
0
|
|
).AddMinutes(1);
|
|
graceTime = 30000;
|
|
break;
|
|
|
|
case GroupedObservationEnum.Regularity.Day:
|
|
nextIntervalTime = new DateTime(
|
|
currentDateTime.Year,
|
|
currentDateTime.Month,
|
|
currentDateTime.Day,
|
|
0,
|
|
0,
|
|
0
|
|
).AddDays(1);
|
|
graceTime = 60000;
|
|
break;
|
|
|
|
default:
|
|
nextIntervalTime = new DateTime(
|
|
currentDateTime.Year,
|
|
currentDateTime.Month,
|
|
currentDateTime.Day,
|
|
currentDateTime.Hour,
|
|
0,
|
|
0
|
|
).AddHours(1);
|
|
graceTime = 30000;
|
|
break;
|
|
}
|
|
|
|
var timeSpanToNextInterval = nextIntervalTime - currentDateTime;
|
|
Log.Information(
|
|
"Obs {Obs}, currentDateTime: {CurrentDateTime}, nextIntervalTime: {NextIntervalTime}, timeSpanToNextInterval: {TimeSpanToNextInterval}, timeSpanToNextInterval.TotalMilliseconds+graceTime: {Tt}",
|
|
Group, currentDateTime, nextIntervalTime, timeSpanToNextInterval,
|
|
(int)timeSpanToNextInterval.TotalMilliseconds + graceTime);
|
|
|
|
return (int)timeSpanToNextInterval.TotalMilliseconds + graceTime;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Provides static extension methods to enhance WebSocket subscriber functionality.
|
|
/// </summary>
|
|
public static class WsSubscriberExtension
|
|
{
|
|
/// <summary>
|
|
/// Determines whether the specified <see cref="GroupedField"/> and patient identifier match the current <see cref="WsSubscriberGrouped"/> instance by comparing regularity, since, start time shift, maximum, result sequence, patient identifier, and names (treating a null <paramref name="r"/>.Name as an empty string).
|
|
/// </summary>
|
|
/// <param name="source">The current <see cref="WsSubscriberGrouped"/> instance being compared.</param>
|
|
/// <param name="r">The <see cref="GroupedField"/> to compare against the source.</param>
|
|
/// <param name="patientId">The patient identifier to match against the source patient identifier.</param>
|
|
/// <returns><c>true</c> if all compared properties are equal; otherwise, <c>false</c>.</returns>
|
|
public static bool Compare(this WsSubscriberGrouped source, GroupedField r, ObjectId patientId)
|
|
{
|
|
return r.Regularity == source.Regularity &&
|
|
r.Since == source.Since &&
|
|
CompareStringShiftList(r.StartTimeShift, source.StartTimeShift) &&
|
|
r.Max == source.Max &&
|
|
r.Result.SequenceEqual(source.Result) &&
|
|
patientId.ToString() == source.PatientId.ToString() &&
|
|
CompareNamesStringList(r.Names, [r.Name ?? string.Empty], source.Names);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Compares two string shift lists for equality, returning true when both lists are null or when both contain the same elements in the same order, and false when only one of the lists is null.
|
|
/// </summary>
|
|
/// <param name="shift">The target shift list to compare.</param>
|
|
/// <param name="sourceShift">The source shift list to compare against.</param>
|
|
/// <returns><c>true</c> if both lists are null or if their sequence of strings is equal; otherwise, <c>false</c>.</returns>
|
|
private static bool CompareStringShiftList(List<string>? shift, List<string>? sourceShift)
|
|
{
|
|
if (shift != null && sourceShift != null) return sourceShift.SequenceEqual(shift);
|
|
if (shift == null && sourceShift == null) return true;
|
|
return false;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Compares the <paramref name="sourceNames"/> list against two candidate name lists and returns whether either one matches.
|
|
/// Falls back to the second list (<paramref name="name"/>) when the first (<paramref name="names"/>) is empty or null, or does not match.
|
|
/// </summary>
|
|
/// <param name="names">The primary candidate list of names to compare against <paramref name="sourceNames"/>.</param>
|
|
/// <param name="name">The fallback candidate list of names to compare against <paramref name="sourceNames"/> when <paramref name="names"/> cannot be used.</param>
|
|
/// <param name="sourceNames">The source list of names being compared.</param>
|
|
/// <returns><c>true</c> if <paramref name="sourceNames"/> is a sequence match for either <paramref name="names"/> or <paramref name="name"/>; otherwise, <c>false</c>.</returns>
|
|
private static bool CompareNamesStringList(List<string> names, List<string> name, List<string> sourceNames)
|
|
{
|
|
if (!CollectionsUtils.IsEmptyOrNull(names) && sourceNames.SequenceEqual(names)) return true;
|
|
return !CollectionsUtils.IsEmptyOrNull(name) && sourceNames.SequenceEqual(name);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Check if the incoming observation affect to the group and is new info or is irrelevant
|
|
/// </summary>
|
|
/// <returns></returns>
|
|
public static bool IsNewObservationRelevantForGroup(this WsSubscriberGrouped group, PatientObservation obs)
|
|
{
|
|
double.TryParse(obs.Value.ToString(), out var pobsValue);
|
|
switch (group.Regularity)
|
|
{
|
|
case GroupedObservationEnum.Regularity.Shift:
|
|
case GroupedObservationEnum.Regularity.Hour:
|
|
var obsInThisHour = group.LastGroupedObservationObs.Where(c =>
|
|
new DateTime(c.Time.Year, c.Time.Month, c.Time.Day, c.Time.Hour, 0, 0) ==
|
|
new DateTime(obs.Time.Year, obs.Time.Month, obs.Time.Day, obs.Time.Hour, 0, 0) && !c.IsFilled);
|
|
//Obs is in grouped time, don't generate group for observation outside of range
|
|
|
|
|
|
//Change to don't cut is obs time is a future
|
|
var obsInTime = DateTime.UtcNow.AddHours(group.Max) > obs.Time &&
|
|
DateTime.UtcNow.AddHours(group.Max * -1) < obs.Time;
|
|
//future data is relevant because maybe nothing is inserted in the same hour after it and we cant lose it
|
|
/* if (DateTime.UtcNow.AddHours(group.Max) > obs.time)
|
|
{
|
|
|
|
}
|
|
*/
|
|
|
|
if (!obsInTime) return false;
|
|
var groupedObservationObsEnumerable = obsInThisHour.ToList();
|
|
if (!groupedObservationObsEnumerable.Any() && obsInTime) return true;
|
|
|
|
//If obs in this hour check if is relevant by Result
|
|
if (groupedObservationObsEnumerable.Any())
|
|
foreach (var result in group.Result)
|
|
switch (result)
|
|
{
|
|
case GroupedObservationEnum.Result.Max:
|
|
var value = groupedObservationObsEnumerable.Max(s => s.Max?.Value);
|
|
var parsed = double.TryParse(value?.ToString(), out var valueParsed);
|
|
//actual value is higher, is relevant
|
|
if (parsed && pobsValue > valueParsed) return true;
|
|
break;
|
|
case GroupedObservationEnum.Result.Min:
|
|
var minValue = groupedObservationObsEnumerable.Min(s => s.Min?.Value);
|
|
var isParsed = double.TryParse(minValue?.ToString(), out var minValueParsed);
|
|
//actual value is higher, is relevant
|
|
if (isParsed && pobsValue < minValueParsed) return true;
|
|
break;
|
|
case GroupedObservationEnum.Result.Last:
|
|
var lastObsInTime = groupedObservationObsEnumerable.OrderBy(t => t.Last?.Time).First();
|
|
//is the new latest in the hour, relevant
|
|
if (lastObsInTime.Time < obs.Time) return true;
|
|
break;
|
|
case GroupedObservationEnum.Result.LastFilled:
|
|
var lastFilledObsInTime = groupedObservationObsEnumerable
|
|
.OrderBy(t => t.LastFilled?.Time).First();
|
|
//is the new latest in the hour, relevant
|
|
if (lastFilledObsInTime.Time < obs.Time) return true;
|
|
break;
|
|
//always is relevant because SUM everything in the hour
|
|
//TODO halfHour should check if any obs already in the halfHour
|
|
case GroupedObservationEnum.Result.Sum:
|
|
case GroupedObservationEnum.Result.Average:
|
|
case GroupedObservationEnum.Result.HalfHour:
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
default:
|
|
|
|
return true;
|
|
}
|
|
}
|
|
} |