File size: 6,888 Bytes
83db774 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | #!/usr/bin/env python3
"""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: # noqa: S310 - fixed HTTPS source
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())
|