Probability Enhanced Attributes (#53)

Author: khozzy

examples/acs2/maze/acs2_in_maze.py

@@ -26,6 +26,7 @@ def maze_metrics(population, environment):
     parser.add_argument("-e", "--environment", default="Maze4-v0")
     parser.add_argument("--epsilon", default=1.0, type=float)
     parser.add_argument("--ga", action="store_true")
+    parser.add_argument("--pee", action="store_true")
     parser.add_argument("--explore-trials", default=50, type=int)
     parser.add_argument("--exploit-trials", default=10, type=int)
     args = parser.parse_args()
@@ -37,6 +38,7 @@ def maze_metrics(population, environment):
     cfg = Configuration(8, 8,
                         epsilon=args.epsilon,
                         do_ga=args.ga,
+                        do_pee=args.pee,
                         metrics_trial_frequency=1,
                         user_metrics_collector_fcn=maze_metrics)
 

lcs/agents/ImmutableSequence.py

@@ -4,15 +4,15 @@
 class ImmutableSequence:
 
     WILDCARD = '#'
-    OK_TYPES = (str, )
+    OK_TYPES = (str, dict)  # PEEs are stored in dict
 
     def __init__(self, observation):
-        assert type(self.WILDCARD) in self.OK_TYPES
+        obs = tuple(observation)
 
-        for attr in observation:
-            assert type(attr) in self.OK_TYPES
+        assert type(self.WILDCARD) in self.OK_TYPES
+        assert all(isinstance(o, self.OK_TYPES) for o in obs)
 
-        self._items = tuple(observation)
+        self._items = obs
 
     @classmethod
     def empty(cls, length: int):
@@ -45,7 +45,7 @@ def __getitem__(self, index):
         return self._items[index]
 
     def __setitem__(self, index, value):
-        assert type(value) in self.OK_TYPES
+        assert isinstance(value, self.OK_TYPES)
         lst = list(self._items)
         lst[index] = value
 

lcs/agents/acs2/Classifier.py

@@ -6,6 +6,7 @@
 
 from lcs import Perception
 from . import Configuration, Condition, Effect, PMark, matching
+from . import ProbabilityEnhancedAttribute
 
 
 logger = logging.getLogger(__name__)
@@ -139,21 +140,26 @@ def fitness(self):
     @property
     def specified_unchanging_attributes(self) -> List[int]:
         """
-        Determines the number of specified unchanging attributes in
-        the classifier. An unchanging attribute is one that is anticipated
-        not to change in the effect part.
+        Determines the specified unchanging attributes in the classifier.
+        An unchanging attribute is one that is anticipated not to change
+        in the effect part.
 
         Returns
         -------
         List[int]
-            list specified unchanging attributes indices
+            list of specified unchanging attributes indices
         """
         indices = []
 
         for idx, (cpi, epi) in enumerate(zip(self.condition, self.effect)):
-            if cpi != self.cfg.classifier_wildcard and \
-                    epi == self.cfg.classifier_wildcard:
-                indices.append(idx)
+            if isinstance(epi, ProbabilityEnhancedAttribute):
+                if cpi != self.cfg.classifier_wildcard and \
+                        epi.does_contain(cpi):
+                    indices.append(idx)
+            else:
+                if cpi != self.cfg.classifier_wildcard and \
+                        epi == self.cfg.classifier_wildcard:
+                    indices.append(idx)
 
         return indices
 
@@ -178,6 +184,9 @@ def is_reliable(self) -> bool:
     def is_inadequate(self) -> bool:
         return self.q < self.cfg.theta_i
 
+    def is_enhanceable(self):
+        return self.ee
+
     def increase_experience(self) -> int:
         self.exp += 1
         return self.exp
@@ -206,16 +215,62 @@ def specialize(self,
             Requires the effect attribute to be a wildcard to specialize it.
             By default false
         """
-        for idx, item in enumerate(situation):
+        for idx in range(len(situation)):
             if leave_specialized:
                 if self.effect[idx] != self.cfg.classifier_wildcard:
                     # If we have a specialized attribute don't change it.
                     continue
 
             if previous_situation[idx] != situation[idx]:
-                self.effect[idx] = situation[idx]
+                if self.effect[idx] == self.cfg.classifier_wildcard:
+                    self.effect[idx] = situation[idx]
+                else:
+                    if not isinstance(self.effect[idx],
+                                      ProbabilityEnhancedAttribute):
+                        self.effect[idx] = ProbabilityEnhancedAttribute(
+                            self.effect[idx])
+                    self.effect[idx].insert_symbol(situation[idx])
+
                 self.condition[idx] = previous_situation[idx]
 
+    def merge_with(self, other_classifier, perception, time):
+        assert self.cfg.do_pee
+
+        result = Classifier(cfg=self.cfg)
+
+        result.condition = Condition(self.condition)
+        result.condition.specialize_with_condition(other_classifier.condition)
+
+        # action is an int, so we can assign directly
+        result.action = self.action
+
+        result.effect = Effect.enhanced_effect(
+            self.effect, other_classifier.effect,
+            self.q, other_classifier.q,
+            perception)
+
+        result.mark = PMark(cfg=self.cfg)
+
+        result.r = (self.r + other_classifier.r) / 2.0
+        result.q = (self.q + other_classifier.q) / 2.0
+
+        # This 0.5 is Q_INI constant in the original C++ code
+        if result.q < 0.5:
+            result.q = 0.5
+
+        result.num = 1
+        result.tga = time
+        result.talp = time
+        result.tav = 0
+        result.exp = 1
+
+        result.ee = False
+
+        return result
+
+    def reverse_increase_quality(self):
+        self.q = (self.q - self.cfg.beta) / (1.0 - self.cfg.beta)
+
     def predicts_successfully(self,
                               p0: Perception,
                               action: int,
@@ -270,14 +325,18 @@ def does_anticipate_correctly(self,
             False otherwise
         """
         def effect_item_is_correct(effect_item, p0_item, p1_item):
-            if effect_item == self.cfg.classifier_wildcard:
-                if p0_item != p1_item:
-                    return False
+            if not isinstance(effect_item, ProbabilityEnhancedAttribute):
+                if effect_item == self.cfg.classifier_wildcard:
+                    if p0_item != p1_item:
+                        return False
+                else:
+                    if p0_item == p1_item:
+                        return False
+
+                    if effect_item != p1_item:
+                        return False
             else:
-                if p0_item == p1_item:
-                    return False
-
-                if effect_item != p1_item:
+                if not effect_item.does_contain(p1_item):
                     return False
 
             # All checks passed

lcs/agents/acs2/ClassifiersList.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import random
+import logging
 from itertools import chain
 from typing import Optional, List
 import lcs.agents.acs2.components.alp as alp_acs2
@@ -66,6 +67,41 @@ def get_maximum_fitness(self) -> float:
 
         return 0.0
 
+    @staticmethod
+    def apply_enhanced_effect_part_check(action_set: ClassifiersList,
+                                         new_list: ClassifiersList,
+                                         previous_situation: Perception,
+                                         time: int,
+                                         cfg: Configuration):
+        # Create a list of candidates.
+        # Every enhanceable classifier is a candidate.
+        candidates = [classifier for classifier in action_set
+                      if classifier.is_enhanceable()]
+
+        logging.debug(
+            "Applying enhanced effect part; number of candidates={}; " +
+            "previous situation: {}".format(
+                len(candidates), previous_situation))
+
+        # If there are less than 2 candidates, don't do it
+        if len(candidates) < 2:
+            return
+
+        for candidate in candidates:
+            candidates2 = [classifier for classifier in candidates
+                           if candidate != classifier]
+            if len(candidates2) > 0:
+                merger = random.choice(candidates2)
+                new_classifier = candidate.merge_with(merger,
+                                                      previous_situation,
+                                                      time)
+                if new_classifier is not None:
+                    candidate.reverse_increase_quality()
+                    alp.add_classifier(new_classifier, action_set, new_list,
+                                       cfg.theta_exp)
+
+        return new_list
+
     @staticmethod
     def apply_alp(population: ClassifiersList,
                   match_set: ClassifiersList,
@@ -125,6 +161,13 @@ def apply_alp(population: ClassifiersList,
                 new_cl.tga = time
                 alp.add_classifier(new_cl, action_set, new_list, theta_exp)
 
+        if cfg.do_pee:
+            ClassifiersList.apply_enhanced_effect_part_check(action_set,
+                                                             new_list,
+                                                             p0,
+                                                             time,
+                                                             cfg)
+
         # No classifier anticipated correctly - generate new one
         if not was_expected_case:
             new_cl = alp_acs2.cover(p0, action, p1, time, cfg)

lcs/agents/acs2/Configuration.py

@@ -12,6 +12,7 @@ def __init__(self,
                  user_metrics_collector_fcn: Callable = None,
                  fitness_fcn=None,
                  metrics_trial_frequency: int = 5,
+                 do_pee: bool = False,
                  do_ga: bool = False,
                  do_subsumption: bool = True,
                  do_action_planning: bool = False,
@@ -38,6 +39,12 @@ def __init__(self,
         :param environment_adapter: EnvironmentAdapter class ACS2 needs to use
             to interact with the environment
         :param fitness_fcn: Custom fitness function
+        :param do_pee: switch *Probability-Enhanced Effects*.
+            This is the mechanism described and implemented in C++
+            in Martin V. Butz, David E. Goldberg, Wolfgang Stolzmann,
+            "Probability-Enhanced Predictions in the Anticipatory Classifier
+             System", University of Illinois at Urbana-Champaign:
+            Illinois Genetic Algorithms Laboratory, Urbana, 2000.
         :param do_ga: switch *Genetic Generalization* module
         :param do_subsumption:
         :param do_action_planning: switch Action Planning phase
@@ -70,6 +77,7 @@ def __init__(self,
         self.environment_adapter = environment_adapter
         self.metrics_trial_frequency = metrics_trial_frequency
         self.user_metrics_collector_fcn = user_metrics_collector_fcn
+        self.do_pee = do_pee
         self.fitness_fcn = fitness_fcn
         self.do_ga = do_ga
         self.do_subsumption = do_subsumption