Feat/mujoco model

demos/fancy_gym/BoxPushingDense.py

@@ -8,7 +8,7 @@
 env = fancy_gym.make(env_name, seed=1)
 obs = env.reset()
 
-publisher = FancyGymPublisher(env_name, env, "127.0.0.1")
+publisher = FancyGymPublisher(env, "127.0.0.1")
 
 for i in range(1000):
     action = env.action_space.sample()

demos/fancy_gym/TableTennis.py

@@ -0,0 +1,19 @@
+import fancy_gym
+import time
+
+from simpub.sim.fancy_gym import FancyGymPublisher
+
+env_name = "TableTennis2D-v0"
+
+env = fancy_gym.make(env_name, seed=1)
+obs = env.reset()
+
+publisher = FancyGymPublisher(env, "127.0.0.1")
+
+while True:
+    action = env.action_space.sample()
+    obs, reward, done, info = env.step(action)
+    env.render()
+    time.sleep(0.01)
+    if done:
+        obs = env.reset()

simpub/core/net_manager.py

@@ -1,21 +1,21 @@
+import concurrent.futures
 import enum
 from typing import List, Dict
 from typing import NewType, Callable, TypedDict, Union
 import asyncio
 from asyncio import sleep as asycnc_sleep
-import threading
 import zmq
 import zmq.asyncio
 import socket
 from socket import AF_INET, SOCK_DGRAM, SOL_SOCKET, SO_BROADCAST
 import struct
 import time
 import json
+from json import dumps
 import uuid
 from .log import logger
-from json import dumps
-import zmq
 import abc
+import concurrent
 
 IPAddress = NewType("IPAddress", str)
 TopicName = NewType("TopicName", str)
@@ -140,8 +140,10 @@ def __init__(
 
     async def callback(self, msg: str):
         result = await asyncio.wait_for(
-            asyncio.to_thread(self.callback_func, msg),
-            timeout=5
+            self.manager.loop.run_in_executor(
+                self.manager.executor, self.callback_func, msg
+            ),
+            timeout=5.0,
         )
         await self.sender(result)
 
@@ -190,19 +192,9 @@ def __init__(
         # setting up thread pool
         self.running: bool = True
         self.loop: asyncio.AbstractEventLoop = None
-        self.start_server_thread()
-
-    def start_server_thread(self) -> None:
-        """
-        Start a thread for service.
-
-        Args:
-            block (bool, optional): main thread stop running and
-            wait for server thread. Defaults to False.
-        """
-        self.server_thread = threading.Thread(target=self.start_event_loop)
-        self.server_thread.daemon = True
-        self.server_thread.start()
+        # start the server in a thread pool
+        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
+        self.server_future = self.executor.submit(self.start_event_loop)
         while self.loop is None:
             time.sleep(0.01)
 
@@ -233,10 +225,10 @@ def submit_task(self, task: Callable, *args) -> asyncio.Future:
     def stop_server(self):
         if self.loop.is_running():
             asyncio.run_coroutine_threadsafe(self.loop.stop(), self.loop)
-        self.server_thread.join()
+        self.executor.shutdown(wait=True)
 
     def join(self):
-        self.server_thread.join()
+        self.executor.shutdown(wait=True)
 
     async def service_loop(self):
         # try:
@@ -258,7 +250,10 @@ async def service_loop(self):
                     )
                     await self.service_socket.send_string("Timeout")
                 except Exception as e:
-                    logger.error(f"Error: {e}")
+                    logger.error(
+                        f"One error ocurred when processing the Service "
+                        f"\"{service}\": {e}"
+                    )
             await asycnc_sleep(0.01)
 
     async def broadcast_loop(self):