Initial commit, basic API working (5fbd8159) · Commits · Smart Buildings / boptest-gym

.gitignore

0 → 100644

+4 −0

Changes for .gitignore: 4 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		.ipynb_checkpoints/
		.vscode/
		__pycache__/
		*.ipynb

env.py

0 → 100644

+189 −0

Changes for env.py: 189 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import json
		from pprint import pformat
		from typing import Dict

		import numpy as np
		import gym
		import requests



		class BOPTestEnv(gym.Env):
		"""
		A wrapper for OpenAI gym around [IBPSA's BOPTest test cases][1]. Currently
		only works with Box spaces.

		[1]: https://github.com/ibpsa/project1-boptest
		"""


		def __init__(self, url='http://localhost:5000', box_spaces=True, dtype=np.float64):
		"""
		Parameters
		----------
		url : str, optional
		The url at which the test case is running (format: http://addr:port),
		by default 'http://localhost:5000'
		box_spaces : bool, optional
		Whether or not to convert input/outputs to gym.spaces.Box, by default True
		dtype : [type], optional
		The data type of numpy arrays for states and actions, by default np.float64
		"""
		super().__init__()
		self.url = url
		self.dtype = np.float64
		self.name = requests.get(self.url + 'name').text
		self.box_spaces = box_spaces
		self.states = self.get_measurements()
		self.actions = self.get_inputs()
		self.setup_spaces()
		self.state_ = None
		self.time_ = 0


		def __repr__(self):
		r = (
		self.name + '\n'
		'State Space\n'
		'-----------\n'
		'' + pformat(self.states) + '\n'
		'Action Space\n'
		'------------\n'
		'{}'.format(str(pformat(self.actions))) + '\n'
		)
		return r


		def describe(self):
		"""
		Print the summary of the environment.
		"""
		print(self.__repr__())


		def setup_spaces(self):
		"""
		Initialize observation_ and action_ spaces by parsing metadata from the
		BOPTest testcase server.
		"""
		minimum = [dim['Minimum'] for _, dim in self.states.items()]
		minimum = [-np.inf if m is None else m for m in minimum]
		maximum = [dim['Maximum'] for _, dim in self.states.items()]
		maximum = [np.inf if m is None else m for m in maximum]
		self.observation_space = gym.spaces.Box(low=np.asarray(minimum, self.dtype),
		high=np.asarray(maximum, self.dtype))

		minimum = [dim['Minimum'] for _, dim in self.actions.items()]
		minimum = [-np.inf if m is None else m for m in minimum]
		maximum = [dim['Maximum'] for _, dim in self.actions.items()]
		maximum = [np.inf if m is None else m for m in maximum]
		self.action_space = gym.spaces.Box(low=np.asarray(minimum, self.dtype),
		high=np.asarray(maximum, self.dtype))


		def to_space(self, d: Dict) -> np.ndarray:
		"""
		Convert a dictionary to an array representing a Box space.

		Parameters
		----------
		d : Dict
		A dictionary of 'name': value pairs of state or action.

		Returns
		-------
		np.ndarray
		An array of values for each key in the dictionary. Key order is
		maintained.
		"""
		return np.asarray([val for _, val in d.items()], self.dtype)


		def from_space(self, s: np.ndarray, which='state') -> Dict:
		"""
		Convert an array representing a Box space into a dictionary for action
		or state.

		Parameters
		----------
		s : np.ndarray
		The array containing action or state values.
		which : str, optional
		Whether the array is a 'state' or 'action', by default 'state'

		Returns
		-------
		Dict
		A dictionary where each value in the array is mapped to the name
		of the dimension in state or action space.
		"""
		if which in ('state', self.states, self.observation_space):
		return {name: value for name, value in zip(self.states.keys(), action)}
		elif which in('action', self.actions, self.action_space):
		return {name: value for name, value in zip(self.actions.keys(), action)}


		def reset(self):
		requests.put(self.url + 'reset')
		self.time_ = 0
		# TODO: return initial state.


		def get_inputs(self):
		"""
		Get metadata describing input/action space. It is stored in self.actions.
		Metadata is a dictionary
		of the form:
		{'name' : {
		'Description':
		'Minimum':
		'Maximum':
		}, ...}
		"""
		return json.loads(requests.get(self.url + 'inputs').content)


		def get_measurements(self):
		"""
		Get metadata describing input/action space. It is stored in self.states.
		Metadata is a dictionary
		of the form:
		{'name' : {
		'Description':
		'Minimum':
		'Maximum':
		}, ...}
		"""
		return json.loads(requests.get(self.url + 'measurements').content)


		def reward(self, state, action, next_state):
		return None


		def done(self, time, state):
		return False


		def step(self, action: np.ndarray) -> np.ndarray:
		"""
		Simulate one time step and return next state.

		Parameters
		----------
		action : np.ndarray
		An array representing action value.

		Returns
		-------
		np.ndarray
		The next state array.
		"""
		self.time_ += 1
		action = {name: value for name, value in zip(self.actions.keys(), action)}
		state = json.loads(requests.post(self.url + 'advance', data=action).content)
		next_state = self.to_space(state)
		reward = self.reward(self.state_, action, next_state)
		done = self.done(self.time_, next_state)
		self.state_ = next_state
		return next_state, reward, done, state

Admin message