import argparse
import collections
import csv
import hashlib
from pathlib import Path
import re
import unicodedata
ROOT = Path(__file__).resolve().parents[1]
INPUT = ROOT / 'data/taxonomy/taxa.tsv'
OUTPUT = ROOT / 'src/taxonomy_paths.rs'
SPECIES_OUTPUT = ROOT / 'src/taxonomy_species.rs'
RANKS = {'domain', 'kingdom', 'phylum', 'class', 'order', 'family', 'genus', 'species'}
KEYWORDS = set('as break const continue crate else enum extern false fn for if impl in let loop match mod move mut pub ref return self static struct super trait true type unsafe use where while async await dyn abstract become box do final macro override priv typeof unsized virtual yield try gen'.split())
def identifier(name):
ascii_name = unicodedata.normalize('NFKD', name).encode('ascii', 'ignore').decode().lower()
value = re.sub('[^a-z0-9]+', '_', ascii_name).strip('_') or 'taxon'
if value[0].isdigit():
value = 'taxon_' + value
if value in KEYWORDS:
value += '_taxon'
return value
def variant_identifier(name):
ascii_name = unicodedata.normalize('NFKD', name).encode('ascii', 'ignore').decode()
value = ''.join(word.capitalize() for word in re.findall('[A-Za-z0-9]+', ascii_name)) or 'Taxon'
if value[0].isdigit():
value = 'Taxon' + value
return value + 'Taxon' if value == 'Self' else value
def species_names(rows):
names = {r['id']: variant_identifier(r['scientific_name']) for r in rows if r['rank'] == 'species'}
groups = collections.defaultdict(list)
for source_id, name in names.items():
groups[name.lower()].append(source_id)
for duplicates in groups.values():
if len(duplicates) > 1:
for source_id in duplicates:
if not re.fullmatch('[A-Za-z0-9]+', source_id):
raise ValueError('Unexpected source ID in species-name suffix')
names[source_id] += 'Col' + source_id.capitalize()
if len({name.lower() for name in names.values()}) != len(names):
raise ValueError('Canonical species names must be unique ignoring ASCII case')
return names
def layout(rows):
by_id = {r['id']: r for r in rows}
if len(by_id) != len(rows) or list(by_id) != sorted(by_id):
raise ValueError('Snapshot IDs must be unique and sorted, matching Taxon indexes')
parents, names = {}, {}
groups = collections.defaultdict(list)
for r in rows:
if r['rank'] not in RANKS:
continue
parent = r['parent_id']
seen = {r['id']}
nearest = None
while parent:
if parent in seen or parent not in by_id:
raise ValueError('Missing or cyclic source ancestry')
seen.add(parent)
if nearest is None and by_id[parent]['rank'] in RANKS:
nearest = parent
parent = by_id[parent]['parent_id']
parents[r['id']] = nearest
name = r['scientific_name']
if r['rank'] == 'species' and nearest and by_id[nearest]['rank'] == 'genus' and name.split()[0] == by_id[nearest]['scientific_name']:
name = name.split()[-1]
names[r['id']] = variant_identifier(name) if r['rank'] == 'species' else identifier(name)
groups[(nearest, names[r['id']])].append(r['id'])
for siblings in groups.values():
if len(siblings) > 1:
for source_id in siblings:
if not re.fullmatch('[A-Za-z0-9]+', source_id):
raise ValueError('Unexpected source ID in disambiguation suffix')
names[source_id] += ('Col' + source_id.capitalize()) if by_id[source_id]['rank'] == 'species' else ('_col_' + source_id.lower())
unique = {(parents[i], name) for i, name in names.items()}
if len(unique) != len(names):
raise ValueError('Identifier collision remains after source-ID disambiguation')
return parents, names
def render(rows, checksum):
parents, names = layout(rows)
variants = species_names(rows)
by_id = {r['id']: (index, r) for index, r in enumerate(rows)}
children = collections.defaultdict(list)
for source_id, parent in parents.items():
children[parent].append(source_id)
lines = ['// Generated by scripts/generate_taxonomy_paths.py; do not edit.',
f'// Source taxa.tsv SHA-256: {checksum}',
'// Classification: CC BY 4.0; see data/taxonomy/NOTICE.md.', '']
def emit(source_id, level):
index, row = by_id[source_id]
indent = ' ' * level
name = names[source_id]
lines.append(f"{indent}/// `{row['scientific_name']}` — {row['rank']}; Catalogue of Life ID `{source_id}`.")
value = f'crate::taxonomy::Taxon({index:_})'
if row['rank'] == 'species':
if children[source_id]:
raise ValueError('A species cannot contain principal-rank descendants')
lines.append(f'{indent}#[doc(no_inline)]')
lines.append(f'{indent}pub use crate::taxonomy::Species::{variants[source_id]} as {name};')
else:
if parents[source_id] is not None and name == names[parents[source_id]]:
lines.append(f'{indent}// The source uses the same name at two consecutive ranks.')
lines.append(f'{indent}#[allow(clippy::module_inception)]')
lines.append(f'{indent}pub mod {name} {{')
lines.append(f'{indent} /// The source record for this classification group.')
lines.append(f'{indent} pub const TAXON: crate::taxonomy::Taxon = {value};')
for child in sorted(children[source_id], key=lambda i: names[i]):
emit(child, level + 1)
lines.append(f'{indent}}}')
for source_id in sorted(children[None], key=lambda i: names[i]):
emit(source_id, 0)
if 'N' in names:
path, current = [], 'N'
while current is not None:
path.append(names[current])
current = parents[current]
lines.extend(['', '/// Animal kingdom; shorthand for the domain-qualified path.',
'#[doc(no_inline)]', 'pub use ' + '::'.join(reversed(path)) + ';'])
return '\n'.join(lines) + '\n'
def render_species(rows, checksum):
names = species_names(rows)
species = [(i, r) for i, r in enumerate(rows) if r['rank'] == 'species']
if len(species) >= 2 ** 32:
raise ValueError('Species count exceeds the u32 representation')
lines = ['// Generated by scripts/generate_taxonomy_paths.py; do not edit.',
f'// Source taxa.tsv SHA-256: {checksum}',
'// Classification: CC BY 4.0; see data/taxonomy/NOTICE.md.', '',
'/// Every selected animal species in the pinned taxonomy snapshot.',
'///',
'/// Variants use normalized scientific names. Hierarchical aliases refer',
'/// to these same values; see the [`taxonomy`](crate::taxonomy) module.',
'/// Numeric discriminants and declaration order are snapshot-specific.',
'/// Persist a source ID with its source version for long-lived records.',
'#[derive(Clone, Copy, PartialEq, Eq, Hash)]', '#[repr(u32)]', 'pub enum Species {']
for _, row in species:
lines.extend([f" /// `{row['scientific_name']}`; Catalogue of Life ID `{row['id']}`.",
f" {names[row['id']]},"])
lines.extend(['}', '', 'impl Species {',
' /// Every selected species in declaration (source-ID) order.',
' // Const promotion gives this array static storage, not a runtime stack allocation.',
' #[allow(clippy::large_stack_arrays)]',
" pub const ALL: &'static [Self] = &["])
lines.extend(f" Self::{names[r['id']]}," for _, r in species)
lines.extend([' ];', '', ' /// The number of selected species.',
f' pub const COUNT: usize = {len(species):_};', '}', '',
'const SPECIES_NAMES: &[&str; Species::COUNT] = &['])
lines.extend(f" \"{names[r['id']]}\"," for _, r in species)
lines.extend(['];', '', 'const SPECIES_TAXA: &[Taxon; Species::COUNT] = &['])
lines.extend(f' Taxon({index:_}),' for index, _ in species)
lines.extend(['];', ''])
return '\n'.join(lines)
def generated(input_path=None, output=None, species_output=None):
input_path = INPUT if input_path is None else input_path
output = OUTPUT if output is None else output
species_output = SPECIES_OUTPUT if species_output is None else species_output
raw = input_path.read_bytes()
with input_path.open(encoding='utf-8', newline='') as source:
rows = list(csv.DictReader(source, delimiter='\t', quoting=csv.QUOTE_NONE))
checksum = hashlib.sha256(raw).hexdigest()
return {output: render(rows, checksum), species_output: render_species(rows, checksum)}
def write(input_path=None, output=None, species_output=None):
for path, content in generated(input_path, output, species_output).items():
path.write_text(content, encoding='utf-8')
print(f'Generated {path.name} ({path.stat().st_size:,} bytes)')
def check(input_path=None, output=None, species_output=None):
for path, content in generated(input_path, output, species_output).items():
if not path.exists() or path.read_text() != content:
raise ValueError(f'{path.name} is stale; run python3 scripts/generate_taxonomy_paths.py')
print('Verified Species and every taxonomy alias against the bundled snapshot')
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--check', action='store_true')
args = parser.parse_args()
if args.check:
check()
else:
write()
if __name__ == '__main__':
main()