Errors sending to UI

notifying client of AI model conversion

Azaion.Common/Services/AnnotationService.cs

@@ -20,7 +20,7 @@ using RabbitMQ.Stream.Client.Reliable;
 
 namespace Azaion.Common.Services;
 
-public class AnnotationService : INotificationHandler<AnnotationsDeletedEvent>
+public class AnnotationService : IAnnotationService, INotificationHandler<AnnotationsDeletedEvent>
 {
     private readonly IDbFactory _dbFactory;
     private readonly FailsafeAnnotationsProducer _producer;
@@ -124,11 +124,17 @@ public class AnnotationService : INotificationHandler<AnnotationsDeletedEvent>
         var fName = originalMediaName.ToTimeName(time);
         var annotation = await _dbFactory.Run(async db =>
         {
-            var ann = await db.Annotations.FirstOrDefaultAsync(x => x.Name == fName, token: token);
+            var ann = await db.Annotations
+                .LoadWith(x => x.Detections)
+                .FirstOrDefaultAsync(x => x.Name == fName, token: token);
+
             status = userRole.IsValidator() && source == SourceEnum.Manual
                 ? AnnotationStatus.Validated
                 : AnnotationStatus.Created;
 
+            if (fromQueue && ann is { AnnotationStatus: AnnotationStatus.Validated })
+                return ann;
+
             await db.Detections.DeleteAsync(x => x.AnnotationName == fName, token: token);
 
             if (ann != null)
@@ -164,6 +170,9 @@ public class AnnotationService : INotificationHandler<AnnotationsDeletedEvent>
             return ann;
         });
 
+        if (fromQueue && annotation is { AnnotationStatus: AnnotationStatus.Validated })
+            return annotation;
+
         if (stream != null)
         {
             var img = System.Drawing.Image.FromStream(stream);
@@ -219,4 +228,12 @@ public class AnnotationService : INotificationHandler<AnnotationsDeletedEvent>
             File.Delete(annotation.ThumbPath);
         }
     }
+}
+
+public interface IAnnotationService
+{
+    Task<Annotation> SaveAnnotation(AnnotationImage a, CancellationToken ct = default);
+    Task<Annotation> SaveAnnotation(string originalMediaName, TimeSpan time, List<Detection> detections, Stream? stream = null, CancellationToken token = default);
+    Task ValidateAnnotations(List<Annotation> annotations, CancellationToken token = default);
+
 }

Azaion.Common/Services/GPSMatcherService.cs

@@ -17,7 +17,7 @@ public interface IGpsMatcherService
 public class GpsMatcherService(IGpsMatcherClient gpsMatcherClient, ISatelliteDownloader satelliteTileDownloader, IOptions<DirectoriesConfig> dirConfig) : IGpsMatcherService
 {
     private const int ZOOM_LEVEL = 18;
-    private const int POINTS_COUNT = 5;
+    private const int POINTS_COUNT = 10;
     private const int DISTANCE_BETWEEN_POINTS_M = 100;
     private const double SATELLITE_RADIUS_M = DISTANCE_BETWEEN_POINTS_M * (POINTS_COUNT + 1);
 
@@ -41,7 +41,7 @@ public class GpsMatcherService(IGpsMatcherClient gpsMatcherClient, ISatelliteDow
         var indexOffset = 0;
         while (routeFiles.Any())
         {
-            await satelliteTileDownloader.GetTiles(currentLat, currentLon, SATELLITE_RADIUS_M, ZOOM_LEVEL, detectToken);
+            //await satelliteTileDownloader.GetTiles(currentLat, currentLon, SATELLITE_RADIUS_M, ZOOM_LEVEL, detectToken);
             gpsMatcherClient.StartMatching(new StartMatchingEvent
             {
                 ImagesCount = POINTS_COUNT,

Azaion.Common/Services/GpsMatcherClient.cs

@@ -23,13 +23,12 @@ public class StartMatchingEvent
     public int    ImagesCount        { get; set; }
     public double Latitude           { get; set; }
     public double Longitude          { get; set; }
-    public string ProcessingType     { get; set; } = "cuda";
     public int Altitude              { get; set; } = 400;
     public double CameraSensorWidth  { get; set; } = 23.5;
     public double CameraFocalLength  { get; set; } = 24;
 
     public override string ToString() =>
-        $"{RouteDir},{SatelliteImagesDir},{ImagesCount},{Latitude},{Longitude},{ProcessingType},{Altitude},{CameraSensorWidth},{CameraFocalLength}";
+        $"{RouteDir},{SatelliteImagesDir},{ImagesCount},{Latitude},{Longitude},{Altitude},{CameraSensorWidth},{CameraFocalLength}";
 }
 
 public class GpsMatcherClient : IGpsMatcherClient

Azaion.Common/Services/InferenceClient.cs

@@ -0,0 +1,150 @@
+using System.Diagnostics;
+using System.IO;
+using System.Text;
+using Azaion.Common.Database;
+using Azaion.Common.Extensions;
+using Azaion.CommonSecurity;
+using Azaion.CommonSecurity.DTO;
+using Azaion.CommonSecurity.DTO.Commands;
+using Azaion.CommonSecurity.Exceptions;
+using Azaion.CommonSecurity.Services;
+using MessagePack;
+using Microsoft.Extensions.Options;
+using NetMQ;
+using NetMQ.Sockets;
+
+namespace Azaion.Common.Services;
+
+public interface IInferenceClient : IDisposable
+{
+    event EventHandler<RemoteCommand> BytesReceived;
+    event EventHandler<RemoteCommand>? InferenceDataReceived;
+    event EventHandler<RemoteCommand>? AIAvailabilityReceived;
+    void Send(RemoteCommand create);
+    void Stop();
+}
+
+public class InferenceClient : IInferenceClient, IResourceLoader
+{
+    private CancellationTokenSource _waitFileCancelSource = new();
+
+    public event EventHandler<RemoteCommand>? BytesReceived;
+    public event EventHandler<RemoteCommand>? InferenceDataReceived;
+    public event EventHandler<RemoteCommand>? AIAvailabilityReceived;
+
+    private readonly DealerSocket _dealer = new();
+    private readonly NetMQPoller _poller = new();
+    private readonly Guid _clientId = Guid.NewGuid();
+    private readonly InferenceClientConfig _inferenceClientConfig;
+
+    public InferenceClient(IOptions<InferenceClientConfig> config, CancellationToken ct)
+    {
+        _inferenceClientConfig = config.Value;
+        Start(ct);
+    }
+
+    private void Start(CancellationToken ct = default)
+    {
+        try
+        {
+            using var process = new Process();
+            process.StartInfo = new ProcessStartInfo
+            {
+                FileName = SecurityConstants.EXTERNAL_INFERENCE_PATH,
+                Arguments = $"--port {_inferenceClientConfig.ZeroMqPort} --api {_inferenceClientConfig.ApiUrl}",
+                //RedirectStandardOutput = true,
+                //RedirectStandardError = true,
+                //CreateNoWindow = true
+            };
+
+            process.OutputDataReceived += (_, e) => { if (e.Data != null) Console.WriteLine(e.Data); };
+            process.ErrorDataReceived += (_, e) => { if (e.Data != null) Console.WriteLine(e.Data); };
+            //process.Start();
+        }
+        catch (Exception e)
+        {
+            Console.WriteLine(e);
+            //throw;
+        }
+
+        _dealer.Options.Identity = Encoding.UTF8.GetBytes(_clientId.ToString("N"));
+        _dealer.Connect($"tcp://{_inferenceClientConfig.ZeroMqHost}:{_inferenceClientConfig.ZeroMqPort}");
+
+        _dealer.ReceiveReady += (_, e) => ProcessClientCommand(e.Socket, ct);
+        _poller.Add(_dealer);
+        _ = Task.Run(() => _poller.RunAsync(), ct);
+    }
+
+    private void ProcessClientCommand(NetMQSocket socket, CancellationToken ct = default)
+    {
+        while (socket.TryReceiveFrameBytes(TimeSpan.Zero, out var bytes))
+        {
+            if (bytes?.Length == 0)
+                continue;
+
+            var remoteCommand = MessagePackSerializer.Deserialize<RemoteCommand>(bytes, cancellationToken: ct);
+            switch (remoteCommand.CommandType)
+            {
+                case CommandType.DataBytes:
+                    BytesReceived?.Invoke(this, remoteCommand);
+                    break;
+                case CommandType.InferenceData:
+                    InferenceDataReceived?.Invoke(this, remoteCommand);
+                    break;
+                case CommandType.AIAvailabilityResult:
+                    AIAvailabilityReceived?.Invoke(this, remoteCommand);
+                    break;
+            }
+
+        }
+    }
+
+    public void Stop()
+    {
+
+    }
+
+    public void Send(RemoteCommand command)
+    {
+        _dealer.SendFrame(MessagePackSerializer.Serialize(command));
+    }
+
+    public MemoryStream LoadFile(string fileName, string? folder = null, TimeSpan? timeout = null)
+    {
+        //TODO: Bad solution, look for better implementation
+        byte[] bytes = [];
+        Exception? exception = null;
+        _waitFileCancelSource = new CancellationTokenSource();
+        Send(RemoteCommand.Create(CommandType.Load, new LoadFileData(fileName, folder)));
+        BytesReceived += OnBytesReceived;
+
+        void OnBytesReceived(object? sender, RemoteCommand command)
+        {
+            if (command.Data is null)
+            {
+                exception = new BusinessException(command.Message ?? "File is empty");
+                _waitFileCancelSource.Cancel();
+            }
+
+            bytes = command.Data;
+            _waitFileCancelSource.Cancel();
+        }
+        _waitFileCancelSource.Token.WaitForCancel(timeout ?? TimeSpan.FromSeconds(15));
+        BytesReceived -= OnBytesReceived;
+        if (exception != null)
+            throw exception;
+        return new MemoryStream(bytes);
+    }
+
+    public void Dispose()
+    {
+        _waitFileCancelSource.Dispose();
+        _poller.Stop();
+        _poller.Dispose();
+
+        _dealer.SendFrame(MessagePackSerializer.Serialize(new RemoteCommand(CommandType.Exit)));
+        _dealer.Disconnect($"tcp://{_inferenceClientConfig.ZeroMqHost}:{_inferenceClientConfig.ZeroMqPort}");
+        _dealer.Close();
+        _dealer.Dispose();
+    }
+}

Azaion.Common/Services/InferenceService.cs

@@ -1,11 +1,11 @@
-using System.Text;
-using Azaion.Common.Database;
+using Azaion.Common.Database;
 using Azaion.Common.DTO.Config;
-using Azaion.CommonSecurity;
+using Azaion.Common.Events;
+using Azaion.Common.Extensions;
 using Azaion.CommonSecurity.DTO.Commands;
 using Azaion.CommonSecurity.Services;
+using MediatR;
 using MessagePack;
-using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
@@ -13,45 +13,74 @@ namespace Azaion.Common.Services;
 
 public interface IInferenceService
 {
-    Task RunInference(List<string> mediaPaths, Func<AnnotationImage, Task> processAnnotation, CancellationToken detectToken = default);
+    Task RunInference(List<string> mediaPaths, CancellationToken ct = default);
     void StopInference();
 }
 
-public class InferenceService(ILogger<InferenceService> logger, IInferenceClient client, IAzaionApi azaionApi, IOptions<AIRecognitionConfig> aiConfigOptions) : IInferenceService
+public class InferenceService : IInferenceService
 {
-    public async Task RunInference(List<string> mediaPaths, Func<AnnotationImage, Task> processAnnotation, CancellationToken detectToken = default)
+    private readonly IInferenceClient _client;
+    private readonly IAzaionApi _azaionApi;
+    private readonly IOptions<AIRecognitionConfig> _aiConfigOptions;
+    private readonly IAnnotationService _annotationService;
+    private readonly IMediator _mediator;
+    private CancellationTokenSource _inferenceCancelTokenSource = new();
+
+    public InferenceService(
+        ILogger<InferenceService> logger,
+        IInferenceClient client,
+        IAzaionApi azaionApi,
+        IOptions<AIRecognitionConfig> aiConfigOptions,
+        IAnnotationService annotationService,
+        IMediator mediator)
     {
-        client.Send(RemoteCommand.Create(CommandType.Login, azaionApi.Credentials));
-        var aiConfig = aiConfigOptions.Value;
+        _client = client;
+        _azaionApi = azaionApi;
+        _aiConfigOptions = aiConfigOptions;
+        _annotationService = annotationService;
+        _mediator = mediator;
 
-        aiConfig.Paths = mediaPaths;
-        client.Send(RemoteCommand.Create(CommandType.Inference, aiConfig));
-
-        while (!detectToken.IsCancellationRequested)
+        client.InferenceDataReceived += async (sender, command) =>
         {
             try
             {
-                var bytes = client.GetBytes(ct: detectToken);
-                if (bytes == null)
-                    throw new Exception("Can't get bytes from inference client");
-
-                if (bytes.Length == 4 && Encoding.UTF8.GetString(bytes) == "DONE")
+                if (command.Message == "DONE")
+                {
+                    _inferenceCancelTokenSource?.Cancel();
                     return;
+                }
 
-                var annotationImage = MessagePackSerializer.Deserialize<AnnotationImage>(bytes, cancellationToken: detectToken);
-
-                await processAnnotation(annotationImage);
+                var annImage = MessagePackSerializer.Deserialize<AnnotationImage>(command.Data);
+                await ProcessDetection(annImage);
             }
             catch (Exception e)
             {
                 logger.LogError(e, e.Message);
-                break;
             }
-        }
+        };
+    }
+
+    private async Task ProcessDetection(AnnotationImage annotationImage, CancellationToken ct = default)
+    {
+        var annotation = await _annotationService.SaveAnnotation(annotationImage, ct);
+        await _mediator.Publish(new AnnotationAddedEvent(annotation), ct);
+    }
+
+    public async Task RunInference(List<string> mediaPaths, CancellationToken ct = default)
+    {
+        _inferenceCancelTokenSource = new CancellationTokenSource();
+        _client.Send(RemoteCommand.Create(CommandType.Login, _azaionApi.Credentials));
+
+        var aiConfig = _aiConfigOptions.Value;
+        aiConfig.Paths = mediaPaths;
+        _client.Send(RemoteCommand.Create(CommandType.Inference, aiConfig));
+
+        using var combinedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(ct, _inferenceCancelTokenSource.Token);
+        await combinedTokenSource.Token.AsTask();
     }
 
     public void StopInference()
     {
-        client.Send(RemoteCommand.Create(CommandType.StopInference));
+        _client.Send(RemoteCommand.Create(CommandType.StopInference));
     }
 }