<reponame>VIGS25/benchmark_mining<gh_stars>0
#!/usr/bin/env python3
"""
Integrative Non-negative Matrix Factorization method (iNMF).
Paper: https://doi.org/10.1093/bioinformatics/btv544
"""

__author__ = "<NAME>"

import numpy as np
from benchmark_mining.methods.core import Method, NMF_obj_eval
from benchmark_mining.data import OmicsDataset
from typing import List, Dict, Tuple
import logging
import os

logger = logging.getLogger(__name__)


class iNMF(Method):
  """
  Integrative NMF (iNMF) is a Non-negative matrix factorization method for multi-
  omics integration. It projects all datasets into a common basis space, and
  also accounts for dataset/view specific heterogeneities in its formulation.
  Update equations are derived for gradient based optimization, and several runs
  of the model are needed before convergence to a local optimum. More details can
  be found at: https://doi.org/10.1093/bioinformatics/btv544
  """

  def __init__(self,
               project_dim: int,
               lambda_sparse: float = 0.0,
               lambda_reg: float = 0.0,
               n_runs: int = 200,
               n_steps: int = 100,
               update_factor: float = 1e-2,
               count_thresh: int = 2000,
               save_dir: str = None,
               log_every: int = 50,
               **kwargs) -> None:
    """
    Parameters
    ----------
    project_dim: int
        Dimensionality of the projection space
    lambda_sparse: float, optional
        Sparsity coefficient
    n_runs: int, optional
        Number of times to run the NMF objective
    n_steps: int, optional
        Number of steps for something?
    update_factor: float, optional
        Scaling factor with respect to start point to make update
    count_thresh: float, optional
        # TODO(SOM): Figure out what this is for
    save_dir: str, default None,
        Directory to save results to
    log_every: int, default 50
        Number of iterations per run to log results
    """
    self._project_dim = project_dim
    self._lambda_reg = lambda_reg
    self._lambda_sparse = lambda_sparse
    self._n_runs = n_runs
    self._n_steps = n_steps
    self._update_factor = update_factor
    self._count_thresh = count_thresh
    self._save_dir = save_dir
    self._log_every = log_every
    super(iNMF, self).__init__(**kwargs)

  def _initialize_run(self,
                      n_samples: int,
                      n_datasets: int,
                      feature_counts: List[int],
                      to_log=False) -> None:
    """Initializations prior to training the model.

    Parameters
    ----------
    n_samples: int
        Number of samples in the multi-omics dataset
    n_datasets: int
        Number of multi-omics datasets
    feature_counts: List[int],
        Number of features in each dataset, organized as List in same order
        as dataset
    """
    self.basis_matrix = np.random.uniform(
        low=0, high=2, size=(n_samples, self.project_dim))
    self.coefficients = [
        np.random.uniform(
            low=0, high=2, size=(self.project_dim, feature_counts[i]))
        for i in range(n_datasets)
    ]
    self.data_heterogeneity = [
        np.random.uniform(low=0, high=2, size=(n_samples, self.project_dim))
        for i in range(n_datasets)
    ]

    self.start_eval = 0
    self.old_eval = np.inf
    self.new_eval = 0
    self.count = 0

  def update_basis_matrix(self, X: List[np.ndarray]) -> None:
    """Updates the basis matrix.

    Parameters
    ----------
    X: list(np.ndarray)
        List of multi-omics datasets
    """
    #logger.info("Updating basis matrix.")
    n_datasets = len(X)
    numerator = np.sum(
        [X[i].dot(self.coefficients[i].T) for i in range(n_datasets)], axis=0)
    denominator = np.sum(
        [(self.basis_matrix + self.data_heterogeneity[i]).dot(
            self.coefficients[i]).dot(self.coefficients[i].T)
         for i in range(n_datasets)],
        axis=0)
    self.basis_matrix *= numerator / denominator

  def update_coefficients(self, X: List[np.ndarray]) -> None:
    """Updates the coefficients matrices.

    Parameters
    ----------
    X: list(np.ndarray)
        List of multi-omics datasets
    """
    #logger.info("Updating coefficients.")
    n_datasets = len(X)
    for i in range(n_datasets):
      WV = self.basis_matrix + self.data_heterogeneity[i]
      denominator = (WV.T.dot(WV) + self.lambda_reg * \
      self.data_heterogeneity[i].T.dot(self.data_heterogeneity[i])).dot(
          self.coefficients[i]) + self.lambda_sparse
      self.coefficients[i] *= (WV.T.dot(X[i])) / denominator

  def update_data_heterogeneity(self, X: List[np.ndarray]) -> None:
    """Updates the dataset specific heterogeneity matrix.

    Parameters
    ----------
    X: list(np.ndarray)
        List of multi-omics datasets
    """
    #logger.info("Updating data heterogeneity.")
    n_datasets = len(X)
    for i in range(n_datasets):
      denominator = (self.basis_matrix +
                     (1 + self.lambda_reg) * self.data_heterogeneity[i]).dot(
                         self.coefficients[i]).dot(self.coefficients[i].T)
      self.data_heterogeneity[i] *= X[i].dot(
          self.coefficients[i].T) / denominator

  def update_counts(self, X: List[np.ndarray]) -> None:
    """Updates the counts, which is used in some constraint. #TODO: Update this

    Parameters
    ----------
    X: list(np.ndarray)
        List of multi-omics datasets
    """
    #logger.info("Updating counts.")
    if self.count == 0:
      self.new_eval = NMF_obj_eval(X, self.basis_matrix, self.coefficients,
                                   self.lambda_reg, self.lambda_sparse,
                                   self.data_heterogeneity)
      self.start_eval = self.new_eval
    if self.count != 0 and self.count % self.n_steps == 0:  ## count > thres, != 0
      self.old_eval = self.new_eval
      self.new_eval = NMF_obj_eval(X, self.basis_matrix, self.coefficients,
                                   self.lambda_reg, self.lambda_sparse,
                                   self.data_heterogeneity)
    self.count += 1

  def train(self, omics_dataset: Dict, save: bool = False):
    """Trains the model on multi-omics data.

    Parameters
    ----------
    omics_dataset: dict
        Multi-omics dataset, keyed by datatype
    save: bool, default False
        Whether to save the results
    """
    if type(omics_dataset) != dict:
      raise ValueError("omics_dataset is supposed to be of type dict")

    X = [omics_dataset[i] for i in omics_dataset]
    datatypes = list(omics_dataset.keys())
    n_datasets = len(X)
    n_samples = X[0].shape[0]
    feature_counts = [X[i].shape[1] for i in range(n_datasets)]

    # Initialize best solutions to zero
    basis_matrix_best = np.zeros((n_samples, self.project_dim))
    coefficients_best = {
        data_type: np.zeros((self.project_dim, feature_counts[i]))
        for i, data_type in enumerate(datatypes)
    }
    data_heterogeneities_best = {
        data_type: np.zeros((n_samples, self.project_dim))
        for data_type in datatypes
    }

    objective_vals = []
    n_iter = []
    logger.info("Starting to train the model.\n")
    for run in range(self.n_runs):
      # Initializing different matrices
      self._initialize_run(
          n_datasets=n_datasets,
          n_samples=n_samples,
          feature_counts=feature_counts)
      old_vs_new = np.abs(self.old_eval - self.new_eval)
      start_vs_new = self.start_eval - self.new_eval

      while (old_vs_new > start_vs_new * self.update_factor) and (
          self.count <= self.count_thresh):
        # Updates for basis matrix, which is same for all datasets
        self.update_basis_matrix(X)
        self.update_coefficients(X)
        self.update_data_heterogeneity(X)
        self.update_counts(X)

        # Update difference between old_eval and new_eval
        old_vs_new = np.abs(self.old_eval - self.new_eval)
        start_vs_new = self.start_eval - self.new_eval

      objective_vals.append(self.new_eval)
      n_iter.append(self.count)
      if run % self._log_every == 0:
        logger.info("Iteration Number: {}, Objective Value: {}".format(
            run, self.new_eval))

      # Update best solutions per run
      if objective_vals[-1] == np.min(objective_vals):
        for i, data_type in enumerate(datatypes):
          basis_matrix_best = self.basis_matrix
          coefficients_best[data_type] = self.coefficients[i]
          data_heterogeneities_best[data_type] = self.data_heterogeneity[i]

    logger.info("Model training complete. Processing results.\n")
    logger.info("Minimum objective value: {0:.4f}".format(
        np.min(objective_vals)))

    best_results = {
        "basis_matrix": basis_matrix_best,
        "coefficients": coefficients_best,
        "data_heterogeneities": data_heterogeneities_best,
        "objective_vals": objective_vals,
        "iterations": n_iter
    }

    if save:
      self.save(results=best_results)

    return best_results

  @property
  def project_dim(self):
    return self._project_dim

  @property
  def lambda_reg(self):
    return self._lambda_reg

  @property
  def lambda_sparse(self):
    return self._lambda_sparse

  @property
  def n_runs(self):
    return self._n_runs

  @property
  def n_steps(self):
    return self._n_steps

  @property
  def update_factor(self):
    return self._update_factor

  @property
  def count_thresh(self):
    return self._count_thresh

#!/usr/bin/env python3
"""
Integrative Non-negative Matrix Factorization method (iNMF).
Paper: https://doi.org/10.1093/bioinformatics/btv544
"""

__author__ = "<NAME>"

import numpy as np
from benchmark_mining.methods.core import Method, NMF_obj_eval
from benchmark_mining.data import OmicsDataset
from typing import List, Dict, Tuple
import logging
import os

logger = logging.getLogger(__name__)


class iNMF(Method):
  """
  Integrative NMF (iNMF) is a Non-negative matrix factorization method for multi-
  omics integration. It projects all datasets into a common basis space, and
  also accounts for dataset/view specific heterogeneities in its formulation.
  Update equations are derived for gradient based optimization, and several runs
  of the model are needed before convergence to a local optimum. More details can
  be found at: https://doi.org/10.1093/bioinformatics/btv544
  """

  def __init__(self,
               project_dim: int,
               lambda_sparse: float = 0.0,
               lambda_reg: float = 0.0,
               n_runs: int = 200,
               n_steps: int = 100,
               update_factor: float = 1e-2,
               count_thresh: int = 2000,
               save_dir: str = None,
               log_every: int = 50,
               **kwargs) -> None:
    """
    Parameters
    ----------
    project_dim: int
        Dimensionality of the projection space
    lambda_sparse: float, optional
        Sparsity coefficient
    n_runs: int, optional
        Number of times to run the NMF objective
    n_steps: int, optional
        Number of steps for something?
    update_factor: float, optional
        Scaling factor with respect to start point to make update
    count_thresh: float, optional
        # TODO(SOM): Figure out what this is for
    save_dir: str, default None,
        Directory to save results to
    log_every: int, default 50
        Number of iterations per run to log results
    """
    self._project_dim = project_dim
    self._lambda_reg = lambda_reg
    self._lambda_sparse = lambda_sparse
    self._n_runs = n_runs
    self._n_steps = n_steps
    self._update_factor = update_factor
    self._count_thresh = count_thresh
    self._save_dir = save_dir
    self._log_every = log_every
    super(iNMF, self).__init__(**kwargs)

  def _initialize_run(self,
                      n_samples: int,
                      n_datasets: int,
                      feature_counts: List[int],
                      to_log=False) -> None:
    """Initializations prior to training the model.

    Parameters
    ----------
    n_samples: int
        Number of samples in the multi-omics dataset
    n_datasets: int
        Number of multi-omics datasets
    feature_counts: List[int],
        Number of features in each dataset, organized as List in same order
        as dataset
    """
    self.basis_matrix = np.random.uniform(
        low=0, high=2, size=(n_samples, self.project_dim))
    self.coefficients = [
        np.random.uniform(
            low=0, high=2, size=(self.project_dim, feature_counts[i]))
        for i in range(n_datasets)
    ]
    self.data_heterogeneity = [
        np.random.uniform(low=0, high=2, size=(n_samples, self.project_dim))
        for i in range(n_datasets)
    ]

    self.start_eval = 0
    self.old_eval = np.inf
    self.new_eval = 0
    self.count = 0

  def update_basis_matrix(self, X: List[np.ndarray]) -> None:
    """Updates the basis matrix.

    Parameters
    ----------
    X: list(np.ndarray)
        List of multi-omics datasets
    """
    #logger.info("Updating basis matrix.")
    n_datasets = len(X)
    numerator = np.sum(
        [X[i].dot(self.coefficients[i].T) for i in range(n_datasets)], axis=0)
    denominator = np.sum(
        [(self.basis_matrix + self.data_heterogeneity[i]).dot(
            self.coefficients[i]).dot(self.coefficients[i].T)
         for i in range(n_datasets)],
        axis=0)
    self.basis_matrix *= numerator / denominator

  def update_coefficients(self, X: List[np.ndarray]) -> None:
    """Updates the coefficients matrices.

    Parameters
    ----------
    X: list(np.ndarray)
        List of multi-omics datasets
    """
    #logger.info("Updating coefficients.")
    n_datasets = len(X)
    for i in range(n_datasets):
      WV = self.basis_matrix + self.data_heterogeneity[i]
      denominator = (WV.T.dot(WV) + self.lambda_reg * \
      self.data_heterogeneity[i].T.dot(self.data_heterogeneity[i])).dot(
          self.coefficients[i]) + self.lambda_sparse
      self.coefficients[i] *= (WV.T.dot(X[i])) / denominator

  def update_data_heterogeneity(self, X: List[np.ndarray]) -> None:
    """Updates the dataset specific heterogeneity matrix.

    Parameters
    ----------
    X: list(np.ndarray)
        List of multi-omics datasets
    """
    #logger.info("Updating data heterogeneity.")
    n_datasets = len(X)
    for i in range(n_datasets):
      denominator = (self.basis_matrix +
                     (1 + self.lambda_reg) * self.data_heterogeneity[i]).dot(
                         self.coefficients[i]).dot(self.coefficients[i].T)
      self.data_heterogeneity[i] *= X[i].dot(
          self.coefficients[i].T) / denominator

  def update_counts(self, X: List[np.ndarray]) -> None:
    """Updates the counts, which is used in some constraint. #TODO: Update this

    Parameters
    ----------
    X: list(np.ndarray)
        List of multi-omics datasets
    """
    #logger.info("Updating counts.")
    if self.count == 0:
      self.new_eval = NMF_obj_eval(X, self.basis_matrix, self.coefficients,
                                   self.lambda_reg, self.lambda_sparse,
                                   self.data_heterogeneity)
      self.start_eval = self.new_eval
    if self.count != 0 and self.count % self.n_steps == 0:  ## count > thres, != 0
      self.old_eval = self.new_eval
      self.new_eval = NMF_obj_eval(X, self.basis_matrix, self.coefficients,
                                   self.lambda_reg, self.lambda_sparse,
                                   self.data_heterogeneity)
    self.count += 1

  def train(self, omics_dataset: Dict, save: bool = False):
    """Trains the model on multi-omics data.

    Parameters
    ----------
    omics_dataset: dict
        Multi-omics dataset, keyed by datatype
    save: bool, default False
        Whether to save the results
    """
    if type(omics_dataset) != dict:
      raise ValueError("omics_dataset is supposed to be of type dict")

    X = [omics_dataset[i] for i in omics_dataset]
    datatypes = list(omics_dataset.keys())
    n_datasets = len(X)
    n_samples = X[0].shape[0]
    feature_counts = [X[i].shape[1] for i in range(n_datasets)]

    # Initialize best solutions to zero
    basis_matrix_best = np.zeros((n_samples, self.project_dim))
    coefficients_best = {
        data_type: np.zeros((self.project_dim, feature_counts[i]))
        for i, data_type in enumerate(datatypes)
    }
    data_heterogeneities_best = {
        data_type: np.zeros((n_samples, self.project_dim))
        for data_type in datatypes
    }

    objective_vals = []
    n_iter = []
    logger.info("Starting to train the model.\n")
    for run in range(self.n_runs):
      # Initializing different matrices
      self._initialize_run(
          n_datasets=n_datasets,
          n_samples=n_samples,
          feature_counts=feature_counts)
      old_vs_new = np.abs(self.old_eval - self.new_eval)
      start_vs_new = self.start_eval - self.new_eval

      while (old_vs_new > start_vs_new * self.update_factor) and (
          self.count <= self.count_thresh):
        # Updates for basis matrix, which is same for all datasets
        self.update_basis_matrix(X)
        self.update_coefficients(X)
        self.update_data_heterogeneity(X)
        self.update_counts(X)

        # Update difference between old_eval and new_eval
        old_vs_new = np.abs(self.old_eval - self.new_eval)
        start_vs_new = self.start_eval - self.new_eval

      objective_vals.append(self.new_eval)
      n_iter.append(self.count)
      if run % self._log_every == 0:
        logger.info("Iteration Number: {}, Objective Value: {}".format(
            run, self.new_eval))

      # Update best solutions per run
      if objective_vals[-1] == np.min(objective_vals):
        for i, data_type in enumerate(datatypes):
          basis_matrix_best = self.basis_matrix
          coefficients_best[data_type] = self.coefficients[i]
          data_heterogeneities_best[data_type] = self.data_heterogeneity[i]

    logger.info("Model training complete. Processing results.\n")
    logger.info("Minimum objective value: {0:.4f}".format(
        np.min(objective_vals)))

    best_results = {
        "basis_matrix": basis_matrix_best,
        "coefficients": coefficients_best,
        "data_heterogeneities": data_heterogeneities_best,
        "objective_vals": objective_vals,
        "iterations": n_iter
    }

    if save:
      self.save(results=best_results)

    return best_results

  @property
  def project_dim(self):
    return self._project_dim

  @property
  def lambda_reg(self):
    return self._lambda_reg

  @property
  def lambda_sparse(self):
    return self._lambda_sparse

  @property
  def n_runs(self):
    return self._n_runs

  @property
  def n_steps(self):
    return self._n_steps

  @property
  def update_factor(self):
    return self._update_factor

  @property
  def count_thresh(self):
    return self._count_thresh
#!/usr/bin/env python3 Integrative Non-negative Matrix Factorization method (iNMF). Paper: https://doi.org/10.1093/bioinformatics/btv544 Integrative NMF (iNMF) is a Non-negative matrix factorization method for multi-   omics integration. It projects all datasets into a common basis space, and   also accounts for dataset/view specific heterogeneities in its formulation.   Update equations are derived for gradient based optimization, and several runs   of the model are needed before convergence to a local optimum. More details can   be found at: https://doi.org/10.1093/bioinformatics/btv544 Parameters     ----------     project_dim: int         Dimensionality of the projection space     lambda_sparse: float, optional         Sparsity coefficient     n_runs: int, optional         Number of times to run the NMF objective     n_steps: int, optional         Number of steps for something?     update_factor: float, optional         Scaling factor with respect to start point to make update     count_thresh: float, optional         # TODO(SOM): Figure out what this is for     save_dir: str, default None,         Directory to save results to     log_every: int, default 50         Number of iterations per run to log results Initializations prior to training the model.      Parameters     ----------     n_samples: int         Number of samples in the multi-omics dataset     n_datasets: int         Number of multi-omics datasets     feature_counts: List[int],         Number of features in each dataset, organized as List in same order         as dataset Updates the basis matrix.      Parameters     ----------     X: list(np.ndarray)         List of multi-omics datasets #logger.info("Updating basis matrix.") Updates the coefficients matrices.      Parameters     ----------     X: list(np.ndarray)         List of multi-omics datasets #logger.info("Updating coefficients.") Updates the dataset specific heterogeneity matrix.      Parameters     ----------     X: list(np.ndarray)         List of multi-omics datasets #logger.info("Updating data heterogeneity.") Updates the counts, which is used in some constraint. #TODO: Update this      Parameters     ----------     X: list(np.ndarray)         List of multi-omics datasets #logger.info("Updating counts.") ## count > thres, != 0 Trains the model on multi-omics data.      Parameters     ----------     omics_dataset: dict         Multi-omics dataset, keyed by datatype     save: bool, default False         Whether to save the results # Initialize best solutions to zero # Initializing different matrices # Updates for basis matrix, which is same for all datasets # Update difference between old_eval and new_eval # Update best solutions per run