using System.Drawing.Imaging; using System.IO; using System.Net; using Azaion.Common.Database; using Azaion.Common.DTO; using Azaion.Common.DTO.Config; using Azaion.Common.DTO.Queue; using Azaion.Common.Events; using Azaion.Common.Extensions; using Azaion.CommonSecurity.DTO; using Azaion.CommonSecurity.Services; using LinqToDB; using LinqToDB.Data; using MediatR; using MessagePack; using Microsoft.Extensions.Options; using Newtonsoft.Json; using RabbitMQ.Stream.Client; using RabbitMQ.Stream.Client.Reliable; namespace Azaion.Common.Services; public class AnnotationService : INotificationHandler { private readonly IDbFactory _dbFactory; private readonly FailsafeAnnotationsProducer _producer; private readonly IGalleryService _galleryService; private readonly IMediator _mediator; private readonly IAzaionApi _api; private readonly QueueConfig _queueConfig; private Consumer _consumer = null!; private readonly UIConfig _uiConfig; private static readonly Guid SaveTaskId = Guid.NewGuid(); public AnnotationService( IDbFactory dbFactory, FailsafeAnnotationsProducer producer, IOptions queueConfig, IOptions uiConfig, IGalleryService galleryService, IMediator mediator, IAzaionApi api) { _dbFactory = dbFactory; _producer = producer; _galleryService = galleryService; _mediator = mediator; _api = api; _queueConfig = queueConfig.Value; _uiConfig = uiConfig.Value; Task.Run(async () => await Init()).Wait(); } private async Task Init(CancellationToken cancellationToken = default) { if (!_api.CurrentUser.Role.IsValidator()) return; var consumerSystem = await StreamSystem.Create(new StreamSystemConfig { Endpoints = new List{new DnsEndPoint(_queueConfig.Host, _queueConfig.Port)}, UserName = _queueConfig.ConsumerUsername, Password = _queueConfig.ConsumerPassword }); var offsets = _api.CurrentUser.UserConfig?.QueueOffsets ?? new UserQueueOffsets(); _consumer = await Consumer.Create(new ConsumerConfig(consumerSystem, Constants.MQ_ANNOTATIONS_QUEUE) { Reference = _api.CurrentUser.Email, OffsetSpec = new OffsetTypeOffset(offsets.AnnotationsOffset + 1), MessageHandler = async (_, _, context, message) => { var msg = MessagePackSerializer.Deserialize(message.Data.Contents); offsets.AnnotationsOffset = context.Offset; ThrottleExt.Throttle(() => { _api.UpdateOffsets(offsets); return Task.CompletedTask; }, SaveTaskId, TimeSpan.FromSeconds(10), scheduleCallAfterCooldown: true); if (msg.CreatedEmail == _api.CurrentUser.Email) //Don't process messages by yourself return; await SaveAnnotationInner( msg.CreatedDate, msg.OriginalMediaName, msg.Time, JsonConvert.DeserializeObject>(msg.Detections) ?? [], msg.Source, new MemoryStream(msg.Image), msg.CreatedRole, msg.CreatedEmail, fromQueue: true, token: cancellationToken); } }); } //AI public async Task SaveAnnotation(AnnotationImage a, CancellationToken ct = default) { a.Time = TimeSpan.FromMilliseconds(a.Milliseconds); return await SaveAnnotationInner(DateTime.Now, a.OriginalMediaName, a.Time, a.Detections.ToList(), SourceEnum.AI, new MemoryStream(a.Image), _api.CurrentUser.Role, _api.CurrentUser.Email, token: ct); } //Manual public async Task SaveAnnotation(string originalMediaName, TimeSpan time, List detections, Stream? stream = null, CancellationToken token = default) => await SaveAnnotationInner(DateTime.UtcNow, originalMediaName, time, detections, SourceEnum.Manual, stream, _api.CurrentUser.Role, _api.CurrentUser.Email, token: token); // Manual save from Validators -> Validated -> stream: azaion-annotations-confirm // AI, Manual save from Operators -> Created -> stream: azaion-annotations private async Task SaveAnnotationInner(DateTime createdDate, string originalMediaName, TimeSpan time, List detections, SourceEnum source, Stream? stream, RoleEnum userRole, string createdEmail, bool fromQueue = false, CancellationToken token = default) { AnnotationStatus status; var fName = originalMediaName.ToTimeName(time); var annotation = await _dbFactory.Run(async db => { var ann = await db.Annotations.FirstOrDefaultAsync(x => x.Name == fName, token: token); status = userRole.IsValidator() && source == SourceEnum.Manual ? AnnotationStatus.Validated : AnnotationStatus.Created; await db.Detections.DeleteAsync(x => x.AnnotationName == fName, token: token); if (ann != null) { await db.Annotations .Where(x => x.Name == fName) .Set(x => x.Source, source) .Set(x => x.AnnotationStatus, status) .Set(x => x.CreatedDate, createdDate) .Set(x => x.CreatedEmail, createdEmail) .Set(x => x.CreatedRole, userRole) .UpdateAsync(token: token); ann.Detections = detections; } else { ann = new Annotation { CreatedDate = createdDate, Name = fName, OriginalMediaName = originalMediaName, Time = time, ImageExtension = Constants.JPG_EXT, CreatedEmail = createdEmail, CreatedRole = userRole, AnnotationStatus = status, Source = source, Detections = detections }; await db.InsertAsync(ann, token: token); } await db.BulkCopyAsync(detections, cancellationToken: token); return ann; }); if (stream != null) { var img = System.Drawing.Image.FromStream(stream); img.Save(annotation.ImagePath, ImageFormat.Jpeg); //todo: check png images coming from queue } await YoloLabel.WriteToFile(detections, annotation.LabelPath, token); await _galleryService.CreateThumbnail(annotation, token); if (_uiConfig.GenerateAnnotatedImage) await _galleryService.CreateAnnotatedImage(annotation, token); if (!fromQueue && !_uiConfig.SilentDetection) //Send to queue only if we're not getting from queue already await _producer.SendToInnerQueue(annotation, token); await _mediator.Publish(new AnnotationCreatedEvent(annotation), token); ThrottleExt.Throttle(async () => { _dbFactory.SaveToDisk(); await Task.CompletedTask; }, SaveTaskId, TimeSpan.FromSeconds(5), true); return annotation; } public async Task ValidateAnnotations(List annotations, CancellationToken token = default) { if (!_api.CurrentUser.Role.IsValidator()) return; var annNames = annotations.Select(x => x.Name).ToHashSet(); await _dbFactory.Run(async db => { await db.Annotations .Where(x => annNames.Contains(x.Name)) .Set(x => x.AnnotationStatus, AnnotationStatus.Validated) .Set(x => x.ValidateDate, DateTime.UtcNow) .Set(x => x.ValidateEmail, _api.CurrentUser.Email) .UpdateAsync(token: token); }); ThrottleExt.Throttle(async () => { _dbFactory.SaveToDisk(); await Task.CompletedTask; }, SaveTaskId, TimeSpan.FromSeconds(5), true); } public async Task Handle(AnnotationsDeletedEvent notification, CancellationToken cancellationToken) { await _dbFactory.DeleteAnnotations(notification.Annotations, cancellationToken); foreach (var annotation in notification.Annotations) { File.Delete(annotation.ImagePath); File.Delete(annotation.LabelPath); File.Delete(annotation.ThumbPath); } } }