feat: add static API for python interop (#93)

kahojyun · Nov 15, 2023 · d195d49 · d195d49
1 parent 89dcf0a
commit d195d49
Show file tree

Hide file tree

Showing 4 changed files with 268 additions and 57 deletions.
diff --git a/src/Qynit.PulseGen.Server/Program.cs b/src/Qynit.PulseGen.Server/Program.cs
@@ -1,59 +1,4 @@
-using MessagePack;
-using MessagePack.Formatters;
-using MessagePack.Resolvers;
-
-using Microsoft.AspNetCore.StaticFiles;
-using Microsoft.Fast.Components.FluentUI;
-
 using Qynit.PulseGen.Server;
-using Qynit.PulseGen.Server.Models;
-using Qynit.PulseGen.Server.Services;
-
-var builder = WebApplication.CreateBuilder(args);
-builder.Services.AddRazorPages();
-builder.Services.AddServerSideBlazor();
-builder.Services.AddFluentUIComponents();
-builder.Services.AddSingleton<IPlotService, PlotService>();
-var app = builder.Build();
-
-var fileExtensionContentTypeProvider = new FileExtensionContentTypeProvider();
-fileExtensionContentTypeProvider.Mappings[".data"] = "application/octet-stream";
-
-app.UseStaticFiles();
-app.UseStaticFiles(new StaticFileOptions
-{
-    ContentTypeProvider = fileExtensionContentTypeProvider
-});
-app.UseRouting();
-
-app.MapBlazorHub();
-app.MapFallbackToPage("/_Host");
-
-var resolver = CompositeResolver.Create(
-    new IMessagePackFormatter[] { new ComplexArrayFormatter() },
-    new[] { StandardResolver.Instance });
-var options = MessagePackSerializerOptions.Standard.WithResolver(resolver);
-
-const string contentType = "application/msgpack";
-
-app.MapPost("/api/schedule", async (HttpRequest request, HttpResponse response, CancellationToken token, IPlotService plotService) =>
-{
-    if (request.ContentType != contentType)
-    {
-        return Results.BadRequest();
-    }
-
-    var pgRequest = await MessagePackSerializer.DeserializeAsync<ScheduleRequest>(request.Body, options, token);
-    var runner = new ScheduleRunner(pgRequest);
-    var waveforms = runner.Run();
-    var arcWaveforms = waveforms.Select(ArcUnsafe.Wrap).ToList();
-    plotService.UpdatePlots(pgRequest.ChannelTable!.Zip(arcWaveforms).ToDictionary(x => x.First.Name, x => new PlotData(x.First.Name, x.Second.Clone(), 1.0 / x.First.SampleRate)));
-    var pgResponse = new PulseGenResponse(arcWaveforms);
-    response.RegisterForDispose(pgResponse);
-    return Results.Extensions.MessagePack(pgResponse, options);
-})
-.WithName("Schedule")
-.Accepts<ScheduleRequest>(contentType)
-.Produces(StatusCodes.Status400BadRequest);
 
-app.Run();
+var server = Server.CreateApp(args, false);
+server.Run();
diff --git a/src/Qynit.PulseGen.Server/PythonApi.cs b/src/Qynit.PulseGen.Server/PythonApi.cs
@@ -0,0 +1,112 @@
+using MessagePack;
+
+using Python.Runtime;
+
+using Qynit.PulseGen.Server.Models;
+using Qynit.PulseGen.Server.Services;
+
+namespace Qynit.PulseGen.Server;
+
+public static class PythonApi
+{
+    internal static Server? ServerInstance;
+    public static void Run(PyObject inRequestMsg, PyObject outWaveforms)
+    {
+        using (Py.GIL())
+        using (inRequestMsg)
+        using (outWaveforms)
+        {
+            var request = DeserializeRequest(inRequestMsg);
+            var waveforms = GenerateWaveforms(request);
+            using var pyDict = new PyDict(outWaveforms);
+            CopyToPyWaveformDict(pyDict, request, waveforms);
+            TryUpdatePlots(request, waveforms);
+        }
+    }
+
+    public static void StartServer()
+    {
+        if (ServerInstance is null)
+        {
+            ServerInstance = Server.CreateApp(Array.Empty<string>(), true);
+            ServerInstance.Start();
+        }
+    }
+
+    public static void StopServer()
+    {
+        if (ServerInstance is not null)
+        {
+            ServerInstance.Stop();
+            ServerInstance.Dispose();
+            ServerInstance = null;
+        }
+    }
+
+    private static void TryUpdatePlots(ScheduleRequest request, List<PooledComplexArray<float>> waveforms)
+    {
+        if (ServerInstance?.GetPlotService() is IPlotService service)
+        {
+            var arcWaveforms = waveforms.Select(ArcUnsafe.Wrap).ToList();
+            service.UpdatePlots(request.ChannelTable!.Zip(arcWaveforms).ToDictionary(x => x.First.Name, x => new PlotData(x.First.Name, x.Second, 1.0 / x.First.SampleRate)));
+        }
+        else
+        {
+            foreach (var waveform in waveforms)
+            {
+                waveform.Dispose();
+            }
+        }
+    }
+
+    private static List<PooledComplexArray<float>> GenerateWaveforms(ScheduleRequest request)
+    {
+        var state = PythonEngine.BeginAllowThreads();
+        try
+        {
+            var runner = new ScheduleRunner(request);
+            return runner.Run();
+        }
+        finally
+        {
+            PythonEngine.EndAllowThreads(state);
+        }
+    }
+
+    private static void CopyToPyWaveformDict(PyDict outWaveforms, ScheduleRequest request, List<PooledComplexArray<float>> waveforms)
+    {
+        foreach (var (channel, waveform) in request.ChannelTable!.Zip(waveforms))
+        {
+            var chName = channel.Name;
+            using var arrayTuple = outWaveforms[chName];
+            using var iArrayObject = arrayTuple[0];
+            CopyToPyBuffer(waveform.DataI, chName, iArrayObject);
+            using var qArrayObject = arrayTuple[1];
+            CopyToPyBuffer(waveform.DataQ, chName, qArrayObject);
+        }
+    }
+
+    private static unsafe ScheduleRequest DeserializeRequest(PyObject inRequestMsg)
+    {
+        using var requestMsg = inRequestMsg.GetBuffer();
+        if (requestMsg.ItemSize != 1 || requestMsg.Dimensions != 1)
+        {
+            throw new ArgumentException("Message must be a byte array");
+        }
+        var msgLength = requestMsg.Length;
+        using var stream = new UnmanagedMemoryStream((byte*)requestMsg.Buffer, msgLength);
+        var options = Server.MessagePackSerializerOptions;
+        return MessagePackSerializer.Deserialize<ScheduleRequest>(stream, options);
+    }
+
+    private static unsafe void CopyToPyBuffer(ReadOnlySpan<float> waveform, string chName, PyObject arrayObject)
+    {
+        using var pyBuffer = arrayObject.GetBuffer();
+        if (pyBuffer.ItemSize != sizeof(float) || pyBuffer.Length != waveform.Length * sizeof(float))
+        {
+            throw new ArgumentException($"Waveform {chName} has wrong shape");
+        }
+        var span = new Span<float>((float*)pyBuffer.Buffer, waveform.Length);
+        waveform.CopyTo(span);
+    }
+}
diff --git a/src/Qynit.PulseGen.Server/Qynit.PulseGen.Server.csproj b/src/Qynit.PulseGen.Server/Qynit.PulseGen.Server.csproj
@@ -4,6 +4,7 @@
     <TargetFramework>net7.0</TargetFramework>
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
 
   <ItemGroup>
@@ -27,6 +28,7 @@
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
     <PackageReference Include="Microsoft.Fast.Components.FluentUI" Version="3.2.0" />
+    <PackageReference Include="pythonnet" Version="3.0.3" />
   </ItemGroup>
 
   <ItemGroup>

diff --git a/src/Qynit.PulseGen.Server/Server.cs b/src/Qynit.PulseGen.Server/Server.cs
@@ -0,0 +1,152 @@
+using System.Reflection;
+
+using MessagePack;
+using MessagePack.Formatters;
+using MessagePack.Resolvers;
+
+using Microsoft.AspNetCore.StaticFiles;
+using Microsoft.Fast.Components.FluentUI;
+
+using Qynit.PulseGen.Server.Models;
+using Qynit.PulseGen.Server.Services;
+
+namespace Qynit.PulseGen.Server;
+
+public sealed class Server : IDisposable
+{
+    internal static MessagePackSerializerOptions MessagePackSerializerOptions { get; } =
+        MessagePackSerializerOptions.Standard.WithResolver(
+            CompositeResolver.Create(
+                new IMessagePackFormatter[] { new ComplexArrayFormatter() },
+                new[] { StandardResolver.Instance }));
+
+    private readonly WebApplication _app;
+    private Server(WebApplication app)
+    {
+        _app = app;
+    }
+
+    public static Server CreateApp(string[] args, bool embedded)
+    {
+        var builder = embedded ? CreateBuilderForEmbedded(args) : WebApplication.CreateBuilder(args);
+        builder.Services.AddRazorPages();
+        builder.Services.AddServerSideBlazor();
+        builder.Services.AddFluentUIComponents();
+        builder.Services.AddSingleton<IPlotService, PlotService>();
+        if (embedded)
+        {
+            builder.Services.AddSingleton<IHostLifetime, NopLifeTime>();
+        }
+        var app = builder.Build();
+
+        app.UseStaticFiles();
+        app.ServeSciChartWasm();
+
+        app.UseRouting();
+
+        app.MapBlazorHub();
+        app.MapFallbackToPage("/_Host");
+
+        app.AddScheduleApi();
+
+        return new Server(app);
+    }
+
+
+
+    private static WebApplicationBuilder CreateBuilderForEmbedded(string[] args)
+    {
+        var assemblyPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location)!;
+        var webRootPath = Path.Combine(assemblyPath, "wwwroot");
+        var env = Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT") ?? Environments.Production;
+
+        var webApplicationOptions = new WebApplicationOptions
+        {
+            EnvironmentName = env,
+            ApplicationName = "Qynit.PulseGen.Server",
+            ContentRootPath = assemblyPath,
+            WebRootPath = webRootPath,
+            Args = args,
+        };
+
+        var builder = WebApplication.CreateBuilder(webApplicationOptions);
+        return builder;
+    }
+
+    public void Run()
+    {
+        _app.Run();
+    }
+
+    public void Start()
+    {
+        _app.Start();
+    }
+
+    public void Stop()
+    {
+        _app.StopAsync().GetAwaiter().GetResult();
+    }
+
+    public void Dispose()
+    {
+        ((IDisposable)_app).Dispose();
+    }
+
+    public IPlotService? GetPlotService()
+    {
+        return _app.Services.GetService<IPlotService>();
+    }
+}
+
+
+internal class NopLifeTime : IHostLifetime
+{
+    public Task StopAsync(CancellationToken cancellationToken)
+    {
+        return Task.CompletedTask;
+    }
+
+    public Task WaitForStartAsync(CancellationToken cancellationToken)
+    {
+        return Task.CompletedTask;
+    }
+}
+
+internal static class BuilderExtensions
+{
+    internal static void AddScheduleApi(this WebApplication app)
+    {
+        const string contentType = "application/msgpack";
+
+        app.MapPost("/api/schedule", async (HttpRequest request, HttpResponse response, CancellationToken token, IPlotService plotService) =>
+        {
+            if (request.ContentType != contentType)
+            {
+                return Results.BadRequest();
+            }
+
+            var pgRequest = await MessagePackSerializer.DeserializeAsync<ScheduleRequest>(request.Body, Server.MessagePackSerializerOptions, token);
+            var runner = new ScheduleRunner(pgRequest);
+            var waveforms = runner.Run();
+            var arcWaveforms = waveforms.Select(ArcUnsafe.Wrap).ToList();
+            plotService.UpdatePlots(pgRequest.ChannelTable!.Zip(arcWaveforms).ToDictionary(x => x.First.Name, x => new PlotData(x.First.Name, x.Second.Clone(), 1.0 / x.First.SampleRate)));
+            var pgResponse = new PulseGenResponse(arcWaveforms);
+            response.RegisterForDispose(pgResponse);
+            return Results.Extensions.MessagePack(pgResponse, Server.MessagePackSerializerOptions);
+        })
+        .WithName("Schedule")
+        .Accepts<ScheduleRequest>(contentType)
+        .Produces(StatusCodes.Status400BadRequest);
+    }
+
+    internal static void ServeSciChartWasm(this WebApplication app)
+    {
+        var fileExtensionContentTypeProvider = new FileExtensionContentTypeProvider();
+        fileExtensionContentTypeProvider.Mappings[".data"] = "application/octet-stream";
+        app.UseStaticFiles(new StaticFileOptions
+        {
+            ContentTypeProvider = fileExtensionContentTypeProvider
+        });
+    }
+}