Add NamedPipeCapture

Baballonia.sln

@@ -47,6 +47,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Baballonia.CaptureBin.IO",
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Baballonia.LibV4L2Capture", "src\Baballonia.LibV4L2Capture\Baballonia.LibV4L2Capture.csproj", "{02E9F6A2-A443-491D-93FF-6F002F3C494F}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Baballonia.NamedPipeCapture", "src\Baballonia.NamedPipeCapture\Baballonia.NamedPipeCapture.csproj", "{917BDD05-B5F1-4F32-993B-30E2BEF08AC6}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -126,6 +128,10 @@ Global
 		{02E9F6A2-A443-491D-93FF-6F002F3C494F}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{02E9F6A2-A443-491D-93FF-6F002F3C494F}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{02E9F6A2-A443-491D-93FF-6F002F3C494F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{917BDD05-B5F1-4F32-993B-30E2BEF08AC6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{917BDD05-B5F1-4F32-993B-30E2BEF08AC6}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{917BDD05-B5F1-4F32-993B-30E2BEF08AC6}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{917BDD05-B5F1-4F32-993B-30E2BEF08AC6}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -142,6 +148,7 @@ Global
 		{4B08BDE8-099A-405F-AB71-094E8FC24AB1} = {82C6A0B4-5014-4BE4-9F52-7DC1989134E3}
 		{8D298C1C-774E-4BF0-A107-D5D3985C7A9B} = {82C6A0B4-5014-4BE4-9F52-7DC1989134E3}
 		{02E9F6A2-A443-491D-93FF-6F002F3C494F} = {82C6A0B4-5014-4BE4-9F52-7DC1989134E3}
+		{917BDD05-B5F1-4F32-993B-30E2BEF08AC6} = {82C6A0B4-5014-4BE4-9F52-7DC1989134E3}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {8D6643ED-DDC1-4236-9471-024DCCFC81F6}

src/Baballonia.Desktop/Baballonia.Desktop.csproj

@@ -55,6 +55,7 @@
     </ItemGroup>
     <ItemGroup>
         <ProjectReference Include="..\Baballonia.CaptureBin.IO\Baballonia.CaptureBin.IO.csproj" />
+        <ProjectReference Include="..\Baballonia.NamedPipeCapture\Baballonia.NamedPipeCapture.csproj" />
         <ProjectReference Include="..\Baballonia.OpenCVCapture\Baballonia.OpenCVCapture.csproj" />
         <ProjectReference Include="..\Baballonia.SDK\Baballonia.SDK.csproj" />
         <ProjectReference Include="..\Baballonia.SerialCameraCapture\Baballonia.SerialCameraCapture.csproj" />
@@ -146,11 +147,13 @@
             <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.OpenCVCapture.dll')" Include="$(OutputPath)Baballonia.OpenCVCapture.dll" />
             <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.IPCameraCapture.dll')" Include="$(OutputPath)Baballonia.IPCameraCapture.dll" />
             <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.VFTCapture.dll')" Include="$(OutputPath)Baballonia.VFTCapture.dll" />
+            <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.NamedPipeCapture.dll')" Include="$(OutputPath)Baballonia.NamedPipeCapture.dll" />
             <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.LibV4L2Capture.pdb')" Include="$(OutputPath)Baballonia.LibV4L2Capture.pdb" />
             <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.SerialCameraCapture.pdb')" Include="$(OutputPath)Baballonia.SerialCameraCapture.pdb" />
             <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.OpenCVCapture.pdb')" Include="$(OutputPath)Baballonia.OpenCVCapture.pdb" />
             <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.IPCameraCapture.pdb')" Include="$(OutputPath)Baballonia.IPCameraCapture.pdb" />
             <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.VFTCapture.pdb')" Include="$(OutputPath)Baballonia.VFTCapture.pdb" />
+            <ModuleDlls Condition="Exists('$(OutputPath)Baballonia.NamedPipeCapture.pdb')" Include="$(OutputPath)Baballonia.NamedPipeCapture.pdb" />
         </ItemGroup>
         <Move SourceFiles="@(ModuleDlls)" DestinationFolder="$(OutputPath)Modules\" />
     </Target>
@@ -162,6 +165,7 @@
             <PublishModuleDlls Condition="Exists('$(OutputPath)\publish\Baballonia.IPCameraCapture.dll')" Include="$(OutputPath)\publish\Baballonia.IPCameraCapture.dll" />
             <PublishModuleDlls Condition="Exists('$(OutputPath)\publish\Baballonia.VFTCapture.dll')" Include="$(OutputPath)\publish\Baballonia.VFTCapture.dll" />
             <PublishModuleDlls Condition="Exists('$(OutputPath)\publish\Baballonia.LibV4L2Capture.dll')" Include="$(OutputPath)\publish\Baballonia.LibV4L2Capture.dll" />
+            <PublishModuleDlls Condition="Exists('$(OutputPath)\publish\Baballonia.NamedPipeCapture.dll')" Include="$(OutputPath)\publish\Baballonia.NamedPipeCapture.dll" />
         </ItemGroup>
         <Move SourceFiles="@(PublishModuleDlls)" DestinationFolder="$(OutputPath)\publish\Modules\" />
     </Target>

src/Baballonia.NamedPipeCapture/Baballonia.NamedPipeCapture.csproj

@@ -0,0 +1,13 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net10.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\Baballonia.SDK\Baballonia.SDK.csproj" />
+    </ItemGroup>
+
+</Project>

src/Baballonia.NamedPipeCapture/NamedPipeCapture.cs

@@ -0,0 +1,200 @@
+using System.IO.Pipes;
+using Baballonia.SDK;
+using Microsoft.Extensions.Logging;
+using OpenCvSharp;
+
+namespace Baballonia.NamedPipeCapture;
+
+/// <summary>
+/// Captures frames from a named pipe source.
+/// Pipe data is expected as a stream of raw BGR frames prefixed with an 8-byte header:
+///   [4 bytes int32 width][4 bytes int32 height] followed by width*height*3 bytes of BGR data.
+/// Cross-platform: works on Windows, Linux, and macOS via System.IO.Pipes.
+///
+/// Address format:  pipe://&lt;pipeName&gt;   (e.g. "pipe://my-camera-feed")
+/// On Windows this resolves to \\.\pipe\&lt;pipeName&gt;.
+/// On Linux/macOS the runtime creates a socket-backed pipe automatically.
+/// </summary>
+public sealed class NamedPipeCapture : Capture
+{
+    private readonly string _pipeName;
+    private CancellationTokenSource? _cts;
+    private Task? _readLoop;
+
+    /// <param name="pipeName">The bare pipe name without any path prefix.</param>
+    /// <param name="logger">Logger instance.</param>
+    public NamedPipeCapture(string pipeName, ILogger logger)
+        : base($"pipe://{pipeName}", logger)
+    {
+        if (string.IsNullOrWhiteSpace(pipeName))
+            throw new ArgumentException("Pipe name must not be empty.", nameof(pipeName));
+
+        _pipeName = pipeName;
+    }
+
+    /// <inheritdoc/>
+    public override async Task<bool> StartCapture()
+    {
+        if (IsReady)
+        {
+            Logger.LogWarning("NamedPipeCapture: already running on pipe '{PipeName}'.", _pipeName);
+            return false;
+        }
+
+        _cts = new CancellationTokenSource();
+        var token = _cts.Token;
+
+        // Perform the initial connection on the calling thread so the caller knows
+        // whether the pipe actually exists before we hand back control.
+        NamedPipeClientStream pipe;
+        try
+        {
+            pipe = new NamedPipeClientStream(
+                serverName: ".",
+                pipeName: _pipeName,
+                direction: PipeDirection.In,
+                options: PipeOptions.Asynchronous);
+
+            Logger.LogInformation("NamedPipeCapture: connecting to pipe '{PipeName}'…", _pipeName);
+            await pipe.ConnectAsync(5_000, cancellationToken: token);
+            Logger.LogInformation("NamedPipeCapture: connected to pipe '{PipeName}'.", _pipeName);
+        }
+        catch (Exception ex)
+        {
+            Logger.LogError(ex, "NamedPipeCapture: failed to connect to pipe '{PipeName}'.", _pipeName);
+            return false;
+        }
+
+        IsReady = true;
+        _readLoop = Task.Run(() => ReadLoopAsync(pipe, token), token);
+        return true;
+    }
+
+    /// <inheritdoc/>
+    public override async Task<bool> StopCapture()
+    {
+        if (!IsReady)
+            return false;
+
+        IsReady = false;
+        _cts?.Cancel();
+
+        if (_readLoop is not null)
+        {
+            try { await _readLoop; }
+            catch (OperationCanceledException) { /* expected */ }
+            catch (Exception ex)
+            {
+                Logger.LogWarning(ex, "NamedPipeCapture: exception during read-loop shutdown.");
+            }
+        }
+
+        _cts?.Dispose();
+        _cts = null;
+        _readLoop = null;
+
+        Logger.LogInformation("NamedPipeCapture: stopped on pipe '{PipeName}'.", _pipeName);
+        return true;
+    }
+
+    // -------------------------------------------------------------------------
+    // Private helpers
+    // -------------------------------------------------------------------------
+
+    private async Task ReadLoopAsync(NamedPipeClientStream pipe, CancellationToken ct)
+    {
+        await using (pipe)
+        {
+            Logger.LogDebug("NamedPipeCapture: read loop started.");
+
+            try
+            {
+                while (!ct.IsCancellationRequested && pipe.IsConnected)
+                {
+                    // --- Read 8-byte header (width + height) -----------------
+                    int width, height;
+                    try
+                    {
+                        width  = await ReadInt32Async(pipe, ct);
+                        height = await ReadInt32Async(pipe, ct);
+                    }
+                    catch (EndOfStreamException)
+                    {
+                        Logger.LogInformation("NamedPipeCapture: pipe closed by server (EOF in header).");
+                        break;
+                    }
+
+                    if (width <= 0 || height <= 0 || width > 16_384 || height > 16_384)
+                    {
+                        Logger.LogWarning(
+                            "NamedPipeCapture: invalid frame dimensions {W}x{H} — skipping.",
+                            width, height);
+                        continue;
+                    }
+
+                    // --- Read BGR payload ------------------------------------
+                    int frameBytes = width * height * 3;
+                    byte[] buffer = new byte[frameBytes];
+
+                    try
+                    {
+                        await ReadExactAsync(pipe, buffer, frameBytes, ct);
+                    }
+                    catch (EndOfStreamException)
+                    {
+                        Logger.LogInformation("NamedPipeCapture: pipe closed by server (EOF in payload).");
+                        break;
+                    }
+
+                    // --- Wrap in Mat and hand off ----------------------------
+                    // Mat.FromArray copies, so the byte[] can be reclaimed by GC normally.
+                    var mat = Mat.FromArray(buffer).Reshape(3, rows: height);
+                    SetRawMat(mat);
+
+                    Logger.LogTrace("NamedPipeCapture: frame {W}x{H} acquired.", width, height);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                Logger.LogDebug("NamedPipeCapture: read loop cancelled.");
+            }
+            catch (Exception ex)
+            {
+                Logger.LogError(ex, "NamedPipeCapture: unexpected error in read loop.");
+            }
+            finally
+            {
+                IsReady = false;
+                Logger.LogDebug("NamedPipeCapture: read loop exited.");
+            }
+        }
+    }
+
+    /// <summary>Reads exactly 4 bytes and returns them as a big-endian int32.</summary>
+    private static async Task<int> ReadInt32Async(Stream stream, CancellationToken ct)
+    {
+        byte[] buf = new byte[4];
+        await ReadExactAsync(stream, buf, 4, ct);
+        if (BitConverter.IsLittleEndian) Array.Reverse(buf);
+        return BitConverter.ToInt32(buf, 0);
+    }
+
+    /// <summary>Reads exactly <paramref name="count"/> bytes into <paramref name="buffer"/>.</summary>
+    private static async Task ReadExactAsync(Stream stream, byte[] buffer, int count, CancellationToken ct)
+    {
+        int offset = 0;
+        while (offset < count)
+        {
+            int read = await stream.ReadAsync(buffer, offset, count - offset, ct);
+            if (read == 0)
+                throw new EndOfStreamException("Pipe closed before all bytes were received.");
+            offset += read;
+        }
+    }
+
+    public override void Dispose()
+    {
+        StopCapture().GetAwaiter().GetResult();
+        base.Dispose();
+    }
+}

src/Baballonia.NamedPipeCapture/NamedPipeCaptureFactory.cs

@@ -0,0 +1,29 @@
+using Baballonia.SDK;
+using Microsoft.Extensions.Logging;
+
+namespace Baballonia.NamedPipeCapture;
+
+/// <summary>
+/// Factory that creates <see cref="NamedPipeCapture"/> instances.
+/// Handles addresses of the form <c>pipe://&lt;pipeName&gt;</c>.
+/// </summary>
+public sealed class NamedPipeCaptureFactory(ILogger<NamedPipeCapture> logger) : ICaptureFactory
+{
+    private const string Scheme = "pipe://";
+
+    public Capture Create(string address)
+    {
+        if (!CanConnect(address))
+            throw new ArgumentException($"Address '{address}' is not a valid pipe:// URI.", nameof(address));
+
+        string pipeName = address[Scheme.Length..];
+        return new NamedPipeCapture(pipeName, logger);
+    }
+
+    public bool CanConnect(string address) =>
+        !string.IsNullOrWhiteSpace(address) &&
+        address.StartsWith(Scheme, StringComparison.OrdinalIgnoreCase) &&
+        address.Length > Scheme.Length;
+
+    public string GetProviderName() => "NamedPipe";
+}