Workflow Classes

Use these classes when you want step-wise control, intermediate state, or save/load support.

scTenifoldNet

scTenifoldNet(
    x_data: ExpressionData,
    y_data: ExpressionData,
    x_label: str,
    y_label: str,
    qc_kws: Optional[Kwargs] = None,
    nc_kws: Optional[Kwargs] = None,
    td_kws: Optional[Kwargs] = None,
    ma_kws: Optional[Kwargs] = None,
    dr_kws: Optional[Kwargs] = None,
)

Bases: scBase

Two-sample scTenifoldNet workflow.

Pipeline order: qc → nc (PC network construction) → td (tensor decomposition) → ma (manifold alignment) → dr (differential regulation). Each step persists its output on the instance so it can be inspected, saved, or rerun individually via :meth:run_step. :meth:build runs the full pipeline and returns the differential regulation table.

Parameters:

Name	Type	Description	Default
`x_data`	`ExpressionData`	Genes-by-cells expression DataFrames (or AnnData-like objects; converted via :func:`anndata_to_dataframe`). The two conditions being compared.	required
`y_data`	`ExpressionData`	Genes-by-cells expression DataFrames (or AnnData-like objects; converted via :func:`anndata_to_dataframe`). The two conditions being compared.	required
`x_label`	`str`	Short labels used as keys in `data_dict`/`QC_dict`/... and to identify each condition in saved output.	required
`y_label`	`str`	Short labels used as keys in `data_dict`/`QC_dict`/... and to identify each condition in saved output.	required
`qc_kws`	`Optional[Kwargs]`	Overrides for :func:`sc_QC` during the QC step.	`None`
`nc_kws`	`Optional[Kwargs]`	Overrides for :func:`make_networks` during PC network construction. Use this dict to pass `backend`, `n_jobs`, `random_state`, etc.	`None`
`td_kws`	`Optional[Kwargs]`	Overrides for :func:`tensor_decomp`.	`None`
`ma_kws`	`Optional[Kwargs]`	Overrides for :func:`manifold_alignment`.	`None`
`dr_kws`	`Optional[Kwargs]`	Overrides for :func:`d_regulation`.	`None`

See class docstring for parameter descriptions.

Source code in scTenifold/core/_base.py

def __init__(self,
             x_data: ExpressionData,
             y_data: ExpressionData,
             x_label: str,
             y_label: str,
             qc_kws: Optional[Kwargs] = None,
             nc_kws: Optional[Kwargs] = None,
             td_kws: Optional[Kwargs] = None,
             ma_kws: Optional[Kwargs] = None,
             dr_kws: Optional[Kwargs] = None) -> None:
    """See class docstring for parameter descriptions."""
    super().__init__(qc_kws=qc_kws, nc_kws=nc_kws, td_kws=td_kws, ma_kws=ma_kws, dr_kws=dr_kws)
    self.x_label, self.y_label = x_label, y_label
    self.data_dict[x_label] = pd.DataFrame() if isinstance(x_data, str) and x_data == "" else anndata_to_dataframe(x_data)
    self.data_dict[y_label] = pd.DataFrame() if isinstance(y_data, str) and y_data == "" else anndata_to_dataframe(y_data)

get_empty_config `classmethod`

get_empty_config() -> Dict[str, object]

Return a blank scTenifoldNet config dict populated with step defaults.

Source code in scTenifold/core/_base.py

@classmethod
def get_empty_config(cls) -> Dict[str, object]:
    """Return a blank scTenifoldNet config dict populated with step defaults."""
    config = {"x_data_path": None, "y_data_path": None,
              "x_label": None, "y_label": None}
    for kw, sig in cls.kw_sigs.items():
        config[kw] = cls.list_kws(kw)
    return config

load_config `classmethod`

load_config(config: Dict[str, object]) -> scTenifoldNet

Build a scTenifoldNet from a config dict, reading data from disk.

x_data_path and y_data_path may be either a 10x folder (loaded via :func:read_folder) or a CSV/TSV file.

Source code in scTenifold/core/_base.py

@classmethod
def load_config(cls, config: Dict[str, object]) -> "scTenifoldNet":
    """Build a scTenifoldNet from a config dict, reading data from disk.

    ``x_data_path`` and ``y_data_path`` may be either a 10x folder
    (loaded via :func:`read_folder`) or a CSV/TSV file.
    """
    x_data_path = Path(config.pop("x_data_path"))
    y_data_path = Path(config.pop("y_data_path"))
    if x_data_path.is_dir():
        x_data = read_folder(x_data_path)
    else:
        x_data = pd.read_csv(x_data_path, sep='\t' if x_data_path.suffix == ".tsv" else ",")
    if y_data_path.is_dir():
        y_data = read_folder(y_data_path)
    else:
        y_data = pd.read_csv(y_data_path, sep='\t' if y_data_path.suffix == ".tsv" else ",")
    return cls(x_data, y_data, **config)

save

save(
    file_dir: Union[str, Path],
    comps: Union[str, List[str]] = "all",
    verbose: bool = True,
    **kwargs: object,
) -> None

Save state plus x_label/y_label so :meth:load can rebuild.

Source code in scTenifold/core/_base.py

def save(self,
         file_dir: Union[str, Path],
         comps: Union[str, List[str]] = "all",
         verbose: bool = True,
         **kwargs: object) -> None:
    """Save state plus ``x_label``/``y_label`` so :meth:`load` can rebuild."""
    super().save(file_dir, comps, verbose,
                 x_data="", y_data="",
                 x_label=self.x_label, y_label=self.y_label)

run_step

run_step(
    step_name: Literal["qc", "nc", "td", "ma", "dr"],
    **kwargs: object,
) -> None

Run a single step of the scTenifoldNet pipeline.

Steps must be invoked in order — each depends on the state produced by the previous one.

Parameters:

Name	Type	Description	Default
`step_name`	`Literal['qc', 'nc', 'td', 'ma', 'dr']`	Which step to run. One of: `"qc"` — quality control + CPM normalisation on both conditions. Reads `self.data_dict`; writes `self.QC_dict`. `"nc"` — PC network construction on the shared gene set. Reads `self.QC_dict`; writes `self.network_dict` and `self.shared_gene_names`. `"td"` — tensor decomposition + symmetrisation. Reads `self.network_dict`; writes `self.tensor_dict`. `"ma"` — manifold alignment between the two decomposed tensors. Writes `self.manifold`. `"dr"` — differential regulation from the aligned manifold. Writes `self.d_regulation`.	required
`**kwargs`	`object`	One-shot overrides for the step. When non-empty these replace the dict stored on the instance (`qc_kws`, `nc_kws`, etc.) for this call only.	`{}`

Raises:

Type	Description
`ValueError`	If `step_name` is not one of the five values above.

Source code in scTenifold/core/_base.py

def run_step(self,
             step_name: Literal["qc", "nc", "td", "ma", "dr"],
             **kwargs: object) -> None:
    """Run a single step of the scTenifoldNet pipeline.

    Steps must be invoked in order — each depends on the state
    produced by the previous one.

    Parameters
    ----------
    step_name
        Which step to run. One of:

        - ``"qc"`` — quality control + CPM normalisation on both
          conditions. Reads ``self.data_dict``; writes
          ``self.QC_dict``.
        - ``"nc"`` — PC network construction on the shared gene
          set. Reads ``self.QC_dict``; writes ``self.network_dict``
          and ``self.shared_gene_names``.
        - ``"td"`` — tensor decomposition + symmetrisation. Reads
          ``self.network_dict``; writes ``self.tensor_dict``.
        - ``"ma"`` — manifold alignment between the two decomposed
          tensors. Writes ``self.manifold``.
        - ``"dr"`` — differential regulation from the aligned
          manifold. Writes ``self.d_regulation``.
    **kwargs
        One-shot overrides for the step. When non-empty these
        replace the dict stored on the instance (``qc_kws``,
        ``nc_kws``, etc.) for this call only.

    Raises
    ------
    ValueError
        If ``step_name`` is not one of the five values above.
    """
    start_time = time.perf_counter()
    if step_name == "qc":
        for label in self.data_dict:
            self._QC(label,
                     **(self.qc_kws if kwargs == {} else kwargs))
            self._norm(label)
            print("finish QC:", label)
    elif step_name == "nc":
        y_gene_names = set(self.QC_dict[self.y_label].index)
        self.shared_gene_names = [gene for gene in self.QC_dict[self.x_label].index if gene in y_gene_names]
        for label, qc_data in self.QC_dict.items():
            self._make_networks(label, data=qc_data.loc[self.shared_gene_names, :],
                                **(self.nc_kws if kwargs == {} else kwargs))
    elif step_name == "td":
        for label, qc_data in self.QC_dict.items():
            self._tensor_decomp(label, self.shared_gene_names, **(self.td_kws if kwargs == {} else kwargs))
        self.tensor_dict[self.x_label] = (self.tensor_dict[self.x_label] + self.tensor_dict[self.x_label].T) / 2
        self.tensor_dict[self.y_label] = (self.tensor_dict[self.y_label] + self.tensor_dict[self.y_label].T) / 2
    elif step_name == "ma":
        self.manifold = manifold_alignment(self.tensor_dict[self.x_label],
                                           self.tensor_dict[self.y_label],
                                           **(self.ma_kws if kwargs == {} else kwargs))
        self.step_comps["ma"] = self.manifold
    elif step_name == "dr":
        self.d_regulation = d_regulation(self.manifold, **(self.dr_kws if kwargs == {} else kwargs))
        self.step_comps["dr"] = self.d_regulation
    else:
        raise ValueError("This step name is not valid, please choose from qc, nc, td, ma, dr")

    print(f"process {step_name} finished in {time.perf_counter() - start_time} secs.")

build

build() -> pd.DataFrame

Run the whole pipeline of scTenifoldNet

Returns:

Name	Type	Description
`d_regulation_df`	`DataFrame`	Differential regulation result dataframe

Source code in scTenifold/core/_base.py

def build(self) -> pd.DataFrame:
    """
    Run the whole pipeline of scTenifoldNet

    Returns
    -------
    d_regulation_df: pd.DataFrame
        Differential regulation result dataframe
    """
    self.run_step("qc")
    self.run_step("nc")
    self.run_step("td")
    self.run_step("ma")
    self.run_step("dr")
    return self.d_regulation

scTenifoldKnk

scTenifoldKnk(
    data: ExpressionData,
    strict_lambda: float = 0,
    ko_method: KOMethod = "default",
    ko_genes: Optional[Union[str, Iterable[str]]] = None,
    qc_kws: Optional[Kwargs] = None,
    nc_kws: Optional[Kwargs] = None,
    td_kws: Optional[Kwargs] = None,
    ma_kws: Optional[Kwargs] = None,
    dr_kws: Optional[Kwargs] = None,
    ko_kws: Optional[Kwargs] = None,
)

Bases: scBase

Single-sample scTenifoldKnk virtual-knockout workflow.

Pipeline order: qc → nc → td → ko → ma → dr. A wild-type PC network is built from data; the ko step produces a perturbed tensor and the remaining steps compare WT vs. KO. Run end-to-end with :meth:build, or step-wise with :meth:run_step.

Parameters:

Name	Type	Description	Default
`data`	`ExpressionData`	Genes-by-cells expression matrix (`pandas.DataFrame` or AnnData-like).	required
`strict_lambda`	`float`	Pruning strength forwarded to :func:`strict_direction` when post-processing the decomposed WT tensor. `0` disables pruning.	`0`
`ko_method`	`KOMethod`	How the KO tensor is generated: `"default"` — zero out the WT tensor rows for `ko_genes`. `"propagation"` — rebuild PC networks with the targeted gene columns masked using :func:`reconstruct_pcnets`, then re-decompose.	`'default'`
`ko_genes`	`Optional[Union[str, Iterable[str]]]`	Gene name or iterable of names to knock out. `None` stores an empty list.	`None`
`qc_kws`	`Optional[Kwargs]`	Overrides for :func:`sc_QC`. If `min_exp_avg` / `min_exp_sum` are missing, `run_step("qc")` injects KO defaults (0.05 and 25).	`None`
`nc_kws`	`Optional[Kwargs]`	Overrides for :func:`make_networks` (`backend`, `n_jobs`, etc.).	`None`
`td_kws`	`Optional[Kwargs]`	Overrides for :func:`tensor_decomp`.	`None`
`ma_kws`	`Optional[Kwargs]`	Overrides for :func:`manifold_alignment`. Defaults to `{"d": 2}` when `None`.	`None`
`dr_kws`	`Optional[Kwargs]`	Overrides for :func:`d_regulation`.	`None`
`ko_kws`	`Optional[Kwargs]`	Extra kwargs forwarded to the KO step (e.g. `degree` for the propagation method).	`None`

See class docstring for parameter descriptions.

Source code in scTenifold/core/_base.py

def __init__(self,
             data: ExpressionData,
             strict_lambda: float = 0,
             ko_method: KOMethod = "default",
             ko_genes: Optional[Union[str, Iterable[str]]] = None,
             qc_kws: Optional[Kwargs] = None,
             nc_kws: Optional[Kwargs] = None,
             td_kws: Optional[Kwargs] = None,
             ma_kws: Optional[Kwargs] = None,
             dr_kws: Optional[Kwargs] = None,
             ko_kws: Optional[Kwargs] = None) -> None:
    """See class docstring for parameter descriptions."""
    ma_kws = {"d": 2} if ma_kws is None else ma_kws
    super().__init__(qc_kws=qc_kws, nc_kws=nc_kws, td_kws=td_kws, ma_kws=ma_kws, dr_kws=dr_kws)
    self.data_dict["WT"] = pd.DataFrame() if isinstance(data, str) and data == "" else anndata_to_dataframe(data)
    self.strict_lambda = strict_lambda
    self.ko_genes = ko_genes if ko_genes is not None else []
    self.ko_method = ko_method
    self.ko_kws = {} if ko_kws is None else ko_kws

get_empty_config `classmethod`

get_empty_config() -> Dict[str, object]

Return a blank scTenifoldKnk config dict populated with step defaults.

Source code in scTenifold/core/_base.py

@classmethod
def get_empty_config(cls) -> Dict[str, object]:
    """Return a blank scTenifoldKnk config dict populated with step defaults."""
    config = {"data_path": None, "strict_lambda": 0,
              "ko_method": "default", "ko_genes": []}
    for kw, sig in cls.kw_sigs.items():
        config[kw] = cls.list_kws(kw)
    return config

load_config `classmethod`

load_config(config: Dict[str, object]) -> scTenifoldKnk

Build a scTenifoldKnk from a config dict, reading data from disk.

Source code in scTenifold/core/_base.py

@classmethod
def load_config(cls, config: Dict[str, object]) -> "scTenifoldKnk":
    """Build a scTenifoldKnk from a config dict, reading data from disk."""
    data_path = Path(config.pop("data_path"))
    if data_path.is_dir():
        data = read_folder(data_path)
    else:
        data = pd.read_csv(data_path, sep='\t' if data_path.suffix == ".tsv" else ",")
    return cls(data, **config)

save

save(
    file_dir: Union[str, Path],
    comps: Union[str, List[str]] = "all",
    verbose: bool = True,
    **kwargs: object,
) -> None

Save state plus KO-specific fields so :meth:load can rebuild.

Source code in scTenifold/core/_base.py

def save(self,
         file_dir: Union[str, Path],
         comps: Union[str, List[str]] = "all",
         verbose: bool = True,
         **kwargs: object) -> None:
    """Save state plus KO-specific fields so :meth:`load` can rebuild."""
    super().save(file_dir, comps, verbose,
                 data="",
                 ko_method=self.ko_method,
                 strict_lambda=self.strict_lambda, ko_genes=self.ko_genes)

run_step

run_step(
    step_name: Literal["qc", "nc", "td", "ko", "ma", "dr"],
    **kwargs: object,
) -> None

Run a single step of the scTenifoldKnk pipeline.

Steps must be invoked in order — each depends on the state produced by the previous one.

Parameters:

Name	Type	Description	Default
`step_name`	`Literal['qc', 'nc', 'td', 'ko', 'ma', 'dr']`	Which step to run. One of: `"qc"` — quality control on the WT sample (no normalisation; injects KO-friendly defaults for `min_exp_avg`/`min_exp_sum` if missing). `"nc"` — PC network construction on the WT QC matrix. Writes `self.network_dict["WT"]` and `self.shared_gene_names`. `"td"` — tensor decomposition of the WT networks plus :func:`strict_direction` pruning controlled by `self.strict_lambda`. `"ko"` — produce the KO tensor according to `self.ko_method`. `kwargs` may contain `ko_genes` to override `self.ko_genes` for this call. `"ma"` — manifold alignment of WT vs. KO tensors. `"dr"` — differential regulation from the aligned manifold.	required
`**kwargs`	`object`	One-shot overrides for the step. When non-empty these replace the corresponding `*_kws` dict on the instance for this call only. For `"ko"` an explicit `ko_genes` kwarg is consumed before the override logic.	`{}`

Raises:

Type	Description
`ValueError`	If `step_name` is not one of the six values above, or if `self.ko_method` is unrecognised during the KO step.

Source code in scTenifold/core/_base.py

def run_step(self,
             step_name: Literal["qc", "nc", "td", "ko", "ma", "dr"],
             **kwargs: object) -> None:
    """Run a single step of the scTenifoldKnk pipeline.

    Steps must be invoked in order — each depends on the state
    produced by the previous one.

    Parameters
    ----------
    step_name
        Which step to run. One of:

        - ``"qc"`` — quality control on the WT sample (no
          normalisation; injects KO-friendly defaults for
          ``min_exp_avg``/``min_exp_sum`` if missing).
        - ``"nc"`` — PC network construction on the WT QC matrix.
          Writes ``self.network_dict["WT"]`` and
          ``self.shared_gene_names``.
        - ``"td"`` — tensor decomposition of the WT networks plus
          :func:`strict_direction` pruning controlled by
          ``self.strict_lambda``.
        - ``"ko"`` — produce the KO tensor according to
          ``self.ko_method``. ``kwargs`` may contain ``ko_genes``
          to override ``self.ko_genes`` for this call.
        - ``"ma"`` — manifold alignment of WT vs. KO tensors.
        - ``"dr"`` — differential regulation from the aligned
          manifold.
    **kwargs
        One-shot overrides for the step. When non-empty these
        replace the corresponding ``*_kws`` dict on the instance
        for this call only. For ``"ko"`` an explicit ``ko_genes``
        kwarg is consumed before the override logic.

    Raises
    ------
    ValueError
        If ``step_name`` is not one of the six values above, or if
        ``self.ko_method`` is unrecognised during the KO step.
    """
    start_time = time.perf_counter()
    if step_name == "qc":
        if "min_exp_avg" not in self.qc_kws:
            self.qc_kws["min_exp_avg"] = 0.05
        if "min_exp_sum" not in self.qc_kws:
            self.qc_kws["min_exp_sum"] = 25
        self._QC("WT", **(self.qc_kws if kwargs == {} else kwargs))
        # no norm
        print("finish QC: WT")
    elif step_name == "nc":
        self._make_networks("WT", self.QC_dict["WT"], **(self.nc_kws if kwargs == {} else kwargs))
        self.shared_gene_names = self.QC_dict["WT"].index.to_list()
    elif step_name == "td":
        self._tensor_decomp("WT", self.shared_gene_names, **(self.td_kws if kwargs == {} else kwargs))
        self.tensor_dict["WT"] = strict_direction(self.tensor_dict["WT"], self.strict_lambda).T.copy()
    elif step_name == "ko":
        self.tensor_dict["WT"] = _fill_dataframe_diagonal(self.tensor_dict["WT"], 0)
        ko_kwargs = dict(self.ko_kws)
        if kwargs.get("ko_genes") is not None:
            ko_genes = kwargs.pop("ko_genes")
        else:
            ko_genes = self.ko_genes
        ko_kwargs.update(kwargs)
        self._get_ko_tensor(ko_genes, **ko_kwargs)
    elif step_name == "ma":
        self.manifold = manifold_alignment(self.tensor_dict["WT"],
                                           self.tensor_dict["KO"],
                                           **(self.ma_kws if kwargs == {} else kwargs))
        self.step_comps["ma"] = self.manifold
    elif step_name == "dr":
        self.d_regulation = d_regulation(self.manifold, **(self.dr_kws if kwargs == {} else kwargs))
        self.step_comps["dr"] = self.d_regulation
    else:
        raise ValueError("No such step")
    print(f"process {step_name} finished in {time.perf_counter() - start_time} secs.")

build

build() -> pd.DataFrame

Run the whole pipeline of scTenifoldKnk

Returns:

Name	Type	Description
`d_regulation_df`	`DataFrame`	Differential regulation result dataframe

Source code in scTenifold/core/_base.py

def build(self) -> pd.DataFrame:
    """
    Run the whole pipeline of scTenifoldKnk

    Returns
    -------
    d_regulation_df: pd.DataFrame
        Differential regulation result dataframe
    """
    self.run_step("qc")
    self.run_step("nc")
    self.run_step("td")
    self.run_step("ko")
    self.run_step("ma")
    self.run_step("dr")
    return self.d_regulation

Workflow Classes

scTenifoldNet

get_empty_config classmethod

load_config classmethod

save

run_step

build

scTenifoldKnk

get_empty_config classmethod

load_config classmethod

save

run_step

build

get_empty_config `classmethod`

load_config `classmethod`

get_empty_config `classmethod`

load_config `classmethod`