JuliaPOMDP / JuliaPOMDP/RLInterface.jl
Interacting via ZMQ
- Dominant language
- Julia
- Stars
- 24
- Forks
- 5
- PR merge metrics
- No merged PRs in 30d
Description
An example:
```python
import zmq
import gym
from gym import spaces
class ZMQConnection:
def __init__(self, ip, port):
self._ip = ip
self._port = port
self._context = zmq.Context()
self._socket = self._context.socket(zmq.REQ)
self._socket.connect("tcp://{}:{}".format(ip, port))
@property
def socket(self):
return self._socket
def sendreq(self, msg):
self.socket.send_json(msg)
respmsg = self.socket.recv_json()
return respmsg
class POMDPsZMQEnv(gym.Env):
def __init__(self, ip='127.0.0.1', port=9393):
self._conn = ZMQConnection(ip, port)
def _step(self, action):
data = self._conn.sendreq({"cmd": "step", "args": int(action + 1)})
assert 'obs' in data
assert 'rew' in data
assert 'done' in data
assert 'info' in data
return data['obs'], data['rew'], data['done'], data['info']
def _reset(self):
data = self._conn.sendreq({"cmd": "reset"})
assert 'obs' in data
return data['obs']
@property
def action_space(self):
data = self._conn.sendreq({"cmd": "n_actions"})
assert 'n_actions' in data
return spaces.Discrete(data['n_actions'])
@property
def observation_space(self):
data = self._conn.sendreq({"cmd": "obs_dimensions"})
assert 'obs_dim' in data
return spaces.Box(-10, 10, shape=tuple(data['obs_dim']))
if __name__ == '__main__':
env = POMDPsZMQEnv()
obs = env.reset()
while True:
action = env.action_space.sample()
ob, reward, done, _ = env.step(action)
print("s ->{}".format(obs))
print("a ->{}".format(action))
print("sp->{}".format(ob))
print("r ->{}".format(reward))
obs = ob
if done:
break
env.close()
```
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.