unity cogstream and cpop clients

Author: Matthias Huerbe

Runtime.meta

@@ -0,0 +1,13 @@
+{
+    "name": "CognitiveXR",
+    "references": [],
+    "includePlatforms": [],
+    "excludePlatforms": [],
+    "allowUnsafeCode": false,
+    "overrideReferences": false,
+    "precompiledReferences": [],
+    "autoReferenced": true,
+    "defineConstraints": [],
+    "versionDefines": [],
+    "noEngineReferences": false
+}

Runtime/CognitiveXR.asmdef

@@ -0,0 +1,31 @@
+using System.Collections.Generic;
+
+namespace CognitiveXR.Cpop
+{
+    [System.Serializable]
+    public struct CpopData
+    {
+        public float Timestamp;
+        public string Type;
+        public Coordinates Position;
+        public List<Coordinates> Shape;
+
+        public override string ToString()
+        {
+            return $"{Timestamp}: {Type}/{Position}";
+        }
+    }
+
+    [System.Serializable]
+    public struct Coordinates
+    {
+        public float X;
+        public float Y;
+        public float Z;
+        
+        public override string ToString()
+        {
+            return $"({X},{Y},{Z})";
+        }
+    }
+}

Runtime/CognitiveXR.asmdef.meta

@@ -0,0 +1,77 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using MQTTnet;
+using MQTTnet.Client;
+using MQTTnet.Client.Options;
+using UnityEngine;
+
+namespace CognitiveXR.Cpop
+{
+    public class CpopServerOptions
+    {
+        public string Server { get; set; }
+        public int? Port { get; set; }
+    }
+
+    public class CpopSubscriber
+    {
+        public ConcurrentQueue<CpopData> Queue { get; }
+        private CpopServerOptions _options;
+        private CancellationTokenSource _cancellationTokenSource;
+        private IMqttClient _client;
+
+        public CpopSubscriber(ConcurrentQueue<CpopData> queue, CpopServerOptions options)
+        {
+            Queue = queue;
+            _options = options;
+            _cancellationTokenSource = new CancellationTokenSource();
+            var factory = new MqttFactory();
+            _client = factory.CreateMqttClient();
+        }
+
+        public CpopSubscriber(CpopServerOptions options) : this(new ConcurrentQueue<CpopData>(), options)
+        {
+        }
+
+        public CpopSubscriber() : this(new CpopServerOptions {Server = "localhost"})
+        {
+        }
+
+        public void Unsubscribe()
+        {
+            _cancellationTokenSource.Cancel();
+        }
+
+        public async void Subscribe()
+        {
+            var options = new MqttClientOptionsBuilder()
+                .WithClientId("CS-Client")
+                .WithTcpServer(_options.Server, _options.Port)
+                .WithCleanSession()
+                .Build();
+            _client.UseApplicationMessageReceivedHandler(DefaultCpopMessageHandlerJson);
+            await _client.ConnectAsync(options, _cancellationTokenSource.Token);
+            await _client.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("cpop").Build());
+        }
+
+        protected void DefaultCpopMessageHandlerJson(MqttApplicationMessageReceivedEventArgs e)
+        {
+            try
+            {
+                var payload = e.ApplicationMessage.Payload;
+             
+                String jsonText = System.Text.Encoding.UTF8.GetString(payload);
+             
+                var cpopData = JsonUtility.FromJson<CpopData>(jsonText);
+
+                Queue.Enqueue(cpopData);
+                
+            }
+            catch (Exception exp)
+            {
+                Debug.LogError(exp);
+            }
+        }
+    }
+}

Runtime/Cpop.meta

@@ -0,0 +1,21 @@
+
+using System.Collections.Generic;
+
+namespace CognitiveXR.CogStream
+{
+    public class DebugReceiveChannel : ResultReceiveChannel
+    {
+        protected override List<EngineResult> ParseResultPacket(ResultPacket resultPacket)
+        {
+            EngineResult engineResult = new EngineResult()
+            {
+                frameId = resultPacket.frameId,
+                seconds = resultPacket.seconds,
+                nanoseconds = resultPacket.nanoseconds,
+                result = System.Text.Encoding.UTF8.GetString(resultPacket.data),
+            };
+
+            return new List<EngineResult>{ engineResult };
+        }
+    }
+}

Runtime/Cpop/CpopData.cs

@@ -0,0 +1,113 @@
+using System;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+
+namespace CognitiveXR.CogStream
+{
+    public class EngineClient
+    {
+        private readonly StreamSpec streamSpec;
+        private NetworkStream networkStream;
+        private NetworkStreamResultPacketScanner packetScanner;
+        private readonly IFrameSendChannel sendChannel;
+        private readonly ResultReceiveChannel resultReceiveChannel;
+        private readonly uint streamId;
+
+        private TcpClient client;
+
+        public EngineClient(StreamSpec streamSpec, IFrameSendChannel sendChannel,
+            ResultReceiveChannel resultReceiveChannel, uint streamId = 0)
+        {
+            this.streamSpec = streamSpec;
+            this.sendChannel = sendChannel;
+            this.resultReceiveChannel = resultReceiveChannel;
+            this.streamId = streamId;
+            this.sendChannel.SetStreamId(this.streamId);
+        }
+
+        public async void Open()
+        {
+            // TODO: implement exception handling
+            string[] parts = streamSpec.engineAddress.Split(':');
+
+            IPAddress ipAddress = IPAddress.Parse(parts[0]);
+            IPEndPoint endPoint = new IPEndPoint(ipAddress, int.Parse(parts[1]));
+
+            try
+            {
+                client = new TcpClient();
+                {
+                    await client.ConnectAsync(endPoint.Address, endPoint.Port);
+
+                    string json = streamSpecToJson(streamSpec);
+
+                    byte[] jsonAsBytes = Encoding.UTF8.GetBytes(json);
+                    byte[] lengthInByte = BitConverter.GetBytes(jsonAsBytes.Length);
+
+                    byte[] data = new byte[4 + jsonAsBytes.Length];
+                    Buffer.BlockCopy(lengthInByte, 0, data, 0, lengthInByte.Length);
+                    Buffer.BlockCopy(jsonAsBytes, 0, data, 4, jsonAsBytes.Length);
+
+                    networkStream = client.GetStream();
+                    await networkStream.WriteAsync(data, 0, data.Length);
+                    await networkStream.FlushAsync();
+
+                    NetworkStreamFrameWriter frameWriter = new NetworkStreamFrameWriter(networkStream);
+                    sendChannel.SetWriter(frameWriter);
+                    packetScanner = new NetworkStreamResultPacketScanner(networkStream);
+                    resultReceiveChannel.SetResultPacketScanner(packetScanner);
+                }
+            }
+            catch (Exception e)
+            {
+                Console.WriteLine(e);
+                throw;
+            }
+
+        }
+
+        public bool isConnected()
+        {
+            return (client != null) && (client.Connected);
+        }
+
+        private string streamSpecToJson(StreamSpec streamSpec)
+        {
+            string attributes = streamSpec.attributes.Aggregate(
+                "\"attributes\": {",
+                (current, streamSpecAttribute)
+                    => current + $"\"{streamSpecAttribute.Key}\": [\"{streamSpecAttribute.Value[0]}\"],");
+
+            attributes = attributes.TrimEnd(',');
+
+            attributes += "}";
+
+            return $"{{" +
+                   $"\"engineAddress\": \"{streamSpec.engineAddress}\"" +
+                   ", " + attributes +
+                   $"}}";
+        }
+
+        /// <summary>
+        /// Returns the send channel of this engine
+        /// </summary>
+        /// <typeparam name="T"></typeparam>
+        /// <returns></returns>
+        public T GetSendChannel<T>() where T : IFrameSendChannel
+        {
+            return (T) sendChannel;
+        }
+
+        /// <summary>
+        /// Returns the receive channel of this engine
+        /// </summary>
+        /// <typeparam name="T">where T is of type ResultReceiveChannel</typeparam>
+        /// <returns></returns>
+        public T GetReceiveChannel<T>() where T : ResultReceiveChannel
+        {
+            return (T) resultReceiveChannel;
+        }
+    }
+}

Runtime/Cpop/CpopData.cs.meta

@@ -0,0 +1,11 @@
+
+namespace CognitiveXR.CogStream
+{
+    public class EngineResult
+    {
+        public uint frameId;
+        public uint seconds;
+        public uint nanoseconds;
+        public string result;
+    }
+}

Runtime/Cpop/CpopSubscriber.cs

@@ -0,0 +1,13 @@
+using System;
+
+namespace CognitiveXR.CogStream
+{
+    public struct Frame
+    {
+        public uint frameId;
+        public DateTime timestamp;
+        public int width;
+        public int height;
+        public byte[] rawFrame;
+    }
+}