mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-10-01 01:38:12 +08:00
Create hosted user annotations [1685] (#1726)
* add function to retrieve latest annotation from db, db updates * read and write tiledb arrays * adding tests
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
from abc import ABCMeta, abstractmethod
|
||||
|
||||
import fastobo
|
||||
import fsspec
|
||||
from server.common.errors import OntologyLoadFailure
|
||||
|
||||
from server.common.utils import series_to_schema
|
||||
|
||||
|
||||
class Annotations(metaclass=ABCMeta):
|
||||
""" baseclass for annotations, including ontologies"""
|
||||
|
||||
""" our default ontology is the PURL for the Cell Ontology.
|
||||
See http://www.obofoundry.org/ontology/cl.html """
|
||||
DefaultOnotology = "http://purl.obolibrary.org/obo/cl.obo"
|
||||
|
||||
def __init__(self):
|
||||
self.ontology_data = None
|
||||
|
||||
def load_ontology(self, path):
|
||||
"""Load and parse ontologies - currently support OBO files only."""
|
||||
if path is None:
|
||||
path = self.DefaultOnotology
|
||||
|
||||
try:
|
||||
with fsspec.open(path) as f:
|
||||
obo = fastobo.iter(f)
|
||||
terms = filter(lambda stanza: type(stanza) is fastobo.term.TermFrame, obo)
|
||||
names = [tag.name for term in terms for tag in term if type(tag) is fastobo.term.NameClause]
|
||||
self.ontology_data = names
|
||||
|
||||
except FileNotFoundError as e:
|
||||
raise OntologyLoadFailure("Unable to find OBO ontology path") from e
|
||||
|
||||
except SyntaxError as e:
|
||||
raise OntologyLoadFailure("Syntax error loading OBO ontology") from e
|
||||
|
||||
except Exception as e:
|
||||
raise OntologyLoadFailure("Error loading OBO file") from e
|
||||
|
||||
def get_schema(self, data_adaptor):
|
||||
schema = []
|
||||
labels = self.read_labels(data_adaptor)
|
||||
if labels is not None and not labels.empty:
|
||||
for col in labels.columns:
|
||||
col_schema = dict(name=col, writable=True)
|
||||
col_schema.update(series_to_schema(labels[col]))
|
||||
schema.append(col_schema)
|
||||
|
||||
return schema
|
||||
|
||||
@abstractmethod
|
||||
def set_collection(self, name):
|
||||
"""set or create a new annotation collection"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def read_labels(self, data_adaptor):
|
||||
"""Return the labels as a pandas.DataFrame"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def write_labels(self, df, data_adaptor):
|
||||
"""Write the labels (df) to a persistent storage such that it can later be read"""
|
||||
pass
|
||||
|
||||
def update_parameters(self, parameters, data_adaptor):
|
||||
"""Update configuration parameters that describe information about the annotations feature"""
|
||||
params = {}
|
||||
params["annotations"] = True
|
||||
|
||||
if self.ontology_data:
|
||||
params["annotations_cell_ontology_enabled"] = True
|
||||
params["annotations_cell_ontology_terms"] = self.ontology_data
|
||||
else:
|
||||
params["annotations_cell_ontology_enabled"] = False
|
||||
|
||||
parameters.update(params)
|
||||
@@ -0,0 +1,113 @@
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
|
||||
import pandas as pd
|
||||
import tiledb
|
||||
from flask import current_app
|
||||
|
||||
from server.common.annotations.annotations import Annotations
|
||||
from server.converters.cxgtool import sanitize_keys, generate_schema_hints_and_convert_value_types, cxg_dtype
|
||||
from server.db.cellxgene_orm import CellxGeneDataset, Annotation
|
||||
|
||||
|
||||
class AnnotationsHostedTileDB(Annotations):
|
||||
CXG_ANNO_COLLECTION = "cxg_anno_collection"
|
||||
|
||||
def __init__(self, directory_path, db):
|
||||
super().__init__()
|
||||
self.db = db
|
||||
self.directory_path = directory_path
|
||||
|
||||
def check_category_names(self, df):
|
||||
sanitize_keys(df.keys().to_list(), False)
|
||||
|
||||
def is_safe_collection_name(self, name):
|
||||
"""
|
||||
return true if this is a safe collection name
|
||||
this is ultra conservative. If we want to allow full legal file name syntax,
|
||||
we could look at modules like `pathvalidate`
|
||||
"""
|
||||
if name is None:
|
||||
return False
|
||||
return re.match(r"^[\w\-]+$", name) is not None
|
||||
|
||||
def set_collection(self, name):
|
||||
self.CXG_ANNO_COLLECTION = name
|
||||
|
||||
def read_labels(self, data_adaptor):
|
||||
user_id = current_app.auth.get_user_id()
|
||||
dataset_name = data_adaptor.get_location()
|
||||
dataset_id = str(self.db.query(
|
||||
table_args=[CellxGeneDataset],
|
||||
filter_args=[CellxGeneDataset.name == dataset_name]
|
||||
)[0].id)
|
||||
|
||||
annotation_object = self.db.query_for_most_recent(
|
||||
Annotation, [Annotation.user_id == user_id, Annotation.dataset_id == dataset_id]
|
||||
)
|
||||
if annotation_object:
|
||||
df = tiledb.open(annotation_object.tiledb_uri)
|
||||
pandas_df = self.convert_to_pandas_df(df)
|
||||
return pandas_df
|
||||
else:
|
||||
return None
|
||||
|
||||
def convert_to_pandas_df(self, tileDBArray):
|
||||
repr_meta = None
|
||||
index_dims = None
|
||||
if '__pandas_attribute_repr' in tileDBArray.meta:
|
||||
# backwards compatibility... unsure if necessary at this point
|
||||
repr_meta = json.loads(tileDBArray.meta['__pandas_attribute_repr'])
|
||||
if '__pandas_index_dims' in tileDBArray.meta:
|
||||
index_dims = json.loads(tileDBArray.meta['__pandas_index_dims'])
|
||||
|
||||
data = tileDBArray[:]
|
||||
indexes = list()
|
||||
|
||||
for col_name, col_val in data.items():
|
||||
if repr_meta and col_name in repr_meta:
|
||||
new_col = pd.Series(col_val, dtype=repr_meta[col_name])
|
||||
data[col_name] = new_col
|
||||
elif index_dims and col_name in index_dims:
|
||||
new_col = pd.Series(col_val, dtype=index_dims[col_name])
|
||||
data[col_name] = new_col
|
||||
indexes.append(col_name)
|
||||
|
||||
new_df = pd.DataFrame.from_dict(data)
|
||||
if len(indexes) > 0:
|
||||
new_df.set_index(indexes, inplace=True)
|
||||
|
||||
return new_df
|
||||
|
||||
def write_labels(self, df, data_adaptor):
|
||||
|
||||
user_id = current_app.auth.get_user_id()
|
||||
timestamp = time.time()
|
||||
dataset_name = data_adaptor.get_location()
|
||||
dataset_id = self.db.get_or_create_dataset(dataset_name)
|
||||
user_id = self.db.get_or_create_user(user_id)
|
||||
|
||||
uri = f"{self.directory_path}-{dataset_name}-{user_id}-{timestamp}"
|
||||
if uri.startswith("s3://"):
|
||||
pass
|
||||
else:
|
||||
os.makedirs(uri, exist_ok=True)
|
||||
schema_hints, values = generate_schema_hints_and_convert_value_types(df)
|
||||
|
||||
annotation = Annotation(
|
||||
tiledb_uri=uri,
|
||||
user_id=user_id,
|
||||
dataset_id=str(dataset_id),
|
||||
schema_hints=json.dumps(schema_hints)
|
||||
)
|
||||
if not df.empty:
|
||||
self.check_category_names(df)
|
||||
# convert to tiledb datatypes
|
||||
for col in df:
|
||||
df[col] = df[col].astype(cxg_dtype(df[col]))
|
||||
tiledb.from_pandas(uri, df)
|
||||
|
||||
self.db.session.add(annotation)
|
||||
self.db.session.commit()
|
||||
@@ -1,86 +1,16 @@
|
||||
import json
|
||||
import uuid
|
||||
import time
|
||||
from datetime import datetime
|
||||
import re
|
||||
import os
|
||||
import pandas as pd
|
||||
from hashlib import blake2b
|
||||
import base64
|
||||
from server import __version__ as cellxgene_version
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
from server.common.errors import AnnotationsError, OntologyLoadFailure
|
||||
from server.common.utils import series_to_schema
|
||||
import fsspec
|
||||
import fastobo
|
||||
from flask import session, current_app, has_request_context
|
||||
from abc import ABCMeta, abstractmethod
|
||||
from datetime import datetime
|
||||
from hashlib import blake2b
|
||||
|
||||
from server.db.cellxgene_orm import CellxGeneDataset, Annotation
|
||||
from server.db.db_utils import DbUtils
|
||||
import pandas as pd
|
||||
from flask import session, has_request_context, current_app
|
||||
|
||||
|
||||
class Annotations(metaclass=ABCMeta):
|
||||
""" baseclass for annotations, including ontologies"""
|
||||
|
||||
""" our default ontology is the PURL for the Cell Ontology.
|
||||
See http://www.obofoundry.org/ontology/cl.html """
|
||||
DefaultOnotology = "http://purl.obolibrary.org/obo/cl.obo"
|
||||
|
||||
def __init__(self):
|
||||
self.ontology_data = None
|
||||
|
||||
def load_ontology(self, path):
|
||||
"""Load and parse ontologies - currently support OBO files only."""
|
||||
if path is None:
|
||||
path = self.DefaultOnotology
|
||||
|
||||
try:
|
||||
with fsspec.open(path) as f:
|
||||
obo = fastobo.iter(f)
|
||||
terms = filter(lambda stanza: type(stanza) is fastobo.term.TermFrame, obo)
|
||||
names = [tag.name for term in terms for tag in term if type(tag) is fastobo.term.NameClause]
|
||||
self.ontology_data = names
|
||||
|
||||
except FileNotFoundError as e:
|
||||
raise OntologyLoadFailure("Unable to find OBO ontology path") from e
|
||||
|
||||
except SyntaxError as e:
|
||||
raise OntologyLoadFailure("Syntax error loading OBO ontology") from e
|
||||
|
||||
except Exception as e:
|
||||
raise OntologyLoadFailure("Error loading OBO file") from e
|
||||
|
||||
def get_schema(self, data_adaptor):
|
||||
schema = []
|
||||
labels = self.read_labels(data_adaptor)
|
||||
if labels is not None and not labels.empty:
|
||||
for col in labels.columns:
|
||||
col_schema = dict(name=col, writable=True)
|
||||
col_schema.update(series_to_schema(labels[col]))
|
||||
schema.append(col_schema)
|
||||
|
||||
return schema
|
||||
|
||||
@abstractmethod
|
||||
def set_collection(self, name):
|
||||
"""set or create a new annotation collection"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def read_labels(self, data_adaptor):
|
||||
"""Return the labels as a pandas.DataFrame"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def write_labels(self, df, data_adaptor):
|
||||
"""Write the labels (df) to a persistent storage such that it can later be read"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def update_parameters(self, parameters, data_adaptor):
|
||||
"""Update configuration parameters that describe information about the annotations feature"""
|
||||
pass
|
||||
from server import __version__ as cellxgene_version
|
||||
from server.common.annotations.annotations import Annotations
|
||||
from server.common.errors import AnnotationsError
|
||||
|
||||
|
||||
class AnnotationsLocalFile(Annotations):
|
||||
@@ -101,7 +31,6 @@ class AnnotationsLocalFile(Annotations):
|
||||
def is_safe_collection_name(self, name):
|
||||
"""
|
||||
return true if this is a safe collection name
|
||||
|
||||
this is ultra conservative. If we want to allow full legal file name syntax,
|
||||
we could look at modules like `pathvalidate`
|
||||
"""
|
||||
@@ -265,55 +194,3 @@ class AnnotationsLocalFile(Annotations):
|
||||
params["annotations-data-collection-name"] = collection
|
||||
|
||||
parameters.update(params)
|
||||
|
||||
|
||||
class AnnotationsHostedTileDB(Annotations):
|
||||
def __init__(self, directory_path: str, db: DbUtils):
|
||||
super().__init__()
|
||||
self.db = db
|
||||
self.directory_path = directory_path
|
||||
|
||||
def set_collection(self, name):
|
||||
pass
|
||||
|
||||
def read_labels(self, data_adaptor):
|
||||
uid = current_app.auth.get_user_id()
|
||||
dataset_name = data_adaptor.get_location()
|
||||
dataset = self.db.query(table_args=[CellxGeneDataset], filter_args=[CellxGeneDataset.name == dataset_name])
|
||||
# Todo @madison retrieve latest based on timestamp
|
||||
annotation_object = self.db.query_for_most_recent( # noqa F841
|
||||
Annotation, [Annotation.user_id == uid, Annotation.dataset == dataset]
|
||||
)
|
||||
# Todo in future pr, retrieve dataframe from tiledb uri
|
||||
|
||||
def write_labels(self, df, data_adaptor):
|
||||
uid = current_app.auth.get_user_id()
|
||||
timestamp = time.time()
|
||||
dataset_name = data_adaptor.get_location()
|
||||
try:
|
||||
dataset_id = self.db.query(
|
||||
table_args=[CellxGeneDataset], filter_args=[CellxGeneDataset.name == dataset_name]
|
||||
)[0].id
|
||||
except IndexError:
|
||||
dataset_id = uuid.uuid4()
|
||||
dataset = CellxGeneDataset(id=dataset_id, name=dataset_name)
|
||||
self.db.session.add(dataset)
|
||||
|
||||
uri = f"{self.directory_path}/{dataset_name}/{uid}/{timestamp}"
|
||||
if "s3" in uri:
|
||||
pass
|
||||
else:
|
||||
os.makedirs(uri, exist_ok=True)
|
||||
schema_hints = {}
|
||||
annotation = Annotation(
|
||||
tiledb_uri=uri,
|
||||
user_id=uid,
|
||||
dataset_id=str(dataset_id),
|
||||
schema_hints=json.dumps(schema_hints)
|
||||
)
|
||||
# todo in future pr -- write df to tiledb, store at uri
|
||||
self.db.session.add(annotation)
|
||||
self.db.session.commit()
|
||||
|
||||
def update_parameters(self, parameters, data_adaptor):
|
||||
pass
|
||||
+45
-32
@@ -12,11 +12,13 @@ from server.common.errors import ConfigurationError, DatasetAccessError, Ontolog
|
||||
from server.data_common.matrix_loader import MatrixDataLoader, MatrixDataCacheManager, MatrixDataType
|
||||
from server.common.utils import find_available_port, is_port_available
|
||||
import warnings
|
||||
from server.common.annotations import AnnotationsLocalFile
|
||||
from server.common.annotations.hosted_tiledb import AnnotationsHostedTileDB
|
||||
from server.common.annotations.local_file_csv import AnnotationsLocalFile
|
||||
from server.common.utils import custom_format_warning
|
||||
import server.compute.diffexp_cxg as diffexp_tiledb
|
||||
from server.common.data_locator import discover_s3_region_name
|
||||
from server.auth.auth import AuthTypeFactory
|
||||
from server.db.db_utils import DbUtils
|
||||
|
||||
DEFAULT_SERVER_PORT = 5005
|
||||
# anything bigger than this will generate a special message
|
||||
@@ -277,6 +279,7 @@ class AppConfig(object):
|
||||
"is_authenticated": auth.is_user_authenticated(),
|
||||
"requires_client_login": auth.requires_client_login(),
|
||||
"username": auth.get_user_name(),
|
||||
"user_id": auth.get_user_id()
|
||||
}
|
||||
if auth.requires_client_login():
|
||||
config["authentication"].update({
|
||||
@@ -736,6 +739,8 @@ class DatasetConfig(BaseConfig):
|
||||
self.user_annotations__local_file_csv__file = dc["user_annotations"]["local_file_csv"]["file"]
|
||||
self.user_annotations__ontology__enable = dc["user_annotations"]["ontology"]["enable"]
|
||||
self.user_annotations__ontology__obo_location = dc["user_annotations"]["ontology"]["obo_location"]
|
||||
self.user_annotations__hosted_tiledb_array__db_uri = dc["user_annotations"]["hosted_tiledb_array"]["db_uri"]
|
||||
self.user_annotations__hosted_tiledb_array__hosted_file_directory = dc["user_annotations"]["hosted_tiledb_array"]["hosted_file_directory"] # noqa E501
|
||||
|
||||
self.embeddings__names = dc["embeddings"]["names"]
|
||||
self.embeddings__enable_reembedding = dc["embeddings"]["enable_reembedding"]
|
||||
@@ -786,6 +791,8 @@ class DatasetConfig(BaseConfig):
|
||||
self.check_attr("user_annotations__local_file_csv__file", (type(None), str))
|
||||
self.check_attr("user_annotations__ontology__enable", bool)
|
||||
self.check_attr("user_annotations__ontology__obo_location", (type(None), str))
|
||||
self.check_attr("user_annotations__hosted_tiledb_array__db_uri", (type(None), str))
|
||||
self.check_attr("user_annotations__hosted_tiledb_array__hosted_file_directory", (type(None), str))
|
||||
|
||||
if self.user_annotations__enable:
|
||||
server_config = self.app_config.server_config
|
||||
@@ -797,43 +804,49 @@ class DatasetConfig(BaseConfig):
|
||||
|
||||
# TODO, replace this with a factory pattern once we have more than one way
|
||||
# to do annotations. currently only local_file_csv
|
||||
if self.user_annotations__type != "local_file_csv":
|
||||
raise ConfigurationError('The only annotation type support is "local_file_csv"')
|
||||
if self.user_annotations__type == "local_file_csv":
|
||||
dirname = self.user_annotations__local_file_csv__directory
|
||||
filename = self.user_annotations__local_file_csv__file
|
||||
|
||||
dirname = self.user_annotations__local_file_csv__directory
|
||||
filename = self.user_annotations__local_file_csv__file
|
||||
if filename is not None and dirname is not None:
|
||||
raise ConfigurationError("'annotations-file' and 'annotations-dir' may not be used together.")
|
||||
|
||||
if filename is not None and dirname is not None:
|
||||
raise ConfigurationError("'annotations-file' and 'annotations-dir' may not be used together.")
|
||||
if filename is not None:
|
||||
lf_name, lf_ext = splitext(filename)
|
||||
if lf_ext and lf_ext != ".csv":
|
||||
raise ConfigurationError(f"annotation file type must be .csv: {filename}")
|
||||
|
||||
if filename is not None:
|
||||
lf_name, lf_ext = splitext(filename)
|
||||
if lf_ext and lf_ext != ".csv":
|
||||
raise ConfigurationError(f"annotation file type must be .csv: {filename}")
|
||||
if dirname is not None and not isdir(dirname):
|
||||
try:
|
||||
os.mkdir(dirname)
|
||||
except OSError:
|
||||
raise ConfigurationError("Unable to create directory specified by --annotations-dir")
|
||||
|
||||
if dirname is not None and not isdir(dirname):
|
||||
try:
|
||||
os.mkdir(dirname)
|
||||
except OSError:
|
||||
raise ConfigurationError("Unable to create directory specified by --annotations-dir")
|
||||
self.user_annotations = AnnotationsLocalFile(dirname, filename)
|
||||
|
||||
self.user_annotations = AnnotationsLocalFile(dirname, filename)
|
||||
|
||||
# if the user has specified a fixed label file, go ahead and validate it
|
||||
# so that we can remove errors early in the process.
|
||||
server_config = self.app_config.server_config
|
||||
if server_config.single_dataset__datapath and self.user_annotations__local_file_csv__file:
|
||||
with server_config.matrix_data_cache_manager.data_adaptor(
|
||||
self.tag, server_config.single_dataset__datapath, self.app_config
|
||||
) as data_adaptor:
|
||||
data_adaptor.check_new_labels(self.user_annotations.read_labels(data_adaptor))
|
||||
|
||||
if self.user_annotations__ontology__enable or self.user_annotations__ontology__obo_location:
|
||||
try:
|
||||
self.user_annotations.load_ontology(self.user_annotations__ontology__obo_location)
|
||||
except OntologyLoadFailure as e:
|
||||
raise ConfigurationError("Unable to load ontology terms\n" + str(e))
|
||||
# if the user has specified a fixed label file, go ahead and validate it
|
||||
# so that we can remove errors early in the process.
|
||||
server_config = self.app_config.server_config
|
||||
if server_config.single_dataset__datapath and self.user_annotations__local_file_csv__file:
|
||||
with server_config.matrix_data_cache_manager.data_adaptor(
|
||||
self.tag, server_config.single_dataset__datapath, self.app_config
|
||||
) as data_adaptor:
|
||||
data_adaptor.check_new_labels(self.user_annotations.read_labels(data_adaptor))
|
||||
|
||||
if self.user_annotations__ontology__enable or self.user_annotations__ontology__obo_location:
|
||||
try:
|
||||
self.user_annotations.load_ontology(self.user_annotations__ontology__obo_location)
|
||||
except OntologyLoadFailure as e:
|
||||
raise ConfigurationError("Unable to load ontology terms\n" + str(e))
|
||||
elif self.user_annotations__type == "hosted_tiledb_array":
|
||||
self.check_attr("user_annotations__hosted_tiledb_array__db_uri", str)
|
||||
self.check_attr("user_annotations__hosted_tiledb_array__hosted_file_directory", str)
|
||||
self.user_annotations = AnnotationsHostedTileDB(
|
||||
directory_path=self.user_annotations__hosted_tiledb_array__hosted_file_directory,
|
||||
db=DbUtils(self.user_annotations__hosted_tiledb_array__db_uri),
|
||||
)
|
||||
else:
|
||||
raise ConfigurationError('The only annotation type support is "local_file_csv" or "hosted_tiledb_array')
|
||||
else:
|
||||
if self.user_annotations__type == "local_file_csv":
|
||||
dirname = self.user_annotations__local_file_csv__directory
|
||||
|
||||
@@ -171,6 +171,9 @@ dataset:
|
||||
user_annotations:
|
||||
enable: true
|
||||
type: local_file_csv
|
||||
hosted_tiledb_array:
|
||||
db_uri: null
|
||||
hosted_file_directory: null
|
||||
local_file_csv:
|
||||
directory: null
|
||||
file: null
|
||||
|
||||
@@ -46,6 +46,11 @@ define_request_exception(
|
||||
"Raised when there is an authentication error",
|
||||
default_status_code=HTTPStatus.UNAUTHORIZED)
|
||||
|
||||
define_request_exception(
|
||||
"AnnotationCategoryNameError",
|
||||
"Raised when an annotation category name cant be saved",
|
||||
default_status_code=HTTPStatus.UNPROCESSABLE_ENTITY)
|
||||
|
||||
define_exception("OntologyLoadFailure", "Raised when reading the ontology file fails")
|
||||
define_exception("ConfigurationError", "Raised when checking configuration errors")
|
||||
define_exception("PrepareError", "Raised when data is misprepared")
|
||||
|
||||
Reference in New Issue
Block a user