- refactored Rbm

- added myStatus with realtime update of reconstruction
- DeepStack: added up/down passes for multilayer Stacks

src/rbm/rbm.py

@@ -2,6 +2,30 @@ import os.path
 import numpy as np
 import cv2 as cv
 from stack_factory import StackFactory
+from status import Status
+from stack_deep import StackDeep
+
+def cv_show(name: str, vec: np.array, shape):
+	img = cv.Mat(np.resize(vec, shape))
+	img_n = cv.normalize(src=img, dst=None, alpha=255, beta=0, norm_type=cv.NORM_MINMAX, dtype=cv.CV_8U)
+	cv.imshow(f"{name}", img_n)
+
+class MyStatus(Status):
+	def __init__(self, _stack: StackDeep, _batch: np.ndarray):
+		Status.__init__(self)
+		self.stack = _stack
+		self.batch = _batch
+
+	def on_change(self, status: dict) -> bool:
+		# Show reconstruction
+		# Shape of training vector
+		shape = self.stack.from_index(0).shape[0:2] + (1,)
+
+		for n in range(self.batch.shape[0]):
+			img = self.stack.pass_down_up(self.batch[n,:])
+			cv_show("img", img, shape)
+			cv.waitKeyEx(1)
+		return True
 
 def read_armadillo(filename: str) -> np.ndarray:
 	result = None
@@ -23,11 +47,6 @@ def read_armadillo(filename: str) -> np.ndarray:
 
 	return result
 
-def cv_show(vec: np.array, shape):
-	img = cv.Mat(np.resize(vec, shape))
-	img_n = cv.normalize(src=img, dst=None, alpha=255, beta=0, norm_type=cv.NORM_MINMAX, dtype=cv.CV_8U)
-	cv.imshow(f"training samples", img_n)
-
 def main(prj_name: str = "test"):
 	work_dir = "../../results"
 	prj_root = "/home/jens/work/repos/Rbm"
@@ -45,21 +64,17 @@ def main(prj_name: str = "test"):
 	training_data_path = os.path.join(prj_root, f"{prj_name}.training.dat")
 	batch = read_armadillo(training_data_path)
 
-	# Shape of training vector
-	shape = stack.from_index(0).shape[0:2] + (1,)
+	# Prepare status listener
+	my_status = MyStatus(stack, batch)
 
 	# Train
-	stack.train(batch)
+	stack.train(batch, status=my_status)
 
 	# Save state
 	stack.state_save()
 
-	for n in range(batch.shape[0]):
-		cv_show(batch[n,:], shape)
-		cv.waitKeyEx(200)
-
 if __name__ == "__main__":
-	main("norb_small_16h")
+	main("prims")
 	cv.destroyAllWindows()
 
 	print("Test: [passed]")

src/rbm/stack_deep.py

@@ -16,10 +16,27 @@ class StackDeep(Stack):
 
 		return _batch
 
-	def train(self, batch: np.ndarray):
+	def train(self, batch: np.ndarray, status=Status()):
 		_batch = np.matrix.copy(batch)
 		for index, layer in enumerate(self.layers):
 			print(f"Train layer {index}")
 			_batch = self.batch_from(batch, index)
-			layer.train(_batch, cd_func=cd_jens, status=Status())
+			layer.train(_batch, cd_func=cd_jens, status=status)
 
+
+	def pass_up(self, visible: np.ndarray, from_layer_id: int = 0):
+		h = np.matrix.copy(visible)
+		for layer in self.layers[from_layer_id:]:
+			h = layer.gibbs_v_to_h(h)
+		return h
+
+	def pass_down(self, hidden: np.ndarray, from_layer_id: int = 0):
+		v = np.matrix.copy(hidden)
+		for layer in list(reversed(self.layers))[from_layer_id:]:
+			v = layer.gibbs_h_to_v(v)
+		return v
+
+	def pass_down_up(self, visible: np.ndarray, from_layer_id: int = 0):
+		h = self.pass_up(visible, from_layer_id)
+		v = self.pass_down(h)
+		return v