55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
from actor import Actor
|
|
import math
|
|
|
|
|
|
class XorScape(Actor):
|
|
def __init__(self):
|
|
super().__init__("XorScape")
|
|
self.data = [
|
|
([-1, -1], [-1]),
|
|
([1, -1], [1]),
|
|
([-1, 1], [1]),
|
|
([1, 1], [-1])
|
|
]
|
|
self.index = 0
|
|
self.error_acc = 0.0
|
|
self.last_actuator = None
|
|
|
|
async def run(self):
|
|
while True:
|
|
msg = await self.inbox.get()
|
|
tag = msg[0]
|
|
|
|
if tag == "sense":
|
|
print("SCAPE: got sensed by sensor...")
|
|
_, sid, from_pid = msg
|
|
self.last_actuator = from_pid
|
|
inputs, correct = self.data[self.index]
|
|
print("SENSOR input /correct: ", inputs, correct)
|
|
await from_pid.send(("percept", inputs))
|
|
|
|
elif tag == "action":
|
|
_, output, from_pid = msg
|
|
_, correct = self.data[self.index]
|
|
step_error = sum((o - c) ** 2 for o, c in zip(output, correct))
|
|
step_rmse = math.sqrt(step_error)
|
|
|
|
print(f"XOR PATTERN: Input={self.data[self.index][0]} → Network={output} → Expected={correct}")
|
|
|
|
self.index += 1
|
|
|
|
if self.index >= len(self.data):
|
|
|
|
total_rmse = math.sqrt(self.error_acc + step_rmse)
|
|
fitness = 1.0 / (total_rmse + 1e-5)
|
|
await from_pid.send(("result", fitness, 1))
|
|
self.index = 0
|
|
self.error_acc = 0.0
|
|
else:
|
|
# Continue episode
|
|
self.error_acc += step_rmse
|
|
await from_pid.send(("result", 0.0, 0))
|
|
|
|
elif tag == "terminate":
|
|
return
|