#27 Add softmax policy

Author: pockerman

src/policies/softmax_policy.py

@@ -0,0 +1,44 @@
+import numpy as np
+from typing import TypeVar, Any
+from src.utils.mixins import WithQTableMixin
+
+QTable = TypeVar('QTable')
+
+
+class SoftMaxPolicy(WithQTableMixin):
+
+    def __init__(self, n_actions: int, tau: float) -> None:
+        self.n_actions = n_actions
+        self.tau = tau
+
+    def __str__(self) -> str:
+        return "SoftMaxPolicy"
+
+    def __call__(self, q_table: QTable, state: Any) -> int:
+        """
+        Execute the policy
+        :param q_table:
+        :param state:
+        :return:
+        """
+        self.q_table = q_table
+        action_values = [q_table[state, a] for a in range(self.n_actions)]
+        softmax = np.exp(np.array(action_values) / self.tau) / np.sum( np.exp(np.array(action_values) / self.tau) )
+
+        # return the action index by choosing from
+        return np.random.choice( [a for a in range(self.n_actions)], p=softmax)
+
+    def on_state(self, state: Any) -> int:
+        """
+        Returns the optimal action on the current state
+        :param state:
+        :return:
+        """
+        action_values = [self.q_table[state, a] for a in range(self.n_actions)]
+        softmax = np.exp(np.array(action_values) / self.tau) / np.sum(np.exp(np.array(action_values) / self.tau))
+
+        # return the action index by choosing from
+        return np.random.choice([a for a in range(self.n_actions)], p=softmax)
+
+    def actions_after_episode(self, episode_idx: int, **options) -> None:
+        pass

src/utils/iteration_control.py

@@ -0,0 +1,46 @@
+"""
+Utility to control iteration
+"""
+
+from src.utils import INFO, VERSION
+
+
+class IterationControl(object):
+    """
+    Helper class to control iteration
+    """
+
+    def __init__(self, n_itrs: int, min_dist: float, max_dist: float) -> None:
+        self.n_itrs = n_itrs
+        self.min_dist = min_dist
+        self.max_dist = max_dist
+        self.iteration_counter = 0
+
+    def continue_itr(self, distortion: float):
+
+        if self.min_dist <= distortion <= self.max_dist:
+            print("{0} Finished iteration with distortion={1} "
+                  "in [{2}, {3}]. Number of iterations={4}".format(INFO, distortion,
+                                                                   self.min_dist, self.max_dist,
+                                                                   self.iteration_counter))
+            return False
+
+        # TODO: There is no way currently we go back
+        # to acceptable distortions
+        if VERSION == '0.0.1-alpha':
+            if distortion > self.max_dist:
+                print("{0} Finished iteration with distortion={1} "
+                      "in [{2}, {3}]. Number of iterations={4}".format(INFO, distortion, self.min_dist,
+                                                                       self.max_dist, self.iteration_counter))
+                return False
+
+        if 0 <= self.n_itrs <= self.iteration_counter:
+            print("{0} Reached maximum number of iterations. With distortion={1} "
+                  "in [{2}, {3}]. Number of iterations={4}".format(INFO, distortion,
+                                                                   self.min_dist, self.max_dist,
+                                                                   self.iteration_counter))
+            return False
+
+        self.iteration_counter += 1
+        print("{0} Iteration={1} Distortion={2}".format(INFO, self.iteration_counter, distortion))
+        return True

src/utils/plot_utils.py

@@ -0,0 +1,31 @@
+import matplotlib.pyplot as plt
+import numpy as np
+
+
+def plot_running_avg(avg_array, steps: int,
+                     xlabel: str, ylabel: str,
+                     title: str, show_grid: bool = True):
+    """
+    Plot a running average of the values in the arra
+    :param title:
+    :param xlabel:
+    :param ylabel:
+    :param show_grid:
+    :param avg_array:
+    :param steps:
+    :return:
+    """
+
+    running_avg = np.empty(avg_array.shape[0])
+    for t in range(avg_array.shape[0]):
+        running_avg[t] = np.mean(avg_array[max(0, t-steps): (t+1)])
+
+    plt.plot(running_avg)
+    plt.xlabel(xlabel)
+    plt.ylabel(ylabel)
+    plt.title(title)
+
+    if show_grid:
+        plt.grid()
+
+    plt.show()