Initial commit as a python_actr package

Author: tcstewar

python_actr/__init__.py

@@ -0,0 +1,6 @@
+from .model import Model, log_everything
+from .production import ProductionSystem
+from .logger import log, finished
+from .runner import run, run_with
+from .actr import *
+from .display import display

python_actr/actr/__init__.py

@@ -0,0 +1,12 @@
+from .buffer import Buffer, Chunk
+from .dm import *
+from .pm import *
+from .compile import PMCompile
+
+from .core import ACTR
+from .timer import Timer
+from .text import TextOutput
+from .imaginal import ImaginalModule
+from .vision import Vision
+from .sosvision import SOSVision
+from .motor import Motor

python_actr/actr/buffer.py

@@ -0,0 +1,70 @@
+import python_actr
+
+from collections import UserDict
+
+class Chunk(UserDict):
+  def __init__(self,contents,bound=None):
+    UserDict.__init__(self)
+    if isinstance(contents,Chunk):
+      self.update(contents)
+    elif isinstance(contents,str):
+      for i,x in enumerate(contents.split()):
+        if ':' in x:
+          i,x=x.split(':',1)
+        if x.startswith('?'):
+          key=x[1:]
+          x=bound[key]
+        self[i]=x
+    elif hasattr(contents,'__dict__'):  
+      for k,v in list(contents.__dict__.items()):
+        if type(v) in [str,float,int,bool]:
+          self[k]=v
+    else:
+      try:
+        self.update(contents)
+      except:  
+        raise Exception('Unknown contents for chunk:',contents)      
+  def __repr__(self):
+    #
+    r=[]
+    keys=list(self.keys())
+    i=0
+    while i in keys:
+      r.append('%s'%self[i])      
+      keys.remove(i)
+      i+=1
+    keys.sort()
+    for k in keys:
+      if k[0]!='_':
+        r.append('%s:%s'%(k,self[k]))
+    return ' '.join(r)  
+      
+
+class Buffer(python_actr.Model):
+  def __init__(self):
+    self.chunk=None
+  def set(self,chunk):
+    try:
+      self.chunk=Chunk(chunk,self.sch.bound)
+    except AttributeError:
+      self.chunk=Chunk(chunk,{})  
+  def modify(self,**args):
+    for k,v in list(args.items()):
+      if k.startswith('_'): k=int(k[1:])
+      if k not in self.chunk:
+        raise Exception('No slot "%s" to modify to "%s"'%(k,v))
+      self.chunk[k]=v
+      self.chunk=self.chunk
+  def __getitem__(self,key):
+    return self.chunk[key]
+  def clear(self):
+    self.chunk=None  
+  def __eq__(self,other):
+    return self.chunk==other
+  def __hash__(self):
+    return id(self)
+  def __len__(self):
+    if self.chunk is None: return 0
+    return len(self.chunk)
+  def isEmpty(self):
+    return len(self)==0

python_actr/actr/compile.py

@@ -0,0 +1,128 @@
+from python_actr.actr.pm import ProceduralSubModule
+from python_actr.production import Production
+from python_actr.pattern import Pattern
+
+class CompiledProduction(Production):
+    def __init__(self,pre,post,keep,retrieve,pre_bound,post_bound):
+        self.name='%s-%s-%d'%(pre.name,post.name,id(self))
+        self.system=pre.system
+        self.base_utility=0
+        self.bound=None
+
+        code1=[]
+        for k,v in list(pre_bound.items()):
+            code1.append(' %s=%s'%(k,repr(v)))
+        code1.append(' if True:  # compiled from %s'%pre.name)    
+        for line in pre.code.splitlines():
+            for k in keep:
+                if line.strip().startswith(k):
+                    code1.append('  '+line)
+                    break
+
+        code1='\n'.join(code1)
+        for k,v in list(pre_bound.items()):
+            code1=code1.replace('?'+k,v)
+
+
+            
+        code2=[]    
+        for k,v in list(post_bound.items()):
+            code2.append(' %s=%s'%(k,repr(v)))
+        code2.append(' if True:  # compiled from %s'%post.name)
+        for line in post.code.splitlines():
+            if len(line.strip())>0:
+                code2.append('  '+line)
+
+        code2='\n'.join(code2)
+        for k,v in list(post_bound.items()):
+            code2=code2.replace('?'+k,v)
+
+        self.code='if True:\n%s\n%s'%(code1,code2)
+
+        self.func=compile(self.code,'<production-%s>'%self.name,'exec')
+
+        
+        keys=list(pre.keys)
+        patterns={}
+        for buf,pat in list(pre.pattern_specs.items()):
+            for k,v in list(pre_bound.items()):
+                pat=pat.replace('?'+k,v)
+            patterns[buf]=pat
+            
+        for m in post.keys:
+            if m==retrieve: pass
+            elif m not in keys:
+                keys.append(m)
+                pat=post.pattern_specs[m]
+                for k,v in list(post_bound.items()):
+                    pat=pat.replace('?'+k,v)
+                patterns[buf]=pat
+                
+        self.keys=keys
+        self.pattern_specs=patterns
+        self.pattern=Pattern(patterns)
+       
+      
+
+        
+
+class PMCompile(ProceduralSubModule):
+    def __init__(self,keep,request,retrieve):
+        if not isinstance(keep,(list,tuple)): keep=(keep,)
+        self.keep=keep
+        self.request=request
+        self.retrieve=retrieve
+        self.log=None
+        self.pre=[]
+        self.post=[]
+        self._previous=None
+        self.compiled={}
+    def create(self,prod,parents=None):
+        if self.retrieve in prod.keys:
+            for m in prod.keys:
+                if m not in self.keep and m!=self.retrieve:
+                    break
+            else:
+                self.post.append(prod)
+
+        code=prod.code
+        good=False
+        for line in code.splitlines():
+            line=line.strip()
+            if len(line)>0:
+                keep=False
+                if line.startswith(self.request):
+                    good=True
+                    continue
+                for k in self.keep:
+                    if line.startswith(k):
+                        keep=True
+                        break
+                if keep:
+                    continue
+                else:
+                    good=False
+                    break
+        if good:
+            self.pre.append(prod)
+    def firing(self,prod):
+        if self._previous is not None and prod in self.post:
+            self.compile(self._previous,self._previousBound,prod,prod.bound)
+        self._previous=None
+        if prod in self.pre:
+            self._previous=prod
+            self._previousBound=dict(prod.bound)
+            
+    def compile(self,pre,pre_bound,post,post_bound):
+        id=(pre,post,tuple(sorted(pre_bound.items())),tuple(sorted(post_bound.items())))
+        p=self.compiled.get(id,None)
+        if p is None:
+            p=CompiledProduction(pre,post,self.keep,self.retrieve,pre_bound,post_bound)
+            self.compiled[id]=p
+            #print p.name
+            #print p.code
+
+            for a in self.parent._adaptors:
+                a.create(p,parents=[pre,post])
+            self.parent._productions.append(p)
+        

python_actr/actr/core.py

@@ -0,0 +1,92 @@
+
+import python_actr
+import collections
+
+class ACTR(python_actr.ProductionSystem):
+    production_time_sd=None
+    production_threshold=None
+    
+    def __init__(self,log=None):
+        #ccm.ProductionSystem.__init__(self,log=log)
+        #print(self,"ACTR")
+        super(ACTR,self).__init__(log=log)
+        self._adaptors=[]
+    
+    def _process_productions(self):
+        self._calc_context()          
+        for i in self._initializers:
+            i.fire(self._context)
+        while True:
+            if self.production_match_delay>0: yield self.production_match_delay 
+            match=[p for p in self._productions if p.match(self._context)]
+            if len(match)==0:
+                yield self._top.changes
+            else:
+                activations=[self.get_activation(p) for p in match]
+                a=max(activations)
+                
+                threshold=self.production_threshold
+                if isinstance(threshold, collections.Callable): threshold=threshold()
+                
+                if threshold is not None and a<threshold:
+                    for a in self._adaptors: a.below_threshold()
+                    yield self._top.changes
+                    continue    
+                options=[p for (i,p) in enumerate(match) if activations[i]==a]
+                choice=self.random.choice(options)
+                
+                for a in self._adaptors: a.selecting(choice)
+                #self.log.selected=choice.name
+                self.log.production=choice.name
+                
+                t=self.production_time
+                if isinstance(t, collections.Callable): t=t()
+                if self.production_time_sd is not None:
+                    t=t+self.random.gauss(0,self.production_time_sd)
+                t-=self.production_match_delay
+                if t<0: t=0
+                yield t
+                
+                if not choice.match(self._context):
+                  #self.log.change_detected='before firing '+choice.name
+                  self.log.production='(changed before firing)'
+                else:  
+                  for a in self._adaptors: a.firing(choice)
+                  self.log.production=None
+                  #self.log.firing=choice.name
+                  choice.fire(self._context)
+                  yield dict(delay=0,priority=-1000)      # delay so we don't try to match again until after the result of production firing has had a chance to occur
+    
+    def add_adaptor(self,module):
+        self._adaptors.append(module)
+        for p in self._productions:
+            module.create(p)
+            
+    def reward(self,value):
+        for a in self._adaptors:
+            a.reward(value)
+    def success(self):
+        self.reward(1)
+    def failure(self):
+        self.reward(-1)            
+                
+                
+
+    def get_activation(self,production=None):
+        if production is None:
+            r={}
+            for p in self._productions:
+                r[p.name]=self.get_activation(p)
+            return r
+        elif isinstance(production,str):
+            for p in self._productions:
+                if p.name==production:
+                    return self.get_activation(p)
+            return None
+        else:
+            activation=production.base_utility
+            for a in self._adaptors:
+                activation+=a.utility(production)
+            return activation
+            
+