import argparse import sys import tarfile import webdataset as wds from torch.utils.data import DataLoader + def log_and_continue(err): if isinstance(err, tarfile.ReadError) and len(err.args) == 3: print(err.args[2]) return True if isinstance(err, ValueError): return True raise err + def eprint(*args, **kwargs): print(*args, file=sys.stderr, **kwargs) + if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument( "--shardlist", required=True, help="Path to all shards in braceexpand format."

parser.add_argument("--batch-size", type=int, default=256, help="Batch size") parser.add_argument("--log-every", type=int, default=100, help="How often to log") args = parser.parse_args() keys = ("__key__", "jpg", "sci.txt", "com.txt", "sci_com.txt", "taxontag_com.txt") - + dataset = wds.DataPipeline( wds.SimpleShardList(args.shardlist), wds.tarfile_to_samples(handler=log_and_continue), wds.decode("torchrgb"), wds.to_tuple(*keys, handler=log_and_continue), ) - + dataloader = DataLoader( dataset, num_workers=args.workers, batch_size=args.batch_size ) itr = iter(dataloader)

batch_size = len(batch[0]) batches += 1 total_examples += batch_size if batches % args.log_every == 0: - eprint(f"{batches} batches / {total_examples} examples (current batch size: {batch_size})") + eprint( + f"{batches} batches / {total_examples} examples (current batch size: {batch_size})" + ) except StopIteration: break - eprint(f"Success! Processed {batches} batches with a total of {total_examples} examples.") + eprint( + f"Success! Processed {batches} batches with a total of {total_examples} examples." + )

df - DataFrame with data source and common as columns. """ missing_common = df.loc[df.common.isna()] num_no_common = len(missing_common) if num_no_common > 0: - logging.warning(f"{num_no_common} entries have null common. They are from {missing_common.data_source.unique()}.") + logging.warning( + f"{num_no_common} entries have null common. They are from {missing_common.data_source.unique()}." + ) # print the first and last 5 entries with missing common if num_no_common > 0: logging.warning(f"First 5 entries with missing common: {missing_common.head()}") logging.warning(f"Last 5 entries with missing common: {missing_common.tail()}") - - def main(): # check for file if len(sys.argv) == 1:

# If split column included, remove "train_small" entries as they are duplicates from "train". if "split" in list(df.columns): df = df.loc[df.split != "train_small"] - # Add data_source column to pin-point missing info + # Add data_source column to pin-point missing info df.loc[df["inat21_filename"].notna(), "data_source"] = "iNat21" df.loc[df["bioscan_filename"].notna(), "data_source"] = "BIOSCAN" df.loc[df["eol_content_id"].notna(), "data_source"] = "EOL" # Run checks

check_hierarchy(df[TAXA]) check_sci_name(df[TAXA]) check_id(df[TAXA]) check_common(df[["data_source", "common"]]) + if __name__ == "__main__": main()

""" Make license files for the various datasets that make up TreeOfLife-10M. """ + import argparse import csv import logging import pathlib import itertools

if not tol10m_id: continue writer.writerow([tol10m_id, eol_content_id, eol_page_id, license, owner]) - if __name__ == "__main__": parser = init_parser() args = parser.parse_args() db = evobio10m.get_db(args.db)

Reads all the files and makes a master mapping to specific files in EvoBio-10M. All files in the 10M have a UUID as the key. This script: 1. Creates the initial mapping (without reading the file contents) 2. Writes all these details to a sqlite database. """ + import argparse import logging import multiprocessing import os import tarfile

for shard in os.listdir(split_dir) if shard.endswith(".tar") ] if not shardlist: - logger.warn(f"No .tar files found in directory {split_dir}. Skipping this directory.") + logger.warn( + f"No .tar files found in directory {split_dir}. Skipping this directory." + ) return dataset = wds.DataPipeline( wds.SimpleShardList(shardlist), # Without this line, you will get num_workers copies

taxon, common_name, _ = taxon_data elif data_id.bioscan_filename: taxon_data = bioscan_name_lookup[key] logger.debug(f"bioscan_name_lookup[{key}] = {taxon_data}") if len(taxon_data) != 2: - logger.error(f"Unexpected value in bioscan_name_lookup: {taxon_data}") + logger.error( + f"Unexpected value in bioscan_name_lookup: {taxon_data}" + ) taxon, common_name, _ = taxon_data elif data_id.inat21_cls_name: taxon_data = inat21_name_lookup[data_id.inat21_cls_num] - logger.debug(f"inat21_name_lookup[{data_id.inat21_cls_num}] = {taxon_data}") + logger.debug( + f"inat21_name_lookup[{data_id.inat21_cls_num}] = {taxon_data}" + ) if len(taxon_data) != 2: - logger.error(f"Unexpected value in inat21_name_lookup: {taxon_data}") + logger.error( + f"Unexpected value in inat21_name_lookup: {taxon_data}" + ) taxon, common_name, _ = taxon_data else: raise ValueError(data_id) # Use get_common to generate the common name if not available

args = parser.parse_args() db = evobio10m.get_db(args.db) logger.info("Started.") - eol_name_lookup = naming_reproduce.load_name_lookup(disk_reproduce.eol_name_lookup_json, keytype=int) + eol_name_lookup = naming_reproduce.load_name_lookup( + disk_reproduce.eol_name_lookup_json, keytype=int + ) inat21_name_lookup = naming_reproduce.load_name_lookup( disk_reproduce.inat21_name_lookup_json, keytype=int ) - bioscan_name_lookup = naming_reproduce.load_name_lookup(disk_reproduce.bioscan_name_lookup_json) + bioscan_name_lookup = naming_reproduce.load_name_lookup( + disk_reproduce.bioscan_name_lookup_json + ) logger.info("Loaded name lookups.") key_lookup = {} # EOL stmt = "SELECT content_id, page_id, evobio10m_id FROM eol"

""" Parses all metadata files into a single format that's easily read by make_wds.py. """ + import argparse import csv import json import logging import os

import json from datetime import datetime from tempfile import TemporaryDirectory from tqdm import tqdm -''' +""" Usage: $ python get_media_manifest.py --output_dir /absolute/path/to/output_dir $ python get_media_manifest.py --output_dir ../relative/path/to/output_dir $ python get_media_manifest.py # saves the files in the current directory

- The column names of the CSV file - The total number of entries in the combined CSV file The script will create the output directory if it does not exist upon confirmation by the user. If the output directory is not specified, the script will save the combined CSV file and log file in the current directory. -''' +""" DOWNLOAD_URL = "https://eol.org/data/media_manifest.tgz" def stream_download_file(file_path): - start_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with requests.get(DOWNLOAD_URL, stream=True) as response: response.raise_for_status() - total_size_in_bytes = int(response.headers.get('content-length', 0)) - progress_bar = tqdm(total=total_size_in_bytes, unit='iB', unit_scale=True, desc="Downloading") + total_size_in_bytes = int(response.headers.get("content-length", 0)) + progress_bar = tqdm( + total=total_size_in_bytes, unit="iB", unit_scale=True, desc="Downloading" + ) with open(file_path, "wb") as file: for chunk in response.iter_content(chunk_size=8192): progress_bar.update(len(chunk)) file.write(chunk) progress_bar.close() - - md5_checksum = hashlib.md5(open(file_path, 'rb').read()).hexdigest() + + md5_checksum = hashlib.md5(open(file_path, "rb").read()).hexdigest() return start_time, md5_checksum + # Extract the numeric part from the chunked manifest file names def extract_numeric_part(file_name): - match = re.search(r'\d+', file_name) + match = re.search(r"\d+", file_name) if match: return int(match.group()) else: raise ValueError(f"No numeric part found in {file_name}") + def generate_output_filename(output_dir): - today = datetime.today().strftime('%Y-%m-%d') + today = datetime.today().strftime("%Y-%m-%d") filename = f"eol_media_manifest_{today}.csv" # If the file exists in this location, append a number to the filename i = 1 while os.path.exists(os.path.join(output_dir, filename)): filename = f"eol_media_manifest_{today}_{i}.csv" i += 1 return os.path.join(output_dir, filename) + def extract_and_combine_csv(tgz_file_path, output_file_path): - with tarfile.open(tgz_file_path, 'r:gz') as tar: - sorted_members = sorted(tar.getmembers(), key=lambda member: extract_numeric_part(member.name)) + with tarfile.open(tgz_file_path, "r:gz") as tar: + sorted_members = sorted( + tar.getmembers(), key=lambda member: extract_numeric_part(member.name) + ) - with open(output_file_path, 'w', encoding='utf-8') as output_file: + with open(output_file_path, "w", encoding="utf-8") as output_file: headers_written = False total_lines = 0 for member in tqdm(sorted_members, desc="Extracting files"): - if member.name.startswith("media_manifest_") and member.name.endswith(".csv"): + if member.name.startswith("media_manifest_") and member.name.endswith( + ".csv" + ): f = tar.extractfile(member) if f: - lines = f.read().decode('utf-8').splitlines() + lines = f.read().decode("utf-8").splitlines() if not headers_written: headers = lines[0] headers_written = True - output_file.write('\n'.join(lines)) - output_file.write('\n') + output_file.write("\n".join(lines)) + output_file.write("\n") total_lines += len(lines) - 1 return headers, total_lines -def create_json_log(log_filename, download_start_time, tgz_md5, csv_md5, output_file_path, headers, total_lines): + +def create_json_log( + log_filename, + download_start_time, + tgz_md5, + csv_md5, + output_file_path, + headers, + total_lines, +): log_data = { "download_start_time": download_start_time, "temp_tgz_md5": tgz_md5, "manifest_csv_md5": csv_md5, "manifest_csv_filepath": os.path.abspath(output_file_path), - "column_names": headers.split(','), - "total_entries_in_media_manifest": total_lines + "column_names": headers.split(","), + "total_entries_in_media_manifest": total_lines, } - with open(log_filename, 'w') as log_file: + with open(log_filename, "w") as log_file: json.dump(log_data, log_file, indent=4) + def main(output_dir): output_dir = os.path.abspath(output_dir) if not os.path.exists(output_dir): - confirm = input(f"The directory {output_dir} does not exist. Do you want it to be created now? (y/n): ") - if confirm.lower() != 'y': + confirm = input( + f"The directory {output_dir} does not exist. Do you want it to be created now? (y/n): " + ) + if confirm.lower() != "y": print("Directory creation cancelled. Exiting script.") return else: os.makedirs(output_dir) - + output_file_path = generate_output_filename(output_dir) base_output_file_name = os.path.splitext(output_file_path)[0] log_filename = f"{base_output_file_name}_log.json" - with TemporaryDirectory() as temp_dir: temp_file_path = os.path.join(temp_dir, "media_manifest.tgz") download_start_time, tgz_md5 = stream_download_file(temp_file_path) headers, total_lines = extract_and_combine_csv(temp_file_path, output_file_path) - csv_md5 = hashlib.md5(open(output_file_path, 'rb').read()).hexdigest() + csv_md5 = hashlib.md5(open(output_file_path, "rb").read()).hexdigest() - create_json_log(log_filename, download_start_time, tgz_md5, csv_md5, output_file_path, headers, total_lines) + create_json_log( + log_filename, + download_start_time, + tgz_md5, + csv_md5, + output_file_path, + headers, + total_lines, + ) print(f"EOL media manifest: {output_file_path}") print(f"Log file: {log_filename}") + if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Download, combine, and log EOL media manifest CSV files.") - parser.add_argument('--output_dir', type=str, default=".", - help="Directory to store the combined CSV file and log file. Defaults to the current directory.") + parser = argparse.ArgumentParser( + description="Download, combine, and log EOL media manifest CSV files." + ) + parser.add_argument( + "--output_dir", + type=str, + default=".", + help="Directory to store the combined CSV file and log file. Defaults to the current directory.", + ) args = parser.parse_args() main(args.output_dir) -

Converts iNat21 to a pretraining webdataset format. * Only includes 9K training classes so there are 1K unseen classes for evaluation * Uses both scientific and common names in the text descriptions """ + import os import torchvision import webdataset from tqdm.auto import tqdm

import requests import json from tqdm import tqdm import argparse -''' + +""" Example usage: python global_names_resolver.py ../data/resolve-please.jsonl ../data/resolved.jsonl --batch_size 300 Note that performance seems to saturate beyond a batch size of 200-300 names.

} } } The output contains the search term used, the name of the provider for the highest scoring result, and the 7-rank hierarchy. -''' +""" + # Function to resolve a batch of names and return structured data def resolve_names(names): - endpoint = 'http://resolver.globalnames.org/name_resolvers.json' - params = { - 'names': '|'.join(names), - 'with_canonical_ranks': 'true' - } + endpoint = "http://resolver.globalnames.org/name_resolvers.json" + params = {"names": "|".join(names), "with_canonical_ranks": "true"} response = requests.get(endpoint, params=params) if response.status_code != 200: return None data = response.json() structured_data_batch = {} - for item_data in data['data']: - if 'results' not in item_data: - continue + for item_data in data["data"]: + if "results" not in item_data: + continue highest_score = 0 highest_score_data = {} - for result in item_data['results']: - score = result.get('score', 0) - ranks = result.get('classification_path_ranks', '') - desired_ranks = ['kingdom', 'phylum', 'class', 'order', 'family', 'genus', 'species'] + for result in item_data["results"]: + score = result.get("score", 0) + ranks = result.get("classification_path_ranks", "") + desired_ranks = [ + "kingdom", + "phylum", + "class", + "order", + "family", + "genus", + "species", + ] if all(rank in ranks.lower() for rank in desired_ranks): - classification_path = result.get('classification_path', '') - data_source_title = result.get('data_source_title', '') - path_list = classification_path.split('|') - rank_list = ranks.lower().split('|') + classification_path = result.get("classification_path", "") + data_source_title = result.get("data_source_title", "") + path_list = classification_path.split("|") + rank_list = ranks.lower().split("|") if len(path_list) != len(rank_list): continue - rank_dict = {rank: term for rank, term in zip(rank_list, path_list) if rank in desired_ranks} + rank_dict = { + rank: term + for rank, term in zip(rank_list, path_list) + if rank in desired_ranks + } if score > highest_score: highest_score = score highest_score_data = {data_source_title: rank_dict} - name = item_data['supplied_name_string'] + name = item_data["supplied_name_string"] if highest_score > 0: structured_data_batch[name] = highest_score_data return structured_data_batch + def main(input_file_path, output_file_path, batch_size=300): # Get total number of lines in the input file for tqdm - with open(input_file_path, 'r') as infile: + with open(input_file_path, "r") as infile: total_lines = sum(1 for line in infile) # Read input file and loop through each line - with open(input_file_path, 'r') as infile: - with open(output_file_path, 'w') as outfile: + with open(input_file_path, "r") as infile: + with open(output_file_path, "w") as outfile: names_batch = [] for line in tqdm(infile, total=total_lines, unit="line"): entry = json.loads(line) name = f"{entry['genus']} {entry['species']}" names_batch.append(name) if len(names_batch) == batch_size: structured_data_batch = resolve_names(names_batch) if structured_data_batch: for name, data in structured_data_batch.items(): - outfile.write(json.dumps({name: data}) + '\n') + outfile.write(json.dumps({name: data}) + "\n") outfile.flush() names_batch = [] # Reset batch # Last batch if it has fewer names than the specified batch size if names_batch: structured_data_batch = resolve_names(names_batch) if structured_data_batch: for name, data in structured_data_batch.items(): - outfile.write(json.dumps({name: data}) + '\n') + outfile.write(json.dumps({name: data}) + "\n") outfile.flush() print("Data resolution complete.") + if __name__ == "__main__": - parser = argparse.ArgumentParser(description='Resolve species names to structured data') - parser.add_argument('input_file', help='Path to the input JSONL file') - parser.add_argument('output_file', help='Path to the output JSONL file') - parser.add_argument('--batch_size', type=int, default=300, help='Number of names to process in each batch') + parser = argparse.ArgumentParser( + description="Resolve species names to structured data" + ) + parser.add_argument("input_file", help="Path to the input JSONL file") + parser.add_argument("output_file", help="Path to the output JSONL file") + parser.add_argument( + "--batch_size", + type=int, + default=300, + help="Number of names to process in each batch", + ) args = parser.parse_args() - + main(args.input_file, args.output_file, args.batch_size) -

import dataclasses import logging import re - logger = logging.getLogger() # ToL ID generated from uuid.uuid4(), uuid regex from https://stackoverflow.com/a/6640851 -tol_filename_pattern = re.compile("^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}.*jpg$") +tol_filename_pattern = re.compile( + "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}.*jpg$" +) @dataclasses.dataclass(frozen=True) class ImageFilename: """

""" Writes the training and validation data to webdataset format. """ + import argparse import collections import json import logging import multiprocessing

""" Writes the training and validation data to webdataset format. """ + import argparse import collections import json import logging import multiprocessing import os import tarfile from PIL import Image, ImageFile -from imageomics import disk_reproduce, eol_reproduce, evobio10m_reproduce, naming_reproduce, wds +from imageomics import ( + disk_reproduce, + eol_reproduce, + evobio10m_reproduce, + naming_reproduce, + wds, +) ######## # CONFIG ########

evobio10m_id: (content_id, page_id) for evobio10m_id, content_id, page_id in db.execute(select_stmt).fetchall() } db.close() - name_lookup = naming_reproduce.load_name_lookup(disk_reproduce.eol_name_lookup_json, keytype=int) + name_lookup = naming_reproduce.load_name_lookup( + disk_reproduce.eol_name_lookup_json, keytype=int + ) # r|gz indcates reading from a gzipped file, streaming only with tarfile.open(imgset_path, "r|gz") as tar: for i, member in enumerate(tar): eol_img = eol_reproduce.ImageFilename.from_filename(member.name) if eol_img.raw in image_blacklist: continue - + # Match on treeoflife_id filename if eol_img.tol_id not in eol_ids_lookup: print(f"Can't find the tol_id {eol_img.tol_id}") logger.warning( "EvoBio10m ID missing. [tol_id: %s]",

continue # fetching page ID content_id, page_id = eol_ids_lookup[eol_img.tol_id] global_id = eol_img.tol_id - + # checking for global id in split if global_id not in splits[args.split] or global_id in finished_ids: continue - + # checking for page id if page_id not in name_lookup: continue - + # using name lookup for taxon, common, classes taxon, common, classes = name_lookup[page_id] if taxon.scientific in species_blacklist: continue

logger.warning( "Error opening file. Skipping. [tar: %s, err: %s]", imgset_path, err ) continue - txt_dct = make_txt(taxon, common) + txt_dct = make_txt(taxon, common) # writing EOL sink.write( - {"__key__": global_id, "jpg": img, **txt_dct} #, "classes": classes} # REMOVED "classes", unused list of numbers, throws an error as an unknown extension with webdataset + { + "__key__": global_id, + "jpg": img, + **txt_dct, + } # , "classes": classes} # REMOVED "classes", unused list of numbers, throws an error as an unknown extension with webdataset ) ######## # INAT21

(filename, cls_num): evobio10m_id for evobio10m_id, filename, cls_num in db.execute(select_stmt).fetchall() } db.close() - name_lookup = naming_reproduce.load_name_lookup(disk_reproduce.inat21_name_lookup_json, keytype=int) + name_lookup = naming_reproduce.load_name_lookup( + disk_reproduce.inat21_name_lookup_json, keytype=int + ) clsdir_path = os.path.join(disk_reproduce.inat21_root_dir, clsdir) for i, filename in enumerate(os.listdir(clsdir_path)): filepath = os.path.join(clsdir_path, filename)

taxon, common, classes = name_lookup[cls_num] if taxon.scientific in species_blacklist: continue - txt_dct = make_txt(taxon, common) #, classes) + txt_dct = make_txt(taxon, common) # , classes) img = load_img(filepath).resize(resize_size) # writing iNat - sink.write({"__key__": global_id, "jpg": img, **txt_dct} #, "classes": classes} # REMOVED "classes", unused list of numbers, throws an error as an unknown extension with webdataset - ) + sink.write( + { + "__key__": global_id, + "jpg": img, + **txt_dct, + } # , "classes": classes} # REMOVED "classes", unused list of numbers, throws an error as an unknown extension with webdataset + ) ######### # BIOSCAN #########

(part, filename): evobio10m_id for evobio10m_id, part, filename in db.execute(select_stmt).fetchall() } db.close() - name_lookup = naming_reproduce.load_name_lookup(disk_reproduce.bioscan_name_lookup_json) + name_lookup = naming_reproduce.load_name_lookup( + disk_reproduce.bioscan_name_lookup_json + ) partdir = os.path.join(disk_reproduce.bioscan_root_dir, f"part{part}") for i, filename in enumerate(os.listdir(partdir)): if (part, filename) not in evobio10m_id_lookup: logger.warning(

taxon, common, classes = name_lookup[global_id] if taxon.scientific in species_blacklist: continue - txt_dct = make_txt(taxon, common) #, classes) + txt_dct = make_txt(taxon, common) # , classes) filepath = os.path.join(partdir, filename) img = load_img(filepath).resize(resize_size) # writing BIOSCAN - sink.write({"__key__": global_id, "jpg": img, **txt_dct} #, "classes": classes} # REMOVED "classes", unused list of numbers, throws an error as an unknown extension with webdataset - ) + sink.write( + { + "__key__": global_id, + "jpg": img, + **txt_dct, + } # , "classes": classes} # REMOVED "classes", unused list of numbers, throws an error as an unknown extension with webdataset + ) ###### # MAIN ######

assert taxon is not None assert not taxon.empty, f"{common} has no taxon!" # test replacing the two if statements with common = naming_reproduce.get_common(taxon, common) - '''if not common: + """if not common: common = taxon.scientific if not common: - common = taxon.taxonomic''' + common = taxon.taxonomic""" # ex: kingdom Animalia phylum Arthropoda class ... tagged = " ".join(f"{tier} {value}" for tier, value in taxon.tagged) # IF YOU UPDATE THE KEYS HERE, BE SURE TO UPDATE check_status() TO LOOK FOR ALL OF

import dataclasses import sqlite3 import os base_dir = os.path.dirname(os.path.abspath(__file__)) + def get_outdir(tag): outdir = "data/TreeOfLife-10M/dataset/evobio10m" return os.path.abspath(f"{outdir}-{tag}") # return f"/fs/ess/PAS2136/open_clip/data/evobio10m-{tag}"

from . import disk, helpers logger = logging.getLogger() + def load_json(filepath): - with open(filepath, 'r') as f: + with open(filepath, "r") as f: return json.load(f) + def dataset_class_to_taxon(cls): tiers = cls.split("_") if cls[0].isdigit(): index, *tiers = cls.split("_") index = int(index, base=10) return Taxon(*tiers, dataset_id=index) else: return Taxon(*cls.split("_")) + @dataclasses.dataclass class Taxon: kingdom: str = "" phylum: str = ""

"order": self.order.capitalize(), "family": self.family.capitalize(), "genus": self.genus.capitalize(), "species": self.species.lower(), } - + @Property def scientific_name(self): - if not self.subspecies == '': - return self.genus.capitalize() + ' ' + self.species.lower() + ' ' + self.subspecies.lower() - elif not self.species == '': - return self.genus.capitalize() + ' ' + self.species.lower() - elif not self.genus == '': + if not self.subspecies == "": + return ( + self.genus.capitalize() + + " " + + self.species.lower() + + " " + + self.subspecies.lower() + ) + elif not self.species == "": + return self.genus.capitalize() + " " + self.species.lower() + elif not self.genus == "": return self.genus.capitalize() else: return None - + @Property def taxonomic_name(self): - if not self.subspecies == '': + if not self.subspecies == "": if self.kingdom == "": - return " ".join([ - self.genus.capitalize(), - self.species.lower(), - self.subspecies.lower() - ]) + return " ".join( + [ + self.genus.capitalize(), + self.species.lower(), + self.subspecies.lower(), + ] + ) else: - return " ".join([ - self.kingdom.capitalize(), - self.phylum.capitalize(), - self.cls.capitalize(), - self.order.capitalize(), - self.family.capitalize(), - self.genus.capitalize(), - self.species.lower(), - self.subspecies.lower() - ]) + return " ".join( + [ + self.kingdom.capitalize(), + self.phylum.capitalize(), + self.cls.capitalize(), + self.order.capitalize(), + self.family.capitalize(), + self.genus.capitalize(), + self.species.lower(), + self.subspecies.lower(), + ] + ) else: if self.kingdom == "": - return " ".join([ - self.genus.capitalize(), - self.species.lower(), - ]) + return " ".join( + [ + self.genus.capitalize(), + self.species.lower(), + ] + ) else: - return " ".join([ - self.kingdom.capitalize(), - self.phylum.capitalize(), - self.cls.capitalize(), - self.order.capitalize(), - self.family.capitalize(), - self.genus.capitalize(), - self.species.lower(), - ]) + return " ".join( + [ + self.kingdom.capitalize(), + self.phylum.capitalize(), + self.cls.capitalize(), + self.order.capitalize(), + self.family.capitalize(), + self.genus.capitalize(), + self.species.lower(), + ] + ) + @Property def get_common_name(self): return self.common_name @Property def sci_common_name(self): - if self.common_name == '': + if self.common_name == "": return self.scientific_name else: - return self.scientific_name + ' with common name ' + self.get_common_name - + return self.scientific_name + " with common name " + self.get_common_name + @Property def taxon_common_name(self): - if self.common_name == '': + if self.common_name == "": return self.taxonomic_name else: - return self.taxonomic_name + ' with common name ' + self.get_common_name + return self.taxonomic_name + " with common name " + self.get_common_name -def to_classes(data,text_type): - if text_type == 'asis': - return data['class'] + +def to_classes(data, text_type): + if text_type == "asis": + return data["class"] data_view = data.drop(columns=list(set(data.keys()) - set(Taxon.__dict__.keys()))) - if text_type == 'sci': - return data_view.apply(lambda x: Taxon(**x.to_dict()).scientific_name, axis=1).values.tolist() - elif text_type == 'taxon': - return data_view.apply(lambda x: Taxon(**x.to_dict()).taxonomic_name, axis=1).values.tolist() - elif text_type == 'com': - return data_view.apply(lambda x: Taxon(**x.to_dict()).get_common_name, axis=1).values.tolist() - elif text_type == 'sci_com': + if text_type == "sci": + return data_view.apply( + lambda x: Taxon(**x.to_dict()).scientific_name, axis=1 + ).values.tolist() + elif text_type == "taxon": + return data_view.apply( + lambda x: Taxon(**x.to_dict()).taxonomic_name, axis=1 + ).values.tolist() + elif text_type == "com": + return data_view.apply( + lambda x: Taxon(**x.to_dict()).get_common_name, axis=1 + ).values.tolist() + elif text_type == "sci_com": return data_view.apply(lambda x: Taxon(**x.to_dict()).sci_common_name, axis=1) - elif text_type == 'taxon_com': + elif text_type == "taxon_com": return data_view.apply(lambda x: Taxon(**x.to_dict()).taxon_common_name, axis=1) else: raise ValueError("text type is not acceptable.")

import dataclasses import json import logging import re - logger = logging.getLogger() taxon_ranks = ("kingdom", "phylum", "cls", "order", "family", "genus", "species") + def clean_rank(value): assert isinstance(value, str) # Remove HTML value = strip_html(value).lower()

else: return current return current + def strip_html(string): """Just removes <i>...</i> tags and similar. Not complete.""" def clean(s): return re.sub(r"<.*?>(.*?)</.*?>", r"\1", s) while string != clean(string): string = clean(string) return string + def clean_name(name): name = strip_html(name.lower()).strip() patterns = [

start, end = match.span() name = name[:start].strip() return name + def load_name_lookup(path, keytype=str): """ Returns dict[key, (taxonomic, common)] """ with open(path) as fd: return { keytype(key): (Taxon(*taxa), common, classes) for key, (taxa, common, classes) in json.load(fd).items() } + def get_common(taxon, common): if common: return common if taxon.scientific: return taxon.scientific

All functions implemented here only input/output the space components, while while calculating the time component according to the Hyperboloid constraint: `x_time = torch.sqrt(1 / curv + torch.norm(x_space) ** 2)` """ + from __future__ import annotations import math import torch

from data.dataset import ImageTextDataset, ImageTextDatasetMERU import pytorch_lightning as pl from pytorch_lightning.callbacks import ModelCheckpoint from argparse import ArgumentParser + def main(args): if args.model == "rcme": from models.rcme import RCME + train_dataset = ImageTextDataset(cfg.dataset_root, cfg.train_csv, mode="train") val_dataset = ImageTextDataset(cfg.dataset_root, cfg.val_csv, mode="val") model = RCME(cfg.rcme, train_dataset, val_dataset) elif args.model == "radial": from models.radial_embed_model import RadialEmbed + train_dataset = ImageTextDataset(cfg.dataset_root, cfg.train_csv, mode="train") val_dataset = ImageTextDataset(cfg.dataset_root, cfg.val_csv, mode="val") model = RadialEmbed(cfg.radial, train_dataset, val_dataset) elif args.model == "meru": from models.meru import MERU - train_dataset = ImageTextDatasetMERU(cfg.dataset_root, cfg.train_csv, mode="train") + + train_dataset = ImageTextDatasetMERU( + cfg.dataset_root, cfg.train_csv, mode="train" + ) val_dataset = ImageTextDatasetMERU(cfg.dataset_root, cfg.val_csv, mode="val") model = MERU(cfg.meru, train_dataset, val_dataset) elif args.model == "atmg": from models.atmg import ATMG - train_dataset = ImageTextDatasetMERU(cfg.dataset_root, cfg.train_csv, mode="train") + + train_dataset = ImageTextDatasetMERU( + cfg.dataset_root, cfg.train_csv, mode="train" + ) val_dataset = ImageTextDatasetMERU(cfg.dataset_root, cfg.val_csv, mode="val") model = ATMG(cfg.atmg, train_dataset, val_dataset) else: raise NotImplementedError(f"Model {args.model} not implemented") - + checkpoint_callback = ModelCheckpoint( - monitor='val_loss', - dirpath=f'checkpoints/{args.model}', - filename='{args.model}-{epoch:02d}-{val_loss:.2f}', + monitor="val_loss", + dirpath=f"checkpoints/{args.model}", + filename="{args.model}-{epoch:02d}-{val_loss:.2f}", save_top_k=3, - mode='min', + mode="min", save_last=True, ) - trainer = pl.Trainer(accelerator='gpu', devices=cfg[args.model].gpus, max_epochs=cfg[args.model].max_epochs, strategy='ddp', callbacks=[checkpoint_callback], - accumulate_grad_batches=cfg[args.model].accumulate_grad_batches, val_check_interval=0.1, limit_val_batches=100) + trainer = pl.Trainer( + accelerator="gpu", + devices=cfg[args.model].gpus, + max_epochs=cfg[args.model].max_epochs, + strategy="ddp", + callbacks=[checkpoint_callback], + accumulate_grad_batches=cfg[args.model].accumulate_grad_batches, + val_check_interval=0.1, + limit_val_batches=100, + ) trainer.fit(model) + if __name__ == "__main__": parser = ArgumentParser() - parser.add_argument("--model", type=str, default="rcme", choices=["rcme", "radial", "meru", "atmg"]) + parser.add_argument( + "--model", type=str, default="rcme", choices=["rcme", "radial", "meru", "atmg"] + ) args = parser.parse_args() - main(args) + main(args)

""" Writes the training and validation data to folder structure organized by taxonomic hierarchy. """ + import sys + sys.path.append("/projects/bdbl/ssastry/bioclip/src") import argparse import collections import csv import json

import tarfile import threading import tqdm from PIL import Image, ImageFile -from imageomics import disk_reproduce, eol_reproduce, evobio10m_reproduce, naming_reproduce, wds +from imageomics import ( + disk_reproduce, + eol_reproduce, + evobio10m_reproduce, + naming_reproduce, + wds, +) ######## # CONFIG ########

def create_taxonomic_folder_name(taxon): """Create folder name from taxonomic hierarchy with incrementing counter.""" global species_counter, species_folder_map - - if not hasattr(taxon, 'tagged') or not taxon.tagged: + + if not hasattr(taxon, "tagged") or not taxon.tagged: return None - + # Extract taxonomic levels from tagged data # Assuming tagged is a list of (level, value) tuples # We need to get all levels from kingdom to species - taxonomic_levels = ['kingdom', 'phylum', 'class', 'order', 'family', 'genus', 'species'] + taxonomic_levels = [ + "kingdom", + "phylum", + "class", + "order", + "family", + "genus", + "species", + ] folder_parts = [] - + for level, value in taxon.tagged: if level.lower() in taxonomic_levels: folder_parts.append(f"{value}") - + # Check if we have all required levels if len(folder_parts) < 7: # Need all 7 levels return None - + # Create species key for mapping - species_key = '_'.join(folder_parts) - + species_key = "_".join(folder_parts) + # Thread-safe access to species counter and map with csv_lock: # Check if we've seen this species before if species_key in species_folder_map: return species_folder_map[species_key] - + # This is a new species, increment counter and create folder name species_counter += 1 folder_name = f"{species_counter:06d}_{species_key}" species_folder_map[species_key] = folder_name - + return folder_name def get_taxonomic_info(taxon): """Extract taxonomic information for CSV.""" - if not hasattr(taxon, 'tagged') or not taxon.tagged: + if not hasattr(taxon, "tagged") or not taxon.tagged: return None, None, None, None, None, None, None - - taxonomic_levels = ['kingdom', 'phylum', 'class', 'order', 'family', 'genus', 'species'] + + taxonomic_levels = [ + "kingdom", + "phylum", + "class", + "order", + "family", + "genus", + "species", + ] taxon_dict = {} - + for level, value in taxon.tagged: if level.lower() in taxonomic_levels: taxon_dict[level.lower()] = value - + return ( - taxon_dict.get('kingdom', ''), - taxon_dict.get('phylum', ''), - taxon_dict.get('class', ''), - taxon_dict.get('order', ''), - taxon_dict.get('family', ''), - taxon_dict.get('genus', ''), - taxon_dict.get('species', '') + taxon_dict.get("kingdom", ""), + taxon_dict.get("phylum", ""), + taxon_dict.get("class", ""), + taxon_dict.get("order", ""), + taxon_dict.get("family", ""), + taxon_dict.get("genus", ""), + taxon_dict.get("species", ""), ) def save_image_to_folder(img, global_id, folder_name, source_dataset, taxon, common): """Save image to appropriate folder and write to CSV.""" # Create folder path folder_path = os.path.join(outdir, folder_name) os.makedirs(folder_path, exist_ok=True) - + # Save image image_filename = f"{global_id}.jpg" image_path = os.path.join(folder_path, image_filename) img.save(image_path, "JPEG", quality=95) - + # Get taxonomic information - kingdom, phylum, class_name, order, family, genus, species = get_taxonomic_info(taxon) - + kingdom, phylum, class_name, order, family, genus, species = get_taxonomic_info( + taxon + ) + # Write to CSV (thread-safe) csv_path = os.path.join(outdir, "image_metadata.csv") with csv_lock: - with open(csv_path, 'a', newline='', encoding='utf-8') as csvfile: + with open(csv_path, "a", newline="", encoding="utf-8") as csvfile: writer = csv.writer(csvfile) - writer.writerow([ - global_id, - folder_name, - image_filename, - source_dataset, - os.path.join(folder_name, image_filename), - taxon.scientific, - common, - kingdom, - phylum, - class_name, - order, - family, - genus, - species - ]) + writer.writerow( + [ + global_id, + folder_name, + image_filename, + source_dataset, + os.path.join(folder_name, image_filename), + taxon.scientific, + common, + kingdom, + phylum, + class_name, + order, + family, + genus, + species, + ] + ) def has_complete_taxonomy(taxon): """Check if taxon has all required taxonomic levels.""" - if not hasattr(taxon, 'tagged') or not taxon.tagged: + if not hasattr(taxon, "tagged") or not taxon.tagged: return False - - required_levels = {'kingdom', 'phylum', 'class', 'order', 'family', 'genus', 'species'} + + required_levels = { + "kingdom", + "phylum", + "class", + "order", + "family", + "genus", + "species", + } found_levels = set() - + for level, value in taxon.tagged: if level.lower() in required_levels and value: found_levels.add(level.lower()) - + return len(found_levels) == len(required_levels) ###################### # Encyclopedia of Life

evobio10m_id: (content_id, page_id) for evobio10m_id, content_id, page_id in db.execute(select_stmt).fetchall() } db.close() - name_lookup = naming_reproduce.load_name_lookup(disk_reproduce.eol_name_lookup_json, keytype=int) + name_lookup = naming_reproduce.load_name_lookup( + disk_reproduce.eol_name_lookup_json, keytype=int + ) print(f"Loaded {len(name_lookup)} EOL names.") # r|gz indicates reading from a gzipped file, streaming only with tarfile.open(imgset_path, "r|gz") as tar: for i, member in tqdm.tqdm(enumerate(tar)): eol_img = eol_reproduce.ImageFilename.from_filename(member.name) if eol_img.raw in image_blacklist: continue - + # Match on treeoflife_id filename if eol_img.tol_id not in eol_ids_lookup: print(f"Can't find the tol_id {eol_img.tol_id}") logger.warning( "EvoBio10m ID missing. [tol_id: %s]",

continue # fetching page ID content_id, page_id = eol_ids_lookup[eol_img.tol_id] global_id = eol_img.tol_id - + # checking for global id in split if global_id not in splits[args.split] or global_id in finished_ids: # logger.info( # "Skipping global ID already in split or finished. [global_id: %s, split: %s]", # global_id, # args.split, # ) continue - + # checking for page id if page_id not in name_lookup: - logger.warning( - "Page ID missing in name lookup. [page_id: %s]", page_id - ) - continue - + logger.warning("Page ID missing in name lookup. [page_id: %s]", page_id) + continue + # using name lookup for taxon, common, classes taxon, common, classes = name_lookup[page_id] if taxon.scientific in species_blacklist: logger.info(

(filename, cls_num): evobio10m_id for evobio10m_id, filename, cls_num in db.execute(select_stmt).fetchall() } db.close() - name_lookup = naming_reproduce.load_name_lookup(disk_reproduce.inat21_name_lookup_json, keytype=int) + name_lookup = naming_reproduce.load_name_lookup( + disk_reproduce.inat21_name_lookup_json, keytype=int + ) clsdir_path = os.path.join(disk_reproduce.inat21_root_dir, clsdir) for i, filename in enumerate(os.listdir(clsdir_path)): filepath = os.path.join(clsdir_path, filename)

(part, filename): evobio10m_id for evobio10m_id, part, filename in db.execute(select_stmt).fetchall() } db.close() - name_lookup = naming_reproduce.load_name_lookup(disk_reproduce.bioscan_name_lookup_json) + name_lookup = naming_reproduce.load_name_lookup( + disk_reproduce.bioscan_name_lookup_json + ) partdir = os.path.join(disk_reproduce.bioscan_root_dir, f"part{part}") for i, filename in enumerate(os.listdir(partdir)): if (part, filename) not in evobio10m_id_lookup: logger.warning(

) parser.add_argument( "--split", choices=["train", "val", "train_small"], default="val" ) parser.add_argument("--tag", default="dev", help="The suffix for the directory.") - #FIXME: Currently, does not work with multiple processes. + # FIXME: Currently, does not work with multiple processes. parser.add_argument( "--workers", type=int, default=1, help="Number of processes to use." ) args = parser.parse_args()

os.makedirs(outdir, exist_ok=True) print(f"Writing images to {outdir}.") # Initialize CSV file csv_path = os.path.join(outdir, "image_metadata.csv") - + # Create CSV header - with open(csv_path, 'w', newline='', encoding='utf-8') as csvfile: + with open(csv_path, "w", newline="", encoding="utf-8") as csvfile: writer = csv.writer(csvfile) - writer.writerow(['global_id', 'folder_name', 'image_filename', 'source_dataset', 'relative_path', 'scientific_name', 'common_name', 'kingdom', 'phylum', 'class', 'order', 'family', 'genus', 'species']) + writer.writerow( + [ + "global_id", + "folder_name", + "image_filename", + "source_dataset", + "relative_path", + "scientific_name", + "common_name", + "kingdom", + "phylum", + "class", + "order", + "family", + "genus", + "species", + ] + ) # db_path = f"{evobio10m_reproduce.get_outdir(args.tag)}/mapping.sqlite" db_path = os.path.abspath(disk_reproduce.db) print(f"Using database at {db_path}.")