#nullable enable using System.Threading; using SharpDX; using T3.Core.Video; using T3.VideoServices; using Utilities = T3.Core.Utils.Utilities; namespace Lib.io.video { [Guid("D9A7233D-5D03-4268-A58B-465972852A5B")] internal sealed class VideoStreamInput : Instance, IStatusProvider { private readonly Lock _lockObject = new(); [Input(Guid = "2E26E552-68D7-4E2F-8208-831F2A75C96B")] public readonly InputSlot Connect = new(); [Input(Guid = "9A240243-71B5-4235-86A9-D5369A3311A9")] public readonly InputSlot Reconnect = new(); [Output(Guid = "2E7E2404-5881-4327-9653-CA9533B856A9", DirtyFlagTrigger = DirtyFlagTrigger.Animated)] public readonly Slot Texture = new(); [Output(Guid = "3F6A960C-906A-4073-A338-ABB785869062", DirtyFlagTrigger = DirtyFlagTrigger.Animated)] public readonly Slot Resolution = new(); [Output(Guid = "B0E4313B-A746-4444-934E-14285D42DADB", DirtyFlagTrigger = DirtyFlagTrigger.Animated)] public new readonly Slot Status = new(); [Input(Guid = "A8B0971B-94CB-4C3D-932D-337581B8D83A")] public readonly InputSlot Url = new("rtsp://your_stream_url"); private CancellationTokenSource? _cancellationTokenSource; private Thread? _captureThread; private Texture2D? _gpuTexture; private bool _lastConnectState; private volatile string _lastStatusMessage = "Not connected."; private string _lastUrl = string.Empty; // The most recent decoded frame as packed RGBA, shared between the capture thread and the render thread. private byte[]? _sharedRgba; private int _sharedWidth; private int _sharedHeight; private bool _hasFrame; public VideoStreamInput() { Texture.UpdateAction = Update; } public IStatusProvider.StatusLevel GetStatusLevel() { return _lastStatusMessage.StartsWith("Error") ? IStatusProvider.StatusLevel.Error : IStatusProvider.StatusLevel.Success; } public string GetStatusMessage() { return _lastStatusMessage; } private void SetStatus(string message) => _lastStatusMessage = message; private void Update(EvaluationContext context) { var url = Url.GetValue(context) ?? string.Empty; var shouldConnect = Connect.GetValue(context); var reconnect = Reconnect.GetValue(context); if (reconnect) Reconnect.SetTypedInputValue(false); var settingsChanged = shouldConnect != _lastConnectState || (shouldConnect && url != _lastUrl); if (shouldConnect) { if (IsCaptureThreadRunning() && (settingsChanged || reconnect)) StopCaptureThread(); if (!IsCaptureThreadRunning()) { _lastUrl = url; StartCaptureThread(url); } } else { StopCaptureThread(); } _lastConnectState = shouldConnect; lock (_lockObject) { if (_hasFrame && _sharedRgba != null) { UploadFrameToGpu(_sharedRgba, _sharedWidth, _sharedHeight); Texture.Value = _gpuTexture; Resolution.Value = new Int2(_sharedWidth, _sharedHeight); } else { Texture.Value = null; } } // Update the status as a side effect. This does NOT trigger a new update. Status.Value = _lastStatusMessage; } private bool IsCaptureThreadRunning() => _captureThread is { IsAlive: true }; private void StartCaptureThread(string url) { if (IsCaptureThreadRunning() || string.IsNullOrWhiteSpace(url)) return; _cancellationTokenSource = new CancellationTokenSource(); var token = _cancellationTokenSource.Token; _captureThread = new Thread(() => CaptureLoop(url, token)) { IsBackground = true, Name = "VideoStream Capture Thread" }; _captureThread.Start(); } private void StopCaptureThread() { if (!IsCaptureThreadRunning()) return; _cancellationTokenSource?.Cancel(); _captureThread?.Join(TimeSpan.FromSeconds(3)); _captureThread = null; _cancellationTokenSource?.Dispose(); _cancellationTokenSource = null; SetStatus("Disconnected"); } private void CaptureLoop(string url, CancellationToken token) { VideoDecoderSession? session = null; SoftwareFrameConverter? converter = null; try { SetStatus($"Connecting to {url}..."); // A live stream is consumed sequentially through this op's own SoftwareFrameConverter, so it needs // CPU-side frames — FastSeeking decodes in software, which provides exactly that. session = VideoDecoderSession.TryOpen(url, VideoPlaybackOptimization.FastSeeking, out var error, BuildDemuxerOptions(url)); if (token.IsCancellationRequested) return; if (session == null) { SetStatus($"Error: Failed to open stream '{url}'. {error}"); return; } converter = new SoftwareFrameConverter(session.IsHdr); SetStatus("Streaming"); while (!token.IsCancellationRequested) { if (!session.TryReadNextFrame(out _)) { SetStatus("Warning: Stream ended or interrupted."); break; } var rgba = converter.Convert(session.CurrentFrame); var byteCount = rgba.Width * rgba.Height * converter.BytesPerPixel; lock (_lockObject) { if (_sharedRgba == null || _sharedRgba.Length < byteCount) _sharedRgba = new byte[byteCount]; rgba.FillImageBuffer(_sharedRgba, 1); _sharedWidth = rgba.Width; _sharedHeight = rgba.Height; _hasFrame = true; } } } catch (Exception e) { if (!token.IsCancellationRequested) { Log.Error($"Video stream thread failed: {e.Message}", this); SetStatus($"Error: {e.Message}"); } } finally { converter?.Dispose(); session?.Dispose(); lock (_lockObject) { _hasFrame = false; } } } // TCP is more reliable than the default UDP, and a socket timeout keeps a dead URL from hanging the open. private static IReadOnlyDictionary? BuildDemuxerOptions(string url) { if (!url.StartsWith("rtsp", StringComparison.OrdinalIgnoreCase)) return null; return new Dictionary { ["rtsp_transport"] = "tcp", ["timeout"] = "5000000", // microseconds }; } private unsafe void UploadFrameToGpu(byte[] buffer, int width, int height) { if (width <= 0 || height <= 0) return; if (_gpuTexture == null || _gpuTexture.Description.Width != width || _gpuTexture.Description.Height != height) { Utilities.Dispose(ref _gpuTexture); _gpuTexture = Texture2D.CreateTexture2D(new Texture2DDescription { Width = width, Height = height, MipLevels = 1, ArraySize = 1, Format = Format.R8G8B8A8_UNorm, SampleDescription = new SampleDescription(1, 0), Usage = ResourceUsage.Default, BindFlags = BindFlags.ShaderResource, CpuAccessFlags = CpuAccessFlags.None, OptionFlags = ResourceOptionFlags.None }); } fixed (byte* pixels = buffer) { var dataBox = new DataBox((IntPtr)pixels, width * 4, 0); ResourceManager.Device.ImmediateContext.UpdateSubresource(dataBox, _gpuTexture, 0); } } protected override void Dispose(bool isDisposing) { if (IsDisposed) return; if (!isDisposing) return; StopCaptureThread(); Utilities.Dispose(ref _gpuTexture); lock (_lockObject) { _sharedRgba = null; _hasFrame = false; } } } }