Initial commit

.gitignore

@@ -0,0 +1,6 @@
+.idea
+build
+results
+images
+*.egg-info
+__pycache__

pyproject.toml

@@ -0,0 +1,21 @@
+requires = [
+    "setuptools>=61.0"
+]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "pyRbm"
+description = "Python Restricted Boltzmann Machine"
+dynamic = ["version"]
+requires-python = ">=3.12"
+authors = [
+    { name = "Jens Ahrensfeld" }
+]
+dependencies = [
+    "numpy"
+]
+readme = "README.md"
+
+[tool.setuptools]
+package-dir = {"" = "src"}
+packages = ["rbm"]

src/rbm/cd_train.py

@@ -0,0 +1,75 @@
+import numpy as np
+
+from params import RbmParams
+from rbm.helper import gaussian
+from state import RbmState
+from helper import prob, sample
+
+class CdTrain:
+	def __init__(self, state: RbmState, params: RbmParams):
+		self.state = state
+		self.params = params
+
+	def v_to_h(self, visible: np.ndarray) -> np.ndarray:
+		return np.vecmat(visible, self.state.w_hv) + self.state.b_h
+
+	def h_to_v(self, hidden: np.ndarray) -> np.ndarray:
+		return np.vecmat(hidden, np.transpose(self.state.w_hv)) + self.state.b_v
+
+	def v_to_h_prob(self, v: np.ndarray) -> np.ndarray:
+		state = self.v_to_h(v)
+		if self.params.do_gaussian_visible:
+			return state
+
+		return prob(state)
+
+	def h_to_v_prob(self, h: np.ndarray) -> np.ndarray:
+		state = self.h_to_v(h)
+		if self.params.do_gaussian_visible:
+			return state
+
+		return prob(state)
+
+
+class CdTrainJens(CdTrain):
+	def __init__(self, state: RbmState, params: RbmParams):
+		CdTrain.__init__(self, state, params)
+
+	def train(self, v_states: np.ndarray):
+		v_probs = v_states
+		h_states = self.v_to_h_prob(v_states)
+		h_probs = h_states
+
+		if self.params.do_gaussian_hidden:
+			h_states += gaussian(h_states.shape)
+		else:
+			h_probs = self.v_to_h_prob(v_states)
+			if self.params.do_rao_blackwell:
+				h_states = h_probs
+			else:
+				h_states = sample(h_probs)
+
+		# Update weights (positive phase)
+		dw = np.transpose(v_states) * h_states
+		dbv = np.sum(v_states, 0)
+		dbh = np.sum(h_states, 0)
+
+		# Gibbs	sampling with training params
+		for i in range(self.params.num_gibbs_samples):
+			if self.params.do_gibbs_sample_hidden:
+				v_probs = self.h_to_v_prob(sample(h_probs))
+			else:
+				v_probs = self.h_to_v_prob(h_probs)
+
+			# Create hidden representation given v
+			if self.params.do_gaussian_visible:
+				h_probs = self.v_to_h_prob(sample(v_probs))
+			else:
+				h_probs = self.v_to_h_prob(v_probs)
+
+		# Update weights (negative phase)
+		dw -= np.transpose(v_probs) * h_probs
+		dbv -= np.sum(v_probs, 0)
+		dbh -= np.sum(h_probs, 0)
+
+		return dw, dbv, dbh

src/rbm/helper.py

@@ -0,0 +1,16 @@
+import numpy as np
+import time
+
+rng = np.random.default_rng(int(time.monotonic()))
+
+def uniform(shape: tuple, mu: float = 0.5, std: float = 1.0) -> np.ndarray:
+	return std * (rng.uniform(size=shape) + mu - 0.5)
+
+def gaussian(shape: tuple, mu: float = 0.5, std: float = 1.0) -> np.ndarray:
+	return rng.normal(size=shape, loc=mu, scale=std)
+
+def sample(src: np.ndarray) -> np.ndarray:
+	return (src > uniform(src.shape)).astype(float)
+
+def prob(src: np.ndarray) -> np.ndarray:
+	return 1.0 / (1 + np.exp(-src))

src/rbm/layer.py

@@ -0,0 +1,50 @@
+import numpy as np
+from collections.abc import Callable
+from params import RbmParams
+from state import RbmState
+from helper import sample, prob, uniform
+
+class RbmLayer:
+	def __init__(self, name: str, num_visible, num_hidden, params: RbmParams):
+		self.name = name
+		self.state = RbmState.from_layer_params(num_visible, num_hidden)
+		self.params = params
+		self.state_filename = f"{self.name}_state.npz"
+
+	def save(self):
+		self.state.to_file(self.state_filename)
+
+	def load(self):
+		self.state = RbmState.from_file(self.state_filename)
+
+	def train_batch(self, v_states: np.ndarray, cd_func: Callable):
+		for epochs in range(self.params.num_epochs):
+			# Contrastive divergence learning: calculate gradients
+			cd_func(v_states, dwhv, dbh, dbv);
+
+
+
+if __name__ == "__main__":
+	p = RbmParams()
+	l1 = RbmLayer("Layer_0", 2, 3, p)
+	l1.save()
+	l1.load()
+	l1.state.init(mu=0.5, std=1.0)
+
+	s1 = RbmState(l1.state.w_hv, l1.state.b_v, l1.state.b_h)
+
+	v0 = uniform((1,2))
+	h0 = uniform((1,3))
+
+	h1 = l1.v_to_h(v0)
+	print(h1.shape)
+	v1 = l1.h_to_v(h1)
+	s_v1 = sample(v1)
+	p_h = prob(s_v1)
+
+	print(v1.shape)
+
+
+
+
+

src/rbm/params.py

@@ -0,0 +1,46 @@
+import json
+
+class Params:
+	def save(self, filename: str):
+		with open(filename, "w") as fp:
+			json.dump(self.__dict__, fp, indent=4)
+
+	def load(self, filename: str):
+		with open(filename, "r") as fp:
+			self.__dict__ = json.load(fp)
+
+class RbmParams(Params):
+	def __init__(self):
+		self.learning_rate = 0
+		self.momentum = 0
+		self.weight_decay = 0
+		self.num_epochs = 1
+		self.num_gibbs_samples = 1
+		self.mini_batch_size = 1
+
+		self.do_gaussian_visible = False
+		self.do_gaussian_hidden = False
+		self.do_rao_blackwell = False
+
+		self.do_gibbs_sample_visible = False
+		self.do_gibbs_sample_hidden = False
+		self.do_batch_sample = False
+
+
+class LayerParams(Params):
+	def __init__(self, num_visible, num_hidden):
+		self.num_visible = num_visible
+		self.num_hidden = num_hidden
+
+
+if __name__ == "__main__":
+	test_filename = "../../test_params.json"
+	p1 = RbmParams()
+	p1.num_epochs = 31101970
+	p1.save(test_filename)
+
+	p2 = RbmParams()
+	p2.load(test_filename)
+
+	assert p2.__dict__ == p1.__dict__
+

src/rbm/state.py

@@ -0,0 +1,41 @@
+import numpy as np
+from helper import uniform
+
+
+class RbmState:
+	def __init__(self, w_hv: np.ndarray, b_v: np.ndarray, b_h: np.ndarray):
+		self.num_visible, self.num_hidden = w_hv.shape
+		self.w_hv = w_hv
+		self.b_v = b_v
+		self.b_h = b_h
+
+	@classmethod
+	def from_layer_params(cls, num_visible, num_hidden):
+		w_hv = np.zeros((num_visible, num_hidden))
+		b_v = np.zeros((1, num_visible))
+		b_h = np.zeros((1, num_hidden))
+		obj = cls(w_hv, b_v, b_h)
+		return obj
+
+	@classmethod
+	def from_file(cls, filename: str):
+		try:
+			with np.load(filename) as X:
+				w_hv, b_v, b_h = [X[i] for i in ('whv', 'bv', 'bh')]
+				print(f"{filename} loaded successfully!")
+		except FileNotFoundError:
+			pass
+		except KeyError:
+			pass
+
+		obj = cls(w_hv, b_v, b_h)
+		return obj
+
+	def to_file(self, filename: str):
+		np.savez(filename, whv=self.w_hv, bv=self.b_v, bh=self.b_h)
+		print(f"{filename} saved successfully!")
+
+	def init(self, mu: float = 0.5, std: float = 1.0):
+		self.w_hv = uniform(self.w_hv.shape, mu, std)
+		self.b_v = uniform(self.b_v.shape, 0, 0)
+		self.b_h = uniform(self.b_h.shape, 0, 0)

test_params.json

@@ -0,0 +1,14 @@
+{
+    "learning_rate": 0,
+    "momentum": 0,
+    "weight_decay": 0,
+    "num_epochs": 31101970,
+    "num_gibbs_samples": 1,
+    "mini_batch_size": 1,
+    "do_gaussian_visible": false,
+    "do_gaussian_hidden": false,
+    "do_rao_blackwell": false,
+    "do_gibbs_sample_visible": false,
+    "do_gibbs_sample_hidden": false,
+    "do_batch_sample": false
+}