mirror of
https://github.com/Novartis/cellxgene-gateway.git
synced 2026-09-25 03:38:11 +08:00
96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
# Copyright 2019 Novartis Institutes for BioMedical Research Inc. Licensed
|
|
# under the Apache License, Version 2.0 (the "License"); you may not use
|
|
# this file except in compliance with the License. You may obtain a copy
|
|
# of the License at http://www.apache.org/licenses/LICENSE-2.0. Unless
|
|
# required by applicable law or agreed to in writing, software distributed
|
|
# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES
|
|
# OR CONDITIONS OF ANY KIND, either express or implied. See the License for
|
|
# the specific language governing permissions and limitations under the License.
|
|
|
|
import os
|
|
|
|
from flask_api import status
|
|
|
|
from cellxgene_gateway import env
|
|
from cellxgene_gateway.cellxgene_exception import CellxgeneException
|
|
from cellxgene_gateway.dir_util import make_h5ad
|
|
from cellxgene_gateway.cache_key import CacheKey
|
|
|
|
def get_key(path):
|
|
if path == "/" or path == "":
|
|
raise CellxgeneException(
|
|
"No matching dataset found.", status.HTTP_404_NOT_FOUND
|
|
)
|
|
|
|
trimmed = path[:-1] if path[-1] == "/" else path
|
|
try:
|
|
# valid paths come in three forms:
|
|
if trimmed.endswith('.h5ad') and data_file_exists(trimmed):
|
|
# 1) somedir/dataset.h5ad: a dataset
|
|
return CacheKey(trimmed, trimmed, None)
|
|
elif trimmed.endswith('.csv'):
|
|
|
|
# 2) somedir/dataset_annotations/saldaal1-T5HMVBNV.csv : an actual annotations file.
|
|
annotations_dir = os.path.split(trimmed)[0]
|
|
dataset = make_h5ad(annotations_dir)
|
|
if data_file_exists(dataset):
|
|
data_dir_ensure(annotations_dir)
|
|
return CacheKey(trimmed, dataset, trimmed)
|
|
elif trimmed.endswith('_annotations') and data_dir_exists(trimmed):
|
|
# 3) somedir/dataset_annotations: an annotation directory. The corresponding h5ad must exist, but the directory may not.
|
|
dataset = make_h5ad(trimmed)
|
|
if data_file_exists(dataset):
|
|
return CacheKey(trimmed, dataset, '')
|
|
except CellxgeneException:
|
|
pass
|
|
split = os.path.split(trimmed)
|
|
return get_key(split[0])
|
|
|
|
def validate_exists(file_path):
|
|
if not os.path.exists(file_path):
|
|
raise CellxgeneException(
|
|
"File does not exist: " + file_path, status.HTTP_400_BAD_REQUEST
|
|
)
|
|
|
|
def validate_is_file(file_path):
|
|
validate_exists(file_path)
|
|
if not os.path.isfile(file_path):
|
|
raise CellxgeneException(
|
|
"Path is not file: " + file_path, status.HTTP_400_BAD_REQUEST
|
|
)
|
|
return
|
|
def validate_is_dir(file_path):
|
|
validate_exists(file_path)
|
|
if not os.path.isdir(file_path):
|
|
raise CellxgeneException(
|
|
"Path is not dir: " + file_path, status.HTTP_400_BAD_REQUEST
|
|
)
|
|
return
|
|
|
|
def data_file_exists(dataset):
|
|
file_path = os.path.join(env.cellxgene_data, dataset)
|
|
validate_is_file(file_path)
|
|
return True
|
|
def data_dir_exists(dataset):
|
|
file_path = os.path.join(env.cellxgene_data, dataset)
|
|
validate_is_dir(file_path)
|
|
return True
|
|
def data_dir_ensure(dataset):
|
|
file_path = os.path.join(env.cellxgene_data, dataset)
|
|
if not os.path.exists(file_path):
|
|
os.makedirs(file_path)
|
|
|
|
def get_file_path(key):
|
|
dataset = key.dataset
|
|
file_path = os.path.join(env.cellxgene_data, dataset)
|
|
validate_is_file(file_path)
|
|
return file_path
|
|
|
|
def get_annotation_file_path(key):
|
|
if key.annotation_file is None:
|
|
return None
|
|
if key.annotation_file == '':
|
|
return ''
|
|
file_path = os.path.join(env.cellxgene_data, key.annotation_file)
|
|
return file_path
|