Merge pull request #9 from pockerman/add_actor_critic_algorithm

Add actor critic algorithm

Author: pockerman

.gitignore

@@ -1,2 +1,7 @@
 src/preprocessor/__pycache__/
 src/exceptions/__pycache__/
+src/utils/__pycache__/
+src/tests/.pytest_cache/
+src/spaces/__pycache__/
+src/__pycache__/
+src/algorithms/__pycache__/

src/__init__.py

@@ -41,20 +41,32 @@ def forward(self, x):
         return pol_out, val_out
 
 
+class A2CConfig(object):
+    """
+    Configuration for A2C algorithm
+    """
+
+    def __init__(self):
+        self.gamma: float = 0.99
+        self.tau: float = 1.2
+        self.n_workers: int = 1
+        self.n_iterations_per_episode: int = 100
+        self.optimizer: Optimizer = None
+        self.loss_function: LossFunction = None
+
+
 class A2C(Generic[Optimizer]):
 
-    def __init__(self, gamma: float, tau: float, n_workers: int,
-                 n_iterations: int,  optimizer: Optimizer,
-                 a2c_net: A2CNet, loss_function: LossFunction):
+    def __init__(self, config: A2CConfig, a2c_net: A2CNet):
 
-        self.gamma = gamma
-        self.tau = tau
-        self.rewards = []
-        self.n_workers = n_workers
-        self.n_iterations = n_iterations
-        self.optimizer = optimizer
+        self.gamma = config.gamma
+        self.tau = config.tau
+        self.n_workers = config.n_workers
+        self.n_iterations_per_episode = config.n_iterations_per_episode
+        self.optimizer = config.optimizer
+        self.loss_function = config.loss_function
         self.a2c_net = a2c_net
-        self.loss_function = loss_function
+        self.rewards = []
         self.name = "A2C"
 
     def _optimize_model(self):
@@ -80,7 +92,8 @@ def train(self, env: Env) -> None:
 
         observation = time_step.observation
 
-        for iteration in range(1, self.n_iterations + 1):
+        # learn over the episode
+        for iteration in range(1, self.n_iterations_per_episode + 1):
 
             # select an action
             action = self.select_action(env=env, observation=observation)

src/algorithms/a2c.py

@@ -33,5 +33,4 @@ def train(self):
             if self.configuration["update_frequency"] % episode == 0:
                 self.agent.update()
 
-
         print("{0} Training finished for agent {1}".format(INFO, self.agent.name))

src/algorithms/trainer.py

@@ -0,0 +1,34 @@
+"""
+Various utilities to calculate the distance
+between two datasets. All distance metrics work
+accumulative
+"""
+
+from typing import TypeVar
+import numpy as np
+
+DataSet = TypeVar("DataSet")
+
+
+def lp_distance(ds1: DataSet, ds2: DataSet, p=None):
+    """
+    Compute the Lp norms between the respective columns in the given data sets.
+    This means that the two datasets should have the same schema. It is
+    up to the application to ensure that the calculation is meaningless
+    :param ds1: Dataset 1
+    :param ds2: Dataset 2
+    :param p:  The order of the norm to calculate
+    :return: The calculated Lp-norm
+    """
+
+    assert ds1.schema == ds2.schema, "Invalid schema for datasets"
+
+    distances = {}
+    cols = ds1.get_columns_names()
+    for col in cols:
+
+        val1 = ds1.get_column(col_name=col)
+        val2 = ds2.get_column(col_name=col)
+        distances[col] = np.linalg.norm(val1 - val2, ord=p)
+
+    return distances, sum(distances.values())

src/datasets/dataset_distances.py

@@ -0,0 +1,4 @@
+"""
+Utilities for calculating the information leakage
+for a dataset
+"""

src/datasets/dataset_information_leakage.py

@@ -111,8 +111,19 @@ def sample_column(self):
         col_idx = np.random.choice(col_names, 1)
         return self.get_column(col_name=col_names[col_idx])
 
-    def apply_transform(self, transform: Transform) -> None:
-        pass
+    def apply_column_transform(self, column_name: str, transform: Transform) -> None:
+        """
+        Apply the given transformation on the underlying dataset
+        :param column_name: The column to transform
+        :param transform: The transformation to apply
+        :return: None
+        """
+
+        # get the column
+        column = self.get_column(col_name=column_name)
+        column = transform.act(**{"data": column})
+        self.ds[transform.column_name] = column
+
 
 
 

src/datasets/dataset_wrapper.py

@@ -1,26 +1,68 @@
+"""
+ActionSpace class. This is a wrapper to the discrete
+actions in the actions.py module
+"""
+
 from gym.spaces.discrete import Discrete
 from src.spaces.actions import ActionBase
 
 
 class ActionSpace(Discrete):
+    """
+    ActionSpace class models a discrete action space of size n
+    """
 
     def __init__(self, n: int) -> None:
 
         super(ActionSpace, self).__init__(n=n)
+
+        # the list of actions the space contains
         self.actions = []
 
+    def __getitem__(self, item) -> ActionBase:
+        """
+        Returns the item-th action
+        :param item: The index of the action to return
+        :return: An action obeject
+        """
+        return self.actions[item]
+
+    def __setitem__(self, key: int, value: ActionBase) -> None:
+        """
+        Update the key-th Action with the new value
+        :param key: The index to the action to update
+        :param value: The new action
+        :return: None
+        """
+        self.actions[key] = value
+
     def add(self, action: ActionBase) -> None:
+        """
+        Add a new action in the space
+        :param action:
+        :return:
+        """
 
         if len(self.actions) >= self.n:
             raise ValueError("Action space is saturated. You cannot add a new action")
 
+        # set a valid id for the action
+        action.idx = len(self.actions)
         self.actions.append(action)
 
-    def add_may(self, *actions) -> None:
+    def add_many(self, *actions) -> None:
+        """
+        Add many actions in one go
+        :param actions: List of actions to add
+        :return: None
+        """
         for a in actions:
             self.add(action=a)
 
     def sample_and_get(self) -> ActionBase:
-
+        """
+        Sample the space and return an action to the application
+        :return: The sampled action
+        """
         action_idx = self.sample()
         return self.actions[action_idx]

src/spaces/action_space.py

@@ -33,8 +33,11 @@ class ActionBase(metaclass=abc.ABCMeta):
     Base class for actions
     """
 
-    def __init__(self, action_type: ActionType) -> None:
+    def __init__(self, column_name: str, action_type: ActionType) -> None:
+        self.column_name = column_name
         self.action_type = action_type
+        self.idx = None
+        self.key = (self.column_name, self.action_type)
 
     @abc.abstractmethod
     def act(self, **ops) -> None:
@@ -77,8 +80,8 @@ class ActionIdentity(ActionBase):
     Implements the identity action
     """
 
-    def __init__(self) -> None:
-        super(ActionIdentity, self).__init__(action_type=ActionType.IDENTITY)
+    def __init__(self, column_name: str) -> None:
+        super(ActionIdentity, self).__init__(column_name=column_name, action_type=ActionType.IDENTITY)
 
     def act(self, **ops):
         """
@@ -93,8 +96,8 @@ class ActionTransform(ActionBase):
     """
     Implements the transform action
     """
-    def __init__(self):
-        super(ActionTransform, self).__init__(action_type=ActionType.TRANSFORM)
+    def __init__(self, column_name: str):
+        super(ActionTransform, self).__init__(column_name=column_name, action_type=ActionType.TRANSFORM)
 
     def act(self, **ops):
         """
@@ -109,8 +112,8 @@ class ActionSuppress(ActionBase, _WithTable):
     """
     Implements the suppress action
     """
-    def __init__(self, suppress_table=None):
-        super(ActionSuppress, self).__init__(action_type=ActionType.SUPPRESS)
+    def __init__(self, column_name: str, suppress_table=None):
+        super(ActionSuppress, self).__init__(column_name=column_name, action_type=ActionType.SUPPRESS)
 
         if suppress_table is not None:
             self.table = suppress_table
@@ -136,36 +139,47 @@ def act(self, **ops) -> None:
         move_next(iterators=self.iterators)
 
 
-class ActionGeneralize(ActionBase):
+class ActionGeneralize(ActionBase, _WithTable):
     """
     Implements the generalization action
     """
 
-    def __init__(self):
-        super(ActionGeneralize, self).__init__(action_type=ActionType.GENERALIZE)
-        self.generalization_table = {}
+    def __init__(self, column_name: str, generalization_table: dict = None):
+        super(ActionGeneralize, self).__init__(column_name=column_name, action_type=ActionType.GENERALIZE)
+
+        if generalization_table is not None:
+            self.table = generalization_table
+
+        # fill in the iterators
+        self.iterators = [iter(self.table[item]) for item in self.table]
 
     def act(self, **ops):
         """
         Perform an action
         :return:
         """
+
+        # get the values of the column
+        col_vals = ops['data'].values
+
         # generalize the data given
-        for item in ops["data"]:
+        for i, item in enumerate(col_vals):
 
+            #print(item)
             # How do we update the generalizations?
-            value = self.generalization_table[item].value
-            item = value
+            value = self.table[item].value
+            col_vals[i] = value
 
-        # update the generalization
-        self._move_next()
+        ops["data"] = col_vals
 
-    def add_generalization(self, key: str, values: HierarchyBase) -> None:
-        self.generalization_table[key] = values
+        # update the generalization iterators
+        # so next time we visit we update according to
+        # the new values
+        move_next(iterators=self.iterators)
+        return ops['data']
 
-    def _move_next(self):
+    def add_generalization(self, key: str, values: HierarchyBase) -> None:
+        self.table[key] = values
 
-        for item in self.generalization_table:
-            next(self.generalization_table[item])
 
 

src/spaces/actions.py

@@ -12,7 +12,7 @@
 import multiprocessing as mp
 
 from src.exceptions.exceptions import Error
-from src.spaces.actions import ActionBase
+from src.spaces.actions import ActionBase, ActionType
 from src.utils.string_distance_calculator import DistanceType, TextDistanceCalculator
 
 DataSet = TypeVar("DataSet")
@@ -169,6 +169,19 @@ def reset(self, **options) -> TimeStep:
                                           observation=self.get_ds_as_tensor().float(), discount=self.gamma)
         return self.current_time_step
 
+    def apply_action(self, action: ActionBase):
+        """
+        Apply the action on the environment
+        :param action: The action to apply on the environment
+        :return:
+        """
+
+        if action.action_type == ActionType.IDENTITY:
+            return
+
+        # apply the transform of the data set
+        self.data_set.apply_column_transform(column_name=action.column_name, transform=action)
+
     def step(self, action: ActionBase) -> TimeStep:
         """
 
@@ -182,11 +195,23 @@ def step(self, action: ActionBase) -> TimeStep:
         `action` will be ignored.
         """
 
-        self.data_set.apply_transform(transform=action)
+        self.apply_action(action=action)
+
+        # if the action is identity don't bother
+        # doing anything
+        #if action.action_type == ActionType.IDENTITY:
+        #    return TimeStep(step_type=StepType.MID, reward=0.0,
+        #                    observation=self.get_ds_as_tensor().float(), discount=self.gamma)
+
+        # apply the transform of the data set
+        #self.data_set.apply_column_transform(transform=action)
 
         # perform the action on the data set
         self.prepare_column_states()
 
+        # calculate the information leakage and establish the reward
+        # to return to the agent
+
         return TimeStep(step_type=StepType.MID, reward=0.0,
                         observation=self.get_ds_as_tensor().float(), discount=self.gamma)