cellrank.kernels.RealTimeKernel

class cellrank.kernels.RealTimeKernel(adata, time_key, couplings=None, policy='sequential', **kwargs)[source]

Kernel which computes transition matrix using optimal transport couplings.

See also

This class should be constructed using either:

  1. the from_moscot() or from_wot() method,

  2. explicitly passing the pre-computed couplings, or

  3. overriding the compute_coupling() method.

Parameters:
  • adata (AnnData) – Annotated data object.

  • time_key (str) – Key in obs containing the experimental time.

  • couplings (Optional[Mapping[tuple[Any, Any], Union[ndarray, spmatrix, AnnData, None]]]) – Pre-computed transport couplings. The keys should correspond to a tuple of categories from the time. If None, the keys will be constructed using the policy and the compute_coupling() method must be overriden.

  • policy (Literal['sequential', 'triu']) –

    How to construct the keys from the time when couplings = None:

    • policy = 'sequential' - the keys will be set to [(t1, t2), (t2, t3), ...].

    • policy = 'triu' - the keys will be set to [(t1, t2), (t1, t3), ..., (t2, t3), ...].

  • kwargs (Any) – Keyword arguments for the Kernel.

Attributes table

adata

Annotated data object.

backward

None.

couplings

Optimal transport couplings.

kernels

Underlying base kernels.

obs

Cell-level metadata.

params

Parameters which are used to compute the transition matrix.

shape

(n_cells, n_cells).

time

Experimental time.

transition_matrix

Row-normalized transition matrix.

Methods table

cbc(source, target, cluster_key, rep[, ...])

Compute cross-boundary correctness score between source and target cluster.

compute_coupling(src, tgt, **kwargs)

Compute coupling for a given time pair.

compute_transition_matrix([threshold, ...])

Compute transition matrix from optimal transport couplings.

copy(*[, deep])

Return a copy of self.

from_adata(adata, key[, copy])

Read the kernel saved using write_to_adata().

from_moscot(problem[, sparse_mode, ...])

Construct the kernel from moscot [Klein et al., 2023].

from_wot(adata, path, time_key, **kwargs)

Construct the kernel from Waddington-OT [Schiebinger et al., 2019].

plot_projection([basis, key_added, ...])

Plot transition_matrix as a stream or a grid plot.

plot_random_walks([n_sims, max_iter, seed, ...])

Plot random walks in an embedding.

plot_single_flow(cluster, cluster_key, time_key)

Visualize outgoing flow from a cluster of cells [Mittnenzweig et al., 2021].

read(fname[, adata, copy])

De-serialize self from a file.

write(fname[, write_adata])

Serialize self to a file using pickle.

write_to_adata([key, copy])

Write the transition matrix and parameters used for computation to the underlying adata object.

Attributes

adata

RealTimeKernel.adata

Annotated data object.

backward

RealTimeKernel.backward

None.

couplings

RealTimeKernel.couplings

Optimal transport couplings.

kernels

RealTimeKernel.kernels

Underlying base kernels.

obs

RealTimeKernel.obs

Cell-level metadata.

params

RealTimeKernel.params

Parameters which are used to compute the transition matrix.

shape

RealTimeKernel.shape

(n_cells, n_cells).

time

RealTimeKernel.time

Experimental time.

transition_matrix

RealTimeKernel.transition_matrix

Row-normalized transition matrix.

Methods

cbc

RealTimeKernel.cbc(source, target, cluster_key, rep, graph_key='distances')

Compute cross-boundary correctness score between source and target cluster.

Parameters:
  • source (str) – Name of the source cluster.

  • target (str) – Name of the target cluster.

  • cluster_key (str) – Key in obs to obtain cluster annotations.

  • rep (str) – Key in obsm to use as data representation.

  • graph_key (str) – Name of graph representation to use from obsp.

Return type:

ndarray

Returns:

: Cross-boundary correctness score for each observation.

compute_coupling

RealTimeKernel.compute_coupling(src, tgt, **kwargs)[source]

Compute coupling for a given time pair.

Note

This implementation only looks up the values in couplings, see from_moscot() or from_wot() on how to initialize them.

Parameters:
  • src (Any) – Source key in time.

  • tgt (Any) – Target key in time.

  • kwargs (Any) – Additional keyword arguments.

Return type:

Union[ndarray, spmatrix, AnnData]

Returns:

: Array of shape (n_source_cells, n_target_cells).

compute_transition_matrix

RealTimeKernel.compute_transition_matrix(threshold='auto', self_transitions='connectivities', conn_weight=None, conn_kwargs=mappingproxy({}), **kwargs)[source]

Compute transition matrix from optimal transport couplings.

Parameters:
  • threshold (Union[int, float, Literal['auto', 'auto_local'], None]) –

    How to remove small non-zero values from the transition matrix. Valid options are:

    • 'auto' - find the maximum threshold value which will not remove every non-zero value from any row.

    • 'auto_local' - same as above, but done for each transport separately.

    • float - value in \([0, 100]\) corresponding to a percentage of non-zeros to remove in the couplings.

    Rows where all values are removed will have a uniform distribution and a warning will be issued.

  • self_transitions (Union[Literal['uniform', 'diagonal', 'connectivities', 'all'], Sequence[Any]]) –

    How to define transitions within the blocks that correspond to transitions within the same key. Valid options are:

    • 'uniform' - row-normalized matrix of \(1\).

    • 'diagonal' - identity matrix.

    • 'connectivities' - transition matrix from the ConnectivityKernel.

    • Sequence` - sequence of source keys defining which blocks should be weighted by the connectivities.

    • 'all' - same as above, but for all keys.

    The first 3 options are applied to the block specified by the reference.

  • conn_weight (Optional[float]) – Weight of connectivities’ self transitions. Only used when self_transitions = 'all' or a sequence of source keys is passed.

  • conn_kwargs (Mapping[str, Any]) – Keyword arguments for neighbors() or compute_transition_matrix() when using self_transitions = 'connectivities'.

  • kwargs (Any) – Keyword arguments for compute_coupling().

Return type:

RealTimeKernel

Returns:

: Returns self and updates transition_matrix, couplings and params.

copy

RealTimeKernel.copy(*, deep=False)

Return a copy of self.

Parameters:

deep (bool) – Whether to use deepcopy().

Return type:

Kernel

Returns:

: Copy of self.

from_adata

classmethod RealTimeKernel.from_adata(adata, key, copy=False)

Read the kernel saved using write_to_adata().

Parameters:
  • adata (AnnData) – Annotated data object.

  • key (str) – Key in obsp where the transition matrix is stored. The parameters should be stored in adata.uns['{key}_params'].

  • copy (bool) – Whether to copy the transition matrix.

Return type:

Kernel

Returns:

: The kernel with explicitly initialized properties:

from_moscot

classmethod RealTimeKernel.from_moscot(problem, sparse_mode=None, sparsify_kwargs=mappingproxy({}), copy=False, **kwargs)[source]

Construct the kernel from moscot [Klein et al., 2023].

Parameters:
Return type:

RealTimeKernel

Returns:

: The kernel.

Examples

import moscot as mt
import cellrank as cr

adata = mt.datasets.hspc()
adata.obs["day"] = adata.obs["day"].astype("category")

problem = mt.problems.TemporalProblem(adata)
problem = problem.prepare(time_key="day").solve()

rtk = cr.kernels.RealTimeKernel.from_moscot(problem)
rtk = rtk.compute_transition_matrix()

from_wot

classmethod RealTimeKernel.from_wot(adata, path, time_key, **kwargs)[source]

Construct the kernel from Waddington-OT [Schiebinger et al., 2019].

Parameters:
  • adata (AnnData) – Annotated data object.

  • path (Union[str, Path]) – Directory where the couplings are stored.

  • time_key (str) – Key in obs containing the experimental time.

  • kwargs (Any) – Keyword arguments for RealTimeKernel.

Return type:

RealTimeKernel

Returns:

: The kernel.

Examples

import wot
import cellrank as cr

adata = cr.datasets.reprogramming_schiebinger(subset_to_serum=True)
adata.obs["day"] = adata.obs["day"].astype(float).astype("category")

ot_model = wot.ot.OTModel(adata, day_field="day")
ot_model.compute_all_transport_maps(tmap_out="tmaps/")

rtk = cr.kernels.RealTimeKernel.from_wot(adata, path="tmaps/", time_key="day")
rtk = rtk.compute_transition_matrix()

plot_projection

RealTimeKernel.plot_projection(basis='umap', key_added=None, recompute=False, stream=True, connectivities=None, **kwargs)

Plot transition_matrix as a stream or a grid plot.

Parameters:
Return type:

None

Returns:

: Nothing, just plots and modifies obsm with a key based on the key_added.

plot_random_walks

RealTimeKernel.plot_random_walks(n_sims=100, max_iter=0.25, seed=None, successive_hits=0, start_ixs=None, stop_ixs=None, basis='umap', cmap='gnuplot', linewidth=1.0, linealpha=0.3, ixs_legend_loc=None, n_jobs=None, backend='loky', show_progress_bar=True, figsize=None, dpi=None, save=None, **kwargs)

Plot random walks in an embedding.

This method simulates random walks on the Markov chain defined though the corresponding transition matrix. The method is intended to give qualitative rather than quantitative insights into the transition matrix. Random walks are simulated by iteratively choosing the next cell based on the current cell’s transition probabilities.

Parameters:
  • n_sims (int) – Number of random walks to simulate.

  • max_iter (Union[int, float]) – Maximum number of steps of a random walk. If a float, it can be specified as a fraction of the number of cells.

  • seed (Optional[int]) – Random seed.

  • successive_hits (int) – Number of successive hits in the stop_ixs required to stop prematurely.

  • start_ixs (Union[Sequence[str], Mapping[str, Union[str, Sequence[str], tuple[float, float]]], None]) –

    Cells from which to sample the starting points. If None, use all cells. Can be specified as:

    • dict - dictionary with 1 key in obs with values corresponding to either 1 or more clusters (if the column is categorical) or a tuple specifying \([min, max]\) interval from which to select the indices.

    • Sequence - sequence of cell ids in obs_names.

    For example {'dpt_pseudotime': [0, 0.1]} means that starting points for random walks will be sampled uniformly from cells whose pseudotime is in \([0, 0.1]\).

  • stop_ixs (Union[Sequence[str], Mapping[str, Union[str, Sequence[str], tuple[float, float]]], None]) –

    Cells which when hit, the random walk is terminated. If None, terminate after max_iters. Can be specified as:

    • dict - dictionary with 1 key in obs with values corresponding to either 1 or more clusters (if the column is categorical) or a tuple specifying \([min, max]\) interval from which to select the indices.

    • Sequence - sequence of cell ids in obs_names.

    For example {'clusters': ['Alpha', 'Beta']} and successive_hits = 3 means that the random walk will stop prematurely after cells in the above specified clusters have been visited successively 3 times in a row.

  • basis (str) – Basis in obsm to use as an embedding.

  • cmap (Union[str, LinearSegmentedColormap]) – Colormap for the random walk lines.

  • linewidth (float) – Width of the random walk lines.

  • linealpha (float) – Alpha value of the random walk lines.

  • ixs_legend_loc (Optional[str]) – Legend location for the start/top indices.

  • show_progress_bar (bool) – Whether to show a progress bar. Disabling it may slightly improve performance.

  • n_jobs (Optional[int]) – Number of parallel jobs. If -1, use all available cores. If None or 1, the execution is sequential.

  • backend (str) – Which backend to use for parallelization. See Parallel for valid options.

  • figsize (Optional[tuple[float, float]]) – Size of the figure.

  • dpi (Optional[int]) – Dots per inch.

  • save (Union[Path, str, None]) – Filename where to save the plot.

  • kwargs (Any) – Keyword arguments for scatter().

Return type:

None

Returns:

: Nothing, just plots the figure. Optionally saves it based on save. For each random walk, the first/last cell is marked by the start/end colors of cmap.

plot_single_flow

RealTimeKernel.plot_single_flow(cluster, cluster_key, time_key, clusters=None, time_points=None, min_flow=0, remove_empty_clusters=True, ascending=False, legend_loc='upper right out', alpha=0.8, xticks_step_size=1, figsize=None, dpi=None, save=None, show=True)

Visualize outgoing flow from a cluster of cells [Mittnenzweig et al., 2021].

Parameters:
  • cluster (str) – Cluster for which to visualize outgoing flow.

  • cluster_key (str) – Key in obs where clustering is stored.

  • time_key (str) – Key in obs where experimental time is stored.

  • clusters (Optional[Sequence[Any]]) – Visualize flow only for these clusters. If None, use all clusters.

  • time_points (Optional[Sequence[Union[float, int]]]) – Visualize flow only for these time points. If None, use all time points.

  • min_flow (float) – Only show flow edges with flow greater than this value. Flow values are always in \([0, 1]\).

  • remove_empty_clusters (bool) – Whether to remove clusters with no incoming flow edges.

  • ascending (Optional[bool]) – Whether to sort the cluster by ascending or descending incoming flow. If None, use the order as in defined by clusters.

  • alpha (Optional[float]) – Alpha value for cell proportions.

  • xticks_step_size (Optional[int]) – Show only every other n-th tick on the x-axis. If None, don’t show any ticks.

  • legend_loc (Optional[str]) – Position of the legend. If None, do not show the legend.

  • figsize (Optional[tuple[float, float]]) – Size of the figure.

  • dpi (Optional[int]) – Dots per inch.

  • figsize – Size of the figure.

  • dpi – Dots per inch.

  • save (Union[Path, str, None]) – Filename where to save the plot.

  • show (bool) – If False, return Axes.

Return type:

Optional[Axes]

Returns:

: The axes object, if show = False. Nothing, just plots the figure. Optionally saves it based on save.

Notes

This function is a Python re-implementation of the following original R function with some minor stylistic differences. This function will not recreate the results from [Mittnenzweig et al., 2021], because there, the Metacell model [Baran et al., 2019] was used to compute the flow, whereas here the transition matrix is used.

read

static RealTimeKernel.read(fname, adata=None, copy=False)

De-serialize self from a file.

Parameters:
  • fname (Union[str, Path]) – Path from which to read the object.

  • adata (Optional[AnnData]) – AnnData object to assign to the saved object. Only used when the saved object has adata and it was saved without it.

  • copy (bool) – Whether to copy adata before assigning it. If adata is a view, it is always copied.

Return type:

IOMixin

Returns:

: The de-serialized object.

write

RealTimeKernel.write(fname, write_adata=True)

Serialize self to a file using pickle.

Parameters:
  • fname (Union[str, Path]) – Path where to save the object.

  • write_adata (bool) – Whether to save adata object.

Return type:

None

Returns:

: Nothing, just writes itself to a file.

write_to_adata

RealTimeKernel.write_to_adata(key=None, copy=False)

Write the transition matrix and parameters used for computation to the underlying adata object.

Parameters:
Return type:

None

Returns:

: Updates the adata with the following fields: