Created Data Transfer Subsystem plugin

- Websocket
- Json conversion both ways
    - Unreal Data Ustructs setup for Evan
- Created data templates with ethan and anuj for the python server
- Created blueprint implementation and example in Test Level

Plugins/LuckyDataTransfer/Source/LuckyDataTransfer/LuckyDataTransfer.Build.cs

@@ -0,0 +1,57 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+using UnrealBuildTool;
+
+public class LuckyDataTransfer : ModuleRules
+{
+	public LuckyDataTransfer(ReadOnlyTargetRules Target) : base(Target)
+	{
+		PCHUsage = ModuleRules.PCHUsageMode.UseExplicitOrSharedPCHs;
+		
+		PublicIncludePaths.AddRange(
+			new string[] {
+				// ... add public include paths required here ...
+			}
+			);
+				
+		
+		PrivateIncludePaths.AddRange(
+			new string[] {
+				// ... add other private include paths required here ...
+			}
+			);
+			
+		
+		PublicDependencyModuleNames.AddRange(
+			new string[]
+			{
+				"Core",
+				"WebSockets",
+				"Json",
+				"JsonUtilities",
+				"ImageWriteQueue"
+				// ... add other public dependencies that you statically link with here ...
+			}
+			);
+			
+		
+		PrivateDependencyModuleNames.AddRange(
+			new string[]
+			{
+				"CoreUObject",
+				"Engine",
+				"Slate",
+				"SlateCore", "ImageWriteQueue",
+				// ... add private dependencies that you statically link with here ...	
+			}
+			);
+		
+		
+		DynamicallyLoadedModuleNames.AddRange(
+			new string[]
+			{
+				// ... add any modules that your module loads dynamically here ...
+			}
+			);
+	}
+}

Plugins/LuckyDataTransfer/Source/LuckyDataTransfer/Private/LuckyDataTransfer.cpp

@@ -0,0 +1,20 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include "LuckyDataTransfer.h"
+
+#define LOCTEXT_NAMESPACE "FLuckyDataTransferModule"
+
+void FLuckyDataTransferModule::StartupModule()
+{
+	// This code will execute after your module is loaded into memory; the exact timing is specified in the .uplugin file per-module
+}
+
+void FLuckyDataTransferModule::ShutdownModule()
+{
+	// This function may be called during shutdown to clean up your module.  For modules that support dynamic reloading,
+	// we call this function before unloading the module.
+}
+
+#undef LOCTEXT_NAMESPACE
+	
+IMPLEMENT_MODULE(FLuckyDataTransferModule, LuckyDataTransfer)

Plugins/LuckyDataTransfer/Source/LuckyDataTransfer/Private/LuckyDataTransferSubsystem.cpp

@@ -0,0 +1,154 @@
+// Fill out your copyright notice in the Description page of Project Settings.
+
+
+#include "LuckyDataTransferSubsystem.h"
+#include "JsonUtilities.h"
+#include "JsonObjectConverter.h"
+#include "WebSocketsModule.h"
+#include "Slate/SceneViewport.h"
+
+ULuckyDataTransferSubsystem::ULuckyDataTransferSubsystem()
+{
+}
+
+void ULuckyDataTransferSubsystem::Initialize(FSubsystemCollectionBase& Collection)
+{
+	Super::Initialize(Collection);
+}
+
+void ULuckyDataTransferSubsystem::Deinitialize()
+{
+	Super::Deinitialize();
+	
+	if (Socket.IsValid() && Socket->IsConnected())
+	{
+		Socket->Close();
+	}
+}
+
+void ULuckyDataTransferSubsystem::Internal_OpenWebsocket(const FString& URL, const FString& Protocol)
+{
+	const FString NewUrl = URL.IsEmpty() ? TEXT("ws://127.0.0.1:3000/ws") : URL;
+	const FString NewProtocol = Protocol.IsEmpty() ? TEXT("ws") : Protocol;
+
+	UE_LOG(LogTemp, Warning, TEXT("Opening WebSocket URL: %s"), *NewUrl);
+	
+	if (!FModuleManager::Get().IsModuleLoaded("WebSockets"))
+	{
+		FModuleManager::Get().LoadModule("WebSockets");
+	}
+	
+	Socket = FWebSocketsModule::Get().CreateWebSocket(NewUrl);
+	Socket->Connect();
+
+	Socket->OnConnected().AddLambda([]()
+	{
+		UE_LOG(LogTemp, Warning, TEXT("WebSocket connected"));
+	});
+
+	Socket->OnConnectionError().AddLambda([](const FString& Message)
+	{
+		UE_LOG(LogTemp, Warning, TEXT("WebSocket connection error"));
+	});
+
+	Socket->OnMessage().AddUObject(this, &ULuckyDataTransferSubsystem::Callback_OnMessage);
+}
+
+//Callbacks / Lambas
+void ULuckyDataTransferSubsystem::Callback_OnConnected()
+{
+}
+
+void ULuckyDataTransferSubsystem::Callback_OnConnectionError(const FString& Error)
+{
+}
+
+void ULuckyDataTransferSubsystem::Callback_OnMessage(const FString& Message)
+{
+	if (!Message.IsEmpty())
+	{
+		CommandReady(InterpretData(Message));
+		return;
+	}
+
+	UE_LOG(LogTemp, Warning, TEXT("The message received from the websocket is invalid"));
+}
+
+void ULuckyDataTransferSubsystem::Internal_OnMessageSent(const FString& Message)
+{
+}
+
+void ULuckyDataTransferSubsystem::Callback_OnConnectionClosed()
+{
+}
+
+//Blueprint Exposed Implementation
+void ULuckyDataTransferSubsystem::ConnectToWebsocket(const FString& URL, const FString& Protocol)
+{
+	Internal_OpenWebsocket(URL, Protocol);
+}
+
+void ULuckyDataTransferSubsystem::SendMessage(const FString& Message)
+{
+	if (Socket.IsValid() && Socket->IsConnected())
+	{
+		Socket->Send(Message);
+		return;
+	}
+
+	UE_LOG(LogTemp, Warning, TEXT("WebSocket message failed"));
+}
+
+FPayload ULuckyDataTransferSubsystem::InterpretData(const FString& Message)
+{
+	FPayload Payload = FPayload();
+	
+	if (!Message.IsEmpty())
+	{
+		TSharedPtr<FJsonObject> JsonObj;
+		TSharedRef<TJsonReader<>> Reader = TJsonReaderFactory<>::Create(Message);
+
+		if (FJsonSerializer::Deserialize(Reader, JsonObj) && JsonObj.IsValid())
+		{
+			for (auto& Elem : JsonObj->Values)
+			{
+				FCommand Command = FCommand();
+				Command.Key = FString(Elem.Key);
+				Command.Value = Elem.Value->AsNumber();
+
+				Payload.Commands.Add(Command);
+			}
+		}
+	}
+
+	return Payload;
+}
+
+void ULuckyDataTransferSubsystem::CommandReady(const FPayload& Payload)
+{
+	if (OnCommandReady.IsBound())
+	{
+		OnCommandReady.Broadcast(Payload);
+	}
+}
+
+//Blueprint Callable function - Use CreateJsonPayload_Observation in C++
+bool ULuckyDataTransferSubsystem::MakeObservationPayload(const FObservationPayload& Data)
+{
+	return CreateJsonPayload_Observation(Data);
+}
+
+bool ULuckyDataTransferSubsystem::CreateJsonPayload_Observation(const FObservationPayload& Data)
+{
+	bool bSuccess = false;
+	ObservationPayloadString = FString();
+	
+	if (!Data.ObservationState.IsEmpty())
+	{
+		FJsonObjectConverter::UStructToJsonObjectString(Data, ObservationPayloadString);
+		UE_LOG(LogTemp, Warning, TEXT("Payload observation: %s"), *ObservationPayloadString);
+		bSuccess = true;
+	}
+
+	return bSuccess;
+}

Plugins/LuckyDataTransfer/Source/LuckyDataTransfer/Public/LuckyDataTransfer.h

@@ -0,0 +1,14 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include "Modules/ModuleManager.h"
+
+class FLuckyDataTransferModule : public IModuleInterface
+{
+public:
+
+	/** IModuleInterface implementation */
+	virtual void StartupModule() override;
+	virtual void ShutdownModule() override;
+};

Plugins/LuckyDataTransfer/Source/LuckyDataTransfer/Public/LuckyDataTransferSubsystem.h

@@ -0,0 +1,118 @@
+// Fill out your copyright notice in the Description page of Project Settings.
+
+#pragma once
+
+#include "CoreMinimal.h"
+#include "IWebSocket.h"
+//#include "LuckyWriteThread.h"
+#include "ObservationData.h"
+#include "Subsystems/WorldSubsystem.h"
+#include "LuckyDataTransferSubsystem.generated.h"
+
+/**
+ * 
+ */
+USTRUCT(BlueprintType)
+struct FCommand
+{
+	GENERATED_BODY()
+
+public:
+	UPROPERTY(BlueprintReadOnly, Category = "Command")
+	FString Key = FString();
+
+	UPROPERTY(BlueprintReadOnly, Category = "Command")
+	float Value = 0.f;
+};
+
+USTRUCT(BlueprintType)
+struct FPayload
+{
+	GENERATED_BODY()
+
+public:
+	UPROPERTY(BlueprintReadOnly, Category = "Command")
+	TArray<FCommand> Commands;
+
+	UPROPERTY(BlueprintReadOnly, Category = "Command")
+	int32 Index = 0;
+};
+
+DECLARE_DYNAMIC_MULTICAST_DELEGATE_OneParam(FCommandReady, const FPayload&, Payload);
+
+UCLASS()
+class LUCKYDATATRANSFER_API ULuckyDataTransferSubsystem : public UWorldSubsystem
+{
+	GENERATED_BODY()
+
+	
+public:
+
+	ULuckyDataTransferSubsystem();
+
+	virtual void Initialize(FSubsystemCollectionBase& Collection);
+	
+	virtual void Deinitialize();
+
+	TSharedPtr<IWebSocket> Socket;
+
+	//internal references
+	
+	void Internal_OpenWebsocket(const FString& URL, const FString& Protocol);
+
+	FPayload InterpretData(const FString& Message);
+
+	UPROPERTY(BlueprintAssignable)
+	FCommandReady OnCommandReady;
+	
+
+	//Callbacks
+	UFUNCTION()
+	void Callback_OnConnected();
+	
+	UFUNCTION()
+	void Callback_OnConnectionError(const FString& Error);
+	
+	UFUNCTION()
+	void Callback_OnMessage(const FString& Message);
+	
+	UFUNCTION()
+	void Internal_OnMessageSent(const FString& Message);
+	
+	UFUNCTION()
+	void Callback_OnConnectionClosed();
+
+	//Exposed Blueprint Functions
+	UFUNCTION(BlueprintCallable, Category = "Websocket")
+	void ConnectToWebsocket(const FString& URL, const FString& Protocol);
+
+	UFUNCTION(BlueprintCallable, Category = "Websocket")
+	void SendMessage(const FString& Message);
+
+	UFUNCTION()
+	void CommandReady(const FPayload& Payload);
+
+	//---Observations (Sent to server from Unreal)--------------//
+	//Image Data declarations 
+	TArray<FString> FeatureNames = {};
+	FString CameraName = TEXT("front_camera");
+	int32 ImageHeight = 480;
+	int32 ImageWidth = 640;
+	int32 Fps = 30;
+	bool bUseVideo = false;
+	
+	//Feature Data declarations
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	FString ObservationPayloadString = FString();
+
+	//Internal Functions
+	bool CreateJsonPayload_Observation(const FObservationPayload& Data);
+
+	//Blueprint Callable Functions
+	UFUNCTION(BlueprintCallable, Category = "Websocket")
+	bool MakeObservationPayload(const FObservationPayload& Data);
+	//---------------------------------------------------------//
+	
+protected:
+	//LuckyWriteThread* WriteThread = nullptr;
+};

Plugins/LuckyDataTransfer/Source/LuckyDataTransfer/Public/ObservationData.h

@@ -0,0 +1,80 @@
+// Fill out your copyright notice in the Description page of Project Settings.
+
+#pragma once
+
+#include "CoreMinimal.h"
+#include "HAL/Runnable.h"
+#include "ObservationData.generated.h"
+
+/**
+ * 
+ */
+USTRUCT(BlueprintType)
+struct FImageShape
+{
+	GENERATED_USTRUCT_BODY()
+
+public:
+	UPROPERTY(EditAnywhere, BlueprintReadWrite, Category = "Observation")
+	float image_width = 640.f;
+
+	UPROPERTY(EditAnywhere, BlueprintReadWrite, Category = "Observation")
+	float image_height = 480.f;
+
+	UPROPERTY(EditAnywhere, BlueprintReadWrite, Category = "Observation")
+	int32 channel = 3;
+};
+
+USTRUCT(BlueprintType)
+struct FObservationCameraInfo
+{
+	GENERATED_USTRUCT_BODY()
+
+public:
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	TMap<FString, FString> info = {
+		{"video_fps", "30"},
+		{"video_codec", "mp4v"},
+		{"video_pix_fmt", "yuv420p"},
+		{"video_is_depth_map", "false"},
+		{"has_audio", "false"}
+	};
+};
+
+USTRUCT(BlueprintType)
+struct FObservationCameraObject
+{
+	GENERATED_USTRUCT_BODY()
+
+public:
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	FString cameraName = TEXT("front_camera");
+	
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	FString dtype = TEXT("image");
+
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	FImageShape shape = FImageShape();
+
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	FString filePath = FString();
+};
+
+USTRUCT(BlueprintType)
+struct FObservationPayload
+{
+	GENERATED_USTRUCT_BODY()
+
+public:
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	FString timeStamp = FString();
+	
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	FString ID = FString();
+	
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	TMap<FString, float> ObservationState;
+	
+	UPROPERTY(BlueprintReadWrite, Category = "Observation")
+	TArray<FObservationCameraObject> ObservationCameras;
+};