mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-09-17 05:47:58 +08:00
5947306ca0
* Add server plugin system Plugins are optional modules loaded at runtime. Specification: * Plugins are loaded from the server.plugins module (directory server/plugins) * The import_plugins method is run as part of the initialization of the server module in __init__.py * Add plugins to the EB build process * Remove bit of dead code * Respond to feedback from @bmccandless
83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
import random
|
|
import shutil
|
|
import string
|
|
import tempfile
|
|
from os import path, popen
|
|
|
|
import pandas as pd
|
|
|
|
from server.common.annotations import AnnotationsLocalFile
|
|
from server.common.data_locator import DataLocator
|
|
from server.common.app_config import AppConfig
|
|
from server.data_common.fbs.matrix import encode_matrix_fbs
|
|
from server.data_common.matrix_loader import MatrixDataLoader, MatrixDataType
|
|
|
|
|
|
PROJECT_ROOT = popen("git rev-parse --show-toplevel").read().strip()
|
|
|
|
|
|
def data_with_tmp_annotations(ext: MatrixDataType, annotations_fixture=False):
|
|
tmp_dir = tempfile.mkdtemp()
|
|
annotations_file = path.join(tmp_dir, "test_annotations.csv")
|
|
if annotations_fixture:
|
|
shutil.copyfile(f"{PROJECT_ROOT}/server/test/test_datasets/pbmc3k-annotations.csv", annotations_file)
|
|
args = {
|
|
"embeddings__names": ["umap"],
|
|
"presentation__max_categories": 100,
|
|
"single_dataset__obs_names": None,
|
|
"single_dataset__var_names": None,
|
|
"diffexp__lfc_cutoff": 0.01,
|
|
}
|
|
fname = {
|
|
MatrixDataType.H5AD: f"{PROJECT_ROOT}/example-dataset/pbmc3k.h5ad",
|
|
MatrixDataType.CXG: "test/test_datasets/pbmc3k.cxg",
|
|
}[ext]
|
|
data_locator = DataLocator(fname)
|
|
config = AppConfig()
|
|
config.update(**args)
|
|
config.update(single_dataset__datapath=data_locator.path)
|
|
config.complete_config()
|
|
data = MatrixDataLoader(data_locator.abspath()).open(config)
|
|
annotations = AnnotationsLocalFile(None, annotations_file)
|
|
return data, tmp_dir, annotations
|
|
|
|
|
|
def make_fbs(data):
|
|
df = pd.DataFrame(data)
|
|
return encode_matrix_fbs(matrix=df, row_idx=None, col_idx=df.columns)
|
|
|
|
|
|
def skip_if(condition, reason: str):
|
|
def decorator(f):
|
|
def wraps(self, *args, **kwargs):
|
|
if condition(self):
|
|
self.skipTest(reason)
|
|
else:
|
|
f(self, *args, **kwargs)
|
|
|
|
return wraps
|
|
|
|
return decorator
|
|
|
|
|
|
def app_config(data_locator, backed=False):
|
|
args = {
|
|
"embeddings__names": ["umap", "tsne", "pca"],
|
|
"presentation__max_categories": 100,
|
|
"single_dataset__obs_names": None,
|
|
"single_dataset__var_names": None,
|
|
"diffexp__lfc_cutoff": 0.01,
|
|
"adaptor__anndata_adaptor__backed": backed,
|
|
"single_dataset__datapath": data_locator,
|
|
"limits__diffexp_cellcount_max": None,
|
|
"limits__column_request_max": None,
|
|
}
|
|
config = AppConfig()
|
|
config.update(**args)
|
|
config.complete_config()
|
|
return config
|
|
|
|
|
|
def random_string(n):
|
|
return "".join(random.choice(string.ascii_letters) for _ in range(n))
|