Skip to content

Commit

Permalink
maint: update some docstrings and and satisfy Flake8
Browse files Browse the repository at this point in the history
  • Loading branch information
AhmetNSimsek committed Oct 13, 2023
1 parent b0474a7 commit 05bd05c
Show file tree
Hide file tree
Showing 8 changed files with 128 additions and 81 deletions.
9 changes: 5 additions & 4 deletions e2e/core/test_parcellation.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,18 +7,19 @@
("julich brain 3", "v1", lambda region: region.name == "Area hOc1 (V1, 17, CalcS)"),
]


@pytest.mark.parametrize("parc_spec,reg_spec,validator", args)
def test_get_region(parc_spec,reg_spec,validator):
def test_get_region(parc_spec, reg_spec, validator):

subclasses_exception = isinstance(validator, type) and issubclass(validator, Exception)
if callable(validator) and not subclasses_exception:
result = siibra.get_region(parc_spec, reg_spec)
assert callable(validator)
assert validator(result)
return

if subclasses_exception:
with pytest.raises(validator) as e:
siibra.get_region(parc_spec, reg_spec)
raise Exception(f"should be either Exception or callable")

raise e("Should be either Exception or callable")
3 changes: 1 addition & 2 deletions e2e/core/test_region.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,12 +82,11 @@ def test_find(parc_spec, region_spec, result_len, check_regions):
])
def test_homologies_related_regions(parc, reg_spec, has_related, has_homology, has_related_ebrains_reg):


reg = siibra.get_region(parc, reg_spec)
related_assessments = [val for val in reg.get_related_regions()]
homology_assessments = [val for val in related_assessments if val.qualification == Qualification.HOMOLOGOUS]
other_v_assessments = [val for val in related_assessments if val.qualification == Qualification.OTHER_VERSION]

assert has_related == (len(other_v_assessments) > 0)
assert has_homology == (len(homology_assessments) > 0)

Expand Down
2 changes: 1 addition & 1 deletion siibra/commons.py
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ def __dir__(self) -> Iterable[str]:
"""List of all object keys in the registry"""
if isinstance(self[0], type):
return list(self._elements.keys())
else:
else:
return ["dataframe"] + list(self._elements.keys())

def __str__(self) -> str:
Expand Down
2 changes: 1 addition & 1 deletion siibra/configuration/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ def extend_configuration(cls, conn: Union[str, RepositoryConnector]):
if not isinstance(conn, RepositoryConnector):
raise RuntimeError("conn needs to be an instance of RepositoryConnector or a valid str")
if conn in cls.CONFIGURATION_EXTENSIONS:
logger.warn(f"The configuration {str(conn)} is already registered.")
logger.warning(f"The configuration {str(conn)} is already registered.")
else:
logger.info(f"Extending configuration with {str(conn)}")
cls.CONFIGURATION_EXTENSIONS.append(conn)
Expand Down
35 changes: 18 additions & 17 deletions siibra/core/concept.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,23 +62,24 @@ def __init__(
Parameters
----------
identifier : str
Unique identifier of the parcellation
name : str
Human-readable name of the parcellation
species: Species or string
Specification of the species
shortname: str
Shortform of human-readable name (optional)
description: str
Textual description of the parcellation
modality : str or None
Specification of the modality underlying this concept
datasets : list
list of datasets corresponding to this concept
publications: list
List of publications, each a dictionary with "doi" and/or "citation" fields
identifier : str
Unique identifier of the parcellation
name : str
Human-readable name of the parcellation
species: Species or string
Specification of the species
shortname: str
Shortform of human-readable name (optional)
description: str
Textual description of the parcellation
modality : str or None
Specification of the modality underlying this concept
datasets : list
list of datasets corresponding to this concept
publications: list
List of publications, each a dictionary with "doi" and/or "citation" fields
spec: dict, default: None
The preconfigured specification.
"""
self._id = identifier
self.name = name
Expand Down
2 changes: 1 addition & 1 deletion siibra/core/parcellation.py
Original file line number Diff line number Diff line change
Expand Up @@ -306,7 +306,7 @@ def get_region(
if isinstance(regionspec, region.Region) and (regionspec.parcellation == self):
return regionspec

# if there exist an exact match of region spec to region name, return
# if there exist an exact match of region spec to region name, return
if isinstance(regionspec, str):
exact_match = [region for region in self if hasattr(region, "name") and region.name == regionspec]
if len(exact_match) == 1:
Expand Down
122 changes: 81 additions & 41 deletions siibra/core/region.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ def __init__(
publications: list = [],
datasets: list = [],
rgb: str = None,
spec = None,
spec=None,
):
"""
Constructs a new Region object.
Expand All @@ -105,6 +105,8 @@ def __init__(
datasets associated with this region
rgb: str, default: None
Hexcode of preferred color of this region (e.g. "#9FE770")
spec: dict, default: None
The preconfigured specification.
"""
anytree.NodeMixin.__init__(self)
concept.AtlasConcept.__init__(
Expand All @@ -131,8 +133,30 @@ def __init__(
self._supported_spaces = None # computed on 1st call of self.supported_spaces
self._CACHED_REGION_SEARCHES = {}
self._str_aliases = None

def get_related_regions(self) -> Iterable["RegionRelationAssessments"]:
"""
Get assements on relations of this region to others defined on EBRAINS.
Yields
------
RegionRelationAssessments
Example
-------
>>> region = siibra.get_region("monkey", "PG")^M
>>> for assesment in region.get_related_regions():
>>> print(assesment)
'PG' is homologous to 'Area PGa (IPL)'
'PG' is homologous to 'Area PGa (IPL) left'
'PG' is homologous to 'Area PGa (IPL) right'
'PG' is homologous to 'Area PGa (IPL)'
'PG' is homologous to 'Area PGa (IPL) left'
'PG' is homologous to 'Area PGa (IPL) right'
'PG' is homologous to 'Area PGa (IPL)'
'PG' is homologous to 'Area PGa (IPL) right'
'PG' is homologous to 'Area PGa (IPL) left'
"""
yield from RegionRelationAssessments.parse_from_region(self)

@property
Expand Down Expand Up @@ -191,25 +215,30 @@ def __eq__(self, other):
return self.id == other.id
if isinstance(other, str):
if not self._str_aliases:

self._str_aliases = {
self.name,
self.key,
self.id,
}
if self.spec:
ebrain_ids = [value for value in self.spec.get("ebrains", {}).values() if isinstance(value, str)]
ebrain_nested_ids = [_id
for value in self.spec.get("ebrains", {}).values() if isinstance(value, list)
for _id in value]
ebrain_ids = [
value
for value in self.spec.get("ebrains", {}).values()
if isinstance(value, str)
]
ebrain_nested_ids = [
_id
for value in self.spec.get("ebrains", {}).values() if isinstance(value, list)
for _id in value
]
assert all(isinstance(_id, str) for _id in ebrain_nested_ids)
all_ebrain_ids = [
*ebrain_ids,
*ebrain_nested_ids
]

self._str_aliases.update(all_ebrain_ids)

return other in self._str_aliases
return False

Expand Down Expand Up @@ -449,11 +478,11 @@ def fetch_regional_map(

for m in parcellationmap.Map.registry():
if (
m.space.matches(fetch_space) and
m.parcellation == self.parcellation and
m.provides_image and
m.maptype == maptype and
self.name in m.regions
m.space.matches(fetch_space)
and m.parcellation == self.parcellation
and m.provides_image
and m.maptype == maptype
and self.name in m.regions
):
result = m.fetch(region=self, format='image')
if (maptype == MapType.STATISTICAL) and (threshold is not None):
Expand Down Expand Up @@ -727,17 +756,21 @@ def __iter__(self):


_get_reg_relation_asmgt_types: Dict[str, Callable] = {}


def _register_region_reference_type(ebrain_type: str):
def outer(fn: Callable):
_get_reg_relation_asmgt_types[ebrain_type] = fn

@wraps(fn)
def inner(*args, **kwargs):
return fn(*args, **kwargs)
return inner
return outer


class RegionRelationAssessments(RelationAssignment[Region]):

anony_client = BucketApiClient()

@staticmethod
Expand All @@ -748,19 +781,19 @@ def get_uuid(long_id: Union[str, Dict]):
long_id = long_id.get("id")
assert isinstance(long_id, str)
else:
raise RuntimeError(f"uuid arg must be str or object")
raise RuntimeError("uuid arg must be str or object")
uuid_search = re.search(r"(?P<uuid>[a-f0-9-]+)$", long_id)
assert uuid_search, f"uuid not found"
assert uuid_search, "uuid not found"
return uuid_search.group("uuid")

@staticmethod
def parse_id_arg(_id: Union[str, List[str]]) -> List[str]:
if isinstance(_id, list):
assert all(isinstance(_i, str) for _i in _id), f"all instances of pev should be str"
assert all(isinstance(_i, str) for _i in _id), "all instances of pev should be str"
elif isinstance(_id, str):
_id = [_id]
else:
raise RuntimeError(f"parse_pev error: arg must be either list of str or str")
raise RuntimeError("parse_pev error: arg must be either list of str or str")
return _id

@classmethod
Expand All @@ -773,16 +806,16 @@ def get_snapshot_factory(cls, type_str: str):
def get_objects(_id: Union[str, List[str]]):
_id = cls.parse_id_arg(_id)
with ThreadPoolExecutor() as ex:
return list(ex.map(
cls.get_object,
[f"ebrainsquery/v3/{type_str}/{_}.json" for _ in _id]
))
return list(
ex.map(
cls.get_object,
[f"ebrainsquery/v3/{type_str}/{_}.json" for _ in _id]
))
return get_objects


@classmethod
def parse_relationship_assessment(cls, src: "Region", assessment):

all_regions = [
region
for p in _parcellation.Parcellation.registry()
Expand All @@ -806,32 +839,36 @@ def parse_relationship_assessment(cls, src: "Region", assessment):
yield cls(query_structure=src, assigned_structure=found_target, qualification=RegionRelationship.parse_relation_assessment(overlap))

if "https://openminds.ebrains.eu/sands/ParcellationEntity" in target.get("type"):
pev_uuids = [cls.get_uuid(has_version)
for pe in cls.get_snapshot_factory("ParcellationEntity")(target_id)
for has_version in pe.get("hasVersion")]
pev_uuids = [
cls.get_uuid(has_version)
for pe in cls.get_snapshot_factory("ParcellationEntity")(target_id)
for has_version in pe.get("hasVersion")
]
for reg in all_regions:
if reg in pev_uuids:
yield cls(query_structure=src, assigned_structure=reg, qualification=RegionRelationship.parse_relation_assessment(overlap))


@classmethod
@_register_region_reference_type("openminds/CustomAnatomicalEntity")
def translate_cae(cls, src: "Region", _id: Union[str, List[str]]):
caes = cls.get_snapshot_factory("CustomAnatomicalEntity")(_id)
for cae in caes:
for ass in cae.get("relationAssessment", []):
yield from cls.parse_relationship_assessment(src, ass)


@classmethod
@_register_region_reference_type("openminds/ParcellationEntityVersion")
def translate_pevs(cls, src: "Region", _id: Union[str, List[str]]):
pe_uuids = [uuid for uuid in
{cls.get_uuid(pe)
for pev in cls.get_snapshot_factory("ParcellationEntityVersion")(_id)
for pe in pev.get("isVersionOf")}]
pe_uuids = [
uuid for uuid in
{
cls.get_uuid(pe)
for pev in cls.get_snapshot_factory("ParcellationEntityVersion")(_id)
for pe in pev.get("isVersionOf")
}
]
pes = cls.get_snapshot_factory("ParcellationEntity")(pe_uuids)

all_regions = [
region
for p in _parcellation.Parcellation.registry()
Expand All @@ -852,17 +889,20 @@ def translate_pevs(cls, src: "Region", _id: Union[str, List[str]]):
found_targets = [
region
for region in all_regions
if region == uuid
if region == uuid
]
if len(found_targets) == 0:
logger.warn(f"other version with uuid {uuid} not found")
logger.warning(f"other version with uuid {uuid} not found")
continue

if len(found_targets) > 1:
logger.warn(f"Found multiple ({len(found_targets)}), returning the first one")

yield cls(query_structure=src, assigned_structure=found_targets[0], qualification=RegionRelationship.OTHER_VERSION)
logger.warning(f"Found multiple ({len(found_targets)}), returning the first one")

yield cls(
query_structure=src,
assigned_structure=found_targets[0],
qualification=RegionRelationship.OTHER_VERSION
)

# homologuous
relations = pe.get("inRelationTo", [])
Expand Down
Loading

0 comments on commit 05bd05c

Please sign in to comment.