101 lines
3.0 KiB
Python
101 lines
3.0 KiB
Python
import numpy as np
|
|
from collections.abc import Callable
|
|
from params import RbmParams
|
|
from state import RbmState
|
|
from matrix import sample, prob, rms_error_accu
|
|
from status import Status
|
|
|
|
class Entity:
|
|
def __init__(self, shape: tuple[int, int], params: RbmParams):
|
|
self.state = RbmState.from_layer_params(shape)
|
|
self.params = params
|
|
|
|
def train(self, batch: np.ndarray, cd_func: Callable, status: Status):
|
|
training_remain = batch.shape[0]
|
|
batch_size = min(self.params.mini_batch_size, training_remain)
|
|
if batch_size == 0:
|
|
batch_size = training_remain
|
|
|
|
status.on_change()
|
|
d_progress = 100.0 / (training_remain*self.params.num_epochs)
|
|
progress = 0
|
|
batch_row_index = 0
|
|
keep_running = True
|
|
while training_remain > 0 and keep_running:
|
|
batch_size_remain = min(batch_size, training_remain)
|
|
mini_batch = batch[batch_row_index:batch_row_index + batch_size_remain]
|
|
training_remain -= batch_size_remain
|
|
batch_row_index += batch_size_remain
|
|
|
|
inc_bv = np.zeros(self.state.b_v.shape)
|
|
inc_bh = np.zeros(self.state.b_h.shape)
|
|
inc_whv = np.zeros(self.state.w_hv.shape)
|
|
|
|
v_states = mini_batch
|
|
if self.params.do_batch_sample:
|
|
v_states = sample(mini_batch)
|
|
|
|
for epochs in range(self.params.num_epochs):
|
|
# Contrastive divergence learning: calculate gradients
|
|
dwhv, dbv, dbh = cd_func(v_states, self.params, self.v_to_ph, self.h_to_pv)
|
|
|
|
# Adjust weight and biases
|
|
kl = self.params.learning_rate/batch_size
|
|
inc_bv = self.params.momentum*inc_bv + kl*dbv
|
|
inc_bh = self.params.momentum*inc_bh + kl*dbh
|
|
inc_whv = self.params.momentum*inc_whv + kl*dwhv - self.params.weight_decay*self.state.w_hv
|
|
|
|
self.state.b_v += inc_bv
|
|
self.state.b_h += inc_bh
|
|
self.state.w_hv += inc_whv
|
|
|
|
# check if status update is needed
|
|
if status.want_report(round(progress)):
|
|
# Calculate error
|
|
err_rms = rms_error_accu(mini_batch - self.h_to_pv(self.v_to_ph(v_states)))
|
|
if not status.on_change({"progress": {"value": round(progress), "unit": "%"},
|
|
"err_rms": {"value": err_rms, "unit": ""}}):
|
|
keep_running = False
|
|
break
|
|
|
|
progress += d_progress*batch_size_remain
|
|
|
|
# Update final status
|
|
err_rms = rms_error_accu(batch - self.h_to_pv(self.v_to_ph(batch)))
|
|
status.on_change({"progress": {"value": round(progress), "unit": "%"},
|
|
"err_rms_total": {"value": err_rms, "unit": ""}})
|
|
|
|
def v_to_ph(self, v: np.ndarray) -> np.ndarray:
|
|
state = self.state.v_to_h(v)
|
|
if self.params.do_gaussian_hidden:
|
|
return state
|
|
|
|
return prob(state)
|
|
|
|
def h_to_pv(self, h: np.ndarray) -> np.ndarray:
|
|
state = self.state.h_to_v(h)
|
|
if self.params.do_gaussian_visible:
|
|
return state
|
|
|
|
return prob(state)
|
|
|
|
|
|
def gibbs_v_to_h(self, v: np.ndarray) -> np.ndarray:
|
|
h = self.v_to_ph(v)
|
|
for i in range(self.params.num_gibbs_samples-1):
|
|
h = self.h_to_pv(h)
|
|
h = self.v_to_ph(h)
|
|
|
|
return h
|
|
|
|
def gibbs_h_to_v(self, h: np.ndarray) -> np.ndarray:
|
|
v = self.h_to_pv(h)
|
|
for i in range(self.params.num_gibbs_samples-1):
|
|
v = self.v_to_ph(v)
|
|
v = self.h_to_pv(v)
|
|
|
|
return v
|
|
|
|
if __name__ == "__main__":
|
|
print("Test: [passed]")
|