Skip to content
Open
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
211 changes: 158 additions & 53 deletions dandi/metadata/util.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

from collections.abc import Callable, Iterable
from dataclasses import dataclass
from datetime import datetime, timedelta
from functools import lru_cache
import re
Expand Down Expand Up @@ -332,84 +333,192 @@ def extract_cellLine(metadata: dict) -> str | None:

NCBITAXON_URI_TEMPLATE = "http://purl.obolibrary.org/obo/NCBITaxon_{}"

# common_names, prefix, uri, name ({current name} - {GenBank common name})
#: Separator between the current scientific name and the GenBank common name
#: within `SpeciesRecord.name`
SPECIES_NAME_SEPARATOR = " - "


@dataclass(frozen=True)
class SpeciesRecord:
"""A single species entry of the `species_map` lookup table.

Attributes
----------
common_names : tuple of str
Additional lower-cased names the species may be referred to by. These
are only consulted if no record matched on `name` or `prefix`.
prefix : str or None
If not `None`, a lower-cased prefix such that any value starting with
it identifies this species, e.g. ``"mus"`` for *Mus musculus*.
taxon_id : str
The numeric NCBITaxon identifier, e.g. ``"10090"``. The `uri` is
derived from it so a record cannot carry a PURL that `extract_species`
would fail to match.
name : str
The canonical DANDI name, formatted as
``"{current scientific name} - {GenBank common name}"``.
"""

common_names: tuple[str, ...]
prefix: str | None
taxon_id: str
name: str

def __post_init__(self) -> None:
# `common_names` is a tuple of one element for most entries, so a dropped
# trailing comma leaves a plain `str` here. That is hashable, iterates as
# lower-cased characters and therefore passes every other check below,
# while making each of its letters match this species. Reject the type
# rather than coercing it, since `tuple("mouse")` would silently produce
# exactly that character tuple.
if not isinstance(self.common_names, tuple):
raise TypeError(
f"Common names {self.common_names!r} of {self.uri} must be a "
f"tuple, got {type(self.common_names).__name__}"
)
for common_name in self.common_names:
if not common_name:
raise ValueError(f"Common name of {self.uri} must not be empty")
if common_name != common_name.lower():
raise ValueError(
f"Common name {common_name!r} of {self.uri} must be lower-cased"
)
# An empty prefix starts every value, so such a record would match every
# lookup and turn every other one into a "multiple species matched" error.
if self.prefix is not None:
if not self.prefix:
raise ValueError(f"Prefix of {self.uri} must not be empty")
if self.prefix != self.prefix.lower():
raise ValueError(
f"Prefix {self.prefix!r} of {self.uri} must be lower-cased"
)
# `extract_species` recovers the id with `NCBITaxon_([0-9]+)`, so a
# non-numeric id would build a URI that can never be matched back.
if not self.taxon_id.isdigit():
raise ValueError(
f"Taxon id {self.taxon_id!r} of {self.name!r} must be numeric"
)
if SPECIES_NAME_SEPARATOR not in self.name:
raise ValueError(
f"Name {self.name!r} of {self.uri} must be formatted as "
f"'{{scientific name}}{SPECIES_NAME_SEPARATOR}{{common name}}'"
)

@property
def uri(self) -> str:
"""The NCBITaxon PURL for the species, see `NCBITAXON_URI_TEMPLATE`"""
return NCBITAXON_URI_TEMPLATE.format(self.taxon_id)

@property
def scientific_name(self) -> str:
"""The current scientific name, i.e. the part of `name` before the separator"""
return self.name.split(SPECIES_NAME_SEPARATOR, 1)[0]

@property
def genbank_common_name(self) -> str:
"""The GenBank common name, i.e. the part of `name` after the separator"""
return self.name.split(SPECIES_NAME_SEPARATOR, 1)[1]

def matches_name(self, value: str) -> bool:
"""Whether `value` identifies this species by its full `name`, either
half of that name, or its `prefix`

`value` is normalized here, so callers need not pre-normalize it.
"""
value = value.strip().lower()
return (
value == self.name.lower()
or (self.prefix is not None and value.startswith(self.prefix))
or value == self.scientific_name.lower()
or value == self.genbank_common_name.lower()
)

def matches_common_name(self, value: str) -> bool:
"""Whether `value` is one of this species' `common_names`

`value` is normalized here, so callers need not pre-normalize it.
"""
value = value.strip().lower()
return any(value == common_name for common_name in self.common_names)


species_map = [
(
["mouse"],
SpeciesRecord(
("mouse",),
"mus",
NCBITAXON_URI_TEMPLATE.format("10090"),
"10090",
"Mus musculus - House mouse",
),
(
["human"],
SpeciesRecord(
("human",),
"homo",
NCBITAXON_URI_TEMPLATE.format("9606"),
"9606",
"Homo sapiens - Human",
),
(
["brown rat", "rat", "norvegicus"],
SpeciesRecord(
("brown rat", "rat", "norvegicus"),
None,
NCBITAXON_URI_TEMPLATE.format("10116"),
"10116",
"Rattus norvegicus - Norway rat",
),
(
["rattus rattus"],
SpeciesRecord(
("rattus rattus",),
None,
NCBITAXON_URI_TEMPLATE.format("10117"),
"10117",
"Rattus rattus - Black rat",
),
(
["mulatta", "rhesus"],
SpeciesRecord(
("mulatta", "rhesus"),
None,
NCBITAXON_URI_TEMPLATE.format("9544"),
"9544",
"Macaca mulatta - Rhesus monkey",
),
(
["jacchus"],
SpeciesRecord(
("jacchus",),
None,
NCBITAXON_URI_TEMPLATE.format("9483"),
"9483",
"Callithrix jacchus - Common marmoset",
),
(
["melanogaster", "fruit fly"],
SpeciesRecord(
("melanogaster", "fruit fly"),
None,
NCBITAXON_URI_TEMPLATE.format("7227"),
"7227",
"Drosophila melanogaster - Fruit fly",
),
(
["danio", "zebrafish", "zebra fish"],
SpeciesRecord(
("danio", "zebrafish", "zebra fish"),
None,
NCBITAXON_URI_TEMPLATE.format("7955"),
"7955",
"Danio rerio - Zebra fish",
),
(
["c. elegans", "caenorhabditis elegans"],
SpeciesRecord(
("c. elegans", "caenorhabditis elegans"),
"caenorhabditis",
NCBITAXON_URI_TEMPLATE.format("6239"),
"6239",
"Caenorhabditis elegans - Roundworm",
),
(
["pig-tailed macaque", "pigtail monkey", "pigtail macaque"],
SpeciesRecord(
("pig-tailed macaque", "pigtail monkey", "pigtail macaque"),
None,
NCBITAXON_URI_TEMPLATE.format("9545"),
"9545",
"Macaca nemestrina - Pig-tailed macaque",
),
(
["bonnet macaque", "bonnet monkey", "radiata"],
SpeciesRecord(
("bonnet macaque", "bonnet monkey", "radiata"),
None,
NCBITAXON_URI_TEMPLATE.format("9548"),
"9548",
"Macaca radiata - Bonnet macaque",
),
(
["mongolian gerbil", "mongolian jird"],
SpeciesRecord(
("mongolian gerbil", "mongolian jird"),
None,
NCBITAXON_URI_TEMPLATE.format("10047"),
"10047",
"Meriones unguiculatus - Mongolian gerbil",
),
(
["common paper wasp"],
SpeciesRecord(
("common paper wasp",),
None,
NCBITAXON_URI_TEMPLATE.format("30207"),
"30207",
"Polistes fuscatus - Common paper wasp",
),
]
Expand Down Expand Up @@ -467,9 +576,9 @@ def extract_species(metadata: dict) -> models.SpeciesType | None:
flags=re.I,
):
normed_value = NCBITAXON_URI_TEMPLATE.format(m[1])
for _common_names, _prefix, uri, name in species_map:
if uri == normed_value:
value_matches.append((uri, name))
for record in species_map:
if record.uri == normed_value:
value_matches.append((record.uri, record.name))
break
else:
value_id = value_orig
Expand All @@ -489,18 +598,14 @@ def extract_species(metadata: dict) -> models.SpeciesType | None:
value_matches.append((value_id, value))
else:
lower_value = value_orig.lower().strip()
for common_names, prefix, uri, name in species_map:
if (
lower_value == name.lower()
or (prefix is not None and lower_value.startswith(prefix))
or any(lower_value == v.lower() for v in name.partition(" - "))
):
value_matches.append((uri, name))
for record in species_map:
if record.matches_name(lower_value):
value_matches.append((record.uri, record.name))
# only if no matches -- try to match by common names within common names
if not value_matches:
for common_names, prefix, uri, name in species_map:
if any(key == lower_value for key in common_names):
value_matches.append((uri, name))
for record in species_map:
if record.matches_common_name(lower_value):
value_matches.append((record.uri, record.name))

value_matches = list(set(value_matches)) # unique values
if not value_matches:
Expand Down
Loading