PythonCMCed/ex2_sandwich_genericmotor.py at main · rlwest/PythonCMCed · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
# uses a generic motor production to do all actions

from utility import Utility
from production_cycle import ProductionCycle

# Initialize memories
working_memory = {'focus_buffer': {'state': 'bread1'}, 'motor_buffer': {'state': 'no_action'}}
environment_memory = {'bread1': {'location': 'counter'},
                      'cheese': {'location': 'counter'},
                      'ham': {'location': 'counter'},
                      'bread2': {'location': 'counter'}
                      }

memories = {
    'working_memory': working_memory,
    'environment_memory': environment_memory,
}

# Initialize productions
ProceduralProductions = []
MotorProductions = []


# Procedural Production to move bread1
def bread1(memories):
    motorbuffer = memories['working_memory']['motor_buffer']
    motorbuffer.update({
        'state': 'do_action',
        'env_object': 'bread1',
        'slot': 'location',
        'newslotvalue': 'plate',
        'delay': 3
    })
    memories['working_memory']['focus_buffer']['state'] = 'cheese'
    print(f"bread1 executed. Updated working_memory: {memories['working_memory']}")
ProceduralProductions.append({
    'matches': {'working_memory': {'focus_buffer': {'state': 'bread1'}, 'motor_buffer': {'state': 'no_action'}}},
    'negations': {},
    'utility': 10,
    'action': bread1,
    'report': "bread1",
})


# Procedural Production to move cheese after bread1
def cheese(memories):
    motorbuffer = memories['working_memory']['motor_buffer']
    motorbuffer.update({
        'state': 'do_action',
        'env_object': 'cheese',
        'slot': 'location',
        'newslotvalue': 'plate',
        'delay': 3
    })
    memories['working_memory']['focus_buffer']['state'] = 'ham'
    print(f"cheese executed. Updated working_memory: {memories['working_memory']}")
ProceduralProductions.append({
    'matches': {'working_memory': {'focus_buffer': {'state': 'cheese'}, 'motor_buffer': {'state': 'no_action'}}},
    'negations': {},
    'utility': 10,
    'action': cheese,
    'report': "cheese",
})


# Procedural Production to move ham after cheese
def ham(memories):
    motorbuffer = memories['working_memory']['motor_buffer']
    motorbuffer.update({
        'state': 'do_action',
        'env_object': 'ham',
        'slot': 'location',
        'newslotvalue': 'plate',
        'delay': 3
    })
    memories['working_memory']['focus_buffer']['state'] = 'bread2'
    print(f"ham executed. Updated working_memory: {memories['working_memory']}")
ProceduralProductions.append({
    'matches': {'working_memory': {'focus_buffer': {'state': 'ham'}, 'motor_buffer': {'state': 'no_action'}}},
    'negations': {},
    'utility': 10,
    'action': ham,
    'report': "ham",
})


# Procedural Production to move bread2 after ham
def bread2(memories):
    motorbuffer = memories['working_memory']['motor_buffer']
    motorbuffer.update({
        'state': 'do_action',
        'env_object': 'bread2',
        'slot': 'location',
        'newslotvalue': 'plate',
        'delay': 3
    })
    memories['working_memory']['focus_buffer']['state'] = 'done'
    print(f"bread2 executed. Updated working_memory: {memories['working_memory']}")
ProceduralProductions.append({
    'matches': {'working_memory': {'focus_buffer': {'state': 'bread2'}, 'motor_buffer': {'state': 'no_action'}}},
    'negations': {},
    'utility': 10,
    'action': bread2,
    'report': "bread2",
})


# Procedural Production to announce the sandwich is ready
def announce_sandwich(memories):
    print("Ham and cheese sandwich is ready!")
ProceduralProductions.append({
    'matches': {'working_memory': {'focus_buffer': {'state': 'done'}, 'motor_buffer': {'state': 'no_action'}}},
    'negations': {},
    'utility': 10,
    'action': announce_sandwich,
    'report': "announce_sandwich",
})


# Motor Production to move item
def move_item(memories):
    motorbuffer = memories['working_memory']['motor_buffer']
    env_object = motorbuffer['env_object']
    slot = motorbuffer['slot']
    newslotvalue = motorbuffer['newslotvalue']
    delay = motorbuffer['delay']

    memories['working_memory']['motor_buffer']['state'] = 'moving'
    print(f"move_item executed for {env_object}. Updated working_memory: {memories['working_memory']}")
    print(f'set action completion for {delay} cycles later')
    return delay


def delayed_action(memories):
    motorbuffer = memories['working_memory']['motor_buffer']
    env_object = motorbuffer['env_object']
    slot = motorbuffer['slot']
    newslotvalue = motorbuffer['newslotvalue']
    memories['environment_memory'][env_object][slot] = newslotvalue
    memories['working_memory']['motor_buffer']['state'] = 'no_action'
    print(f"delayed_action executed. Updated environment_memory: {memories['environment_memory']}")
    print(f"delayed_action executed. Updated working_memory: {memories['working_memory']}")
MotorProductions.append({
    'matches': {'working_memory': {'motor_buffer': {'state': 'do_action'}}},
    'negations': {},
    'utility': 10,
    'action': move_item,
    'report': "move_item",
    'delayed_action': delayed_action
})


# Production system delays in ticks
ProductionSystem1_Countdown = 1
ProductionSystem2_Countdown = 1

# Stores the number of cycles for a production system to fire and reset
DelayResetValues = {
    'ProductionSystem1': ProductionSystem1_Countdown,
    'ProductionSystem2': ProductionSystem2_Countdown
}

# Dictionary of all production systems and delays
AllProductionSystems = {
    'ProductionSystem1': [ProceduralProductions, ProductionSystem1_Countdown],
    'ProductionSystem2': [MotorProductions, ProductionSystem2_Countdown]
}

# Initialize ProductionCycle
ps = ProductionCycle()

# Run the cycle with custom parameters
ps.run_cycles(memories, AllProductionSystems, DelayResetValues, cycles=22, millisecpercycle=10)