refactor external clients

put model batch size as parameter in config

Azaion.Common/Services/FailsafeProducer.cs

@@ -100,26 +100,40 @@ public class FailsafeAnnotationsProducer
                 .ToListAsync(token: cancellationToken);
 
             var messages = new List<AnnotationCreatedMessage>();
+            var badImages = new List<string>();
             foreach (var annotation in annotations)
             {
-                var image = await File.ReadAllBytesAsync(annotation.ImagePath, cancellationToken);
-                var annCreateMessage = new AnnotationCreatedMessage
+                try
                 {
-                    Name = annotation.Name,
-                    OriginalMediaName = annotation.OriginalMediaName,
-                    Time = annotation.Time,
-                    CreatedRole = annotation.CreatedRole,
-                    CreatedEmail = annotation.CreatedEmail,
-                    CreatedDate = annotation.CreatedDate,
-                    Status = annotation.AnnotationStatus,
+                    var image = await File.ReadAllBytesAsync(annotation.ImagePath, cancellationToken);
+                    var annCreateMessage = new AnnotationCreatedMessage
+                    {
+                        Name = annotation.Name,
+                        OriginalMediaName = annotation.OriginalMediaName,
+                        Time = annotation.Time,
+                        CreatedRole = annotation.CreatedRole,
+                        CreatedEmail = annotation.CreatedEmail,
+                        CreatedDate = annotation.CreatedDate,
+                        Status = annotation.AnnotationStatus,
 
-                    ImageExtension = annotation.ImageExtension,
-                    Image = image,
-                    Detections = JsonConvert.SerializeObject(annotation.Detections),
-                    Source = annotation.Source,
+                        ImageExtension = annotation.ImageExtension,
+                        Image = image,
+                        Detections = JsonConvert.SerializeObject(annotation.Detections),
+                        Source = annotation.Source,
+                    };
+                    messages.Add(annCreateMessage);
+                }
+                catch (Exception e)
+                {
+                    _logger.LogError(e, e.Message);
+                    badImages.Add(annotation.Name);
+                }
+            }
 
-                };
-                messages.Add(annCreateMessage);
+            if (badImages.Any())
+            {
+                await db.AnnotationsQueue.Where(x => badImages.Contains(x.Name)).DeleteAsync(token: cancellationToken);
+                _dbFactory.SaveToDisk();
             }
             return messages;
         });

Azaion.Common/Services/InferenceService.cs

@@ -2,12 +2,12 @@
 using Azaion.Common.Database;
 using Azaion.Common.DTO.Config;
 using Azaion.CommonSecurity;
-using Azaion.CommonSecurity.DTO;
 using Azaion.CommonSecurity.DTO.Commands;
+using Azaion.CommonSecurity.Services;
+using MessagePack;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using NetMQ;
-using NetMQ.Sockets;
 
 namespace Azaion.Common.Services;
 
@@ -16,30 +16,29 @@ public interface IInferenceService
     Task RunInference(List<string> mediaPaths, Func<AnnotationImage, Task> processAnnotation, CancellationToken ct = default);
 }
 
-public class PythonInferenceService(ILogger<PythonInferenceService> logger, IOptions<PythonConfig> pythonConfigOptions, IOptions<AIRecognitionConfig> aiConfigOptions) : IInferenceService
+public class InferenceService(ILogger<InferenceService> logger, [FromKeyedServices(SecurityConstants.EXTERNAL_INFERENCE_PATH)] IExternalClient externalClient, IOptions<AIRecognitionConfig> aiConfigOptions) : IInferenceService
 {
     public async Task RunInference(List<string> mediaPaths, Func<AnnotationImage, Task> processAnnotation, CancellationToken ct = default)
     {
-        var pythonConfig = pythonConfigOptions.Value;
         var aiConfig = aiConfigOptions.Value;
 
-        using var dealer = new DealerSocket();
-        var clientId = Guid.NewGuid();
-        dealer.Options.Identity = Encoding.UTF8.GetBytes(clientId.ToString("N"));
-        dealer.Connect($"tcp://{pythonConfig.ZeroMqHost}:{pythonConfig.ZeroMqPort}");
-
         aiConfig.Paths = mediaPaths;
-        dealer.SendFrame(RemoteCommand.Serialize(CommandType.Inference, aiConfig));
+        externalClient.Send(RemoteCommand.Create(CommandType.Inference, aiConfig));
 
         while (!ct.IsCancellationRequested)
         {
             try
             {
-                var annotationStream = dealer.Get<AnnotationImage>(bytes => bytes.Length == 4 && Encoding.UTF8.GetString(bytes) == "DONE", ct: ct);
-                if (annotationStream == null)
-                    break;
+                var bytes = externalClient.GetBytes(ct: ct);
+                if (bytes == null)
+                    throw new Exception("Can't get bytes from inference client");
 
-                await processAnnotation(annotationStream);
+                if (bytes.Length == 4 && Encoding.UTF8.GetString(bytes) == "DONE")
+                    return;
+
+                var annotationImage = MessagePackSerializer.Deserialize<AnnotationImage>(bytes, cancellationToken: ct);
+
+                await processAnnotation(annotationImage);
             }
             catch (Exception e)
             {