Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 15 additions & 2 deletions app/endpoints/declared_endpoints.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from http import HTTPStatus
from typing import Annotated
from typing import Annotated, Literal

import entitysdk.client
import entitysdk.exception
Expand All @@ -17,6 +17,7 @@
get_electrophysiology_metrics,
)
from obi_one.scientific.morphology_metrics.morphology_metrics import (
MORPHOLOGY_METRICS,
MorphologyMetricsOutput,
get_morphology_metrics,
)
Expand All @@ -30,15 +31,27 @@ def activate_declared_endpoints(router: APIRouter) -> APIRouter:
morphology.",
)
def neuron_morphology_metrics_endpoint(
db_client: Annotated[entitysdk.client.Client, Depends(get_client)],
reconstruction_morphology_id: str,
db_client: Annotated[entitysdk.client.Client, Depends(get_client)],
requested_metrics: Annotated[
list[Literal[*MORPHOLOGY_METRICS]] | None, # type: ignore[misc]
Query(
description="List of requested metrics",
),
] = None,
) -> MorphologyMetricsOutput:
"""Calculates neuron morphology metrics for a given reconstruction morphology.

- **reconstruction_morphology_id**: ID of the reconstruction morphology.
- **requested_metrics**: List of requested metrics (optional).
"""
L.info("get_morphology_metrics")

try:
metrics = get_morphology_metrics(
reconstruction_morphology_id=reconstruction_morphology_id,
db_client=db_client,
requested_metrics=requested_metrics,
)
except entitysdk.exception.EntitySDKError as err:
raise HTTPException(
Expand Down
13 changes: 12 additions & 1 deletion obi_one/core/block_reference.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,18 @@ class Config:
@staticmethod
def json_schema_extra(schema: dict, model: "BlockReference") -> None:
# Dynamically get allowed_block_types from subclass
schema["allowed_block_types"] = [t.__name__ for t in model.allowed_block_types_union()]
allowed_types = model.allowed_block_types_union()
if isinstance(allowed_types, tuple):
schema["allowed_block_types"] = [t.__name__ for t in allowed_types]
elif hasattr(allowed_types, "__name__"):
schema["allowed_block_types"] = [allowed_types.__name__]
else:
# Handle UnionType or other types without __name__
schema["allowed_block_types"] = [
t.__name__
for t in get_args(model.allowed_block_types)
if hasattr(t, "__name__")
]
schema["is_block_reference"] = True

@property
Expand Down
143 changes: 88 additions & 55 deletions obi_one/scientific/morphology_metrics/morphology_metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,31 @@

L = logging.getLogger(__name__)

MORPHOLOGY_METRICS = [
"aspect_ratio",
"circularity",
"length_fraction_above_soma",
"max_radial_distance",
"number_of_neurites",
"soma_radius",
"soma_surface_area",
"total_length",
"total_height",
"total_width",
"total_depth",
"total_area",
"total_volume",
"section_lengths",
"segment_radii",
"number_of_sections",
"local_bifurcation_angles",
"remote_bifurcation_angles",
"section_path_distances",
"section_radial_distances",
"section_branch_orders",
"section_strahler_orders",
]


class MorphologyMetricsForm(Form):
single_coord_class_name: ClassVar[str] = "MorphologyMetrics"
Expand All @@ -32,182 +57,190 @@ class Initialize(Block):

class MorphologyMetricsOutput(BaseModel):
aspect_ratio: Annotated[
float,
float | None,
Field(
title="aspect_ratio",
description="Calculates the min/max ratio of the principal direction extents \
along the plane.",
default=None,
),
]
circularity: Annotated[
float,
float | None,
Field(
title="circularity",
description="Calculates the circularity of the morphology points along the plane.",
default=None,
),
]
length_fraction_above_soma: Annotated[
float,
float | None,
Field(
title="length_fraction_above_soma",
description="Returns the length fraction of the segments that have their midpoints \
higher than the soma.",
default=None,
),
]
max_radial_distance: Annotated[
float,
float | None,
Field(
title="max_radial_distance",
description="The maximum radial distance from the soma in micrometers.",
default=None,
),
]
number_of_neurites: Annotated[
int, Field(title="number_of_neurites", description="Number of neurites in a morphology.")
int | None,
Field(
title="number_of_neurites",
description="Number of neurites in a morphology.",
default=None,
),
]

soma_radius: Annotated[
float, Field(title="soma_radius [μm]", description="The radius of the soma in micrometers.")
float | None,
Field(
title="soma_radius [μm]",
description="The radius of the soma in micrometers.",
default=None,
),
]
soma_surface_area: Annotated[
float,
float | None,
Field(
title="soma_surface_area [μm^2]",
description="The surface area of the soma in square micrometers.",
default=None,
),
]
total_length: Annotated[
float,
float | None,
Field(
title="total_length [μm]",
description="The total length of the morphology neurites in micrometers.",
default=None,
),
]
total_height: Annotated[
float,
float | None,
Field(
title="total_height [μm]",
description="The total height (Y-range) of the morphology in micrometers.",
default=None,
),
]
total_height: Annotated[
float,
float | None,
Field(
title="total_width [μm]",
description="The total width (X-range) of the morphology in micrometers.",
default=None,
),
]
total_depth: Annotated[
float,
float | None,
Field(
title="total_depth [μm]",
description="The total depth (Z-range) of the morphology in micrometers.",
default=None,
),
]
total_area: Annotated[
float,
float | None,
Field(
title="total_area [μm^2]",
description="The total surface area of all sections in square micrometers.",
default=None,
),
]
total_volume: Annotated[
float,
float | None,
Field(
title="total_volume [μm^3]",
description="The total volume of all sections in cubic micrometers.",
default=None,
),
]
section_lengths: Annotated[
list[float],
list[float] | None,
Field(
title="section_lengths [μm]",
description="The distribution of lengths per section in micrometers.",
default=None,
),
]
segment_radii: Annotated[
list[float],
list[float] | None,
Field(
title="segment_radii [μm]",
description="The distribution of radii of the morphology in micrometers.",
default=None,
),
]
number_of_sections: Annotated[
float,
float | None,
Field(
title="number_of_sections",
description="The number of sections in the morphology.",
default=None,
),
]
local_bifurcation_angles: Annotated[
list[float],
list[float] | None,
Field(
title="local_bifurcation_angles [rad]",
description="Angles between two sections computed at the bifurcation (local).",
default=None,
),
]
remote_bifurcation_angles: Annotated[
list[float],
list[float] | None,
Field(
title="remote_bifurcation_angles [rad]",
description="Angles between two sections computed at the end of the sections (remote).",
default=None,
),
]
section_path_distances: Annotated[
list[float],
list[float] | None,
Field(
title="section_path_distances [μm]",
description="Path distances from the soma to section endpoints in micrometers.",
default=None,
),
]
section_radial_distances: Annotated[
list[float],
list[float] | None,
Field(
title="section_radial_distances [μm]",
description="Radial distance from the soma to section endpoints in micrometers.",
default=None,
),
]
section_branch_orders: Annotated[
list[float],
list[float] | None,
Field(
title="section_branch_orders",
description="The distribution of branch orders of sections, computed from soma.",
default=None,
),
]
section_strahler_orders: Annotated[
list[float],
list[float] | None,
Field(
title="section_strahler_orders",
description="The distribution of strahler branch orders of sections, computed from \
terminals.",
default=None,
),
]

@classmethod
def from_morphology(cls, neurom_morphology: neurom.core.Morphology) -> Self:
return cls(
aspect_ratio=neurom.get("aspect_ratio", neurom_morphology),
circularity=neurom.get("circularity", neurom_morphology),
length_fraction_above_soma=neurom.get("length_fraction_above_soma", neurom_morphology),
max_radial_distance=neurom.get("max_radial_distance", neurom_morphology),
number_of_neurites=neurom.get("number_of_neurites", neurom_morphology),
soma_radius=neurom.get("soma_radius", neurom_morphology),
soma_surface_area=neurom.get("soma_surface_area", neurom_morphology),
total_length=neurom.get("total_length", neurom_morphology),
total_height=neurom.get("total_height", neurom_morphology),
total_width=neurom.get("total_width", neurom_morphology),
total_depth=neurom.get("total_depth", neurom_morphology),
total_area=neurom.get("total_area", neurom_morphology),
total_volume=neurom.get("total_volume", neurom_morphology),
section_lengths=neurom.get("section_lengths", neurom_morphology),
segment_radii=neurom.get("segment_radii", neurom_morphology),
number_of_sections=neurom.get("number_of_sections", neurom_morphology),
local_bifurcation_angles=neurom.get("local_bifurcation_angles", neurom_morphology),
remote_bifurcation_angles=neurom.get("remote_bifurcation_angles", neurom_morphology),
section_path_distances=neurom.get("section_path_distances", neurom_morphology),
section_radial_distances=neurom.get("section_radial_distances", neurom_morphology),
section_branch_orders=neurom.get("section_branch_orders", neurom_morphology),
section_strahler_orders=neurom.get("section_strahler_orders", neurom_morphology),
)
values = {metric: neurom.get(metric, neurom_morphology) for metric in MORPHOLOGY_METRICS}
return cls(**values)


class MorphologyMetrics(MorphologyMetricsForm, SingleCoordinateMixin):
Expand All @@ -226,28 +259,28 @@ def run(self, db_client: entitysdk.client.Client = None) -> MorphologyMetricsOut


def get_morphology_metrics(
reconstruction_morphology_id: str, db_client: entitysdk.client.Client
reconstruction_morphology_id: str,
db_client: entitysdk.client.Client,
requested_metrics: list[str] | None = None,
) -> MorphologyMetricsOutput:
morphology = db_client.get_entity(
entity_id=reconstruction_morphology_id, entity_type=ReconstructionMorphology
)

# Iterate through the assets of the morphology to find the one with content
# type "application/asc"
for asset in morphology.assets:
if asset.content_type == "application/swc":
# Download the content into memory
content = db_client.download_content(
entity_id=morphology.id,
entity_type=ReconstructionMorphology,
asset_id=asset.id,
).decode(encoding="utf-8")

# Use StringIO to create a file-like object in memory from the string content
neurom_morphology = load_morphology(io.StringIO(content), reader="swc")

# Calculate the metrics using neurom
morphology_metrics = MorphologyMetricsOutput.from_morphology(neurom_morphology)

return morphology_metrics
if requested_metrics:
values = {
metric: neurom.get(metric, neurom_morphology) for metric in requested_metrics
}
return MorphologyMetricsOutput(**values)
return MorphologyMetricsOutput.from_morphology(neurom_morphology)
return None