Skip to content

api.common.data_handlers

compounds

ROLE = os.environ.get('SIIBRA_API_ROLE', 'all') module-attribute

ROLE

data_decorator(role)

data decorator

Extensively used in api.common.data_handlers. Most of the business logic, including data fetching, serialization, etc.

Parameters:

Name Type Description Default
role ROLE_TYPE

Role of this process

required

Raises:

Type Description
ImportError

Celery not installed, but role is set to either worker or server

Source code in api/common/decorators.py
def data_decorator(role: ROLE_TYPE):
    """data decorator

    Extensively used in api.common.data_handlers. Most of the business logic, including data fetching, serialization, etc.

    Args:
        role: Role of this process

    Raises:
        ImportError: Celery not installed, but role is set to either `worker` or `server`
    """
    def outer_wrapper(fn):
        if role == "all":
            return fn
        if role == "worker" or role == "server":
            try:
                from api.worker import app

                def celery_task_wrapper(self, *args, **kwargs):
                    return fn(*args, **kwargs)

                return app.task(bind=True)(
                    wraps(fn)(celery_task_wrapper)
                )
            except ImportError as e:
                errmsg = f"For worker role, celery must be installed as a dep"
                logger.critical(errmsg)
                raise ImportError(errmsg) from e
        raise FaultyRoleException(f"role must be 'all', 'server' or 'worker', but it is {role}")
    return outer_wrapper

download_all(space_id=None, parcellation_id=None, region_id=None, feature_id=None, bbox=None, strict_mode=None)

Create a download bundle (zip) for the provided specification

Parameters:

Name Type Description Default
space_id str

lookup id of the space requested

None
parcellation_id str

lookup_id of the parcellation requested

None
region_id str

lookup_id of the region requested

None

Returns:

Type Description
str

Path to the zip file

Source code in api/common/data_handlers/compounds/download.py
@data_decorator(ROLE)
def download_all(space_id: str=None, parcellation_id: str=None, region_id: str=None, feature_id: str=None, bbox: str=None, strict_mode: str=None) -> str:
    """Create a download bundle (zip) for the provided specification

    Args:
        space_id: lookup id of the space requested
        parcellation_id: lookup_id of the parcellation requested
        region_id: lookup_id of the region requested

    Returns:
        Path to the zip file

    """
    zipfilename = Path(SIIBRA_API_SHARED_DIR, f"atlas-download-{str(uuid4())}.zip")
    output_credit = 1 if strict_mode else 1e10

    import siibra

    with ZipFile(zipfilename, "w") as zipfile:

        if feature_id:
            try:
                path_to_feature_export = Path(SIIBRA_API_SHARED_DIR, f"export-{feature_id}.zip")
                if not path_to_feature_export.exists():
                    feature = siibra.features.Feature.get_instance_by_id(feature_id)
                    feature.export(path_to_feature_export)
                zipfile.write(path_to_feature_export, f"export-{feature_id}.zip")
            except Exception as e:
                zipfile.writestr(f"{feature_id}.error.txt", f"Feature exporting failed: {str(e)}")

        injected_content=f"space_id={space_id}, parcellation_id={parcellation_id}, region_id={region_id}, bbox={bbox}"

        readme_txt = README.format(siibra_api_version=__version__,
                                    timestamp=str(datetime.now()),
                                    injected_content=injected_content)
        zipfile.writestr("README.md", readme_txt)
        zipfile.writestr("LICENCE.txt", LICENSE)

        if region_id is not None:
            if output_credit >= 1:
                download_region(space_id, parcellation_id, region_id, bbox=bbox, zipfile=zipfile)
                output_credit -= 1

        if parcellation_id is not None:
            if output_credit >= 1:
                download_parcellation(space_id, parcellation_id, bbox=bbox, zipfile=zipfile)
                output_credit -= 1

        if space_id is not None:
            if output_credit >= 1:
                download_space(space_id, bbox=bbox, zipfile=zipfile)
                output_credit -= 1

    return str(zipfilename)

download

download_all(space_id=None, parcellation_id=None, region_id=None, feature_id=None, bbox=None, strict_mode=None)

Create a download bundle (zip) for the provided specification

Parameters:

Name Type Description Default
space_id str

lookup id of the space requested

None
parcellation_id str

lookup_id of the parcellation requested

None
region_id str

lookup_id of the region requested

None

Returns:

Type Description
str

Path to the zip file

Source code in api/common/data_handlers/compounds/download.py
@data_decorator(ROLE)
def download_all(space_id: str=None, parcellation_id: str=None, region_id: str=None, feature_id: str=None, bbox: str=None, strict_mode: str=None) -> str:
    """Create a download bundle (zip) for the provided specification

    Args:
        space_id: lookup id of the space requested
        parcellation_id: lookup_id of the parcellation requested
        region_id: lookup_id of the region requested

    Returns:
        Path to the zip file

    """
    zipfilename = Path(SIIBRA_API_SHARED_DIR, f"atlas-download-{str(uuid4())}.zip")
    output_credit = 1 if strict_mode else 1e10

    import siibra

    with ZipFile(zipfilename, "w") as zipfile:

        if feature_id:
            try:
                path_to_feature_export = Path(SIIBRA_API_SHARED_DIR, f"export-{feature_id}.zip")
                if not path_to_feature_export.exists():
                    feature = siibra.features.Feature.get_instance_by_id(feature_id)
                    feature.export(path_to_feature_export)
                zipfile.write(path_to_feature_export, f"export-{feature_id}.zip")
            except Exception as e:
                zipfile.writestr(f"{feature_id}.error.txt", f"Feature exporting failed: {str(e)}")

        injected_content=f"space_id={space_id}, parcellation_id={parcellation_id}, region_id={region_id}, bbox={bbox}"

        readme_txt = README.format(siibra_api_version=__version__,
                                    timestamp=str(datetime.now()),
                                    injected_content=injected_content)
        zipfile.writestr("README.md", readme_txt)
        zipfile.writestr("LICENCE.txt", LICENSE)

        if region_id is not None:
            if output_credit >= 1:
                download_region(space_id, parcellation_id, region_id, bbox=bbox, zipfile=zipfile)
                output_credit -= 1

        if parcellation_id is not None:
            if output_credit >= 1:
                download_parcellation(space_id, parcellation_id, bbox=bbox, zipfile=zipfile)
                output_credit -= 1

        if space_id is not None:
            if output_credit >= 1:
                download_space(space_id, bbox=bbox, zipfile=zipfile)
                output_credit -= 1

    return str(zipfilename)

core

atlas

all_atlases()

Get all atlases

Returns:

Type Description
List[Dict]

List of all serialized atlases.

Source code in api/common/data_handlers/core/atlas.py
@data_decorator(ROLE)
def all_atlases() -> List[Dict]:
    """Get all atlases

    Returns:
        List of all serialized atlases."""
    import siibra
    from api.serialization.util import instance_to_model
    return [instance_to_model(atlas).dict() for atlas in siibra.atlases]

single_atlas(atlas_id)

Get a single atlas

Parameters:

Name Type Description Default
atlas_id str

id of the atlas

required

Returns:

Type Description
Dict

The atlas specified by the provided id, serialized into dict

Source code in api/common/data_handlers/core/atlas.py
@data_decorator(ROLE)
def single_atlas(atlas_id: str) -> Dict:
    """Get a single atlas

    Args:
        atlas_id: id of the atlas

    Returns:
        The atlas specified by the provided id, serialized into dict
    """
    import siibra
    from api.serialization.util import instance_to_model
    atlas = siibra.atlases[atlas_id]
    return instance_to_model(atlas).dict()

misc

cache_region_statistic_map(parcellation_id, region_id, space_id)

Retrieve and save regional statistical map (if necessary), and then return the path of the map.

Parameters:

Name Type Description Default
parcellation_id str

lookup id of the parcellation of the map

required
region_id str

lookup id of the region of the map

required
space_id str

lookup id of the space of the map

required

Returns:

Type Description
Tuple[str, bool]

path to statistical map, if a cached file is returned

Source code in api/common/data_handlers/core/misc.py
def cache_region_statistic_map(parcellation_id: str, region_id: str, space_id: str) -> Tuple[str, bool]:
    """Retrieve and save regional statistical map (if necessary), and then return the path of the map.

    Args:
        parcellation_id: lookup id of the parcellation of the map
        region_id: lookup id of the region of the map
        space_id: lookup id of the space of the map

    Returns:
        path to statistical map, if a cached file is returned
    """
    import os
    full_filename = get_filename("statistical_map", parcellation_id, region_id, space_id, ext=".nii.gz")
    if os.path.isfile(full_filename):
        return full_filename, True

    import siibra
    import nibabel as nib
    error_text = f"Map with parc id '{parcellation_id}', space id '{space_id}'"

    stat_map = siibra.get_map(parcellation_id, space_id, siibra.MapType.STATISTICAL)
    assert stat_map is not None, f"{error_text} returns None"

    volume_data = stat_map.fetch(region=region_id)

    error_text = f"{error_text}, with region_id '{region_id}'"
    assert isinstance(volume_data, nib.Nifti1Image), f"{error_text}, volume provided is not of type Nifti1Image"

    nib.save(volume_data, full_filename)
    import json
    import time
    with open(f"{full_filename}.{str(int(time.time()))}.json", "w") as fp:
        json.dump({
            "prefix": "statistical_map",
            "parcellation_id": parcellation_id,
            "region_id": region_id,
            "space_id": space_id,
        }, fp=fp, indent="\t")
    return full_filename, False

get_map(parcellation_id, space_id, maptype)

Get a map instance, based on specification

Parameters:

Name Type Description Default
parcellation_id str

lookup id of the parcellation of the map

required
space_id str

lookup id of the space of the map

required
maptype Union[MapType, str]

maptype, either LABELLED or STATISTICAL

required

Returns:

Type Description
Dict

Requested map instance, serialized into dict

Raises:

Type Description
AssertionError

if the supplied maptype is invalid type

NotFound

Map with the specification not found

Source code in api/common/data_handlers/core/misc.py
@data_decorator(ROLE)
def get_map(parcellation_id: str, space_id: str, maptype: Union[MapType, str]) -> Dict:
    """Get a map instance, based on specification

    Args:
        parcellation_id: lookup id of the parcellation of the map
        space_id: lookup id of the space of the map
        maptype: maptype, either LABELLED or STATISTICAL

    Returns:
        Requested map instance, serialized into dict

    Raises:
        AssertionError: if the supplied maptype is invalid type
        NotFound: Map with the specification not found
    """
    import siibra
    from api.serialization.util import instance_to_model

    maptype_string = parse_maptype(maptype)

    assert maptype_string is not None, f"maptype is neither MapType nor str"

    siibra_maptype = siibra.MapType[maptype_string]
    assert siibra_maptype.name == maptype_string, f"Expecting maptype.name to match"

    returned_map = siibra.get_map(parcellation_id, space_id, siibra_maptype)

    if returned_map is None:
        raise NotFound

    return instance_to_model(
        returned_map, detail=True
    ).dict()

get_parcellation_labelled_map(parcellation_id, space_id, region_id=None)

Retrieve and save labelled map / regional mask (if necessary), and then return the path of the map.

Parameters:

Name Type Description Default
parcellation_id str

lookup id of the parcellation of the map

required
region_id str

lookup id of the region of the map

None
space_id str

lookup id of the space of the map

required

Returns:

Type Description

path to labelled map/regional mask, if a cached file is returned

Source code in api/common/data_handlers/core/misc.py
@data_decorator(ROLE)
def get_parcellation_labelled_map(parcellation_id: str, space_id: str, region_id:str=None):
    """Retrieve and save labelled map / regional mask (if necessary), and then return the path of the map.

    Args:
        parcellation_id: lookup id of the parcellation of the map
        region_id: lookup id of the region of the map
        space_id: lookup id of the space of the map

    Returns:
        path to labelled map/regional mask, if a cached file is returned
    """
    import os
    full_filename = get_filename("labelled_map", parcellation_id, space_id, region_id if region_id else "", ext=".nii.gz")
    if os.path.isfile(full_filename):
        return full_filename, True

    import siibra
    import nibabel as nib
    error_text = f"Map with parc id '{parcellation_id}', space id '{space_id}'"

    volume_data = None
    if region_id is not None:
        region = siibra.get_region(parcellation_id, region_id)
        volume_data = region.fetch_regional_map(space_id, siibra.MapType.LABELLED)
    else:
        labelled_map = siibra.get_map(parcellation_id, space_id, siibra.MapType.LABELLED)
        assert labelled_map is not None, f"{error_text} returns None"
        volume_data = labelled_map.fetch()

    assert isinstance(volume_data, nib.Nifti1Image), f"{error_text}, volume provided is not of type Nifti1Image"

    nib.save(volume_data, full_filename)
    import json
    import time
    with open(f"{full_filename}.{str(int(time.time()))}.json", "w") as fp:
        json.dump({
            "prefix": "labelled_map",
            "parcellation_id": parcellation_id,
            "space_id": space_id,
            "region_id": region_id,
        }, fp=fp, indent="\t")
    return full_filename, False

get_region_statistic_map(parcellation_id, region_id, space_id)

Retrieve and save regional statistical map (if necessary), and then return the path of the map.

Parameters:

Name Type Description Default
parcellation_id str

lookup id of the parcellation of the map

required
region_id str

lookup id of the region of the map

required
space_id str

lookup id of the space of the map

required

Returns:

Type Description

path to statistical map, if a cached file is returned

Source code in api/common/data_handlers/core/misc.py
@data_decorator(ROLE)
def get_region_statistic_map(parcellation_id: str, region_id: str, space_id: str):
    """Retrieve and save regional statistical map (if necessary), and then return the path of the map.

    Args:
        parcellation_id: lookup id of the parcellation of the map
        region_id: lookup id of the region of the map
        space_id: lookup id of the space of the map

    Returns:
        path to statistical map, if a cached file is returned
    """
    return cache_region_statistic_map(parcellation_id, region_id, space_id)

get_region_statistic_map_info(parcellation_id, region_id, space_id)

Retrieve and save regional statistical map (if necessary), and then return the path of the map.

Parameters:

Name Type Description Default
parcellation_id str

lookup id of the parcellation of the map

required
region_id str

lookup id of the region of the map

required
space_id str

lookup id of the space of the map

required

Returns:

Type Description

dict of min an max of the statistical map

Source code in api/common/data_handlers/core/misc.py
@data_decorator(ROLE)
def get_region_statistic_map_info(parcellation_id: str, region_id: str, space_id: str):
    """Retrieve and save regional statistical map (if necessary), and then return the path of the map.

    Args:
        parcellation_id: lookup id of the parcellation of the map
        region_id: lookup id of the region of the map
        space_id: lookup id of the space of the map

    Returns:
        dict of min an max of the statistical map
    """
    full_filename, _cache_flag = cache_region_statistic_map(parcellation_id, region_id, space_id)

    import nibabel as nib
    import numpy as np

    nii = nib.load(full_filename)
    data = nii.get_fdata()
    return {
        "min": np.min(data),
        "max": np.max(data),
    }

get_resampled_map(parcellation_id, space_id, name=None)

Retrieve and save a labelled map, resampled in space (if necessary), and then return the path of the map.

Parameters:

Name Type Description Default
parcellation_id str

lookup id of the parcellation of the map

required
space_id str

lookup id of the target space of the sampled map

required

Returns:

Type Description

path to statistical map, if a cached file is returned

Source code in api/common/data_handlers/core/misc.py
@data_decorator(ROLE)
def get_resampled_map(parcellation_id: str, space_id: str, name: str=None):
    """Retrieve and save a labelled map, resampled in space (if necessary), and then return the path of the map.

    Args:
        parcellation_id: lookup id of the parcellation of the map
        space_id: lookup id of the target space of the sampled map

    Returns:
        path to statistical map, if a cached file is returned
    """
    import os
    full_filename = get_filename("resampled_map", parcellation_id, space_id, name or "none", ext=".nii.gz")
    if os.path.isfile(full_filename):
        return full_filename, True

    import siibra
    import nibabel as nib
    parcellation: siibra.core.parcellation.Parcellation = siibra.parcellations[parcellation_id]
    parcellation_map = parcellation.get_map(siibra.spaces[space_id], siibra.MapType.LABELLED)
    if len(parcellation_map.fragments) > 0:
        nii = parcellation_map.get_resampled_template(fragment=name)
    else:
        nii = parcellation_map.get_resampled_template()

    assert isinstance(nii, nib.Nifti1Image), f"resample failed... returned not of type nii"

    import time
    import json
    nib.save(nii, full_filename)
    with open(f"{full_filename}.{str(int(time.time()))}.json", "w") as fp:
        json.dump({
            "prefix": "resampled_map",
            "parcellation_id": parcellation_id,
            "space_id": space_id,
        }, indent="\t", fp=fp)
    return full_filename, False

parcellation

all_parcellations()

Get all parcellations

Returns:

Type Description

List of all serialized parcellations

Source code in api/common/data_handlers/core/parcellation.py
@data_decorator(ROLE)
def all_parcellations():
    """Get all parcellations

    Returns:
        List of all serialized parcellations"""
    import siibra
    from api.serialization.util import instance_to_model
    return [instance_to_model(parc).dict() for parc in siibra.parcellations]

single_parcellation(parc_id)

Get a single parcellation

Parameters:

Name Type Description Default
parc_id str

id of the parcellation

required

Returns:

Type Description

The parcellation specified by the provided id, serialized into dict

Source code in api/common/data_handlers/core/parcellation.py
@data_decorator(ROLE)
def single_parcellation(parc_id: str):
    """Get a single parcellation

    Args:
        parc_id: id of the parcellation

    Returns:
        The parcellation specified by the provided id, serialized into dict
    """
    import siibra
    from api.serialization.util import instance_to_model
    return instance_to_model(
        siibra.parcellations[parc_id]
    ).dict()

region

all_regions(parcellation_id, find=None)

Get all regions, categorised under the parcellation specified by parcellation_id, optionally filtered by find.

Parameters:

Name Type Description Default
parcellation_id str

id of the parcellation, under which regions will be fetched

required
find str

string to search for

None

Returns:

Type Description
List

List of all serialized regions

Source code in api/common/data_handlers/core/region.py
@data_decorator(ROLE)
def all_regions(parcellation_id: str, find:str=None) -> List:
    """Get all regions, categorised under the parcellation specified by `parcellation_id`, optionally filtered by find.

    Args:
        parcellation_id: id of the parcellation, under which regions will be fetched
        find: string to search for

    Returns:
        List of all serialized regions"""
    import siibra
    from api.serialization.util import instance_to_model
    parc = siibra.parcellations[parcellation_id]
    regions = [r for r in (parc.find(find) if find else parc) ]
    return [instance_to_model(r).dict() for r in regions if r is not parc]

Get related regions, including the relationship qualification.

Parameters:

Name Type Description Default
parcellation_id str

id of the parcellation, under which the regions will be fetched

required
region_id str

lookup id of the region

required

Returns:

Type Description

List of RegionalRelationshipAssessments, serialized

Source code in api/common/data_handlers/core/region.py
@data_decorator(ROLE)
def get_related_regions(parcellation_id: str, region_id: str):
    """Get related regions, including the relationship qualification.

    Args:
        parcellation_id: id of the parcellation, under which the regions will be fetched
        region_id: lookup id of the region

    Returns:
        List of RegionalRelationshipAssessments, serialized"""
    import siibra
    from api.serialization.util import instance_to_model

    region = siibra.get_region(parcellation_id, region_id)
    all_assessments = [ass for ass in region.get_related_regions()]
    return [instance_to_model(ass).dict() for ass in all_assessments]

single_region(parcellation_id, region_id, space_id=None)

Get a single region, categorised under parcellation specified by parcellation_id, defined by region_id.

If space_id optional parameter is supplied, additional information about the region in the space will also be returned.

Parameters:

Name Type Description Default
parcellation_id str

id of the parcellation, under which the regions will be fetched

required
region_id str

lookup id of the region

required
space_id str

additional information about the region in the provided space

None

Returns:

Type Description

The region, specified by the criteria, serialized.

Source code in api/common/data_handlers/core/region.py
@data_decorator(ROLE)
def single_region(parcellation_id: str, region_id: str, space_id: str=None):
    """Get a single region, categorised under parcellation specified by `parcellation_id`, defined by `region_id`.

    If space_id optional parameter is supplied, additional information about the region in the space will also be returned.

    Args:
        parcellation_id: id of the parcellation, under which the regions will be fetched
        region_id: lookup id of the region
        space_id: additional information about the region in the provided space

    Returns:
        The region, specified by the criteria, serialized."""
    import siibra
    from api.serialization.util import instance_to_model

    space = siibra.spaces[space_id] if space_id else None
    return instance_to_model(
        siibra.get_region(parcellation_id, region_id),
        space=space
    ).dict()

space

all_spaces()

Get all spaces

Returns:

Type Description
List

List of all serialized spaces.

Source code in api/common/data_handlers/core/space.py
@data_decorator(ROLE)
def all_spaces() -> List:
    """Get all spaces

    Returns:
        List of all serialized spaces."""
    import siibra
    from api.serialization.util import instance_to_model
    return [instance_to_model(space).dict() for space in siibra.spaces]

single_space(space_id)

Get a single space

Parameters:

Name Type Description Default
space_id str

lookup id of the space

required

Returns:

Type Description
Dict

The space specified by the provided id, serialized into Dict

Source code in api/common/data_handlers/core/space.py
@data_decorator(ROLE)
def single_space(space_id: str) -> Dict:
    """Get a single space

    Args:
        space_id: lookup id of the space

    Returns:
        The space specified by the provided id, serialized into Dict"""
    import siibra
    from api.serialization.util import instance_to_model
    return instance_to_model(
        siibra.spaces[space_id]
    ).dict()

features

misc

get_genes(find=None)

Get all genes

Returns:

Type Description

List of the genes.

Source code in api/common/data_handlers/features/misc.py
@data_decorator(ROLE)
def get_genes(find:str=None):
    """Get all genes

    Args:
        string to find in vocabularies

    Returns:
        List of the genes."""

    from siibra.vocabularies import GENE_NAMES
    if find == None:
        return_list = [v for v in GENE_NAMES]
    else:
        return_list = GENE_NAMES.find(find)

    return [GeneModel(
        symbol=v.get("symbol"),
        description=v.get("description")
        ).dict() for v in return_list]

types

all_feature_types()

Get all feature types

Returns:

Type Description
List[Dict[str, str]]

List of all features, including their name, and inheritance from Feature

Source code in api/common/data_handlers/features/types.py
@data_decorator(ROLE)
def all_feature_types() -> List[Dict[str, str]]:
    """Get all feature types

    Returns:
        List of all features, including their name, and inheritance from Feature"""
    from siibra.features import Feature


    def get_hierarchy_type(Cls: Type[Any]) -> str:
        """Get inherited class in the form of `Feature.{LessGenericCls}.{SpecificCls}.{Cls}`

        Args:
            Cls: subclass of Feature

        Returns:
            string representing class hierarchy"""
        return ".".join([
            BaseCls.__name__
            for BaseCls in Cls.__mro__
            if issubclass(BaseCls, Feature)
        ][::-1])

    return [
        {
            'name': get_hierarchy_type(Cls),
            'category': Cls.category,
        } for Cls in Feature.SUBCLASSES
    ]

vocabularies

gene

get_genes(find=None)

Get all genes

Returns:

Type Description

List of the genes.

Source code in api/common/data_handlers/vocabularies/gene.py
@data_decorator(ROLE)
def get_genes(find:str=None):
    """Get all genes

    Args:
        string to find in vocabularies

    Returns:
        List of the genes."""

    from siibra.vocabularies import GENE_NAMES
    if find == None:
        return_list = [v for v in GENE_NAMES]
    else:
        return_list = GENE_NAMES.find(find)

    return [GeneModel(
        symbol=v.get("symbol"),
        description=v.get("description")
        ).dict() for v in return_list]