Merge pull request wbap#6 from wbap/python-to-unity-with-messagepack

sending action from python to unity with messagepack

python-agent/server.py

@@ -43,6 +43,10 @@ class AgentServer(WebSocket):
     depth_image_dim = 32 * 32
     depth_image_count = 1
 
+    def send_action(self, action):
+        dat = msgpack.packb([{"command": str(action)}])
+        self.send(dat, binary=True)
+
     def received_message(self, m):
         payload = m.data
         dat = msgpack.unpackb(payload)
@@ -67,7 +71,7 @@ def received_message(self, m):
                 depth_image_dim=self.depth_image_dim * self.depth_image_count)
 
             action = self.agent.agent_start(observation)
-            self.send(str(action))
+            self.send_action(action)
             with open(self.log_file, 'w') as the_file:
                 the_file.write('cycle, episode_reward_sum \n')
         else:
@@ -78,14 +82,14 @@ def received_message(self, m):
             if end_episode:
                 self.agent.agent_end(reward)
                 action = self.agent.agent_start(observation)  # TODO
-                self.send(str(action))
+                self.send_action(action)
                 with open(self.log_file, 'a') as the_file:
                     the_file.write(str(self.cycle_counter) +
                                    ',' + str(self.reward_sum) + '\n')
                 self.reward_sum = 0
             else:
                 action, eps, q_now, obs_array = self.agent.agent_step(reward, observation)
-                self.send(str(action))
+                self.send_action(action)
                 self.agent.agent_step_update(reward, action, eps, q_now, obs_array)
 
         self.thread_event.set()

unity-sample-environment/Assets/Scripts/Action.cs

@@ -1,5 +1,5 @@
 using UnityEngine;
-using System.Collections;
+using System.Collections.Generic;
 
 namespace MLPlayer {
 	public class Action {
@@ -11,9 +11,25 @@ public void Clear() {
 			forward = 0;
 			jump = false;
 		}
-		public void Set(string msg) {
+
+		public void Set(Dictionary<System.Object, System.Object> action) {
+
+			// make hash table (because the 2 data arrays with equal content do not provide the same hash)
+			var originalKey = new Dictionary<string, byte[]>();
+			foreach (byte[] key in action.Keys) {
+				originalKey.Add (System.Text.Encoding.UTF8.GetString(key), key);
+				//Debug.Log ("key:" + System.Text.Encoding.UTF8.GetString(key) + " value:" + action[key]);
+			}
+
+			// string:
+			string command = System.Text.Encoding.UTF8.GetString((byte[])action [originalKey ["command"]]);
+			// int:
+			//int i = (int)action [originalKey ["command"]];
+			// float:
+			//float f = float.Parse (System.Text.Encoding.UTF8.GetString((byte[])action [originalKey ["value"]]));
+
 			Clear ();
-			switch (msg) {
+			switch (command) {
 			case "0":
 				rotate = 1;
 				break;

unity-sample-environment/Assets/Scripts/SceneController.cs

@@ -35,8 +35,13 @@ public static SceneController Instance {
 		public static ManualResetEvent received = new ManualResetEvent(false);
 		private Vector3 FirstLocation;
 
-		void OnMassage(string msg) {
-			agent.action.Set (msg);
+		void OnMassage(byte[] msg) {
+			var packer = new MsgPack.BoxingPacker();
+			var res = packer.Unpack(msg);
+			System.Object[] actions = (System.Object[])packer.Unpack(msg);
+
+			agent.action.Set ((Dictionary<System.Object, System.Object>)actions[0]);
+
 			received.Set();
 		}
 
@@ -64,10 +69,7 @@ void Start () {
 			ws = new WebSocketSharp.WebSocket (url);
 			Debug.Log("connecting... " + url);
 			ws.Connect ();
-
-			ws.OnMessage += (sender, e) => OnMassage(e.Data);
-			// if using binary 
-			//ws.OnMessage += (sender, e) => OnMassage(e.RawData);
+			ws.OnMessage += (sender, e) => OnMassage(e.RawData);
 
 			lastSendTime = -cycleTimeStepSize;
 		}

unity-sample-environment/Assets/Scripts/SceneControllerAsync.cs

@@ -8,7 +8,7 @@ namespace MLPlayer {
 	public class AIClient {
 
 		private Queue<byte[]> agentMessageQueue;
-		private Queue<string> aiMessageQueue;
+		private Queue<byte[]> aiMessageQueue;
 		private string url;
 		private Thread th;
 		private Mutex mutAgent;
@@ -21,7 +21,7 @@ public AIClient (string _url) {
 			mutAi = new Mutex();
 			packer = new MsgPack.CompiledPacker();
 			agentMessageQueue = new Queue<byte[]>();
-			aiMessageQueue = new Queue<string>();
+			aiMessageQueue = new Queue<byte[]>();
 			th = new Thread(new ThreadStart(ExecuteInForeground));
 			th.Start(this);
 		}
@@ -31,9 +31,7 @@ private void ExecuteInForeground() {
 			WebSocketSharp.WebSocket ws = new WebSocketSharp.WebSocket (url);
 			Debug.Log("connecting... " + url);
 
-			ws.OnMessage += (sender, e) => OnMassage(e.Data);
-			// if using binary 
-			//ws.OnMessage += (sender, e) => OnMassage(e.RawData);
+			ws.OnMessage += (sender, e) => OnMassage(e.RawData);
 
 			while (true) {
 				ws.Connect ();
@@ -52,7 +50,7 @@ private void ExecuteInForeground() {
 			}
 		}
 
-		private void OnMassage(string msg) {
+		private void OnMassage(byte[] msg) {
 			PushAIMessage(msg);
 		}
 
@@ -63,14 +61,14 @@ public void PushAgentState(State s) {
 			mutAgent.ReleaseMutex();
 		}
 
-		public void PushAIMessage(string msg) {
+		public void PushAIMessage(byte[] msg) {
 			mutAi.WaitOne();
 			aiMessageQueue.Enqueue(msg);
 			mutAi.ReleaseMutex();
 		}
 
-		public string PopAIMessage() {
-			string received = null;
+		public byte[] PopAIMessage() {
+			byte[] received = null;
 
 			mutAi.WaitOne();
 			if( aiMessageQueue.Count > 0 ) {
@@ -155,9 +153,12 @@ void StartNewEpisode() {
 		void Update() {
 			Application.targetFrameRate = (int)Mathf.Max(60.0f, 60.0f * timeScale);
 
-			string msg = client.PopAIMessage();
+			byte[] msg = client.PopAIMessage();
 			if(msg != null) {
-				agent.action.Set(msg);
+				var packer = new MsgPack.BoxingPacker();
+				var res = packer.Unpack(msg);
+				System.Object[] actions = (System.Object[])packer.Unpack(msg);
+				agent.action.Set((Dictionary<System.Object, System.Object>)actions[0]);
 				OnCycleUpdateAfterReceiveAction();
 				Time.timeScale = timeScale;
 			}