voxel mountain
学校は最悪なこともある。多くの人がそう思っている。でも、学校は学ぶための最良の方法ではない。伝統的な学校はたった200年前に始まっただけで、人類は何千年も存在していた。学校は個々の興味に合わないことも多い、硬直したカリキュラムを押し付ける。『無駄』なことを強制的に学ばせるのは自然な好奇心を殺す。自分で学ぶことは、自分の情熱や目的を発見する助けになる—学校はむしろその逆をすることが多い。授業で受け身で聞くだけよりも、実際に手を動かして学ぶ方がはるかに効果的。宿題や成績は余計なストレスを生み、学ぶことが義務のように感じられてしまう。好奇心があると、強制的に暗記するよりずっと速く情報を吸収できる。独学で得たスキル（例えばプログラミング！）は充実したキャリアに繋がることが多い。自由に探求することで創造力が育まれる—これは学校ではよく抑えられてしまう。詳しくはこちら：
なぜ遊ぶ本能を解放すると子どもはより幸せで自立し、長く良い学習者になるのか
啓蒙時代後の自己教育
学生の教育の自由の権利
教育に反対する理由
ホームスクーリングに関する研究レビュー
doesn't work, A1p can't predict in correct position. no categories in A1. LIPp doesn't even predict position correctly. it seems like connection from LIPp to LIPct interferes with learning, but it is needed for LIPct to represent predicted position.
import torch
from torch import tensor
import torch.nn as nn
import torch.nn.functional as F
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('macosx')
import math
lr = 0.01
adaptrate = 0.03
class DenseConnection:
def __init__(self, w,scale):
self.w=w
self.scale = scale
self.off = False
def make(sender,receiver,scale=1.0):
con = DenseConnection(torch.rand(sender.size,receiver.size),scale/sender.size)
receiver.connect(sender,con)
return con
def make2(sender,receiver,scale=1.0):
w=torch.rand(sender.size,receiver.size)
con = DenseConnection(w,scale/sender.size)
con2 = DenseConnection(w.T,scale/receiver.size)
receiver.connect(sender,con)
sender.connect(receiver,con2)
return con,con2
def forward(self, x): return [email protected]*self.scale
def updateWeight(self, sender,receiver,adaptAct=True):
change = sender.act[:,None]*receiver.act[None,:]*1.01 - sender.minusPhaseAct[:,None]*receiver.minusPhaseAct[None,:]*0.99
self.w += change*lr
self.w.clamp(0,1)
if adaptAct:
self.w += (torch.where(receiver.targetDiff[None,:]<0.0, 0.0, 1.0) - self.w) * (receiver.targetDiff[None,:].abs()) * adaptrate
def updateWeightSpecific(self, senderMinusPhase,senderPlusPhase,receiver,adaptAct=True):
change = senderPlusPhase[:,None]*receiver.act[None,:]*1.01 - senderMinusPhase[:,None]*receiver.minusPhaseAct[None,:]*0.99
self.w += change*lr
self.w.clamp(0,1)
if adaptAct:
self.w += (torch.where(receiver.targetDiff[None,:]<0.0, 0.0, 1.0) - self.w) * (receiver.targetDiff[None,:].abs()) * adaptrate
class LocalPoolConnection:
"""
Supports different strides, different pool sizes, and different neuron counts
in sender and receiver layers.
Structure:
- Sender: (S_px, S_py, S_neurons)
- Receiver: (R_px, R_py, R_neurons)
- Strides: (stride_x, stride_y) in terms of receiver pools per sender pool.
Actually, it's easier to define:
Sender pool (i, j) connects to Receiver pools in a local neighborhood.
But usually, local connection means Receiver pool (r, c) connects to
Sender pools (r+dx, c+dy).
Let's define strides as:
- Sender stride: distance between sender pool centers in terms of sender pool grid.
- Receiver stride: distance between receiver pool centers.
Actually, standard "local connection" with different strides:
If stride_x=1, receiver pool r connects to sender pools r, r+1, r+2...
If stride_x=2, receiver pool r connects to sender pools r*2, r*2+1, r*2+2...
Let's use:
- Sender grid size: (S_px, S_py)
- Receiver grid size: (R_px, R_py)
- Neurons per sender pool: S_n
- Neurons per receiver pool: R_n
- Neighborhood size: neighborhood_size (e.g., 1 for 3x3, 2 for 5x5)
For each receiver pool (rx, ry), it connects to sender pools:
(rx * stride_x + dx, ry * stride_y + dy) for dx, dy in neighborhood
Where stride_x, stride_y are the strides.
"""
def __init__(self, lw, sw, scale, sender_indices, receiver_indices,
s_px, s_py, s_neurons, r_px, r_py, r_neurons,
stride_x, stride_y, neighborhood_size=1):
"""
Args:
w: 1D tensor of weights
scale: scaling factor
sender_indices: 1D long tensor, sender neuron indices
receiver_indices: 1D long tensor, receiver neuron indices
s_px, s_py: sender grid dimensions
s_neurons: neurons per sender pool
r_px, r_py: receiver grid dimensions
r_neurons: neurons per receiver pool
stride_x, stride_y: strides
neighborhood_size: size of neighborhood
"""
assert float(r_px)*float(stride_x)==float(s_px) and float(r_py)*float(stride_y)==float(s_py), "stride is wrong"
self.lw = lw
self.sw = sw
self.w = lw*sw
self.scale = scale
self.sender_indices = sender_indices
self.receiver_indices = receiver_indices
self.s_px = s_px
self.s_py = s_py
self.s_neurons = s_neurons
self.r_px = r_px
self.r_py = r_py
self.r_neurons = r_neurons
self.stride_x = stride_x
self.stride_y = stride_y
self.neighborhood_size = neighborhood_size
self.off = False
@classmethod
def make(cls, sender, receiver,
stride_x=1, stride_y=1, scale=1.0, neighborhood_size=1):
"""
Creates a local pool connection.
"""
#center_size_x = 1 if stride_x<1 else int(stride_x)
#center_size_y = 1 if stride_y<1 else int(stride_y)
#neighborhood_size_sender_x = neighborhood_size if stride_x<1 else neighborhood_size*stride_x
#neighborhood_size_sender_y = neighborhood_size if stride_y<1 else neighborhood_size*stride_y
offsets = [(dx, dy) for dx in range(-neighborhood_size, neighborhood_size+1)
for dy in range(-neighborhood_size, neighborhood_size+1)]
sender_indices_list = []
receiver_indices_list = []
r_px, r_py, r_neurons = receiver.shape
s_px, s_py, s_neurons = sender.shape
for rx in range(r_px):
for ry in range(r_py):
for dx, dy in offsets:
# Sender pool coordinates
spx = math.floor(rx * stride_x) + dx
spy = math.floor(ry * stride_y) + dy
if 0 <= spx < s_px and 0 <= spy < s_py:
# Connect each neuron in sender pool to each neuron in receiver pool
# This creates a dense local connection within the neighborhood
# But if s_neurons != r_neurons, we need to handle it.
# Let's assume full connectivity between sender and receiver neurons in the neighborhood
# Or pointwise? The original code did pointwise: n sender -> n receiver
# So we connect sender neuron (spx, spy, n) to receiver neuron (rx, ry, n)
for n in range(s_neurons):
# Calculate base indices for the pool connection
base_sender_idx = spx * (s_py * s_neurons) + spy * s_neurons
base_receiver_idx = rx * (r_py * r_neurons) + ry * r_neurons
# Add indices for all receiver neurons for each sender neuron
for r_n in range(r_neurons):
sender_indices_list.append(base_sender_idx + n)
receiver_indices_list.append(base_receiver_idx + r_n)
sender_indices = torch.tensor(sender_indices_list, dtype=torch.long)
receiver_indices = torch.tensor(receiver_indices_list, dtype=torch.long)
w = torch.rand(len(sender_indices_list))
# Calculate scale factor
total_receiver_neurons = r_px * r_py * r_neurons
fan_in = torch.zeros(total_receiver_neurons, dtype=torch.float)
fan_in.index_add_(0, receiver_indices, torch.ones_like(w))
avg_fan_in = fan_in.mean()
scale_factor = scale / avg_fan_in
con = cls(torch.full_like(w,0.5), w, scale_factor, sender_indices, receiver_indices,
s_px, s_py, s_neurons, r_px, r_py, r_neurons,
stride_x, stride_y, neighborhood_size)
receiver.connect(sender,con)
return con
@classmethod
def make2(cls, sender, receiver,
stride_x=1, stride_y=1, scale=1.0, neighborhood_size=1):
#return DenseConnection.make2( s_px* s_py* s_neurons, r_px* r_py* r_neurons,scale)
"""
Creates transposed connection.
"""
conn = cls.make(sender, receiver, stride_x, stride_y, scale, neighborhood_size)
# Transposed: swap sender and receiver
# But we need to adjust strides and grid sizes
# If original: receiver pool rx connects to sender pool rx*stride + dx
# Transposed: sender pool (rx*stride + dx) connects to receiver pool rx
# New receiver (old sender) grid: (s_px, s_py)
# New sender (old receiver) grid: (r_px, r_py)
# New strides: inverse of original strides?
# Actually, let's just swap indices and keep the structure.
# The connectivity graph is reversed.
new_sender_indices = conn.receiver_indices
new_receiver_indices = conn.sender_indices
new_w = conn.lw.clone()
# New grid dimensions
new_r_px, new_r_py = conn.s_px, conn.s_py
new_s_px, new_s_py = conn.r_px, conn.r_py
new_r_neurons = conn.s_neurons
new_s_neurons = conn.r_neurons
# New strides:
# If original stride was sx, sy, the transposed stride is 1/sx, 1/sy?
# Not exactly, because the grid sizes are different.
# Let's keep the same neighborhood size and offsets.
# Recalculate scale
total_new_receiver_neurons = new_r_px * new_r_py * new_r_neurons
fan_in = torch.zeros(total_new_receiver_neurons, dtype=torch.float)
fan_in.index_add_(0, new_receiver_indices, torch.ones_like(new_w))
avg_fan_in = fan_in.mean()
scale_factor = scale / avg_fan_in
con2 = cls(torch.full_like(new_w,0.5), new_w, scale_factor, new_sender_indices, new_receiver_indices,
new_s_px, new_s_py, new_s_neurons, new_r_px, new_r_py, new_r_neurons,
1/conn.stride_x, 1/conn.stride_y, conn.neighborhood_size)
sender.connect(receiver,con2)
return conn,con2
def forward(self, x):
"""
Forward pass.
x: sender activations, shape (S_px * S_py * S_n,)
Returns: receiver activations, shape (R_px * R_py * R_n,)
"""
assert len(x.shape)==1 and x.shape[0]==self.s_px*self.s_py*self.s_neurons, x.shape
sender_values = x[self.sender_indices]
weighted_values = self.w * sender_values
receiver_flat = torch.zeros(self.r_px * self.r_py * self.r_neurons)
receiver_flat.index_add_(0, self.receiver_indices, weighted_values)
return receiver_flat * self.scale
def updateWeight(self, sender, receiver, adaptAct=True):
"""
Update weights.
sender: object with .act, .minusPhaseAct
receiver: object with .act, .minusPhaseAct, .targetDiff
"""
sender_plus = sender.act
sender_minus = sender.minusPhaseAct
receiver_act = receiver.act
receiver_minus = receiver.minusPhaseAct
target_diff = receiver.targetDiff
sender_plus_vals = sender_plus[self.sender_indices]
receiver_act_vals = receiver_act[self.receiver_indices]
sender_minus_vals = sender_minus[self.sender_indices]
receiver_minus_vals = receiver_minus[self.receiver_indices]
change_plus = sender_plus_vals * receiver_act_vals
change_minus = sender_minus_vals * receiver_minus_vals
change = change_plus*1.01 - change_minus*0.99
self.lw += change * lr
self.lw.clamp_(0, 1)
if adaptAct:
target_diff_vals = target_diff[self.receiver_indices]
target_mask = torch.where(target_diff_vals < 0, 0.0, 1.0)
target_update = (target_mask - self.lw) * target_diff_vals.abs() * adaptrate
self.lw += target_update
self.w=self.lw*self.sw
def updateWeightSpecific(self, senderMinusPhase, senderPlusPhase, receiver, adaptAct=True):
sender_plus = senderPlusPhase
sender_minus = senderMinusPhase
receiver_act = receiver.act
receiver_minus = receiver.minusPhaseAct
target_diff = receiver.targetDiff
sender_plus_vals = sender_plus[self.sender_indices]
receiver_act_vals = receiver_act[self.receiver_indices]
sender_minus_vals = sender_minus[self.sender_indices]
receiver_minus_vals = receiver_minus[self.receiver_indices]
change_plus = sender_plus_vals * receiver_act_vals
change_minus = sender_minus_vals * receiver_minus_vals
change = change_plus*1.01 - change_minus*0.99
self.lw += change * lr
self.lw.clamp_(0, 1)
if adaptAct:
target_diff_vals = target_diff[self.receiver_indices]
target_mask = torch.where(target_diff_vals < 0, 0.0, 1.0)
target_update = (target_mask - self.lw) * target_diff_vals.abs() * adaptrate
self.lw += target_update
self.w=self.lw*self.sw
class SameUnitConnection(LocalPoolConnection):
""" each unit connects to the same unit index in every pool """
@classmethod
def make(cls, sender, receiver, scale=1.0):
sender_indices_list = []
receiver_indices_list = []
r_px, r_py, r_neurons = receiver.shape
s_px, s_py, s_neurons = sender.shape
assert r_neurons==s_neurons
for rx in range(r_px):
for ry in range(r_py):
for ni in range(r_neurons):
for sx in range(s_px):
for sy in range(s_py):
if sx == rx and sy == ry: continue
# Calculate base indices for the pool connection
base_sender_idx = sx * (s_py * s_neurons) + sy * s_neurons
base_receiver_idx = rx * (r_py * r_neurons) + ry * r_neurons
sender_indices_list.append(base_sender_idx + ni)
receiver_indices_list.append(base_receiver_idx + ni)
sender_indices = torch.tensor(sender_indices_list, dtype=torch.long)
receiver_indices = torch.tensor(receiver_indices_list, dtype=torch.long)
w = torch.ones(len(sender_indices_list))
# Calculate scale factor
total_receiver_neurons = r_px * r_py * r_neurons
fan_in = torch.zeros(total_receiver_neurons, dtype=torch.float)
fan_in.index_add_(0, receiver_indices, torch.ones_like(w))
avg_fan_in = fan_in.mean()
scale_factor = scale / avg_fan_in if avg_fan_in > 0 else scale
con = cls(torch.full_like(w,0.5), w, scale_factor, sender_indices, receiver_indices,
s_px, s_py, s_neurons, r_px, r_py, r_neurons,
1,1,None)
receiver.connect(sender,con)
return con
class Layer:
def __init__(self, shape, adaptAct=True, target=None, targetSameUnitAllPools=False, poolAvgAct=False):
self.shape = shape
if isinstance(shape, tuple):
prod = 1
for s in shape: prod *= s
self.size = prod
else:
self.size = shape
self.shape = (shape,)
self.inConnections=[]
self.inConnectionsSender=[]
self.inConnectionsUsePrevAct=[]
self.minusPhaseAct=None
self.act=None
self.prevAct=torch.zeros(self.size)
self.avgAct = torch.zeros(self.size)
self.input = None
self.adaptAct = adaptAct
self.target = target
self.targetSameUnitAllPools = targetSameUnitAllPools
self.poolAvgAct = poolAvgAct
self.off=False
def connect(self, sender, connection, usePrevAct=False):
if connection in self.inConnections:
i = self.inConnections.index(connection)
self.inConnectionsUsePrevAct[i] = usePrevAct
else:
self.inConnections.append(connection)
self.inConnectionsSender.append(sender)
self.inConnectionsUsePrevAct.append(usePrevAct)
def update(self):
if self.off:
self.act = None
return
input = None
if self.input is not None: input = self.input.clone()
for connection,sender,usePrevAct in zip(self.inConnections,self.inConnectionsSender,self.inConnectionsUsePrevAct):
if connection.off or sender.off: continue
senderAct = sender.prevAct if usePrevAct else sender.act
if senderAct is not None:
thisInput = connection.forward(senderAct)
if input is None: input = thisInput
else: input += thisInput
assert input is not None
feedforwardInhibition = input.mean()*0.7 + input.amax()*0.3
if len(self.shape)>1 and self.shape[-1]>1:
inPoolDim=len(self.shape)-1
inputView = input.reshape(self.shape)
feedforwardInhibitionPool = inputView.mean(inPoolDim,keepdim=True)*0.7 + inputView.amax(inPoolDim,keepdim=True)*0.3
feedforwardInhibition = torch.maximum(feedforwardInhibition,feedforwardInhibitionPool.expand(self.shape).flatten())
self.feedforwardInhibition=feedforwardInhibition
input = (input - feedforwardInhibition)
self.act = (input*64.0).tanh_().clamp_(0)
return self.act
def updatePlusPhase(self, input):
self.update(input)
#input = (input - self.feedforwardInhibition)
#self.act = (input*64.0).tanh_().clamp_(0)
return self.act
def slowUpdate(self):
if self.off: return
self.avgAct += (self.act-self.avgAct)*adaptrate
if self.poolAvgAct:
assert len(self.shape)==3, "it needs pools to use poolAvgAct"
# this doesn't prevent pools from becoming too active, but doesn't keep pushing up less active ones at the edges
self.avgActTarget = self.avgAct.reshape(self.shape).mean(2,keepdim=True).expand(self.shape).flatten()
else:
self.avgActTarget = self.avgAct.mean()
self.targetDiff = self.avgActTarget - self.avgAct
for connection,sender,usePrevAct in zip(self.inConnections,self.inConnectionsSender,self.inConnectionsUsePrevAct):
if connection.off or sender.off: continue
if usePrevAct: connection.updateWeightSpecific(sender.prevAct,sender.prevAct,self,self.adaptAct)
else: connection.updateWeight(sender,self,self.adaptAct)
inpLayer=Layer((8,8,3))
objVel=Layer((8,8))
V1=Layer((8,8,16))
V2=Layer((8,8,40))
V3=Layer((4,4,40))
V4=Layer((4,4,40))
V1p=Layer(V1.shape, adaptAct=False, target=V1)
V2p=Layer(V1.shape, adaptAct=False, target=V1)
V3p=Layer(V1.shape, adaptAct=False, target=V1)
V4p=Layer(V1.shape, adaptAct=False, target=V1)
V3ct=Layer(V3.shape)
V4ct=Layer(V4.shape)
MtPos=Layer((8,8,1), poolAvgAct=True)
LIP=Layer((8,8,16), poolAvgAct=True)
LIPct=Layer((8,8,16), poolAvgAct=True)
LIPP=Layer((8,8,1), adaptAct=False, target=MtPos)
DP=Layer((10,10))
DPct=Layer((10,10))
DPp=Layer(V1.shape, adaptAct=False, target=V1)
A1=Layer((4,4,40))
A2=Layer((4,4,40))
A1ct=Layer(A1.shape)
A2ct=Layer(A2.shape)
A1p=Layer((4,4,40+16), adaptAct=False, target=[V1,V4], targetSameUnitAllPools=True)
A2p=Layer((4,4,40+16), adaptAct=False, target=[V1,V4], targetSameUnitAllPools=True)
v1ToV2,V2ToV1=LocalPoolConnection.make2(V1,V2, scale=8.0)
v2ToV3,V3ToV2=LocalPoolConnection.make2(V2,V3, stride_x=2,stride_y=2, scale=8.0)
v2ToV4,V4ToV2=LocalPoolConnection.make2(V2,V4, stride_x=2,stride_y=2, scale=8.0)
# helps to make the same units in different pools more similar
SameUnitConnection.make(V2,V2, scale=0.1*8.0)
SameUnitConnection.make(V3,V3, scale=0.1*8.0)
SameUnitConnection.make(V4,V4, scale=0.1*8.0)
V2p.connect(V2,LocalPoolConnection.make(V2,V2p, scale=8.0),usePrevAct=True)
V1p.connect(V1,LocalPoolConnection.make(V1,V1p, neighborhood_size=0, scale=8.0),usePrevAct=True)
V4ct.connect(V4,LocalPoolConnection.make(V4,V4ct, scale=8.0),usePrevAct=True)
V4ct.connect(V4ct,LocalPoolConnection.make(V4ct,V4ct, scale=0.25*8.0),usePrevAct=True)
LocalPoolConnection.make2(V4ct,V4p, stride_x=0.5,stride_y=0.5, scale=8.0)
LocalPoolConnection.make2(V2p,V4p, scale=8.0)
V3ct.connect(V3,LocalPoolConnection.make(V3,V3ct, scale=8.0),usePrevAct=True)
V3ct.connect(V3ct,LocalPoolConnection.make(V3ct,V3ct, scale=0.25*8.0),usePrevAct=True)
LocalPoolConnection.make2(V3ct,V3p, stride_x=0.5,stride_y=0.5, scale=8.0)
LocalPoolConnection.make(V1p,V1, scale=0.1*8.0)
LocalPoolConnection.make(V2p,V2, scale=0.1*8.0)
LocalPoolConnection.make(V3p,V3, stride_x=2,stride_y=2, scale=0.1*8.0)
LocalPoolConnection.make(V4p,V4, stride_x=2,stride_y=2, scale=0.1*8.0)
LocalPoolConnection.make(V1p,V2, scale=0.1*8.0)
LocalPoolConnection.make(V1p,V3, stride_x=2,stride_y=2, scale=0.1*8.0)
LocalPoolConnection.make(V1p,V4, stride_x=2,stride_y=2, scale=0.1*8.0)
LocalPoolConnection.make(V4ct,V3ct, scale=8.0)
LocalPoolConnection.make(V3ct,V2p, stride_x=0.5,stride_y=0.5, scale=8.0)
LocalPoolConnection.make(V2p,V1p, scale=8.0)
v1ToMtPos=LocalPoolConnection.make(V1,MtPos, scale=1.0, neighborhood_size=0)
MtPosToLIP=LocalPoolConnection.make(MtPos, LIP, scale=1.0, neighborhood_size=0)
DenseConnection.make(objVel,LIP, scale=1.0)
LIPct.connect(LIP,LocalPoolConnection.make(LIP,LIPct, scale=8.0),usePrevAct=True)
DenseConnection.make(objVel,LIPct, scale=8.0)
DenseConnection.make(LIPct, LIPP, scale=8.0)
LocalPoolConnection.make(LIPP, LIPct, scale=0.5*8.0) #back connection, helps LIPct represent new position
LocalPoolConnection.make(LIPP,LIP, scale=0.1*1.0, neighborhood_size=0)
LIPToV2=LocalPoolConnection.make(LIP,V2, scale=8.0, neighborhood_size=0)
LIPToV3=LocalPoolConnection.make(LIP,V3, stride_x=2,stride_y=2, scale=8.0, neighborhood_size=0)
LocalPoolConnection.make(LIPct,V1p, scale=8.0)#todo change to LIPct
LocalPoolConnection.make(LIPct,V2p, scale=8.0)
LocalPoolConnection.make(LIPct,V3ct, stride_x=2,stride_y=2, scale=8.0) # larger neighborhood_size because the connection is not centered currently
#LocalPoolConnection.make(MtPos,V4p, stride_x=2,stride_y=2, neighborhood_size=2, scale=8.0)
#LocalPoolConnection.make(MtPos,A1ct, stride_x=2,stride_y=2, neighborhood_size=2, scale=8.0)
DenseConnection.make2(V3,DP, scale=8.0)
DenseConnection.make(DPp,DP, scale=0.1*8.0)
DenseConnection.make(V3p,DP, scale=0.1*8.0)
DenseConnection.make(A1,DP, scale=8.0)
DenseConnection.make(A1p,DP, scale=0.1*8.0)
DPct.connect(DP,DenseConnection.make(DP,DPct, scale=8.0), usePrevAct=True)
DenseConnection.make(A1ct,DPct, scale=8.0)
DenseConnection.make2(DPct,DPp, scale=8.0)
DenseConnection.make(A1ct,DPp, scale=8.0)
DenseConnection.make2(V3ct,DPp, scale=8.0)
DenseConnection.make(DPp,V3, scale=0.1*8.0)
DenseConnection.make(DPct,V3ct, scale=8.0)
DenseConnection.make(DP,V3ct, scale=8.0)
DenseConnection.make(DPct,V3p, scale=8.0)
v4ToA1,A1ToV4=DenseConnection.make2(V4,A1, scale=8.0) # almost full connection
LocalPoolConnection.make(A1p,A1, neighborhood_size=2, scale=0.1*8.0)
LocalPoolConnection.make(V1p,A1, stride_x=2,stride_y=2, neighborhood_size=2, scale=0.1*8.0)
A1ct.connect(A1,LocalPoolConnection.make(A1,A1ct, scale=8.0),usePrevAct=True)
A1ct.connect(A1ct,LocalPoolConnection.make(A1ct,A1ct, neighborhood_size=0, scale=8.0),usePrevAct=True)
LocalPoolConnection.make2(V4ct,A1p, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make(V3ct,A1p, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make2(A1ct,A1p, scale=8.0)
LocalPoolConnection.make2(A2ct,A1p, scale=8.0)
LocalPoolConnection.make(A1p,A1, scale=0.1*8.0)
LocalPoolConnection.make(A1ct,V4p, stride_x=0.5,stride_y=0.5, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make(A1ct,V3, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make(A1ct,V3p, stride_x=0.5,stride_y=0.5, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make(A1ct,V2, stride_x=0.5,stride_y=0.5, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make2(A1ct,V1p, stride_x=0.5,stride_y=0.5, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make(A1,V3ct, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make(A1,V2p, stride_x=0.5,stride_y=0.5, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make2(A1,A2, scale=8.0)
LocalPoolConnection.make(A2p,A2, neighborhood_size=2, scale=0.1*8.0)
LocalPoolConnection.make(V1p,A2, stride_x=2,stride_y=2, neighborhood_size=2, scale=0.1*8.0)
A2ct.connect(A2,LocalPoolConnection.make(A2,A2ct, scale=8.0),usePrevAct=True)
A2ct.connect(A2ct,LocalPoolConnection.make(A2ct,A2ct, neighborhood_size=0, scale=8.0),usePrevAct=True)
LocalPoolConnection.make2(V4ct,A2p, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make(V3ct,A2p, neighborhood_size=2, scale=8.0)
LocalPoolConnection.make2(A1ct,A2p, scale=8.0)
LocalPoolConnection.make2(A2ct,A1ct, scale=8.0)
i=0
fig, axs = plt.subplots(7,3)
plt.tight_layout(pad=0)
plt.subplots_adjust(wspace=0.05, hspace=0.05)
for ax in axs.flat: ax.tick_params(labelbottom=False, labelleft=False)
def showAt(ax, v, title=None, vmin=0,vmax=1):
ax.clear()
ax.axis('off')
if v is None or v is False: return
if hasattr(v,"minusPhaseAct"):
if v.minusPhaseAct is None: return
v = v.minusPhaseAct.reshape(v.shape)
if len(v.shape)==1: v=v.unsqueeze(0)
if len(v.shape)==3: v=v.flatten(1)
ax.imshow(v, vmin=vmin, vmax=vmax)
if title: ax.set_title(title)
for layer in [V2,V2p,V3,V3ct,V3p,V4,V4ct,V4p,DP,DPct,DPp,A1,A1ct,A1p,A2,A2ct,A2p]:
layer.off = True
layer.prevAct=torch.zeros(layer.size)
video_sigs=[]
for j in range(1000):
i+=1
input_image,input_velx,input_vely=next_frame(i,moveSpeed=1.0)
v1Input=F.conv2d(
input=input_image,
weight=gabor_filters_tensor,
bias=None,
stride=1,
padding=4 # (9-1)/2 = 4, to keep spatial dimensions same
)
V1.input=F.avg_pool2d(v1Input, kernel_size=4, stride=4).permute(0,2,3,1).flatten() / 8.0
objVel.act=objVel.minusPhaseAct=gaussian_blob_tensor((8,8),input_velx*4+4,input_vely*4+4,2).flatten()
for layer in [V1,V1p,V2,V2p,V3,V3ct,V3p,V4,V4ct,V4p,MtPos,LIP,LIPct,LIPP,DP,DPct,DPp,A1,A1ct,A1p,A2,A2ct,A2p]:
layer.act = None
#V1.update();MtPos.update();LIP.update()# temp
V1.update()
MtPos.input = V1.act.reshape(V1.shape).mean(2).flatten()/4.0 # normal connection doesn't work because the ones at the edge are less active, so they get pushed up and become too strong when active
# update ct and p layers first because they use prevAct
for layer in [A1ct,A2ct,A1ct,DPct,LIPct,V4ct,V3ct]:
layer.update()
for layer in [A2p,A1p,DPp,LIPP,V4p,V3p,V2p,V1p]:
layer.update()
#forward
for layer in [V1,V2,V3,V4,MtPos,LIP,A1,A2,DP]:
layer.update()
# update p layers because some connect to super layers which were just updated. again update ct because it depends on lower p layers, and for backwards from same level p to ct
for layer in [A1ct,A2ct,A1ct,DPct,LIPct,V4ct,V3ct]:
layer.update()
for layer in [A2p,A1p,DPp,LIPP,V4p,V3p,V2p,V1p]:
layer.update()
#spread backward so it spreads more completely
for layer in [A1,V4,V3,V2,V1]:
layer.update()
#forward again
for layer in [V1,V2,V3,V4,MtPos,LIP,A1,A2,DP]:
layer.update()
# do again but clamp p layers
for layer in [V1,V1p,V2,V2p,V3,V3ct,V3p,V4,V4ct,V4p,MtPos,LIP,LIPct,LIPP,DP,DPct,DPp,A1,A1ct,A1p,A2,A2ct,A2p]:
layer.minusPhaseAct = layer.act
for layer in [V1p,V2p,V3p,LIPP,DPp,A1p,A2p]:
if layer.targetSameUnitAllPools:
assert len(layer.shape) == 3
offset = 0
newAct = torch.zeros(layer.shape)
for target in layer.target:
assert len(target.shape) == 3
if target.act is None: continue
targetView = target.act.reshape(target.shape)
sameUnitAllPools = targetView.mean((0,1), keepdim=True)#*0.5 + targetView.amax((0,1), keepdim=True)*0.5
# this is kind of wrong way to do it, but its simpler
feedforwardInhibition = sameUnitAllPools.mean()*0.7 + sameUnitAllPools.amax()*0.3
sameUnitAllPools = (sameUnitAllPools - feedforwardInhibition)
sameUnitAllPools=(sameUnitAllPools*64.0).tanh_().clamp_(0)
newAct[:,:,offset:offset+target.shape[2]] = sameUnitAllPools
offset += target.shape[2]
layer.act = newAct.flatten()
else:
layer.act = layer.target.act
# also update ct layers without updating p layers
for layer in [A1ct,A2ct,A1ct,DPct,LIPct,V4ct,V3ct]:
layer.update()
#forward
for layer in [V1,V2,V3,V4,MtPos,LIP,A1,A2,DP]:
layer.update()
# again update ct because it depends on lower layers
for layer in [A2ct,A1ct,A2ct,DPct,LIPct,V4ct,V3ct]:
layer.update()
#spread backward so it spreads more completely
for layer in [A1,V4,V3,V2,V1]:
layer.update()
#forward again
for layer in [V1,V2,V3,V4,MtPos,LIP,A1,A2,DP]:
layer.update()
#update weight
for layer in [V1,V1p,V2,V2p,V3,V3ct,V3p,V4,V4ct,V4p,MtPos,LIP,LIPct,LIPP,DP,DPct,DPp,A1,A1ct,A1p,A2,A2ct,A2p]:
layer.slowUpdate()
#update prevAct
for layer in [V1,V1p,V2,V2p,V3,V3ct,V3p,V4,V4ct,V4p,MtPos,LIP,LIPct,LIPP,DP,DPct,DPp,A1,A1ct,A1p,A2,A2ct,A2p]:
layer.prevAct = layer.act
if j>0:
#showAt(axs[0,0], input_image.view(32,32))
#showAt(axs[0,0], inp.view(inpLayer.shape))
#showAt(axs[0,1], inppa.view(inpp.shape))
#showAt(axs[0,2], -10*inpp.targetDiff.view(inpp.shape))
showAt(axs[0,0], V1)
showAt(axs[0,1], V1p)
showAt(axs[0,2], V1p.minusPhaseAct is not None and (V1p.act.view(V1p.shape).amax(dim=2)-V1p.minusPhaseAct.view(V1p.shape).amax(dim=2)), None, -1,1)
axs[0,2].text(0.1,0.1, f"err: {F.mse_loss(V1p.minusPhaseAct,V1p.act).item()}")
#conw=torch.zeros(V1.size)
#contoshow=MtPos.inConnections[0]
#selw=contoshow.receiver_indices==36
#conw=contoshow.w[selw]#;conw=(conw-conw.mean())*10+0.5
showAt(axs[1,0], V2)
#showAt(axs[1,0], conw.reshape(1,-1))
showAt(axs[1,1], V2p)
showAt(axs[1,2], V3p.minusPhaseAct is not None and (V3p.act.view(V3p.shape).amax(dim=2)-V3p.minusPhaseAct.view(V3p.shape).amax(dim=2)), None, -1,1)
showAt(axs[2,0], V3)
showAt(axs[2,1], V3p)
showAt(axs[2,2], V4p,"v4p")
showAt(axs[3,0], V4)
showAt(axs[3,1], A1p,"A1p")
showAt(axs[3,2], A2p,"A2p")
showAt(axs[4,0], MtPos,"MtPos")
#showAt(axs[4,1], V1.act.view(8,8,16).mean(2),"should be")
showAt(axs[4,2], LIPP,"LIPP")
#V1.act.view(8,8,16).mean(2)
showAt(axs[4,1], LIPct,"LIPct")
showAt(axs[5,2], LIP,"LIP")
showAt(axs[5,0], A1,"A1")
showAt(axs[5,1], A2,"A2")
showAt(axs[5,1], objVel,"objVel")
#showAt(axs[5,2], A1p.minusPhaseAct is not None and (A1p.act.view(A1p.shape).amax(dim=2)-A1p.minusPhaseAct.view(A1p.shape).amax(dim=2)), None, -1,1)
#showAt(axs[4,0], A1ct,"A1ct")
#showAt(axs[4,1], A2ct,"A2ct")
showAt(axs[6,0], DP,"DP")
showAt(axs[6,1], DPct,"DPct")
showAt(axs[6,2], DPp,"DPp")
if A1.act is not None: video_sigs.append(A1.act)
if len(video_sigs)>50:video_sigs.pop(0)
if 1 and len(video_sigs) >= 2:
sigs = torch.stack(video_sigs)
sigs = sigs / sigs.norm(dim=1, keepdim=True).clamp(min=1e-8)
sim = (sigs @ sigs.T).numpy()
axs[4,2]
axs[4,2].imshow(sim, vmin=-1, vmax=1, cmap='RdBu', aspect='auto')
plt.pause(1)
plt.pause(100000)
import torch
import torch.nn.functional as F
import matplotlib.pyplot as plt
def make_obj(rows):
return torch.tensor(rows, dtype=torch.float32)
objects = [
# Pyramid 1
make_obj([
[0,0,0,0,0,0,0,0],
[0,0,0,0,0,0,0,0],
[0,0,0,0,0,0,0,0],
[0,0,0,0,0,0,0,0],
[0,0,0,1,1,0,0,0],
[0,0,1,1,1,1,0,0],
[0,1,1,1,1,1,1,0],
[1,1,1,1,1,1,1,1]
]),
# Round 1
make_obj([
[0,0,0,1,1,0,0,0],
[0,0,1,1,1,1,0,0],
[0,1,1,1,1,1,1,0],
[1,1,1,1,1,1,1,1],
[1,1,1,1,1,1,1,1],
[0,1,1,1,1,1,1,0],
[0,0,1,1,1,1,0,0],
[0,0,0,1,1,0,0,0]
]),
# Box 1
make_obj([
[1,1,1,1,1,1,1,1],
[1,0,0,0,0,0,0,1],
[1,0,0,0,0,0,0,1],
[1,0,0,0,0,0,0,1],
[1,0,0,0,0,0,0,1],
[1,0,0,0,0,0,0,1],
[1,0,0,0,0,0,0,1],
[1,1,1,1,1,1,1,1]
]),
# Horizontal 1
make_obj([
[0,0,0,0,0,0,0,0],
[0,0,0,0,0,0,0,0],
[0,0,0,0,0,0,0,0],
[1,1,1,1,1,1,1,1],
[1,1,1,1,1,1,1,1],
[0,0,0,0,0,0,0,0],
[0,0,0,0,0,0,0,0],
[0,0,0,0,0,0,0,0]
]),
]
def next_frame(i, moveSpeed=1.0):
# Grid for output frame coordinates normalized to [-1, 1]
y_out = torch.linspace(-4, 4, 32)
x_out = torch.linspace(-4, 4, 32)
y_out_grid, x_out_grid = torch.meshgrid(y_out, x_out, indexing='ij')
objNum = round(i/10-0.5)%len(objects)
obj_bitmap = objects[objNum]
i=torch.tensor(i+objNum*33)
rot = i*0.4
# Map position to [-1, 1]
pos_x_norm = torch.sin(i*0.5*moveSpeed)*2.0
pos_y_norm = torch.sin(i*0.3*moveSpeed)*2.0
velx = pos_x_norm - torch.sin((i-1)*0.5*moveSpeed)*2.0
vely = pos_y_norm - torch.sin((i-1)*0.3*moveSpeed)*2.0
# Relative coordinates in frame space
x_rel = x_out_grid - pos_x_norm
y_rel = y_out_grid - pos_y_norm
# Inverse rotation to map frame coordinates to object local space
cos_neg = torch.cos(-rot)
sin_neg = torch.sin(-rot)
x_local = x_rel * cos_neg - y_rel * sin_neg
y_local = x_rel * sin_neg + y_rel * cos_neg
# Normalize local coordinates to [-1, 1] for grid_sample
# Assuming the object bitmap is centered at 0 and scaled to fit [-1, 1]
# Since the bitmap is 8x8, we assume it maps 0..8 to -1..1
x_norm = x_local
y_norm = y_local
grid = torch.stack([x_norm, y_norm], dim=-1).unsqueeze(0)
obj_expanded = obj_bitmap.unsqueeze(0).unsqueeze(0)
frame = F.grid_sample(
obj_expanded,
grid,
mode='bilinear',
padding_mode='zeros',
align_corners=True
)
frame = torch.clamp(frame, 0.0, 1.0)
return frame, velx,vely
#plt.imshow(next_frame(100).reshape(32,32))
def gaussian_blob_tensor(shape, mean_x, mean_y, sigma):
"""
Generates a 2D tensor with a Gaussian blob.
Args:
shape (tuple): The spatial dimensions of the output tensor (height, width).
mean_x (float): The x-coordinate (column) of the center.
mean_y (float): The y-coordinate (row) of the center.
sigma (float): The standard deviation (width) of the Gaussian.
Returns:
torch.Tensor: A 2D tensor of shape (height, width) with values from
a normalized Gaussian distribution centered at (mean_x, mean_y).
"""
height, width = shape
# Create coordinate grids
# y corresponds to rows (0 to height-1)
# x corresponds to columns (0 to width-1)
y_coords = torch.arange(height, dtype=torch.float32)
x_coords = torch.arange(width, dtype=torch.float32)
# Create meshgrid
# yy has shape (height, 1), xx has shape (1, width)
yy, xx = torch.meshgrid(y_coords, x_coords, indexing='ij')
# Calculate squared distance from the center
dist_sq = ((xx - mean_x) ** 2 + (yy - mean_y) ** 2)
# Calculate Gaussian values
# Normalization factor for 2D Gaussian is 1 / (2 * pi * sigma^2)
# However, often we just want the relative intensities.
# If you need it to sum to 1, divide by the sum.
# If you need peak intensity 1, don't normalize by sum.
exponent = -dist_sq / (2 * sigma ** 2)
blob = torch.exp(exponent)
return blob
import torch
import matplotlib.pyplot as plt
def create_gabor_kernel(length, wavelength, theta, sigma_len, sigma_wd, phase_deg):
"""Generate a Gabor kernel matching the Go code's structure."""
half_length = length // 2
# Create coordinate tensors
x = torch.arange(-half_length, half_length + 1, dtype=torch.float32)
y = torch.arange(-half_length, half_length + 1, dtype=torch.float32)
xx, yy = torch.meshgrid(x, y, indexing='ij')
# Rotation to align with orientation theta
theta=torch.tensor(theta)
cos_theta = torch.cos(theta)
sin_theta = torch.sin(theta)
nx = xx * cos_theta + yy * sin_theta
ny = -xx * sin_theta + yy * cos_theta
# Gaussian envelope
len_norm = 1.0 / (2.0 * sigma_len**2)
wd_norm = 1.0 / (2.0 * sigma_wd**2)
gauss = torch.exp(-(len_norm * nx**2 + wd_norm * ny**2))
# Sinusoidal wave
two_pi_norm = (2.0 * torch.pi) / wavelength
phase_rad = torch.tensor(phase_deg * torch.pi / 180.0) # Convert degrees to radians
sin_wave = torch.sin(two_pi_norm * ny + phase_rad)
val = gauss * sin_wave
return val
filters = []
num_filters = 16
for i in range(num_filters):
kernel = create_gabor_kernel(9,6,i/16.0*torch.pi*2.0,0.3*9,0.2*9,0)
filters.append(kernel)
# Stack into [16, 1, 9, 9]
#gabor_filters_tensor = torch.stack(filters).unsqueeze(1)
#input_image=next_frame(1)
#v1Input=F.conv2d(
# input=input_image,
# weight=gabor_filters_tensor,
# bias=None,
# stride=1,
# padding=4 # (9-1)/2 = 4, to keep spatial dimensions same
#)
#plt.imshow(v1Input[0,5])