commit b6e94e75acb9686e19afa4f1a71c5582d0176d10 Author: Jens Ahrensfeld Date: Tue Dec 16 14:52:36 2025 +0100 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2e27c8c --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +.idea +build +results +images +*.egg-info +__pycache__ diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..832dab2 --- /dev/null +++ b/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"] diff --git a/src/rbm/__init__.py b/src/rbm/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/rbm/cd_train.py b/src/rbm/cd_train.py new file mode 100644 index 0000000..669445b --- /dev/null +++ b/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 \ No newline at end of file diff --git a/src/rbm/helper.py b/src/rbm/helper.py new file mode 100644 index 0000000..23559b6 --- /dev/null +++ b/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)) diff --git a/src/rbm/layer.py b/src/rbm/layer.py new file mode 100644 index 0000000..e750911 --- /dev/null +++ b/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) + + + + + diff --git a/src/rbm/params.py b/src/rbm/params.py new file mode 100644 index 0000000..b41d1b5 --- /dev/null +++ b/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__ + diff --git a/src/rbm/rbm.py b/src/rbm/rbm.py new file mode 100644 index 0000000..e69de29 diff --git a/src/rbm/state.py b/src/rbm/state.py new file mode 100644 index 0000000..3b71cd0 --- /dev/null +++ b/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) diff --git a/test_params.json b/test_params.json new file mode 100644 index 0000000..864df3b --- /dev/null +++ b/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 +} \ No newline at end of file