mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-09-16 05:07:55 +08:00
3ebbb0ccbf
* move common code into server, update tests and makefile remove backend directory, refactor update smoke tests
57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
from enum import Enum
|
|
|
|
from server.common.utils.data_locator import DataLocator
|
|
from server.common.errors import DatasetAccessError
|
|
from http import HTTPStatus
|
|
|
|
|
|
class MatrixDataType(Enum):
|
|
H5AD = "h5ad"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
class MatrixDataLoader(object):
|
|
def __init__(self, location, matrix_data_type=None, app_config=None):
|
|
""" location can be a string or DataLocator """
|
|
region_name = None if app_config is None else app_config.server_config.data_locator__s3__region_name
|
|
self.location = DataLocator(location, region_name=region_name)
|
|
if not self.location.exists():
|
|
raise DatasetAccessError("Dataset does not exist.", HTTPStatus.NOT_FOUND)
|
|
|
|
# matrix_data_type is an enum value of type MatrixDataType
|
|
self.matrix_data_type = matrix_data_type
|
|
# matrix_type is a DataAdaptor type, which corresonds to the matrix_data_type
|
|
self.matrix_type = None
|
|
|
|
if matrix_data_type is None:
|
|
self.matrix_data_type = self.__matrix_data_type()
|
|
|
|
if not self.__matrix_data_type_allowed(app_config):
|
|
raise DatasetAccessError("Dataset does not have an allowed type.")
|
|
|
|
if self.matrix_data_type == MatrixDataType.H5AD:
|
|
from server.data_anndata.anndata_adaptor import AnndataAdaptor
|
|
|
|
self.matrix_type = AnndataAdaptor
|
|
|
|
def __matrix_data_type(self):
|
|
if self.location.path.endswith(".h5ad"):
|
|
return MatrixDataType.H5AD
|
|
else:
|
|
return MatrixDataType.UNKNOWN
|
|
|
|
def __matrix_data_type_allowed(self, app_config):
|
|
return self.matrix_data_type != MatrixDataType.UNKNOWN
|
|
|
|
def pre_load_validation(self):
|
|
if self.matrix_data_type == MatrixDataType.UNKNOWN:
|
|
raise DatasetAccessError("Dataset does not have a recognized type: .h5ad")
|
|
self.matrix_type.pre_load_validation(self.location)
|
|
|
|
def file_size(self):
|
|
return self.matrix_type.file_size(self.location)
|
|
|
|
def open(self, app_config, dataset_config=None):
|
|
# create and return a DataAdaptor object
|
|
return self.matrix_type.open(self.location, app_config, dataset_config)
|