Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_environment(self):
self.start_tests(name='getting-started-environment')
environment = Environment.create(
environment='gym', level='CartPole', max_episode_timesteps=500
)
self.finished_test()
environment = Environment.create(environment='gym', level='CartPole-v1')
self.finished_test()
environment = Environment.create(
environment='test/data/environment.json', max_episode_timesteps=500
)
self.finished_test()
environment = Environment.create(
environment='test.data.custom_env.CustomEnvironment', max_episode_timesteps=10
)
self.finished_test()
import time
from multiprocessing.pool import ThreadPool
from tensorforce.environments import Environment
import numpy as np
POOL = ThreadPool(processes=10)
class DistributedRewardsEnvironment(Environment):
def execute_async(self, async_func, args):
'''
Return an async reward.
'''
return POOL.apply_async(async_func, args)
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
from collections import OrderedDict
import itertools
import numpy as np
from tensorforce.environments import Environment
class ViZDoom(Environment):
"""
[ViZDoom](https://github.com/mwydmuch/ViZDoom) environment adapter (specification key:
`vizdoom`).
May require:
```bash
sudo apt-get install g++ build-essential libsdl2-dev zlib1g-dev libmpg123-dev libjpeg-dev \
libsndfile1-dev nasm tar libbz2-dev libgtk2.0-dev make cmake git chrpath timidity \
libfluidsynth-dev libgme-dev libopenal-dev timidity libwildmidi-dev unzip libboost-all-dev \
liblua5.1-dev
pip3 install vizdoom
```
Args:
level (string): ViZDoom configuration file
yield data[i]
@dataclass
class EpisodeInfo:
"""
Information class.
"""
__slots__ = ["target_id", "time", "normalized_hamming_distance"]
target_id: int
time: float
normalized_hamming_distance: float
class RnaDesignEnvironment(Environment):
"""
The environment for RNA design using deep reinforcement learning.
"""
def __init__(self, dot_brackets, env_config):
"""TODO
Initialize an environemnt.
Args:
env_config: The configuration of the environment.
"""
self._env_config = env_config
targets = [_Target(dot_bracket, self._env_config) for dot_bracket in dot_brackets]
self._target_gen = _random_epoch_gen(targets)
def __del__(self):
if isinstance(self.tf_agent, Agent):
self.tf_agent.close()
if isinstance(self.tf_environment, Environment):
self.tf_environment.close()
)
states = environment.states()
actions = environment.actions()
self.environments.append(environment)
for environment in environments[1:]:
assert isinstance(environment, Environment) == self.is_environment_external
environment = Environment.create(
environment=environment, max_episode_timesteps=max_episode_timesteps
)
assert environment.states() == states
assert environment.actions() == actions
self.environments.append(environment)
else:
assert num_parallel is not None and environments is None
assert not isinstance(environment, Environment)
self.is_environment_external = False
for _ in range(num_parallel):
environment = Environment.create(
environment=environment, max_episode_timesteps=max_episode_timesteps
)
self.environments(environment)
if evaluation_environment is None:
self.evaluation_environment = None
else:
self.is_eval_environment_external = isinstance(evaluation_environment, Environment)
self.evaluation_environment = Environment.create(
environment=evaluation_environment, max_episode_timesteps=max_episode_timesteps
)
assert self.evaluation_environment.states() == environment.states()
assert self.evaluation_environment.actions() == environment.actions()
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""
Deepmind Pycolab Integration: https://github.com/deepmind/pycolab.
"""
import numpy as np
import copy
from tensorforce import TensorforceError
from tensorforce.environments import Environment
class DMPycolab(Environment):
"""
Bindings for Deepmind Pycolab environment https://github.com/deepmind/pycolab
"""
def __init__(self, game, ui, visualize=False):
"""
Initialize Pycolab environment.
Args:
game: Pycolab Game Engine object. See https://github.com/deepmind/pycolab/tree/master/pycolab/examples
ui: Pycolab CursesUI object. See https://github.com/deepmind/pycolab/tree/master/pycolab/examples
visualize: If set True, the program will visualize the trainings of Pycolab game # TODO
"""
self.game = game
self.init_game = copy.deepcopy(self.game)
self.ui = ui
import gym
import numpy as np
from tensorforce import TensorForceError
from tensorforce.environments import Environment
from trading_gym import TradingGym
class TradingEnv(Environment):
def __init__(self, query, lags):
self.gym = TradingGym(query, lags)
# self._states = dict(type='float', shape=self.gym.observation_space.shape)
# self._actions = self.actions_ = dict(type='int', shape=(), num_actions=len(self.gym.actions))
self._states = TradingEnv.state_from_space(space=self.gym.observation_space)
self._actions = TradingEnv.action_from_space(space=self.gym.action_space)
def __str__(self):
return 'Simulation Environment: {}'.format(self.gym.env_id)
@property
def states(self):
return self._states
@property
max_episode_timesteps=None, evaluation_environment=None, save_best_agent=None
):
self.environments = list()
if environment is None:
assert num_parallel is None and environments is not None
if not util.is_iterable(x=environments):
raise TensorforceError.type(
name='parallel-runner', argument='environments', value=environments
)
elif len(environments) == 0:
raise TensorforceError.value(
name='parallel-runner', argument='environments', value=environments
)
num_parallel = len(environments)
environment = environments[0]
self.is_environment_external = isinstance(environment, Environment)
environment = Environment.create(
environment=environment, max_episode_timesteps=max_episode_timesteps
)
states = environment.states()
actions = environment.actions()
self.environments.append(environment)
for environment in environments[1:]:
assert isinstance(environment, Environment) == self.is_environment_external
environment = Environment.create(
environment=environment, max_episode_timesteps=max_episode_timesteps
)
assert environment.states() == states
assert environment.actions() == actions
self.environments.append(environment)
else: