In [1]:
from transitions import Machine, State
import random
import numpy as np
from collections import Counter
In [2]:
class TrialSeq(object):
def __init__(self,N):
_CueTypes = [5,6]
_nCueTypes = len(_CueTypes)
_GoalIDs = [3,4,5,6]
_GoalIDsByCue = {5:[3,4],6:[5,6]}
_nGoalsByCue = {k: len(v) for k, v in _GoalIDsByCue.items()}
# increase N if not divisible by number of conditions
_nConds = 4
while N%_nConds != 0:
N = N+1
# seq refers to the cue sequence, indicating trial type.
_nTrialsPerCue = int(N/_nCueTypes)
_seq = _CueTypes*_nTrialsPerCue # create sequence of length N for CueTypes
_seq = np.array(np.random.permutation(_seq))
# assign goal to cues
_GoalSeq = np.zeros(N)-1
for cue in _CueTypes:
_sublist = _seq==cue
_nTrialsPerCueGoal = int(_nTrialsPerCue/_nGoalsByCue[cue])
_cuegoalseq = np.array(_GoalIDsByCue[cue]*_nTrialsPerCueGoal)
_cuegoalseq = np.random.permutation(_cuegoalseq)
_GoalSeq[_sublist] = _cuegoalseq
# another implementation of goal allocation, no restrain for # of goal
# for ii in range(N):
# for jj in _CueTypes:
# rr = random.random():
# for kk in range(_GoalIDsByCue[jj]):
# if rr < (kk+1)*(1/_nGoalsByCue[jj]):
# self.GoalSeq[ii] = _GoalIDsByCue[jj][kk]
# break
# del temp variables
del _sublist, _cuegoalseq
# visible variables
self.N = N
self.CueSeq = _seq
self.GoalSeq = _GoalSeq.astype(int)
self.CueIDs = _CueTypes
self.GoalIDsByCue = _GoalIDsByCue
self.CurrentTrial = 0
def CueID(self):
return self.CueSeq[self.CurrentTrial]
def GoalID(self):
return self.GoalSeq[self.CurrentTrial]
def NextTrial(self):
self.CurrentTrial = self.CurrentTrial + 1
def SeqCounts(self):
print("Number of trials = ", self.N)
print("Trial Cue Counts: ")
print(Counter(self.CueSeq))
print("Trial Goal Counts: ")
print(Counter(self.GoalSeq))
In [12]:
ts = TrialSeq(303)
ts.GoalSeq
Out[12]:
In [2]:
class Maze(object):
nWells = 6
nCues = 6
def __init__(self):
self.Act_Well = np.zeros(self.nWells,dtype=bool)
self.PrevAct_Well = np.zeros(self.nWells,dtype=bool)
self.Act_Cue = 0
self.DetectedWellNum = 0
self.WellDetectSeq = []
self.ValidWellDetectSeq = []
self.StateChangeFlag = 0
def activate_cue(self):
# send command to activate relevant cue
#self.Act_Cue
pass
def inactivate_cue(self):
# send command to inactivate cue
self.Act_Cue = 0
def activate_well(self,well):
if self.Act_Well[well]:
print('activated well', well+1)
#send command to arduino
def detect(self,well):
#well = event.kwargs.get('well',0)
well = well-1 # zero indexing the wells
self.DetectedWellNum = well
self.WellDetectSeq.append(well+1)
if self.Act_Well[well]==True:
self.Act_Well[well]=False
self.ValidWellDetectSeq.append(well+1)
def update_states(self):
well = int(self.state[2:])
self.PrevAct_Well = np.array(self.Act_Well)
if (well==99): #activate all
if all(self.Act_Well)==False:
self.Act_Well[:] = True
elif (well>=1 and well<=6):
if self.Act_Well[well-1] == False:
self.Act_Well[well-1] = True
elif (well==34):
if (self.Act_Well[2]==False):
self.Act_Well[2]=True
if (self.Act_Well[3]==False):
self.Act_Well[3]=True
elif (well==56):
if (self.Act_Well[4]==False):
self.Act_Well[4]=True
if (self.Act_Well[5]==False):
self.Act_Well[5]=True
else: # inactivate all
if any(self.Act_Well)==True:
self.Act_Well[:] = False
change_wells = np.array((self.PrevAct_Well != self.Act_Well).nonzero()).flatten()
for ii in change_wells:
self.activate_well(ii)
def next_trial(self):
pass
def G34(self):
if self.Act_Cue==5:
return True
return False
def G56(self):
if self.Act_Cue==6:
return True
return False
def G3(self):
pass
# if sequence.goal(current trial) == 3
# return true
def G4(self): pass
def G5(self): pass
def G6(self): pass
In [28]:
N = 100;
GoalSeq = []
RightCues = [1,2,5]
LeftCues = [3,4,6]
trialTypes = ['']
states = [State(name='AW0',on_enter=['inactivate_cue'],ignore_invalid_triggers=True),
State(name='AW1',on_enter='next_trial',on_exit=['activate_cue'],ignore_invalid_triggers=True),
State(name='AW2',ignore_invalid_triggers=True),
State(name='AW3',ignore_invalid_triggers=True),
State(name='AW4',ignore_invalid_triggers=True),
State(name='AW5',ignore_invalid_triggers=True),
State(name='AW6',ignore_invalid_triggers=True),
State(name='AW34',ignore_invalid_triggers=True),
State(name='AW56',ignore_invalid_triggers=True)
]
conditions = ['G3','G4','G5','G6','G34','G56']
transitions = [
{'trigger':'D1','source':'AW1','dest':'AW2'},
{'trigger':'D2','source':'AW2','dest':'AW34', 'conditions':'G34','after':'inactivate_cue'},
{'trigger':'D2','source':'AW2','dest':'AW56', 'conditions':'G56','after':'inactivate_cue'},
{'trigger':'D2','source':'AW2','dest':'AW0'},
{'trigger':'D3','source':'AW34','dest':'AW4'},
{'trigger':'D3','source':'AW3','dest':'AW1'},
{'trigger':'D4','source':'AW34','dest':'AW3'},
{'trigger':'D4','source':'AW4','dest':'AW1'},
{'trigger':'D5','source':'AW5','dest':'AW1'},
{'trigger':'D5','source':'AW56','dest':'AW6'},
{'trigger':'D6','source':'AW6','dest':'AW1'},
{'trigger':'D6','source':'AW56','dest':'AW5'},
{'trigger':'stop','source':'*','dest':'AW0'}]
In [55]:
MS = Maze()
machine = Machine(MS,states,transitions=transitions, ignore_invalid_triggers=True ,initial='AW0',
after_state_change='update_states')
# list of detections
dlist = [1,2,3,4,1,2,4,3,1,2,3,5,6]
goals = [1,2,3,4,5,6]
cuelist = [5,5,6,6,5,6]
detect_callback = []
MS.Act_Cue=5
for ii in goals:
detect_callback.append(getattr(MS,'D'+str(ii)))
MS.to_AW1()
trialcount =0
for ii in dlist:
print(MS.state)
if ii==1 and MS.state==1:
MS.activate_cue=cuelist(trialcount)
trialcount=trialcount+1
MS.detect(ii)
detect_callback[ii-1]()
In [54]:
help(MS.trigger)
In [34]:
# thread 1
# gets arduino data
# runs continouesly, no printing to command line
# writes events codes and timing to text file
# thread 2
# get user input data
# if there is user input trigger a flag
# store user commands
# thread 3
# ongoing loop getting detect events from GPIO
# act on user inputs, if any (checks input flag), turn off flag
# x = dected well
# f=getattr(machine,'D'+str(x));
# f(x); triggers the transition and passes which well was detected.
In [36]:
Out[36]:
In [38]:
Out[38]: