annotations/Azaion.Common/Services/InferenceService.cs

using System.Text;
using Azaion.Common.Database;
using Azaion.Common.DTO.Config;
using Azaion.CommonSecurity;
using Azaion.CommonSecurity.DTO;
using Azaion.CommonSecurity.DTO.Commands;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using NetMQ;
using NetMQ.Sockets;

namespace Azaion.Common.Services;

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 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));

        while (!ct.IsCancellationRequested)
        {
            try
            {
                var annotationStream = dealer.Get<AnnotationImage>(bytes => bytes.Length == 4 && Encoding.UTF8.GetString(bytes) == "DONE", ct: ct);
                if (annotationStream == null)
                    break;

                await processAnnotation(annotationStream);
            }
            catch (Exception e)
            {
                logger.LogError(e, e.Message);
                break;
            }
        }
    }
}