mirror of
https://github.com/wassname/kair_algorithms_draft.git
synced 2026-09-09 11:25:10 +08:00
Convert code to python 2.7 (#35)
* Convert code format to python2.7 (SAC) * Convert code format python2.7 (TD3, all fD) * Remove no use import and black setting * Change SAC param * Change env name Reacher-v2 to v1 * Remove old version reacher training script * Convert code format python2.7 * Modify .travis.yml * Add install command python3.6 & black on Makefile * Fix seperator to tab on Makefile * Modify Makefile * Fix little error * Change td3 gamma parameter
This commit is contained in:
Kyunghwan Kim
@@ -5,18 +5,16 @@
|
||||
- Contact: curt.park@medipixel.io
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import subprocess
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Tuple
|
||||
from abc import ABCMeta, abstractmethod
|
||||
|
||||
import gym
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
|
||||
class AbstractAgent(ABC):
|
||||
class AbstractAgent:
|
||||
"""Abstract Agent used for all agents.
|
||||
|
||||
Attributes:
|
||||
@@ -27,7 +25,9 @@ class AbstractAgent(ABC):
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, env: gym.Env, args: argparse.Namespace):
|
||||
__metaclass__ = ABCMeta
|
||||
|
||||
def __init__(self, env, args):
|
||||
"""Initialization.
|
||||
|
||||
Args:
|
||||
@@ -52,11 +52,11 @@ class AbstractAgent(ABC):
|
||||
)
|
||||
|
||||
@abstractmethod
|
||||
def select_action(self, state: np.ndarray):
|
||||
def select_action(self, state):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def step(self, action: torch.Tensor) -> Tuple[np.ndarray, np.float64, bool]:
|
||||
def step(self, action):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
@@ -68,7 +68,7 @@ class AbstractAgent(ABC):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def save_params(self, params: dict, n_episode: int):
|
||||
def save_params(self, params, n_episode):
|
||||
if not os.path.exists("./save"):
|
||||
os.mkdir("./save")
|
||||
|
||||
@@ -77,7 +77,7 @@ class AbstractAgent(ABC):
|
||||
path = os.path.join("./save/" + save_name + "_ep_" + str(n_episode) + ".pt")
|
||||
torch.save(params, path)
|
||||
|
||||
print("[INFO] Saved the model and optimizer to", path)
|
||||
print ("[INFO] Saved the model and optimizer to", path)
|
||||
|
||||
@abstractmethod
|
||||
def write_log(self, *args):
|
||||
@@ -106,7 +106,7 @@ class AbstractAgent(ABC):
|
||||
score += reward
|
||||
step += 1
|
||||
|
||||
print(
|
||||
print (
|
||||
"[INFO] episode %d\tstep: %d\ttotal score: %d"
|
||||
% (i_episode, step, score)
|
||||
)
|
||||
@@ -118,7 +118,7 @@ class AbstractAgent(ABC):
|
||||
class NormalizedActions(gym.ActionWrapper):
|
||||
"""Rescale and relocate the actions."""
|
||||
|
||||
def action(self, action: np.ndarray) -> np.ndarray:
|
||||
def action(self, action):
|
||||
"""Change the range (-1, 1) to (low, high)."""
|
||||
low = self.action_space.low
|
||||
high = self.action_space.high
|
||||
@@ -131,7 +131,7 @@ class NormalizedActions(gym.ActionWrapper):
|
||||
|
||||
return action
|
||||
|
||||
def reverse_action(self, action: np.ndarray) -> np.ndarray:
|
||||
def reverse_action(self, action):
|
||||
"""Change the range (low, high) to (-1, 1)."""
|
||||
low = self.action_space.low
|
||||
high = self.action_space.high
|
||||
|
||||
@@ -8,7 +8,6 @@
|
||||
"""
|
||||
|
||||
import random
|
||||
from typing import Tuple
|
||||
|
||||
import numpy as np
|
||||
import torch
|
||||
@@ -35,7 +34,7 @@ class PrioritizedReplayBuffer(ReplayBuffer):
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, buffer_size: int, batch_size: int, alpha: float = 0.6):
|
||||
def __init__(self, buffer_size, batch_size, alpha=0.6):
|
||||
"""Initialization.
|
||||
|
||||
Args:
|
||||
@@ -59,27 +58,22 @@ class PrioritizedReplayBuffer(ReplayBuffer):
|
||||
self.min_tree = MinSegmentTree(tree_capacity)
|
||||
self._max_priority = 1.0
|
||||
|
||||
def add(
|
||||
self,
|
||||
state: np.ndarray,
|
||||
action: np.ndarray,
|
||||
reward: np.float64,
|
||||
next_state: np.ndarray,
|
||||
done: bool,
|
||||
):
|
||||
def add(self, state, action, reward, next_state, done):
|
||||
"""Add experience and priority."""
|
||||
idx = self.tree_idx
|
||||
self.tree_idx = (self.tree_idx + 1) % self.buffer_size
|
||||
super().add(state, action, reward, next_state, done)
|
||||
super(PrioritizedReplayBuffer, self).add(
|
||||
state, action, reward, next_state, done
|
||||
)
|
||||
|
||||
self.sum_tree[idx] = self._max_priority ** self.alpha
|
||||
self.min_tree[idx] = self._max_priority ** self.alpha
|
||||
|
||||
def extend(self, transitions: list):
|
||||
def extend(self, transitions):
|
||||
"""Add experiences to memory."""
|
||||
raise NotImplementedError
|
||||
|
||||
def _sample_proportional(self, batch_size: int) -> list:
|
||||
def _sample_proportional(self, batch_size):
|
||||
"""Sample indices based on proportional."""
|
||||
indices = []
|
||||
p_total = self.sum_tree.sum(0, len(self.buffer) - 1)
|
||||
@@ -92,7 +86,7 @@ class PrioritizedReplayBuffer(ReplayBuffer):
|
||||
indices.append(idx)
|
||||
return indices
|
||||
|
||||
def sample(self, beta: float = 0.4) -> Tuple[torch.Tensor, ...]:
|
||||
def sample(self, beta=0.4):
|
||||
"""Sample a batch of experiences."""
|
||||
assert beta > 0
|
||||
|
||||
@@ -127,7 +121,7 @@ class PrioritizedReplayBuffer(ReplayBuffer):
|
||||
|
||||
return experiences
|
||||
|
||||
def update_priorities(self, indices: list, priorities: np.ndarray):
|
||||
def update_priorities(self, indices, priorities):
|
||||
"""Update priorities of sampled transitions."""
|
||||
assert len(indices) == len(priorities)
|
||||
|
||||
@@ -153,14 +147,7 @@ class PrioritizedReplayBufferfD(PrioritizedReplayBuffer):
|
||||
epsilon_d (float) : epsilon_d parameter to update priority using demo
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
buffer_size: int,
|
||||
batch_size: int,
|
||||
demo: list,
|
||||
alpha: float = 0.6,
|
||||
epsilon_d: float = 1.0,
|
||||
):
|
||||
def __init__(self, buffer_size, batch_size, demo, alpha=0.6, epsilon_d=1.0):
|
||||
"""Initialization.
|
||||
Args:
|
||||
buffer_size (int): size of replay buffer for experience
|
||||
@@ -181,14 +168,7 @@ class PrioritizedReplayBufferfD(PrioritizedReplayBuffer):
|
||||
self.min_tree[self.tree_idx] = self._max_priority ** self.alpha
|
||||
self.tree_idx += 1
|
||||
|
||||
def add(
|
||||
self,
|
||||
state: np.ndarray,
|
||||
action: np.ndarray,
|
||||
reward: np.float64,
|
||||
next_state: np.ndarray,
|
||||
done: bool,
|
||||
):
|
||||
def add(self, state, action, reward, next_state, done):
|
||||
"""Add experience and priority."""
|
||||
idx = self.tree_idx
|
||||
# buffer is full
|
||||
@@ -196,7 +176,9 @@ class PrioritizedReplayBufferfD(PrioritizedReplayBuffer):
|
||||
self.tree_idx = self.demo_size
|
||||
else:
|
||||
self.tree_idx = self.tree_idx + 1
|
||||
super().add(state, action, reward, next_state, done)
|
||||
super(PrioritizedReplayBuffer, self).add(
|
||||
state, action, reward, next_state, done
|
||||
)
|
||||
|
||||
self.sum_tree[idx] = self._max_priority ** self.alpha
|
||||
self.min_tree[idx] = self._max_priority ** self.alpha
|
||||
@@ -204,7 +186,7 @@ class PrioritizedReplayBufferfD(PrioritizedReplayBuffer):
|
||||
# update current total size
|
||||
self.total_size = self.demo_size + len(self.buffer)
|
||||
|
||||
def sample(self, beta: float = 0.4) -> Tuple[torch.Tensor, ...]:
|
||||
def sample(self, beta=0.4):
|
||||
"""Sample a batch of experiences."""
|
||||
assert beta > 0
|
||||
|
||||
@@ -266,7 +248,7 @@ class PrioritizedReplayBufferfD(PrioritizedReplayBuffer):
|
||||
|
||||
return experiences
|
||||
|
||||
def update_priorities(self, indices: list, priorities: np.ndarray):
|
||||
def update_priorities(self, indices, priorities):
|
||||
"""Update priorities of sampled transitions."""
|
||||
assert len(indices) == len(priorities)
|
||||
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
"""Replay buffer for baselines."""
|
||||
|
||||
from collections import deque
|
||||
from typing import Any, Deque, List, Tuple
|
||||
|
||||
import numpy as np
|
||||
import torch
|
||||
@@ -25,7 +24,7 @@ class ReplayBuffer:
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, buffer_size: int, batch_size: int):
|
||||
def __init__(self, buffer_size, batch_size):
|
||||
"""Initialize a ReplayBuffer object.
|
||||
|
||||
Args:
|
||||
@@ -33,19 +32,12 @@ class ReplayBuffer:
|
||||
batch_size (int): size of a batched sampled from replay buffer for training
|
||||
|
||||
"""
|
||||
self.buffer: list = list()
|
||||
self.buffer = list()
|
||||
self.buffer_size = buffer_size
|
||||
self.batch_size = batch_size
|
||||
self.idx = 0
|
||||
|
||||
def add(
|
||||
self,
|
||||
state: np.ndarray,
|
||||
action: np.ndarray,
|
||||
reward: np.float64,
|
||||
next_state: np.ndarray,
|
||||
done: bool,
|
||||
):
|
||||
def add(self, state, action, reward, next_state, done):
|
||||
"""Add a new experience to memory."""
|
||||
data = (state, action, reward, next_state, done)
|
||||
|
||||
@@ -55,14 +47,12 @@ class ReplayBuffer:
|
||||
else:
|
||||
self.buffer.append(data)
|
||||
|
||||
def extend(self, transitions: list):
|
||||
def extend(self, transitions):
|
||||
"""Add experiences to memory."""
|
||||
for transition in transitions:
|
||||
self.add(*transition)
|
||||
|
||||
def sample(
|
||||
self
|
||||
) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
|
||||
def sample(self):
|
||||
"""Randomly sample a batch of experiences from memory."""
|
||||
idxs = np.random.choice(len(self.buffer), size=self.batch_size, replace=False)
|
||||
|
||||
@@ -84,7 +74,7 @@ class ReplayBuffer:
|
||||
|
||||
return states, actions, rewards, next_states, dones
|
||||
|
||||
def __len__(self) -> int:
|
||||
def __len__(self):
|
||||
"""Return the current size of internal memory."""
|
||||
return len(self.buffer)
|
||||
|
||||
@@ -100,7 +90,7 @@ class NStepTransitionBuffer:
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, buffer_size: int, n_step: int, gamma: float, demo: list = None):
|
||||
def __init__(self, buffer_size, n_step, gamma, demo=None):
|
||||
"""Initialize a ReplayBuffer object.
|
||||
|
||||
Args:
|
||||
@@ -110,9 +100,9 @@ class NStepTransitionBuffer:
|
||||
"""
|
||||
assert buffer_size > 0
|
||||
|
||||
self.n_step_buffer: Deque = deque(maxlen=n_step)
|
||||
self.n_step_buffer = deque(maxlen=n_step)
|
||||
self.buffer_size = buffer_size
|
||||
self.buffer: list = list()
|
||||
self.buffer = list()
|
||||
self.n_step = n_step
|
||||
self.gamma = gamma
|
||||
self.demo_size = 0
|
||||
@@ -125,7 +115,7 @@ class NStepTransitionBuffer:
|
||||
|
||||
self.buffer.extend([None] * self.buffer_size)
|
||||
|
||||
def add(self, transition: Tuple[np.ndarray, ...]) -> Tuple[Any, ...]:
|
||||
def add(self, transition):
|
||||
"""Add a new transition to memory."""
|
||||
self.n_step_buffer.append(transition)
|
||||
|
||||
@@ -146,7 +136,7 @@ class NStepTransitionBuffer:
|
||||
# return a single step transition to insert to replay buffer
|
||||
return self.n_step_buffer[0]
|
||||
|
||||
def sample(self, indices: List[int]) -> Tuple[torch.Tensor, ...]:
|
||||
def sample(self, indices):
|
||||
"""Randomly sample a batch of experiences from memory."""
|
||||
states, actions, rewards, next_states, dones = [], [], [], [], []
|
||||
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
"""Segment tree for Proirtized Replay Buffer."""
|
||||
|
||||
import operator
|
||||
from typing import Callable
|
||||
|
||||
|
||||
class SegmentTree:
|
||||
@@ -18,7 +17,7 @@ class SegmentTree:
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, capacity: int, operation: Callable, init_value: float):
|
||||
def __init__(self, capacity, operation, init_value):
|
||||
"""Initialization.
|
||||
|
||||
Args:
|
||||
@@ -34,9 +33,7 @@ class SegmentTree:
|
||||
self.tree = [init_value for _ in range(2 * capacity)]
|
||||
self.operation = operation
|
||||
|
||||
def _operate_helper(
|
||||
self, start: int, end: int, node: int, node_start: int, node_end: int
|
||||
) -> float:
|
||||
def _operate_helper(self, start, end, node, node_start, node_end):
|
||||
"""Returns result of operation in segment."""
|
||||
if start == node_start and end == node_end:
|
||||
return self.tree[node]
|
||||
@@ -52,7 +49,7 @@ class SegmentTree:
|
||||
self._operate_helper(mid + 1, end, 2 * node + 1, mid + 1, node_end),
|
||||
)
|
||||
|
||||
def operate(self, start: int = 0, end: int = 0) -> float:
|
||||
def operate(self, start=0, end=0):
|
||||
"""Returns result of applying `self.operation`."""
|
||||
if end <= 0:
|
||||
end += self.capacity
|
||||
@@ -60,7 +57,7 @@ class SegmentTree:
|
||||
|
||||
return self._operate_helper(start, end, 1, 0, self.capacity - 1)
|
||||
|
||||
def __setitem__(self, idx: int, val: float):
|
||||
def __setitem__(self, idx, val):
|
||||
"""Set value in tree."""
|
||||
idx += self.capacity
|
||||
self.tree[idx] = val
|
||||
@@ -70,7 +67,7 @@ class SegmentTree:
|
||||
self.tree[idx] = self.operation(self.tree[2 * idx], self.tree[2 * idx + 1])
|
||||
idx //= 2
|
||||
|
||||
def __getitem__(self, idx: int) -> float:
|
||||
def __getitem__(self, idx):
|
||||
"""Get real value in leaf node of tree."""
|
||||
assert 0 <= idx < self.capacity
|
||||
|
||||
@@ -85,7 +82,7 @@ class SumSegmentTree(SegmentTree):
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, capacity: int):
|
||||
def __init__(self, capacity):
|
||||
"""Initialization.
|
||||
|
||||
Args:
|
||||
@@ -96,11 +93,11 @@ class SumSegmentTree(SegmentTree):
|
||||
capacity=capacity, operation=operator.add, init_value=0.0
|
||||
)
|
||||
|
||||
def sum(self, start: int = 0, end: int = 0) -> float:
|
||||
def sum(self, start=0, end=0):
|
||||
"""Returns arr[start] + ... + arr[end]."""
|
||||
return super(SumSegmentTree, self).operate(start, end)
|
||||
|
||||
def retrieve(self, upperbound: float) -> int:
|
||||
def retrieve(self, upperbound):
|
||||
"""Find the highest index `i` about upper bound in the tree"""
|
||||
assert 0 <= upperbound <= self.sum() + 1e-5
|
||||
|
||||
@@ -125,7 +122,7 @@ class MinSegmentTree(SegmentTree):
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, capacity: int):
|
||||
def __init__(self, capacity):
|
||||
"""Initialization.
|
||||
|
||||
Args:
|
||||
@@ -136,6 +133,6 @@ class MinSegmentTree(SegmentTree):
|
||||
capacity=capacity, operation=min, init_value=float("inf")
|
||||
)
|
||||
|
||||
def min(self, start: int = 0, end: int = 0) -> float:
|
||||
def min(self, start=0, end=0):
|
||||
"""Returns min(arr[start], ..., arr[end])."""
|
||||
return super(MinSegmentTree, self).operate(start, end)
|
||||
|
||||
@@ -7,28 +7,25 @@
|
||||
|
||||
import random
|
||||
from collections import deque
|
||||
from typing import Deque, List, Tuple
|
||||
|
||||
import gym
|
||||
import numpy as np
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
|
||||
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
|
||||
|
||||
|
||||
def identity(x: torch.Tensor) -> torch.Tensor:
|
||||
def identity(x):
|
||||
"""Return input without any change."""
|
||||
return x
|
||||
|
||||
|
||||
def soft_update(local: nn.Module, target: nn.Module, tau: float):
|
||||
def soft_update(local, target, tau):
|
||||
"""Soft-update: target = tau*local + (1-tau)*target."""
|
||||
for t_param, l_param in zip(target.parameters(), local.parameters()):
|
||||
t_param.data.copy_(tau * l_param.data + (1.0 - tau) * t_param.data)
|
||||
|
||||
|
||||
def set_random_seed(seed: int, env: gym.Env):
|
||||
def set_random_seed(seed, env):
|
||||
"""Set random seed"""
|
||||
env.seed(seed)
|
||||
torch.manual_seed(seed)
|
||||
@@ -36,16 +33,14 @@ def set_random_seed(seed: int, env: gym.Env):
|
||||
random.seed(seed)
|
||||
|
||||
|
||||
def get_n_step_info_from_demo(
|
||||
demo: List, n_step: int, gamma: float
|
||||
) -> Tuple[List, List]:
|
||||
def get_n_step_info_from_demo(demo, n_step, gamma):
|
||||
"""Return 1 step and n step demos."""
|
||||
assert demo
|
||||
assert n_step > 1
|
||||
|
||||
demos_1_step = list()
|
||||
demos_n_step = list()
|
||||
n_step_buffer: Deque = deque(maxlen=n_step)
|
||||
n_step_buffer = deque(maxlen=n_step)
|
||||
|
||||
for transition in demo:
|
||||
n_step_buffer.append(transition)
|
||||
@@ -63,9 +58,7 @@ def get_n_step_info_from_demo(
|
||||
return demos_1_step, demos_n_step
|
||||
|
||||
|
||||
def get_n_step_info(
|
||||
n_step_buffer: Deque, gamma: float
|
||||
) -> Tuple[np.int64, np.ndarray, bool]:
|
||||
def get_n_step_info(n_step_buffer, gamma):
|
||||
"""Return n step reward, next state, and done."""
|
||||
# info of the last transition
|
||||
reward, next_state, done = n_step_buffer[-1][-3:]
|
||||
|
||||
@@ -5,9 +5,6 @@
|
||||
- Contact: whikwon@gmail.com
|
||||
"""
|
||||
|
||||
from typing import Callable
|
||||
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
import torch.nn.functional as F
|
||||
|
||||
@@ -30,13 +27,13 @@ class LSTM(nn.Module):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
input_size: int,
|
||||
output_size: int,
|
||||
hidden_sizes: list,
|
||||
hidden_activation: Callable = F.relu,
|
||||
output_activation: Callable = identity,
|
||||
use_output_layer: bool = True,
|
||||
init_w: float = 3e-3,
|
||||
input_size,
|
||||
output_size,
|
||||
hidden_sizes,
|
||||
hidden_activation=F.relu,
|
||||
output_activation=identity,
|
||||
use_output_layer=True,
|
||||
init_w=3e-3,
|
||||
):
|
||||
"""Initialization.
|
||||
|
||||
@@ -59,7 +56,7 @@ class LSTM(nn.Module):
|
||||
self.output_activation = output_activation
|
||||
self.use_output_layer = use_output_layer
|
||||
|
||||
self.hidden_layers: list = []
|
||||
self.hidden_layers = []
|
||||
in_size = self.input_size
|
||||
for i, next_size in enumerate(hidden_sizes):
|
||||
lstm = nn.LSTM(in_size, next_size, batch_first=True)
|
||||
@@ -73,14 +70,14 @@ class LSTM(nn.Module):
|
||||
self.output_layer.weight.data.uniform_(-init_w, init_w)
|
||||
self.output_layer.bias.data.uniform_(-init_w, init_w)
|
||||
|
||||
def get_last_activation(self, x: torch.Tensor) -> torch.Tensor:
|
||||
def get_last_activation(self, x):
|
||||
"""Get the activation of the last hidden layer."""
|
||||
for hidden_layer in self.hidden_layers:
|
||||
x, _ = hidden_layer(x)
|
||||
x = self.hidden_activation(x)
|
||||
return x
|
||||
|
||||
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
||||
def forward(self, x):
|
||||
"""Forward method implementation."""
|
||||
assert self.use_output_layer
|
||||
|
||||
|
||||
@@ -5,8 +5,6 @@
|
||||
- Contact: kh.kim@medipixel.io
|
||||
"""
|
||||
|
||||
from typing import Callable, Tuple
|
||||
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
import torch.nn.functional as F
|
||||
@@ -31,13 +29,13 @@ class MLP(nn.Module):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
input_size: int,
|
||||
output_size: int,
|
||||
hidden_sizes: list,
|
||||
hidden_activation: Callable = F.relu,
|
||||
output_activation: Callable = identity,
|
||||
use_output_layer: bool = True,
|
||||
init_w: float = 3e-3,
|
||||
input_size,
|
||||
output_size,
|
||||
hidden_sizes,
|
||||
hidden_activation=F.relu,
|
||||
output_activation=identity,
|
||||
use_output_layer=True,
|
||||
init_w=3e-3,
|
||||
):
|
||||
"""Initialization.
|
||||
|
||||
@@ -61,7 +59,7 @@ class MLP(nn.Module):
|
||||
self.use_output_layer = use_output_layer
|
||||
|
||||
# set hidden layers
|
||||
self.hidden_layers: list = []
|
||||
self.hidden_layers = []
|
||||
in_size = self.input_size
|
||||
for i, next_size in enumerate(hidden_sizes):
|
||||
fc = nn.Linear(in_size, next_size)
|
||||
@@ -75,13 +73,13 @@ class MLP(nn.Module):
|
||||
self.output_layer.weight.data.uniform_(-init_w, init_w)
|
||||
self.output_layer.bias.data.uniform_(-init_w, init_w)
|
||||
|
||||
def get_last_activation(self, x: torch.Tensor) -> torch.Tensor:
|
||||
def get_last_activation(self, x):
|
||||
"""Get the activation of the last hidden layer."""
|
||||
for hidden_layer in self.hidden_layers:
|
||||
x = self.hidden_activation(hidden_layer(x))
|
||||
return x
|
||||
|
||||
def forward(self, x: torch.Tensor) -> torch.Tensor:
|
||||
def forward(self, x):
|
||||
"""Forward method implementation."""
|
||||
assert self.use_output_layer
|
||||
|
||||
@@ -96,7 +94,7 @@ class MLP(nn.Module):
|
||||
class FlattenMLP(MLP):
|
||||
"""Baseline of Multilayer perceptron for Flatten input."""
|
||||
|
||||
def forward(self, *args: torch.Tensor) -> torch.Tensor:
|
||||
def forward(self, *args):
|
||||
"""Forward method implementation."""
|
||||
states, actions = args
|
||||
flat_inputs = torch.cat((states, actions), dim=-1)
|
||||
@@ -116,14 +114,14 @@ class GaussianDist(MLP):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
input_size: int,
|
||||
output_size: int,
|
||||
hidden_sizes: list,
|
||||
hidden_activation: Callable = F.relu,
|
||||
mu_activation: Callable = torch.tanh,
|
||||
log_std_min: float = -20,
|
||||
log_std_max: float = 2,
|
||||
init_w: float = 3e-3,
|
||||
input_size,
|
||||
output_size,
|
||||
hidden_sizes,
|
||||
hidden_activation=F.relu,
|
||||
mu_activation=torch.tanh,
|
||||
log_std_min=-20,
|
||||
log_std_max=2,
|
||||
init_w=3e-3,
|
||||
):
|
||||
"""Initialization."""
|
||||
super(GaussianDist, self).__init__(
|
||||
@@ -149,7 +147,7 @@ class GaussianDist(MLP):
|
||||
self.mu_layer.weight.data.uniform_(-init_w, init_w)
|
||||
self.mu_layer.bias.data.uniform_(-init_w, init_w)
|
||||
|
||||
def get_dist_params(self, x: torch.Tensor) -> Tuple[torch.Tensor, ...]:
|
||||
def get_dist_params(self, x):
|
||||
"""Return gausian distribution parameters."""
|
||||
hidden = super(GaussianDist, self).get_last_activation(x)
|
||||
|
||||
@@ -165,7 +163,7 @@ class GaussianDist(MLP):
|
||||
|
||||
return mu, log_std, std
|
||||
|
||||
def forward(self, x: torch.Tensor) -> Tuple[torch.Tensor, ...]:
|
||||
def forward(self, x):
|
||||
"""Forward method implementation."""
|
||||
mu, _, std = self.get_dist_params(x)
|
||||
|
||||
@@ -181,11 +179,9 @@ class TanhGaussianDistParams(GaussianDist):
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
"""Initialization."""
|
||||
super(TanhGaussianDistParams, self).__init__(**kwargs, mu_activation=identity)
|
||||
super(TanhGaussianDistParams, self).__init__(mu_activation=identity, **kwargs)
|
||||
|
||||
def forward(
|
||||
self, x: torch.Tensor, epsilon: float = 1e-6
|
||||
) -> Tuple[torch.Tensor, ...]:
|
||||
def forward(self, x, epsilon=1e-6):
|
||||
"""Forward method implementation."""
|
||||
mu, _, std = super(TanhGaussianDistParams, self).get_dist_params(x)
|
||||
|
||||
|
||||
@@ -13,20 +13,14 @@ class GaussianNoise:
|
||||
Taken from https://github.com/vitchyr/rlkit
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
action_dim: int,
|
||||
min_sigma: float = 1.0,
|
||||
max_sigma: float = 1.0,
|
||||
decay_period: int = 1000000,
|
||||
):
|
||||
def __init__(self, action_dim, min_sigma=1.0, max_sigma=1.0, decay_period=1000000):
|
||||
"""Initialization."""
|
||||
self.action_dim = action_dim
|
||||
self.min_sigma = min_sigma
|
||||
self.max_sigma = max_sigma
|
||||
self.decay_period = decay_period
|
||||
|
||||
def sample(self, t: int = 0) -> float:
|
||||
def sample(self, t=0):
|
||||
"""Get an action with gaussian noise."""
|
||||
sigma = self.max_sigma - (self.max_sigma - self.min_sigma) * min(
|
||||
1.0, t / self.decay_period
|
||||
@@ -42,9 +36,7 @@ class OUNoise:
|
||||
ddpg-pendulum/ddpg_agent.py
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self, size: int, mu: float = 0.0, theta: float = 0.15, sigma: float = 0.2
|
||||
):
|
||||
def __init__(self, size, mu=0.0, theta=0.15, sigma=0.2):
|
||||
"""Initialize parameters and noise process."""
|
||||
self.state = np.float64(0.0)
|
||||
self.mu = mu * np.ones(size)
|
||||
@@ -56,7 +48,7 @@ class OUNoise:
|
||||
"""Reset the internal state (= noise) to mean (mu)."""
|
||||
self.state = copy.copy(self.mu)
|
||||
|
||||
def sample(self) -> float:
|
||||
def sample(self):
|
||||
"""Update internal state and return it as a noise sample."""
|
||||
x = self.state
|
||||
dx = self.theta * (self.mu - x) + self.sigma * np.array(
|
||||
|
||||
Reference in New Issue
Block a user