- added training params as list for model.train()

- added gaussian sample
- introduced layout concept
- updated README

README.md

@@ -0,0 +1,7 @@
+# GRBM
+## Paper
+https://medium.com/@rtdcunha/gaussian-bernoulli-restricted-boltzmann-machines-4a68b8765485
+https://journals.plos.org/plosone/article/file?id=10.1371/journal.pone.0171015&type=printable
+
+## Code
+https://github.com/DSL-Lab/GRBM/tree/main

src/rbm/layout.py

@@ -0,0 +1,29 @@
+import numpy as np
+
+from .entity import Entity
+from .matrix import Mat
+from .train import train, cd_jens, TrainingParams
+from .status import Status
+
+class Horizontal:
+	def __init__(self, entities: list[Entity]):
+		self.units: list[Entity] = entities
+
+	def forward(self, x: Mat):
+		res: Mat = Mat([])
+		for unit in self.units:
+			x = unit.forward(x)
+			res = np.concat((res, x))
+
+	def reconstruct(self, x: Mat):
+		res: Mat = Mat([])
+		for unit in self.units:
+			x = unit.forward(x)
+			res = np.concat((res, x))
+
+	def train(self, batch: Mat, params: TrainingParams):
+		_batch = np.copy(batch)
+		for unit in self.units:
+			train(unit, _batch, params, Status(), cd_jens)
+			_batch = unit.forward(_batch, num_gibbs=params.num_gibbs_samples)
+

src/rbm/matrix.py

@@ -18,11 +18,18 @@ def uniform(shape: tuple, mu: float = 0.5, std: float = 1.0) -> Mat:
 	return std * (np.random.rand(shape[0], shape[1]) + mu - 0.5)
 
 def gaussian(shape: tuple, mu: float = 0.0, std: float = 1.0) -> Mat:
+	if shape.__len__() == 1:
+		return std * (np.random.randn(shape[0]) + mu)
+	else:
 		return std * (np.random.randn(shape[0], shape[1]) + mu)
 
+
 def sample(src: Mat) -> Mat:
 	return (src > uniform(src.shape)).astype(float)
 
+def gaussian_sample(src: Mat, mu=0, std=1.0) -> Mat:
+	return gaussian(src.shape, mu, std)
+
 def prob(src: Mat) -> Mat:
 	return 1.0 / (1 + np.exp(-src))
 

src/rbm/model.py

@@ -20,11 +20,14 @@ class Model(ABC):
 				obj_list.append(value)
 		return obj_list
 
-	def train(self, batch: Mat, params: TrainingParams):
+	def train(self, batch: Mat, params: TrainingParams|list[TrainingParams]):
+		entities = self.objects(Entity)
+		if isinstance(params, TrainingParams):
+			params = [params]*len(entities)
 		_batch = np.copy(batch)
-		for entity in self.objects(Entity):
+		for entity, param in zip(entities, params):
-			train(entity, _batch, params, Status(), cd_jens)
+			train(entity, _batch, param, Status(), cd_jens)
-			_batch = entity.forward(_batch, num_gibbs=params.num_gibbs_samples)
+			_batch = entity.forward(_batch, num_gibbs=param.num_gibbs_samples)
 
 	@abstractmethod
 	def forward(self, x: Mat) -> Mat:

src/tests/test_deep_norbs.py

@@ -0,0 +1,65 @@
+import os
+import matplotlib.pyplot as plt
+
+from rbm.model import Model
+from rbm.entity import Entity, EntityParams
+from rbm.matrix import Mat, np, read_armadillo
+from rbm.train import TrainingParams
+
+class TestModel(Model):
+	def __init__(self, name: str, work_dir: str = '.'):
+		super().__init__(name, work_dir)
+		self.unit1 = Entity((96*96, 16), EntityParams(do_gaussian_visible=True, do_gaussian_hidden=True))
+		self.unit2 = Entity((16, 16), EntityParams())
+
+	def forward(self, x: Mat):
+		x = self.unit1.forward(x)
+		x = self.unit2.forward(x)
+		return x
+
+	def backward(self, x: Mat):
+		x = self.unit2.reconstruct(x)
+		x = self.unit1.reconstruct(x)
+		return x
+
+if __name__ == "__main__":
+	work_dir = "results"
+	prj_name = "deep_norb_small_16h_v2"
+	prj_root = "/home/jens/work/repos/Rbm"
+
+	# Create model
+	model = TestModel(prj_name, "results")
+
+	# Init state
+	model.init(0.01)
+
+	# load state
+	model.load()
+
+	# Load train data
+	train_batch = read_armadillo(os.path.join(prj_root, f"norb_small_16h_v2.training.dat"))
+
+	# Load test data
+	test_batch = read_armadillo(os.path.join(prj_root, f"norb_small_16h_v2.test.dat"))
+
+	# Train
+	model.train(train_batch,[
+		TrainingParams(learning_rate=0.00001, momentum=0.9, do_rao_blackwell=True, num_epochs=1000, num_gibbs_samples=3),
+		TrainingParams(learning_rate=0.01, momentum=0.9, do_rao_blackwell=True, num_epochs=1000, num_gibbs_samples=1)
+		])
+
+	# save state
+	model.save()
+
+	fig, axes = plt.subplots(1, len(test_batch), figsize=(12, 3))
+	for index, inp in enumerate(test_batch):
+		out_normalized = model.backward(model.forward(inp))
+		img = 2*(out_normalized + 0.5)
+		img = np.reshape(img, (96, 96))
+		axes[index].imshow(img)
+		axes[index].axis('off')
+
+	plt.show()
+
+
+

src/tests/test_learn_norbs_labels.py

@@ -0,0 +1,63 @@
+import os
+import matplotlib.pyplot as plt
+
+from rbm.model import Model
+from rbm.entity import Entity, EntityParams
+from rbm.matrix import Mat, np, read_armadillo
+from rbm.train import TrainingParams
+from rbm.layout import Horizontal
+
+class TestModel(Model):
+	def __init__(self, name: str, work_dir: str = '.'):
+		super().__init__(name, work_dir)
+		self.unit1 = Horizontal([
+			Entity((96*96, 16), EntityParams(do_gaussian_visible=True, do_gaussian_hidden=True)),
+			Entity((16, 16), EntityParams())
+			])
+
+	def forward(self, x: Mat):
+		x = self.unit1.forward(x)
+		return x
+
+	def reconstruct(self, x: Mat):
+		x = self.unit1.reconstruct(x)
+		return x
+
+if __name__ == "__main__":
+	work_dir = "results"
+	prj_name = "norb_small_16h_v2"
+	prj_root = "/home/jens/work/repos/Rbm"
+
+	# Create model
+	model = TestModel("norb_small_16h_v2", "results")
+
+	# Init state
+	model.init(0.01)
+
+	# load state
+	model.load()
+
+	# Load train data
+	train_batch = read_armadillo(os.path.join(prj_root, f"{prj_name}.training.dat"))
+
+	# Load test data
+	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))
+
+	# save state
+	model.save()
+
+	fig, axes = plt.subplots(1, len(test_batch), figsize=(12, 3))
+	for index, inp in enumerate(test_batch):
+		out_normalized = model.reconstruct(model.forward(inp))
+		img = 2*(out_normalized + 0.5)
+		img = np.reshape(img, (96, 96))
+		axes[index].imshow(img)
+		axes[index].axis('off')
+
+	plt.show()
+
+
+