From 6a4b10f8753989ecc3238f3f8f4eb385f1db2dd6 Mon Sep 17 00:00:00 2001 From: "David S. Reiner" <420230+chronicgiardia@users.noreply.github.com> Date: Thu, 13 Nov 2025 01:04:03 -0800 Subject: [PATCH 1/2] Add Pylint workflow for Python code analysis --- .github/workflows/pylint.yml | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) create mode 100644 .github/workflows/pylint.yml diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml new file mode 100644 index 0000000..c73e032 --- /dev/null +++ b/.github/workflows/pylint.yml @@ -0,0 +1,23 @@ +name: Pylint + +on: [push] + +jobs: + build: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.8", "3.9", "3.10"] + steps: + - uses: actions/checkout@v4 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v3 + with: + python-version: ${{ matrix.python-version }} + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install pylint + - name: Analysing the code with pylint + run: | + pylint $(git ls-files '*.py') From 888f5c345a3e54e5703013d5c38ec69936e8f5a3 Mon Sep 17 00:00:00 2001 From: heisenberg Date: Tue, 30 Jun 2026 21:29:28 -0700 Subject: [PATCH 2/2] Add --LR_model and --demo options to gene_vec_model.py Refactor gene_vec_model.py to expose the embedding pipeline via main() and add helpers for the new CLI arguments: - --LR_model: train (or load) a logistic regression model over gene-pair embedding features and persist it as gene_pair_lr.pkl in the directory. - --demo: export a copy of the learned embeddings into the demo directory. Add a unittest suite covering argument parsing, feature construction, LR model save/load, and demo export. Document both arguments in the README. Co-Authored-By: Oz --- README.md | 9 ++ src/gene_vec_model.py | 193 +++++++++++++++++++++++---------- src/test_gene_vec_model.py | 213 +++++++++++++++++++++++++++++++++++++ 3 files changed, 358 insertions(+), 57 deletions(-) create mode 100644 src/test_gene_vec_model.py diff --git a/README.md b/README.md index 847fef9..c6d6b2e 100644 --- a/README.md +++ b/README.md @@ -98,11 +98,20 @@ python gene_vec_model.py -h A TAB separated file where each line starts with a GO/ARCHS4 category ID, following by the number of genes in the category, following by the genes associated with the category separated by comma. 3. Path to save the learned embeddings (`--outfile`) +4. Path to a directory to save or load a logistic regression (LR) model (`--LR_model`, optional) + + When specified, after training the gene embeddings, a logistic regression model is trained on the learned embeddings to predict whether a pair of genes is associated (using the element-wise product of the two genes' embeddings as features). The model is saved as `gene_pair_lr.pkl` inside this directory. If a model already exists at that path, it is loaded instead of being retrained. This argument requires `scikit-learn`. + +5. Path to the demo directory (`--demo`, optional) + + When specified, a copy of the learned embeddings file (`--outfile`) is exported into this directory so the demo scripts can consume it. The file is placed in the `data/` subdirectory when one exists, otherwise directly in the given directory. ### Output 1. A gene embedding file will be saved as the specified `--outfile`. A comma separated file where each line indicates a gene embedding. The first item in the line is the gene ID, following by a list of float numbers which is the embedding representation of the gene. +2. If `--LR_model` is specified, the trained LR model is saved as `gene_pair_lr.pkl` in that directory. +3. If `--demo` is specified, a copy of the embeddings file is written into the demo directory (its `data/` subdirectory when present). ### Generate gene signature embeddings Use the following command to run examples of generating gene signature embeddings from embeddings of individual genes: diff --git a/src/gene_vec_model.py b/src/gene_vec_model.py index a0ebb62..18e7822 100644 --- a/src/gene_vec_model.py +++ b/src/gene_vec_model.py @@ -1,6 +1,9 @@ import numpy as np import tensorflow as tf import gc +import os +import shutil +import pickle from utils.random_walk import random_walk_w_restart as rwr from utils.sampling_util import rw_sampling from tensorflow.python.keras import backend as K @@ -17,6 +20,10 @@ def parse_args(): help='Path to a file containing GO term gene associations or ARCHS4 experiments gene associations') parser.add_argument('--outfile', default='../data/gene_vec_go.csv', help='Path to save the learned embeddings') + parser.add_argument('--LR_model', default=None, + help='Path to a directory to save or load the logistic regression model') + parser.add_argument('--demo', default=None, + help='Path to the demo directory to export the learned embeddings into') return parser.parse_args() def get_model(vocab_size, latent_dim, ori_dim): @@ -36,60 +43,132 @@ def get_model(vocab_size, latent_dim, ori_dim): return encoder, model -args = parse_args() -datatype, association_file, outfile = args.datatype, args.association_file, args.outfile - -geneids, association_mat, diffusion_state = rwr(datatype, association_file) -nonzeroidx = np.where(np.sum(association_mat, axis=1) > 0)[0] - -geneids = np.array(geneids) -fp_array = association_mat[nonzeroidx] -geneids = geneids[nonzeroidx] - -ori_dim = fp_array.shape[-1] -latent_dim = 256 - -encoder, autoencoder = get_model(len(geneids), latent_dim, ori_dim) -optimizer = keras.optimizers.Adam() -autoencoder.compile( - optimizer=optimizer, - loss=[losses.BinaryCrossentropy(from_logits=True)] -) - -rs = rw_sampling(diffusion_state) -epochs = 3000 -for e in range(epochs): - print('epoch:', e) - left_input = [] - right_input = [] - targets = [] - pos_pairs = 1 - neg_pairs = 5 - #sample training pairs - left_input, right_input, targets = rs.sampling(np.arange(len(geneids)), pos_pairs, neg_pairs) - - left_input = np.squeeze(np.array(left_input)) - right_input = np.squeeze(np.array(right_input)) - targets = np.squeeze(np.array(targets)) - sample_size = len(left_input) - smaple_idx = np.arange(sample_size) - np.random.shuffle(smaple_idx) - left_input = left_input[smaple_idx] - right_input = right_input[smaple_idx] - targets = targets[smaple_idx] - - autoencoder.fit([left_input[:int(0.8*sample_size)] * 1.0, right_input[:int(0.8*sample_size)] * 1.0], [targets[:int(0.8*sample_size)]], - batch_size=1000, - epochs=1, - verbose=1, - validation_data=([left_input[int(0.8*sample_size):] * 1.0, right_input[int(0.8*sample_size):] * 1.0], [targets[int(0.8*sample_size):]])) - gc.collect() - -encoded_genes = encoder.predict(np.arange(len(geneids)) * 1.0) -fw = open(outfile, 'w') -for i in range(len(encoded_genes)): - fw.write(geneids[i]) - for j in range(len(encoded_genes[0])): - fw.write(',' + str(encoded_genes[i, j])) - fw.write('\n') -fw.close() +def build_pair_features(encoded_genes, left_idx, right_idx): + '''Build feature vectors for gene pairs from the learned embeddings. + + The feature for a pair is the element-wise product of the two genes' + embedding vectors. ``left_idx`` and ``right_idx`` are arrays of gene + indices into ``encoded_genes``. + ''' + encoded_genes = np.asarray(encoded_genes) + left_idx = np.asarray(left_idx, dtype=int) + right_idx = np.asarray(right_idx, dtype=int) + return encoded_genes[left_idx] * encoded_genes[right_idx] + +def train_or_load_lr_model(X, y, lr_dir, filename='gene_pair_lr.pkl'): + '''Train, or load if already present, a logistic regression model. + + The model predicts whether a pair of genes is associated based on the + gene-pair features ``X`` and labels ``y``. It is persisted as a pickle + file inside ``lr_dir``; if that file already exists it is loaded instead + of being retrained. Returns a tuple ``(model, model_path)``. + ''' + from sklearn.linear_model import LogisticRegression + + os.makedirs(lr_dir, exist_ok=True) + lr_path = os.path.join(lr_dir, filename) + + if os.path.exists(lr_path): + with open(lr_path, 'rb') as f: + lr = pickle.load(f) + print('Loaded LR model from', lr_path) + else: + lr = LogisticRegression(max_iter=1000) + lr.fit(X, y) + with open(lr_path, 'wb') as f: + pickle.dump(lr, f) + print('Saved LR model to', lr_path) + + return lr, lr_path + +def export_embeddings_for_demo(outfile, demo_dir): + '''Copy the learned embeddings file into the demo directory. + + When ``demo_dir`` contains a ``data`` subdirectory the file is placed + there (where the demo scripts look for it), otherwise it is copied into + ``demo_dir`` directly. Returns the destination path. + ''' + demo_data_dir = os.path.join(demo_dir, 'data') + dest_dir = demo_data_dir if os.path.isdir(demo_data_dir) else demo_dir + os.makedirs(dest_dir, exist_ok=True) + demo_outfile = os.path.join(dest_dir, os.path.basename(outfile)) + shutil.copyfile(outfile, demo_outfile) + print('Copied embeddings for demo to', demo_outfile) + return demo_outfile + +def main(): + args = parse_args() + datatype, association_file, outfile = args.datatype, args.association_file, args.outfile + LR_model, demo = args.LR_model, args.demo + + geneids, association_mat, diffusion_state = rwr(datatype, association_file) + nonzeroidx = np.where(np.sum(association_mat, axis=1) > 0)[0] + + geneids = np.array(geneids) + fp_array = association_mat[nonzeroidx] + geneids = geneids[nonzeroidx] + + ori_dim = fp_array.shape[-1] + latent_dim = 256 + + encoder, autoencoder = get_model(len(geneids), latent_dim, ori_dim) + optimizer = keras.optimizers.Adam() + autoencoder.compile( + optimizer=optimizer, + loss=[losses.BinaryCrossentropy(from_logits=True)] + ) + + rs = rw_sampling(diffusion_state) + epochs = 3000 + for e in range(epochs): + print('epoch:', e) + left_input = [] + right_input = [] + targets = [] + pos_pairs = 1 + neg_pairs = 5 + #sample training pairs + left_input, right_input, targets = rs.sampling(np.arange(len(geneids)), pos_pairs, neg_pairs) + + left_input = np.squeeze(np.array(left_input)) + right_input = np.squeeze(np.array(right_input)) + targets = np.squeeze(np.array(targets)) + sample_size = len(left_input) + smaple_idx = np.arange(sample_size) + np.random.shuffle(smaple_idx) + left_input = left_input[smaple_idx] + right_input = right_input[smaple_idx] + targets = targets[smaple_idx] + + autoencoder.fit([left_input[:int(0.8*sample_size)] * 1.0, right_input[:int(0.8*sample_size)] * 1.0], [targets[:int(0.8*sample_size)]], + batch_size=1000, + epochs=1, + verbose=1, + validation_data=([left_input[int(0.8*sample_size):] * 1.0, right_input[int(0.8*sample_size):] * 1.0], [targets[int(0.8*sample_size):]])) + gc.collect() + + encoded_genes = encoder.predict(np.arange(len(geneids)) * 1.0) + fw = open(outfile, 'w') + for i in range(len(encoded_genes)): + fw.write(geneids[i]) + for j in range(len(encoded_genes[0])): + fw.write(',' + str(encoded_genes[i, j])) + fw.write('\n') + fw.close() + + # Train (or load) a logistic regression model that predicts whether a pair + # of genes is associated, using the learned gene embeddings as features. + # The model is persisted in the directory given by --LR_model. + if LR_model is not None: + X_lr = build_pair_features(encoded_genes, left_input, right_input) + y_lr = np.asarray(targets, dtype=int) + lr, _ = train_or_load_lr_model(X_lr, y_lr, LR_model) + print('LR gene-pair prediction accuracy:', lr.score(X_lr, y_lr)) + + # Export a copy of the learned embeddings into the demo directory so that + # the demo scripts (e.g. classifier.py) can consume the trained embeddings. + if demo is not None: + export_embeddings_for_demo(outfile, demo) + +if __name__ == '__main__': + main() diff --git a/src/test_gene_vec_model.py b/src/test_gene_vec_model.py new file mode 100644 index 0000000..d8d1a68 --- /dev/null +++ b/src/test_gene_vec_model.py @@ -0,0 +1,213 @@ +"""Tests for the --LR_model and --demo functionality of gene_vec_model.py. + +The heavy training pipeline in gene_vec_model.py lives inside ``main()`` and is +guarded by ``if __name__ == '__main__'``, so importing the module does not +trigger any training. These tests exercise the small, self-contained helpers +that back the --LR_model and --demo command line arguments. + +These tests use only the standard-library ``unittest`` framework so they can be +run without pytest: + + # from the src/ directory + python3 -m unittest test_gene_vec_model -v + +(pytest, if installed, can also collect and run this file.) + +The logistic-regression tests require scikit-learn; they are skipped +automatically when it is not installed. +""" +import importlib +import os +import pickle +import sys +import tempfile +import unittest + +import numpy as np + +# Make sure the module under test (and its ``utils`` package) is importable +# regardless of the directory the tests are launched from. +SRC_DIR = os.path.dirname(os.path.abspath(__file__)) +if SRC_DIR not in sys.path: + sys.path.insert(0, SRC_DIR) + +gvm = importlib.import_module("gene_vec_model") + +try: + from sklearn.linear_model import LogisticRegression + HAS_SKLEARN = True +except ImportError: # scikit-learn is an optional dependency for these tests + HAS_SKLEARN = False + + +def _separable_dataset(): + rng = np.random.RandomState(0) + X_pos = rng.normal(loc=2.0, scale=0.2, size=(40, 3)) + X_neg = rng.normal(loc=-2.0, scale=0.2, size=(40, 3)) + X = np.vstack([X_pos, X_neg]) + y = np.array([1] * 40 + [0] * 40) + return X, y + + +class ParseArgsTest(unittest.TestCase): + def test_accepts_lr_model_and_demo(self): + argv = [ + "gene_vec_model.py", + "--datatype", "data/", + "--association_file", "results/", + "--outfile", "saved_model/", + "--LR_model", "LR_model/", + "--demo", "demo/", + ] + old_argv = sys.argv + try: + sys.argv = argv + args = gvm.parse_args() + finally: + sys.argv = old_argv + self.assertEqual(args.LR_model, "LR_model/") + self.assertEqual(args.demo, "demo/") + + def test_lr_model_and_demo_default_to_none(self): + old_argv = sys.argv + try: + sys.argv = ["gene_vec_model.py"] + args = gvm.parse_args() + finally: + sys.argv = old_argv + self.assertIsNone(args.LR_model) + self.assertIsNone(args.demo) + + +class BuildPairFeaturesTest(unittest.TestCase): + def test_is_elementwise_product(self): + encoded_genes = np.array([[1.0, 2.0], + [3.0, 4.0], + [5.0, 6.0]]) + feats = gvm.build_pair_features(encoded_genes, [0, 2], [1, 1]) + expected = np.array([[1.0 * 3.0, 2.0 * 4.0], + [5.0 * 3.0, 6.0 * 4.0]]) + self.assertEqual(feats.shape, (2, 2)) + np.testing.assert_allclose(feats, expected) + + def test_casts_float_indices(self): + # Indices arrive as floats after np.squeeze on sampled data. + encoded_genes = np.arange(6.0).reshape(3, 2) + feats = gvm.build_pair_features( + encoded_genes, np.array([0.0, 1.0]), np.array([2.0, 2.0])) + expected = np.vstack([encoded_genes[0] * encoded_genes[2], + encoded_genes[1] * encoded_genes[2]]) + np.testing.assert_allclose(feats, expected) + + +@unittest.skipUnless(HAS_SKLEARN, "scikit-learn is required for the LR model tests") +class TrainOrLoadLRModelTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.tmp = self._tmp.name + + def tearDown(self): + self._tmp.cleanup() + + def test_saves_model_when_absent(self): + X, y = _separable_dataset() + lr_dir = os.path.join(self.tmp, "LR_model") + + model, path = gvm.train_or_load_lr_model(X, y, lr_dir) + + self.assertEqual(path, os.path.join(lr_dir, "gene_pair_lr.pkl")) + self.assertTrue(os.path.exists(path)) + self.assertIsInstance(model, LogisticRegression) + self.assertAlmostEqual(model.score(X, y), 1.0) + with open(path, "rb") as f: + reloaded = pickle.load(f) + np.testing.assert_allclose(reloaded.coef_, model.coef_) + + def test_creates_directory_if_missing(self): + X, y = _separable_dataset() + lr_dir = os.path.join(self.tmp, "nested", "LR_model") + self.assertFalse(os.path.exists(lr_dir)) + + gvm.train_or_load_lr_model(X, y, lr_dir) + + self.assertTrue(os.path.isdir(lr_dir)) + self.assertTrue(os.path.exists(os.path.join(lr_dir, "gene_pair_lr.pkl"))) + + def test_loads_existing_model_without_retraining(self): + lr_dir = os.path.join(self.tmp, "LR_model") + os.makedirs(lr_dir) + path = os.path.join(lr_dir, "gene_pair_lr.pkl") + + # Pre-train and persist a model on one dataset. + X_a, y_a = _separable_dataset() + original = LogisticRegression(max_iter=1000).fit(X_a, y_a) + with open(path, "wb") as f: + pickle.dump(original, f) + + # Call with a *different* dataset; because a model already exists it + # must be loaded as-is rather than retrained on the new data. + X_b = X_a + 100.0 + y_b = 1 - y_a + model, returned_path = gvm.train_or_load_lr_model(X_b, y_b, lr_dir) + + self.assertEqual(returned_path, path) + np.testing.assert_allclose(model.coef_, original.coef_) + np.testing.assert_allclose(model.intercept_, original.intercept_) + + def test_custom_filename_is_respected(self): + X, y = _separable_dataset() + lr_dir = os.path.join(self.tmp, "LR_model") + model, path = gvm.train_or_load_lr_model(X, y, lr_dir, filename="custom.pkl") + self.assertEqual(path, os.path.join(lr_dir, "custom.pkl")) + self.assertTrue(os.path.exists(path)) + + +class ExportEmbeddingsForDemoTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.tmp = self._tmp.name + + def tearDown(self): + self._tmp.cleanup() + + def _write_embeddings(self, path): + content = "1234,0.1,0.2\n5678,0.3,0.4\n" + with open(path, "w") as f: + f.write(content) + return content + + def test_copies_into_data_subdir_when_present(self): + outfile = os.path.join(self.tmp, "gene_vec_go.csv") + content = self._write_embeddings(outfile) + demo_dir = os.path.join(self.tmp, "demo") + os.makedirs(os.path.join(demo_dir, "data")) + + dest = gvm.export_embeddings_for_demo(outfile, demo_dir) + + self.assertEqual(dest, os.path.join(demo_dir, "data", "gene_vec_go.csv")) + self.assertTrue(os.path.exists(dest)) + with open(dest) as f: + self.assertEqual(f.read(), content) + + def test_copies_into_demo_dir_when_no_data_subdir(self): + outfile = os.path.join(self.tmp, "gene_vec_go.csv") + content = self._write_embeddings(outfile) + demo_dir = os.path.join(self.tmp, "demo") # no data/ subdirectory + + dest = gvm.export_embeddings_for_demo(outfile, demo_dir) + + self.assertEqual(dest, os.path.join(demo_dir, "gene_vec_go.csv")) + self.assertTrue(os.path.exists(dest)) + with open(dest) as f: + self.assertEqual(f.read(), content) + + def test_preserves_outfile_basename(self): + outfile = os.path.join(self.tmp, "my_embeddings.csv") + self._write_embeddings(outfile) + demo_dir = os.path.join(self.tmp, "demo") + dest = gvm.export_embeddings_for_demo(outfile, demo_dir) + self.assertEqual(os.path.basename(dest), "my_embeddings.csv") + + +if __name__ == "__main__": + unittest.main()