mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-09-17 05:47:58 +08:00
80f6137528
* add function to retrieve latest annotation from db, db updates * dont create directory in s3
320 lines
12 KiB
Python
320 lines
12 KiB
Python
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 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 server.db.cellxgene_orm import CellxGeneDataset, Annotation
|
|
from server.db.db_utils import DbUtils
|
|
|
|
|
|
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
|
|
|
|
|
|
class AnnotationsLocalFile(Annotations):
|
|
|
|
CXG_ANNO_COLLECTION = "cxg_anno_collection"
|
|
|
|
def __init__(self, output_dir, output_file):
|
|
super().__init__()
|
|
self.output_dir = output_dir
|
|
self.output_file = output_file
|
|
# lock used to protect label file write ops
|
|
self.label_lock = threading.RLock()
|
|
|
|
# cache the most recent annotations
|
|
self.last_fname = None
|
|
self.last_labels = None
|
|
|
|
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):
|
|
session[self.CXG_ANNO_COLLECTION] = name
|
|
session.permanent = True
|
|
|
|
def get_collection(self):
|
|
if session is None:
|
|
return None
|
|
return session.get(self.CXG_ANNO_COLLECTION)
|
|
|
|
def read_labels(self, data_adaptor):
|
|
if has_request_context():
|
|
if not current_app.auth.is_user_authenticated():
|
|
return pd.DataFrame()
|
|
|
|
fname = self._get_filename(data_adaptor)
|
|
with self.label_lock:
|
|
if fname is not None and os.path.exists(fname) and os.path.getsize(fname) > 0:
|
|
# returned the cached labels if possible, otherwise read them from the file
|
|
if fname == self.last_fname:
|
|
return self.last_labels
|
|
else:
|
|
labels = pd.read_csv(
|
|
fname, dtype="category", index_col=0, header=0, comment="#", keep_default_na=False
|
|
)
|
|
# update the cache
|
|
self.last_fname = fname
|
|
self.last_labels = labels
|
|
return labels
|
|
else:
|
|
return pd.DataFrame()
|
|
|
|
def write_labels(self, df, data_adaptor):
|
|
# update our internal state and save it. Multi-threading often enabled,
|
|
# so treat this as a critical section.
|
|
with self.label_lock:
|
|
lastmod = data_adaptor.get_last_mod_time()
|
|
lastmodstr = "'unknown'" if lastmod is None else lastmod.isoformat(timespec="seconds")
|
|
header = (
|
|
f"# Annotations generated on {datetime.now().isoformat(timespec='seconds')} "
|
|
f"using cellxgene version {cellxgene_version}\n"
|
|
f"# Input data file was {data_adaptor.get_location()}, "
|
|
f"which was last modified on {lastmodstr}\n"
|
|
)
|
|
|
|
fname = self._get_filename(data_adaptor)
|
|
self._backup(fname)
|
|
if not df.empty:
|
|
with open(fname, "w", newline="") as f:
|
|
if header is not None:
|
|
f.write(header)
|
|
df.to_csv(f)
|
|
else:
|
|
open(fname, "w").close()
|
|
|
|
# update the cache
|
|
self.last_fname = fname
|
|
self.last_labels = df
|
|
|
|
def _get_userdata_idhash(self, data_adaptor):
|
|
"""
|
|
Return a short hash that weakly identifies the user and dataset.
|
|
Used to create safe annotations output file names.
|
|
"""
|
|
uid = current_app.auth.get_user_id()
|
|
id = (uid + data_adaptor.get_location()).encode()
|
|
idhash = base64.b32encode(blake2b(id, digest_size=5).digest()).decode("utf-8")
|
|
return idhash
|
|
|
|
def _get_output_dir(self):
|
|
if self.output_dir:
|
|
return self.output_dir
|
|
|
|
if self.output_file:
|
|
return os.path.dirname(self.path.abspath(self.output_dir))
|
|
|
|
return os.getcwd()
|
|
|
|
def _get_filename(self, data_adaptor):
|
|
""" return the current annotation file name """
|
|
if self.output_file:
|
|
return self.output_file
|
|
|
|
# we need to generate a file name, which we can only do if we have a UID and collection name
|
|
if session is None:
|
|
raise AnnotationsError("unable to determine file name for annotations")
|
|
|
|
collection = self.get_collection()
|
|
if collection is None:
|
|
return None
|
|
|
|
if data_adaptor is None:
|
|
raise AnnotationsError("unable to determine file name for annotations")
|
|
|
|
idhash = self._get_userdata_idhash(data_adaptor)
|
|
return os.path.join(self._get_output_dir(), f"{collection}-{idhash}.csv")
|
|
|
|
def _backup(self, fname, max_backups=9):
|
|
"""
|
|
save N backups of file to backup_dir.
|
|
1. fname -> backup_dir/fname-TIME
|
|
2. delete excess files in backup_dir
|
|
"""
|
|
root, ext = os.path.splitext(fname)
|
|
backup_dir = f"{root}-backups"
|
|
|
|
# Make sure there is work to do
|
|
if not os.path.exists(fname):
|
|
return
|
|
|
|
# Ensure backup_dir exists
|
|
if not os.path.exists(backup_dir):
|
|
os.mkdir(backup_dir)
|
|
|
|
# Save current file to backup_dir
|
|
fname_base = os.path.basename(fname)
|
|
fname_base_root, fname_base_ext = os.path.splitext(fname_base)
|
|
# don't use ISO standard time format, as it contains characters illegal on some filesytems.
|
|
nowish = datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
|
|
backup_fname = os.path.join(backup_dir, f"{fname_base_root}-{nowish}{fname_base_ext}")
|
|
if os.path.exists(backup_fname):
|
|
os.remove(backup_fname)
|
|
os.rename(fname, backup_fname)
|
|
|
|
# prune the backup_dir to max number of backup files, keeping the most recent backups
|
|
backups = list(filter(lambda s: s.startswith(fname_base_root), os.listdir(backup_dir)))
|
|
excess_count = len(backups) - max_backups
|
|
if excess_count > 0:
|
|
backups.sort()
|
|
for bu in backups[0:excess_count]:
|
|
os.remove(os.path.join(backup_dir, bu))
|
|
|
|
def update_parameters(self, parameters, data_adaptor):
|
|
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
|
|
|
|
if self.output_file is not None:
|
|
# user has hard-wired the name of the annotation data collection
|
|
fname = os.path.basename(self.output_file)
|
|
collection_fname = os.path.splitext(fname)[0]
|
|
params["annotations-data-collection-is-read-only"] = True
|
|
params["annotations-data-collection-name"] = collection_fname
|
|
|
|
elif session is not None:
|
|
collection = self.get_collection()
|
|
if current_app.auth.is_user_authenticated():
|
|
params["annotations-user-data-idhash"] = self._get_userdata_idhash(data_adaptor)
|
|
params["annotations-data-collection-is-read-only"] = False
|
|
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
|