- initroduced types: BB-RBM GB-RBM, GG-RBM as property of entity

- training simplification and for clarity: added specialized cd_funcs for each entity type

src/rbm/entity.py

@@ -1,6 +1,6 @@
 from .state import RbmState
 from .matrix import prob, Mat
-
+from enum import Enum
 class EntityParams:
 	def __init__(self, do_gaussian_visible: bool = False, do_gaussian_hidden: bool = False, num_gibbs_samples: int = 1):
 		# Entity parameters
@@ -21,11 +21,29 @@ class EntityParams:
 		return obj
 
 class Entity:
+	class Type(Enum):
+		BB_RBM = "BB-RBM"
+		BG_RBM = "BG-RBM"
+		GB_RBM = "GB-RBM"
+		GG_RBM = "GG-RBM"
+
 	def __init__(self, shape: tuple[int, int], params: EntityParams):
 		self.shape = shape
 		self.params = params
 		self.state = RbmState.from_layer_params(shape)
 		self.grad = RbmState.from_layer_params(shape)
+		self.type = None
+		if params.do_gaussian_visible:
+			if params.do_gaussian_hidden:
+				self.type = Entity.Type.GG_RBM
+			else:
+				self.type = Entity.Type.GB_RBM
+		else:
+			if params.do_gaussian_hidden:
+				self.type = Entity.Type.BG_RBM
+			else:
+				self.type = Entity.Type.BB_RBM
+
 
 	def __call__(self, x: Mat):
 		return self.forward(x)
@@ -72,6 +90,7 @@ class Entity:
 
 		return prob(state)
 
+
 	def _h_to_pv(self, h: Mat) -> Mat:
 		state = self.state.h_to_v(h)
 		if self.params.do_gaussian_visible:
@@ -79,5 +98,14 @@ class Entity:
 
 		return prob(state)
 
+	def h_given_v(self, v: Mat) -> Mat:
+		state = self.state.v_to_h(v)
+		return state
+
+
+	def v_given_h(self, h: Mat) -> Mat:
+		state = self.state.h_to_v(h)
+		return state
+
 if __name__ == "__main__":
 	print("Test: [passed]")

src/rbm/layout.py

@@ -24,6 +24,6 @@ class Horizontal:
 	def train(self, batch: Mat, params: TrainingParams):
 		_batch = np.copy(batch)
 		for unit in self.units:
-			train(unit, _batch, params, Status(), cd_jens)
+			train(unit, _batch, params, Status())
 			_batch = unit.forward(_batch, num_gibbs=params.num_gibbs_samples)
 

src/rbm/matrix.py

@@ -27,7 +27,7 @@ def gaussian(shape: tuple, mu: float = 0.0, std: float = 1.0) -> Mat:
 def sample(src: Mat) -> Mat:
 	return (src > uniform(src.shape)).astype(float)
 
-def gaussian_sample(src: Mat, mu=0, std=1.0) -> Mat:
+def sample_gaussian(src: Mat, mu=0, std=1.0) -> Mat:
 	return gaussian(src.shape, mu, std)
 
 def prob(src: Mat) -> Mat:

src/rbm/model.py

@@ -26,7 +26,7 @@ class Model(ABC):
 			params = [params]*len(entities)
 		_batch = np.copy(batch)
 		for entity, param in zip(entities, params):
-			train(entity, _batch, param, Status(), cd_jens)
+			train(entity, _batch, param, Status())
 			_batch = entity.forward(_batch, num_gibbs=param.num_gibbs_samples)
 
 	@abstractmethod

src/rbm/train.py

@@ -1,5 +1,5 @@
 from collections.abc import Callable
-from .matrix import gaussian, sample, prob, rms_error_accu, Mat, np
+from .matrix import sample, sample_gaussian, prob, rms_error_accu, Mat, np
 from .entity import Entity
 from .status import Status
 
@@ -50,7 +50,7 @@ def cd_jens(entity: Entity, v_states: Mat, params: TrainingParams):
 	h_probs = h_states
 
 	if entity.params.do_gaussian_hidden:
-		h_states += gaussian(h_states.shape)
+		h_states += sample_gaussian(h_states)
 	else:
 		h_probs = entity.forward(v_states)
 		if params.do_rao_blackwell:
@@ -83,6 +83,81 @@ def cd_jens(entity: Entity, v_states: Mat, params: TrainingParams):
 
 	return dw, dbv, dbh
 
+def cd_binary_binary(entity: Entity, data_pos: Mat, params: TrainingParams):
+	# Positive phase
+	h_probs_pos = entity.h_given_v(data_pos)
+
+	# Update weights (positive phase)
+	dw = np.dot(np.transpose(data_pos), h_probs_pos)
+	dbh = np.sum(h_probs_pos, 0)
+	dbv = np.sum(data_pos, 0)
+
+	# Sample hidden states
+	if params.do_rao_blackwell:
+		h_states_pos = h_probs_pos
+	else:
+		h_states_pos = sample(h_probs_pos)
+
+	# Negative phase
+	data_neg = entity.v_given_h(h_states_pos)
+	h_probs_neg = entity.h_given_v(data_neg)
+
+	# Gibbs	sampling with training params
+	# ToDo
+
+	# Update weights (negative phase)
+	dw -= np.dot(np.transpose(data_neg), h_probs_neg)
+	dbh -= np.sum(h_probs_neg, 0)
+	dbv -= np.sum(data_neg, 0)
+
+	return dw, dbv, dbh
+
+def cd_gaussian_binary(entity: Entity, data_pos: Mat, params: TrainingParams):
+	# Positive phase
+	h_probs_pos = entity.h_given_v(data_pos)
+
+	# Update weights (positive phase)
+	dw = np.dot(np.transpose(data_pos), h_probs_pos)
+	dbh = np.sum(h_probs_pos, 0)
+	dbv = np.sum(data_pos, 0)
+
+	# Sample hidden states
+	h_states_pos = sample(h_probs_pos)
+
+	# Negative phase
+	data_neg = entity.v_given_h(h_states_pos)
+	h_probs_neg = entity.h_given_v(data_neg)
+
+	# Update weights (negative phase)
+	dw -= np.dot(np.transpose(data_neg), h_probs_neg)
+	dbh -= np.sum(h_probs_neg, 0)
+	dbv -= np.sum(data_neg, 0)
+
+	return dw, dbv, dbh
+
+def cd_gaussian_gaussian(entity: Entity, data_pos: Mat, params: TrainingParams):
+	# Positive phase
+	h_probs_pos = entity.h_given_v(data_pos)
+
+	# Sample hidden states
+	h_states_pos = h_probs_pos + sample_gaussian(h_probs_pos)
+
+	# Update weights (positive phase)
+	dw = np.dot(np.transpose(data_pos), h_states_pos)
+	dbh = np.sum(h_states_pos, 0)
+	dbv = np.sum(data_pos, 0)
+
+	# Negative phase
+	data_neg = entity.v_given_h(h_states_pos)
+	h_probs_neg = entity.h_given_v(data_neg)
+
+	# Update weights (negative phase)
+	dw -= np.dot(np.transpose(data_neg), h_probs_neg)
+	dbh -= np.sum(h_probs_neg, 0)
+	dbv -= np.sum(data_neg, 0)
+
+	return dw, dbv, dbh
+
 def to_mini_batch(batch: Mat, mini_batch_size: int):
 	mini_batches = []
 	remain = batch.shape[0]
@@ -95,12 +170,20 @@ def to_mini_batch(batch: Mat, mini_batch_size: int):
 
 	return mini_batches
 
-def train(entity: Entity, batch: Mat, params: TrainingParams, status: Status, cd_func: Callable = cd_jens):
+def train(entity: Entity, batch: Mat, params: TrainingParams, status: Status):
 	mini_batch_size = min(params.mini_batch_size, batch.shape[0]) if params.mini_batch_size > 0 else batch.shape[0]
 	d_progress = 100.0 / (batch.shape[0]*params.num_epochs)
 	progress = 0
 	keep_running = True
 	status.on_change()
+	cd_func = None
+	if entity.type == Entity.Type.GB_RBM:
+		cd_func = cd_gaussian_binary
+	if entity.type == Entity.Type.BB_RBM:
+		cd_func = cd_binary_binary
+	if entity.type == Entity.Type.GG_RBM:
+		cd_func = cd_gaussian_gaussian
+
 	for mini_batch in to_mini_batch(batch, mini_batch_size):
 		if not keep_running:
 			break

src/tests/test_norbs.py

@@ -40,7 +40,7 @@ if __name__ == "__main__":
 	test_batch = read_armadillo(os.path.join(prj_root, f"{prj_name}.test.dat"))
 
 	# Train
-	model.train(train_batch, TrainingParams(learning_rate=0.00001, momentum=0.9, do_rao_blackwell=True, num_epochs=100, num_gibbs_samples=3))
+	model.train(train_batch, TrainingParams(learning_rate=0.000001, momentum=0.9, do_rao_blackwell=False, num_epochs=1000, num_gibbs_samples=3))
 
 	# save state
 	model.save()