| |
| """Build an exact-CAS Good Scents odor subset from the Pyrfume mirror.""" |
| from __future__ import annotations |
|
|
| from collections import defaultdict |
| import csv |
| import io |
| import json |
| from pathlib import Path |
| import re |
| from urllib.request import Request, urlopen |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| DATA = ROOT / "data" |
| PROFILES = DATA / "material_profiles_v11.jsonl" |
| OUTPUT = DATA / "goodscents_odor_descriptions_pyrfume_v11.jsonl" |
| BASE = "https://raw.githubusercontent.com/pyrfume/pyrfume-data/main/goodscents" |
| OPL_URL = f"{BASE}/data_rw_opl.csv" |
| ODOR_URL = f"{BASE}/data_rw_odor.csv" |
| MOLECULES_URL = f"{BASE}/molecules.csv" |
| CAS_RE = re.compile(r"^\d{2,7}-\d{2}-\d$") |
|
|
|
|
| def valid_cas(value: str) -> bool: |
| value = value.strip() |
| if not CAS_RE.fullmatch(value): |
| return False |
| digits = value.replace("-", "") |
| return sum(int(digit) * weight for weight, digit in enumerate(reversed(digits[:-1]), 1)) % 10 == int(digits[-1]) |
|
|
|
|
| def download_csv(url: str) -> list[dict[str, str]]: |
| request = Request(url, headers={"User-Agent": "PinoDataIngest/1.0"}) |
| with urlopen(request, timeout=60) as response: |
| text = response.read().decode("utf-8-sig") |
| return list(csv.DictReader(io.StringIO(text))) |
|
|
|
|
| def inchikey(smiles: str) -> str | None: |
| from rdkit import Chem |
| from rdkit.Chem import inchi |
|
|
| if not smiles or smiles.startswith("NATURAL:"): |
| return None |
| molecule = Chem.MolFromSmiles(smiles) |
| return inchi.MolToInchiKey(molecule) if molecule is not None else None |
|
|
|
|
| def build_rows( |
| opl_rows: list[dict[str, str]], |
| odor_rows: list[dict[str, str]], |
| molecule_rows: list[dict[str, str]], |
| profiles: list[dict[str, object]], |
| ) -> tuple[list[dict[str, object]], dict[str, int]]: |
| profile_identities = {str(row.get("cas") or "") for row in profiles} |
| profile_by_inchikey: dict[str, set[str]] = defaultdict(set) |
| for profile in profiles: |
| smiles = str(profile.get("smiles") or "").removeprefix("SMILES:") |
| key = inchikey(smiles) if smiles else None |
| if key: |
| profile_by_inchikey[key].add(str(profile.get("cas") or profile.get("profile_id"))) |
| unique_profile_by_inchikey = { |
| key: next(iter(values)) |
| for key, values in profile_by_inchikey.items() |
| if len(values) == 1 |
| } |
| cid_to_inchikey = { |
| str(row.get("CID") or "").strip(): key |
| for row in molecule_rows |
| if (key := inchikey(str(row.get("IsomericSMILES") or ""))) |
| } |
|
|
| candidates: dict[str, set[tuple[str, str, str]]] = defaultdict(set) |
| for row in opl_rows: |
| cas = str(row.get("CAS Number") or "").strip() |
| tgsc_id = str(row.get("TGSC ID") or "").strip() |
| if not tgsc_id or not valid_cas(cas): |
| continue |
| if cas in profile_identities: |
| candidates[tgsc_id].add((cas, cas, "tgsc-id-to-unique-exact-cas")) |
| key = cid_to_inchikey.get(str(row.get("CID") or "").strip()) |
| identity = unique_profile_by_inchikey.get(key or "") |
| if identity: |
| candidates[tgsc_id].add( |
| (identity, cas, "tgsc-cid-to-exact-structure-inchikey") |
| ) |
| structure_source_cas: dict[str, set[str]] = defaultdict(set) |
| for values in candidates.values(): |
| for identity, source_cas, method in values: |
| if method == "tgsc-cid-to-exact-structure-inchikey": |
| structure_source_cas[identity].add(source_cas) |
| ambiguous_structure_identities = { |
| identity for identity, values in structure_source_cas.items() if len(values) > 1 |
| } |
| if ambiguous_structure_identities: |
| candidates = { |
| tgsc_id: { |
| value |
| for value in values |
| if not ( |
| value[2] == "tgsc-cid-to-exact-structure-inchikey" |
| and value[0] in ambiguous_structure_identities |
| ) |
| } |
| for tgsc_id, values in candidates.items() |
| } |
| tgsc_to_identity: dict[str, tuple[str, str, str]] = {} |
| ambiguous_tgsc_ids = 0 |
| for tgsc_id, values in candidates.items(): |
| exact_cas = {value for value in values if value[2] == "tgsc-id-to-unique-exact-cas"} |
| usable = exact_cas or values |
| if not usable: |
| continue |
| if len(usable) == 1: |
| tgsc_to_identity[tgsc_id] = next(iter(usable)) |
| else: |
| ambiguous_tgsc_ids += 1 |
|
|
| output: list[dict[str, object]] = [] |
| seen: set[tuple[str, str]] = set() |
| for row in odor_rows: |
| tgsc_id = str(row.get("TGSC ID") or "").strip() |
| resolved = tgsc_to_identity.get(tgsc_id) |
| identity, source_cas, join_method = resolved or (None, None, None) |
| description = " ".join(str(row.get("Description") or "").split()) |
| signature = (identity or "", description.casefold()) |
| if not identity or not description or signature in seen: |
| continue |
| seen.add(signature) |
| item: dict[str, object] = { |
| "cas": identity, |
| "source_cas": source_cas, |
| "text": description, |
| "tgsc_id": tgsc_id, |
| "join_method": join_method, |
| "provenance": "goodscents-via-pyrfume-mirror", |
| "source_dataset": ODOR_URL, |
| } |
| for source_key, output_key in ( |
| ("Source", "original_source"), |
| ("Source Year", "source_year"), |
| ("Sample Supplier", "sample_supplier"), |
| ): |
| value = str(row.get(source_key) or "").strip() |
| if value: |
| item[output_key] = value |
| output.append(item) |
| output.sort(key=lambda row: (str(row["cas"]), str(row["text"]).casefold())) |
| return output, { |
| "profiles_with_rows": len({str(row["cas"]) for row in output}), |
| "description_rows": len(output), |
| "ambiguous_tgsc_ids_rejected": ambiguous_tgsc_ids, |
| "exact_structure_profiles": len({ |
| str(row["cas"]) |
| for row in output |
| if row["join_method"] == "tgsc-cid-to-exact-structure-inchikey" |
| }), |
| "ambiguous_structure_profiles_rejected": len(ambiguous_structure_identities), |
| "identity_policy": ( |
| "exact CAS first; otherwise exact full InChIKey with one profile and one source CAS; " |
| "ambiguous structures rejected" |
| ), |
| } |
|
|
|
|
| def main() -> int: |
| profiles = [json.loads(line) for line in PROFILES.read_text().splitlines() if line.strip()] |
| rows, report = build_rows( |
| download_csv(OPL_URL), download_csv(ODOR_URL), download_csv(MOLECULES_URL), profiles |
| ) |
| OUTPUT.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows)) |
| print(json.dumps({"output": str(OUTPUT.relative_to(ROOT)), **report}, indent=2)) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
