mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-09-15 12:47:56 +08:00
3ebbb0ccbf
* move common code into server, update tests and makefile remove backend directory, refactor update smoke tests
424 lines
15 KiB
Python
424 lines
15 KiB
Python
from abc import ABCMeta, abstractmethod
|
|
from os.path import basename, splitext
|
|
import numpy as np
|
|
import pandas as pd
|
|
from scipy import sparse
|
|
from server_timing import Timing as ServerTiming
|
|
|
|
from server.common.config.app_config import AppConfig
|
|
from server.common.constants import Axis, XApproximateDistribution
|
|
from server.common.errors import FilterError, JSONEncodingValueError, ExceedsLimitError, UnsupportedSummaryMethod
|
|
from server.common.utils.utils import jsonify_strict
|
|
from server.common.fbs.matrix import encode_matrix_fbs
|
|
from server.common.genesets import validate_gene_sets
|
|
|
|
|
|
class DataAdaptor(metaclass=ABCMeta):
|
|
"""Base class for loading and accessing matrix data"""
|
|
|
|
def __init__(self, data_locator, app_config, dataset_config=None):
|
|
if not isinstance(app_config, AppConfig):
|
|
raise TypeError("config expected to be of type AppConfig")
|
|
|
|
# location to the dataset
|
|
self.data_locator = data_locator
|
|
|
|
# config is the application configuration
|
|
self.app_config = app_config
|
|
self.server_config = self.app_config.server_config
|
|
self.dataset_config = dataset_config or app_config.dataset_config
|
|
|
|
# parameters set by this data adaptor based on the data.
|
|
self.parameters = {}
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def pre_load_validation(data_locator):
|
|
pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def open(data_locator, app_config, dataset_config):
|
|
pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def file_size(data_locator):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_name(self):
|
|
"""return a string name for this data adaptor"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_library_versions(self):
|
|
"""return a dictionary of library name to library versions"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_embedding_names(self):
|
|
"""return a list of pre-computed embedding names"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_embedding_array(self, ename, dims=2):
|
|
"""return an numpy array for the given pre-computed embedding name."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_X_array(self, obs_mask=None, var_mask=None):
|
|
"""return the X array, possibly filtered by obs_mask or var_mask.
|
|
the return type is either ndarray or scipy.sparse.spmatrix."""
|
|
pass
|
|
|
|
def get_X_approximate_distribution(self) -> XApproximateDistribution:
|
|
"""return the approximate distribution of the X matrix."""
|
|
return XApproximateDistribution.NORMAL
|
|
|
|
@abstractmethod
|
|
def get_shape(self):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def query_var_array(self, term_var):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def query_obs_array(self, term_var):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_colors(self):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_obs_index(self):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_obs_columns(self):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_obs_keys(self):
|
|
# return list of keys
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_var_keys(self):
|
|
# return list of keys
|
|
pass
|
|
|
|
@abstractmethod
|
|
def cleanup(self):
|
|
pass
|
|
|
|
def get_data_locator(self):
|
|
return self.data_locator
|
|
|
|
def get_location(self):
|
|
return self.data_locator.uri_or_path
|
|
|
|
def get_about(self):
|
|
return None
|
|
|
|
def get_title(self):
|
|
# default to file name
|
|
location = self.get_location()
|
|
if location.endswith("/"):
|
|
location = location[:-1]
|
|
return splitext(basename(location))[0]
|
|
|
|
def get_corpora_props(self):
|
|
return None
|
|
|
|
@abstractmethod
|
|
def get_schema(self):
|
|
"""
|
|
Return current schema
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def annotation_to_fbs_matrix(self, axis, field=None, uid=None):
|
|
"""
|
|
Gets annotation value for each observation
|
|
:param axis: string obs or var
|
|
:param fields: list of keys for annotation to return, returns all annotation values if not set.
|
|
:return: flatbuffer: in fbs/matrix.fbs encoding
|
|
"""
|
|
pass
|
|
|
|
def update_parameters(self, parameters):
|
|
parameters.update(self.parameters)
|
|
|
|
def _index_filter_to_mask(self, filter, count):
|
|
mask = np.zeros((count,), dtype=np.bool)
|
|
for i in filter:
|
|
if isinstance(i, list):
|
|
mask[i[0] : i[1]] = True
|
|
else:
|
|
mask[i] = True
|
|
return mask
|
|
|
|
def _axis_filter_to_mask(self, axis, filter, count):
|
|
mask = np.ones((count,), dtype=np.bool)
|
|
if "index" in filter:
|
|
mask = np.logical_and(mask, self._index_filter_to_mask(filter["index"], count))
|
|
if "annotation_value" in filter:
|
|
mask = np.logical_and(mask, self._annotation_filter_to_mask(axis, filter["annotation_value"], count))
|
|
|
|
return mask
|
|
|
|
def _annotation_filter_to_mask(self, axis, filter, count):
|
|
mask = np.ones((count,), dtype=np.bool)
|
|
for v in filter:
|
|
name = v["name"]
|
|
if axis == Axis.VAR:
|
|
anno_data = self.query_var_array(name)
|
|
elif axis == Axis.OBS:
|
|
anno_data = self.query_obs_array(name)
|
|
|
|
if anno_data.dtype.name in ["boolean", "category", "object"]:
|
|
values = v.get("values", [])
|
|
key_idx = np.in1d(anno_data, values)
|
|
mask = np.logical_and(mask, key_idx)
|
|
|
|
else:
|
|
min_ = v.get("min", None)
|
|
max_ = v.get("max", None)
|
|
if min_ is not None:
|
|
key_idx = (anno_data >= min_).ravel()
|
|
mask = np.logical_and(mask, key_idx)
|
|
if max_ is not None:
|
|
key_idx = (anno_data <= max_).ravel()
|
|
mask = np.logical_and(mask, key_idx)
|
|
|
|
return mask
|
|
|
|
def _filter_to_mask(self, filter):
|
|
"""
|
|
Return the filter as a row and column selection list.
|
|
No filter on a dimension means 'all'
|
|
"""
|
|
shape = self.get_shape()
|
|
var_selector = None
|
|
obs_selector = None
|
|
if filter is not None:
|
|
if Axis.OBS in filter:
|
|
obs_selector = self._axis_filter_to_mask(Axis.OBS, filter["obs"], shape[0])
|
|
|
|
if Axis.VAR in filter:
|
|
var_selector = self._axis_filter_to_mask(Axis.VAR, filter["var"], shape[1])
|
|
|
|
return (obs_selector, var_selector)
|
|
|
|
def check_new_labels(self, labels_df):
|
|
"""Check the new annotations labels, then set the labels_df index"""
|
|
if labels_df is None or labels_df.empty:
|
|
return
|
|
|
|
labels_df.index = self.get_obs_index()
|
|
if labels_df.index.name is None:
|
|
labels_df.index.name = "index"
|
|
|
|
# all labels must have a name, which must be unique and not used in obs column names
|
|
if not labels_df.columns.is_unique:
|
|
raise KeyError("All column names specified in user annotations must be unique.")
|
|
|
|
# the label index must be unique, and must have same values the anndata obs index
|
|
if not labels_df.index.is_unique:
|
|
raise KeyError("All row index values specified in user annotations must be unique.")
|
|
|
|
obs_columns = self.get_obs_columns()
|
|
|
|
duplicate_columns = list(set(labels_df.columns) & set(obs_columns))
|
|
if len(duplicate_columns) > 0:
|
|
raise KeyError(
|
|
"Labels file may not contain column names which overlap " f"with h5ad obs columns {duplicate_columns}"
|
|
)
|
|
|
|
# labels must have same count as obs annotations
|
|
shape = self.get_shape()
|
|
if labels_df.shape[0] != shape[0]:
|
|
raise ValueError("Labels file must have same number of rows as data file.")
|
|
|
|
# This will convert a float column that contains integer data into an integer type.
|
|
# This case can occur when a user makes a copy of a category that originally contained integer data.
|
|
# The client always copies array data to floats, therefore the copy will contain floats instead of integers.
|
|
# float data is not allowed as a categorical type.
|
|
if any([np.issubdtype(coltype.type, np.floating) for coltype in labels_df.dtypes]):
|
|
labels_df = labels_df.convert_dtypes()
|
|
for col, dtype in zip(labels_df, labels_df.dtypes):
|
|
if isinstance(dtype, pd.Int32Dtype):
|
|
labels_df[col] = labels_df[col].astype("int32")
|
|
if isinstance(dtype, pd.Int64Dtype):
|
|
labels_df[col] = labels_df[col].astype("int64")
|
|
|
|
if any([np.issubdtype(coltype.type, np.floating) for coltype in labels_df.dtypes]):
|
|
raise ValueError("Columns may not have floating point types")
|
|
|
|
return labels_df
|
|
|
|
def check_new_gene_sets(self, genesets, context=None):
|
|
var_names = set(self.query_var_array(self.parameters.get("var_names")))
|
|
return validate_gene_sets(genesets, var_names)
|
|
|
|
def data_frame_to_fbs_matrix(self, filter, axis):
|
|
"""
|
|
Retrieves data 'X' and returns in a flatbuffer Matrix.
|
|
:param filter: filter: dictionary with filter params
|
|
:param axis: string obs or var
|
|
:return: flatbuffer Matrix
|
|
|
|
Caveats:
|
|
* currently only supports access on VAR axis
|
|
* currently only supports filtering on VAR axis
|
|
"""
|
|
if axis != Axis.VAR:
|
|
raise ValueError("Only VAR dimension access is supported")
|
|
|
|
try:
|
|
obs_selector, var_selector = self._filter_to_mask(filter)
|
|
except (KeyError, IndexError, TypeError, AttributeError):
|
|
raise FilterError("Error parsing filter")
|
|
|
|
if obs_selector is not None:
|
|
raise FilterError("filtering on obs unsupported")
|
|
|
|
num_columns = self.get_shape()[1] if var_selector is None else np.count_nonzero(var_selector)
|
|
if self.server_config.exceeds_limit("column_request_max", num_columns):
|
|
raise ExceedsLimitError("Requested dataframe columns exceed column request limit")
|
|
|
|
X = self.get_X_array(obs_selector, var_selector)
|
|
col_idx = np.nonzero([] if var_selector is None else var_selector)[0]
|
|
return encode_matrix_fbs(X, col_idx=col_idx, row_idx=None)
|
|
|
|
def diffexp_topN(self, obsFilterA, obsFilterB, top_n=None):
|
|
"""
|
|
Computes the top N differentially expressed variables between two observation sets. If mode
|
|
is "TOP_N", then stats for the top N
|
|
dataframes
|
|
contain a subset of variables, then statistics for all variables will be returned, otherwise
|
|
only the top N vars will be returned.
|
|
:param obsFilterA: filter: dictionary with filter params for first set of observations
|
|
:param obsFilterB: filter: dictionary with filter params for second set of observations
|
|
:param top_n: Limit results to top N (Top var mode only)
|
|
:return: top N genes and corresponding stats
|
|
"""
|
|
if Axis.VAR in obsFilterA or Axis.VAR in obsFilterB:
|
|
raise FilterError("Observation filters may not contain variable conditions")
|
|
try:
|
|
shape = self.get_shape()
|
|
obs_mask_A = self._axis_filter_to_mask(Axis.OBS, obsFilterA["obs"], shape[0])
|
|
obs_mask_B = self._axis_filter_to_mask(Axis.OBS, obsFilterB["obs"], shape[0])
|
|
except (KeyError, IndexError):
|
|
raise FilterError("Error parsing filter")
|
|
if top_n is None:
|
|
top_n = self.dataset_config.diffexp__top_n
|
|
|
|
if self.server_config.exceeds_limit(
|
|
"diffexp_cellcount_max", np.count_nonzero(obs_mask_A) + np.count_nonzero(obs_mask_B)
|
|
):
|
|
raise ExceedsLimitError("Diffexp request exceeds max cell count limit")
|
|
|
|
result = self.compute_diffexp_ttest(
|
|
maskA=obs_mask_A,
|
|
maskB=obs_mask_B,
|
|
top_n=top_n,
|
|
lfc_cutoff=self.dataset_config.diffexp__lfc_cutoff,
|
|
)
|
|
|
|
try:
|
|
return jsonify_strict(result)
|
|
except ValueError:
|
|
raise JSONEncodingValueError("Error encoding differential expression to JSON")
|
|
|
|
@abstractmethod
|
|
def compute_diffexp_ttest(self, maskA, maskB, top_n, lfc_cutoff):
|
|
pass
|
|
|
|
@staticmethod
|
|
def normalize_embedding(embedding):
|
|
"""Normalize embedding layout to meet client assumptions.
|
|
Embedding is an ndarray, shape (n_obs, n)., where n is normally 2
|
|
"""
|
|
|
|
# scale isotropically
|
|
try:
|
|
min = np.nanmin(embedding, axis=0)
|
|
max = np.nanmax(embedding, axis=0)
|
|
except RuntimeError:
|
|
# indicates entire array was NaN, which should propagate
|
|
min = np.NaN
|
|
max = np.NaN
|
|
|
|
scale = np.amax(max - min)
|
|
normalized_layout = (embedding - min) / scale
|
|
|
|
# translate to center on both axis
|
|
translate = 0.5 - ((max - min) / scale / 2)
|
|
normalized_layout = normalized_layout + translate
|
|
|
|
normalized_layout = normalized_layout.astype(dtype=np.float32)
|
|
return normalized_layout
|
|
|
|
def layout_to_fbs_matrix(self, fields):
|
|
"""
|
|
return specified embeddings as a flatbuffer, using the cellxgene matrix fbs encoding.
|
|
|
|
* returns only first two dimensions, with name {ename}_0 and {ename}_1,
|
|
where {ename} is the embedding name.
|
|
* client assumes each will be individually centered & scaled (isotropically)
|
|
to a [0, 1] range.
|
|
* does not support filtering
|
|
|
|
"""
|
|
embeddings = self.get_embedding_names() if fields is None or len(fields) == 0 else fields
|
|
layout_data = []
|
|
with ServerTiming.time("layout.query"):
|
|
for ename in embeddings:
|
|
embedding = self.get_embedding_array(ename, 2)
|
|
normalized_layout = DataAdaptor.normalize_embedding(embedding)
|
|
layout_data.append(pd.DataFrame(normalized_layout, columns=[f"{ename}_0", f"{ename}_1"]))
|
|
|
|
with ServerTiming.time("layout.encode"):
|
|
if layout_data:
|
|
df = pd.concat(layout_data, axis=1, copy=False)
|
|
else:
|
|
df = pd.DataFrame()
|
|
fbs = encode_matrix_fbs(df, col_idx=df.columns, row_idx=None)
|
|
|
|
return fbs
|
|
|
|
def get_last_mod_time(self):
|
|
try:
|
|
lastmod = self.get_data_locator().lastmodtime()
|
|
except RuntimeError:
|
|
lastmod = None
|
|
return lastmod
|
|
|
|
def summarize_var(self, method, filter, query_hash):
|
|
if method != "mean":
|
|
raise UnsupportedSummaryMethod("Unknown gene set summary method.")
|
|
|
|
obs_selector, var_selector = self._filter_to_mask(filter)
|
|
if obs_selector is not None:
|
|
raise FilterError("filtering on obs unsupported")
|
|
|
|
# if no filter, just return zeros. We don't have a use case
|
|
# for summarizing the entire X without a filter, and it would
|
|
# potentially be quite compute / memory intensive.
|
|
if var_selector is None or np.count_nonzero(var_selector) == 0:
|
|
mean = np.zeros((self.get_shape()[0], 1), dtype=np.float32)
|
|
else:
|
|
X = self.get_X_array(obs_selector, var_selector)
|
|
if sparse.issparse(X):
|
|
mean = X.mean(axis=1).A
|
|
else:
|
|
mean = X.mean(axis=1, keepdims=True)
|
|
|
|
col_idx = pd.Index([query_hash])
|
|
return encode_matrix_fbs(mean, col_idx=col_idx, row_idx=None)
|