#!/usr/bin/env python3 """Build the v8 26-class odour taxonomy and coverage census. This script intentionally stops at data/census artifacts. It does not assemble a training dataset or launch training. """ from __future__ import annotations from collections import Counter, defaultdict import json from pathlib import Path import re import sqlite3 from typing import Any ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / "data" ARTIFACTS = ROOT / "artifacts" TAXONOMY_PATH = DATA / "odour_taxonomy_26.json" ANCHOR_REPORT_PATH = ARTIFACTS / "odour_taxonomy_anchor_report_v8.json" CROSSWALK_PATH = ARTIFACTS / "odour_character_crosswalk_v8.json" CROSSWALK_V81_PATH = ARTIFACTS / "odour_character_crosswalk_v8_1.json" CROSSWALK_CHANGELOG_PATH = ARTIFACTS / "odour_character_crosswalk_v8_1_changelog.json" LABELS_PATH = DATA / "odour_character_labels_v8.jsonl" CENSUS_PATH = ARTIFACTS / "odour_taxonomy_coverage_census_v8.json" SUMMARY_PATH = ARTIFACTS / "odour_taxonomy_coverage_census_v8.md" SOURCE_COVERAGE_PATH = ARTIFACTS / "odour_character_source_coverage_v8_1.json" FORMULA_DATASETS = [ DATA / "empirical_dataset_v8.jsonl", ] ANCHOR_RESOLUTION_DATASETS = [ DATA / "tgsc_training_formulas_v8.jsonl", DATA / "wisemoor_training_records_v9.jsonl", DATA / "fraterworks_free_formulas.jsonl", ] SUBSTANTIVITY_PATH = DATA / "substantivity_targets_poucher_enriched_v8_phase1.jsonl" ARCTANDER_PATH = DATA / "literature_flat" / "arctander_monographs.jsonl" SOLVENT_CAS = {"64-17-5", "67-63-0", "25265-71-8", "84-66-2", "110-98-5", "57-55-6"} TAXONOMY_ROWS: list[dict[str, Any]] = [ {"code": "A", "name": "ALI-FAT-IC", "description": "Fatty, waxy, soapy, clean", "reference_materials": ["Aliphatic aldehydes", "alcohols"]}, {"code": "B", "name": "Berg-ICEBERG", "description": "Cooling, borneol, mint, camphor", "reference_materials": ["Menthol", "camphor", "eucalyptol"]}, {"code": "C", "name": "CITRUS", "description": "Sour, sharp, citrus peel", "reference_materials": ["Citral", "orange", "lemon", "lime"]}, {"code": "D", "name": "DAIRY", "description": "Milky, cream, butter, cheese", "reference_materials": ["Diacetyl", "butyrate", "lactone", "valerate"]}, {"code": "E", "name": "EDIBLE", "description": "Vegetable, nut, fish, meat", "reference_materials": ["Thiazoles", "pyrazines", "sulphides"]}, {"code": "F", "name": "FRUIT", "description": "Sour, sweet fruits, strawberry", "reference_materials": ["Allyl caproate", "Verdox"]}, {"code": "G", "name": "GREEN", "description": "Cut-grass, leaves", "reference_materials": ["cis-3-Hexenol", "Triplal"]}, {"code": "H", "name": "HERB (Cool)", "description": "Cool herbaceous notes", "reference_materials": ["Lavender", "sage"]}, {"code": "I", "name": "IRIS", "description": "Orris, violet", "reference_materials": ["Ionones", "methyl ionone"]}, {"code": "J", "name": "JASMIN", "description": "Fruity, oily, narcotic, jasmin", "reference_materials": ["HCA", "benzyl acetate"]}, {"code": "K", "name": "KONIFER", "description": "Pine, pine needle", "reference_materials": ["Terpineol", "bornyl acetate"]}, {"code": "L", "name": "LIGHT Chemical Floral", "description": "Fresh light floral chemical", "reference_materials": ["Linalool", "Vertenex", "DMBC"]}, {"code": "M", "name": "MUGUET", "description": "Lily of the valley, green, fresh", "reference_materials": ["Hydroxy", "Lilial", "Lyral"]}, {"code": "N", "name": "NARCOTIC", "description": "Heavy sweet florals, absolutes", "reference_materials": ["Narcissus", "ylang ylang", "tuberose"]}, {"code": "O", "name": "ORCHID", "description": "Aromatic, deep floral", "reference_materials": ["Salicylates", "benzoates"]}, {"code": "P", "name": "PHENOL", "description": "Phenol, medicinal, honey", "reference_materials": ["p-Cresol", "ethyl phenyl acetate"]}, {"code": "Q", "name": "Queen of the ORIENT", "description": "Resin, balsam", "reference_materials": ["Benzoin", "tolu", "terpenes"]}, {"code": "R", "name": "ROSE", "description": "Rose otto, absolute, geranium", "reference_materials": ["Citronellol", "PEA", "rhodinol"]}, {"code": "S", "name": "SPICE (Hot)", "description": "Hot culinary, spice", "reference_materials": ["Clove", "cinnamon", "thyme"]}, {"code": "T", "name": "TAR SMOKE", "description": "Smoke, tar, burnt", "reference_materials": ["Cade", "birch tar"]}, {"code": "U", "name": "Urine Faecal ANIMAL", "description": "Animal, faecal, leather", "reference_materials": ["Civet", "castoreum", "ambergris"]}, {"code": "V", "name": "VANILLA", "description": "Sweet edible, vanilla", "reference_materials": ["Vanillin", "coumarin", "heliotropin"]}, {"code": "W", "name": "WOOD", "description": "Wood, oily", "reference_materials": ["Cedar", "santal", "vetivert", "patchouli"]}, {"code": "X", "name": "X-rated MUSK", "description": "Sexy, musk, sensual, sweet", "reference_materials": ["Musk ketone", "Galaxolide"]}, {"code": "Y", "name": "EARTHY MOSSY", "description": "Yeast, fungal, moss, marine", "reference_materials": ["Oakmoss", "Calone"]}, {"code": "Z", "name": "ZOLVENTS", "description": "Odourless solvents, solubilisers", "reference_materials": ["DEP", "DPG", "IPP", "ethanol", "PG"]}, ] # Only these non-specific anchors are treated as family/natural-complex anchors. FAMILY_ANCHORS = { "aliphatic aldehydes", "alcohols", "orange", "lemon", "lime", "butyrate", "lactone", "valerate", "thiazoles", "pyrazines", "sulphides", "lavender", "sage", "ionones", "narcissus", "ylang ylang", "tuberose", "salicylates", "benzoates", "benzoin", "tolu", "terpenes", "clove", "cinnamon", "thyme", "cade", "birch tar", "civet", "castoreum", "ambergris", "cedar", "santal", "vetivert", "oakmoss", } # Explicit single-material aliases. These avoid substring matching and document # abbreviations present in the source table. ANCHOR_ALIASES = { "hca": "alpha hexyl cinnamic aldehyde", "pea": "phenylethyl alcohol", "dmbc": "dimethyl benzyl carbinyl acetate", "hydroxy": "hydroxycitronellal", "ipp": "isopropyl palmitate", "pg": "propylene glycol", "diacetyl": "diacetyl", "eucalyptol": "eucalyptol", "terpineol": "terpineol", "methyl ionone": "methylionone", "allyl caproate": "allyl hexanoate", "patchouli": "patchouli alcohol", "ethyl phenyl acetate": "ethyl phenylacetate", } # Exact CAS fallbacks for source-table single anchors absent from the local exact # alias index. They are only used after exact normalized lookup fails. EXACT_SINGLE_ANCHOR_FALLBACKS = { "diacetyl": {"cas": "431-03-8", "name": "Diacetyl", "smiles": "CC(=O)C(C)=O"}, "hca": {"cas": "101-86-0", "name": "alpha-Hexylcinnamaldehyde", "smiles": "CCCCCCC(=CC1=CC=CC=C1)C=O"}, "pea": {"cas": "60-12-8", "name": "Phenylethyl alcohol", "smiles": "C1=CC=C(C=C1)CCO"}, "hydroxy": {"cas": "107-75-5", "name": "Hydroxycitronellal", "smiles": "CC(CO)CCC=C(C)C=O"}, "ipp": {"cas": "142-91-6", "name": "Isopropyl palmitate", "smiles": "CCCCCCCCCCCCCCCC(=O)OC(C)C"}, "pg": {"cas": "57-55-6", "name": "Propylene glycol", "smiles": "CC(O)CO"}, } DESCRIPTOR_CROSSWALK = { "A": ["fatty", "waxy", "soapy", "clean", "aldehydic"], "B": ["cooling", "borneol", "mint", "minty", "menthol", "camphor", "camphoraceous"], "C": ["citrus", "orange", "lemon", "lime", "bergamot", "citrus peel"], "D": ["milky", "cream", "creamy", "butter", "buttery", "cheese", "cheesy", "dairy"], "E": ["vegetable", "nut", "nutty", "fish", "fishy", "meat", "meaty", "sulphide", "sulfide", "pyrazine", "thiazole"], "F": ["fruit", "fruity", "strawberry", "apple", "peach", "pear", "berry", "apricot", "pineapple"], "G": ["green", "cut grass", "grassy", "leaf", "leafy", "foliage"], "H": ["herb", "herbal", "herbaceous", "lavender", "sage"], "I": ["iris", "orris", "violet", "ionone"], "J": ["jasmin", "jasmine"], "K": ["pine", "pine needle", "conifer", "terpineol"], "L": ["fresh floral", "light floral", "linalool"], "M": ["muguet", "lily of the valley"], "N": ["heavy floral", "ylang", "tuberose", "narcissus", "narcotic"], "O": ["orchid", "deep floral"], "P": ["phenol", "phenolic", "medicinal", "honey"], "Q": ["resin", "resinous", "balsam", "balsamic", "benzoin", "tolu"], "R": ["rose", "rosy", "geranium", "citronellol", "rhodinol"], "S": ["spice", "spicy", "clove", "cinnamon", "thyme", "hot spice"], "T": ["smoke", "smoky", "tar", "burnt", "burned", "cade", "birch tar"], "U": ["animal", "animalic", "faecal", "fecal", "leather", "leathery", "civet", "castoreum", "ambergris", "urine"], "V": ["vanilla", "vanillic", "coumarin", "coumarinic", "heliotropin", "tonka"], "W": ["wood", "woody", "cedar", "sandalwood", "santal", "vetiver", "vetivert", "patchouli"], "X": ["musk", "musky", "sensual", "galaxolide"], "Y": ["earthy", "mossy", "moss", "fungal", "marine", "oakmoss", "calone", "seaweed"], } AMBIGUOUS_UNMAPPED_DESCRIPTORS = { "aromatic": ["H", "O", "Q"], "floral": ["J", "L", "M", "N", "O", "R"], "fresh": ["A", "C", "G", "L", "M"], "oily": ["J", "W"], "powdery": ["I", "V", "X"], "sweet": ["F", "N", "V", "X"], } CROSSWALK_CHANGELOG = [ { "descriptor": "narcotic", "old": "J", "new": "N", "reason": "The taxonomy defines N as NARCOTIC; mapping the descriptor to JASMIN made a specific N-class descriptor collapse into class J.", }, { "descriptor": "powdery", "old": "I", "new": None, "reason": "Powdery is multi-class-capable across iris, heliotrope/vanilla, and musk contexts; it is masked unless accompanied by a more specific descriptor.", }, { "descriptor": "aromatic", "old": "O", "new": None, "reason": "Aromatic is multi-class-capable in source prose and is too broad to force into ORCHID without a deep-floral/orchid qualifier.", }, ] def norm(text: str) -> str: return re.sub(r"[^a-z0-9]+", " ", str(text).lower()).strip() def compact_name(text: str) -> str: """Conservative exact-match alias cleanup for trade names and dilutions.""" text = str(text) text = re.sub(r"[®™]", "", text) text = re.sub(r"[\"“”]", "", text) text = re.sub(r"\([^)]*\)", " ", text) text = re.sub(r"\b\d+(?:\.\d+)?\s*%.*$", " ", text, flags=re.IGNORECASE) text = re.sub( r"\b(signature|supra|crist|crystal|pure|extra|absolute|resinoid|oil|eo)\b", " ", text, flags=re.IGNORECASE, ) return norm(text) def alias_keys(text: str) -> set[str]: keys = {norm(text), compact_name(text)} return {key for key in keys if key} def word_pattern(term: str) -> re.Pattern[str]: parts = [re.escape(p) for p in norm(term).split()] return re.compile(r"(? str | None: if not smiles or smiles.startswith("NATURAL:"): return None try: from rdkit import Chem from rdkit.Chem import inchi mol = Chem.MolFromSmiles(smiles.removeprefix("SMILES:")) return inchi.MolToInchiKey(mol) if mol is not None else None except Exception: return None def formula_components(record: dict[str, Any]) -> list[dict[str, Any]]: return record.get("metadata", {}).get("initial_components") or record.get("formula", []) def load_jsonl(path: Path) -> list[dict[str, Any]]: with path.open() as f: return [json.loads(line) for line in f if line.strip()] def load_formula_records() -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] for path in FORMULA_DATASETS: records.extend(load_jsonl(path)) return records def build_alias_index(records: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: aliases: dict[str, dict[str, Any]] = {} db = ROOT / "src" / "pino" / "registry.db" con = sqlite3.connect(db) con.row_factory = sqlite3.Row for row in con.execute("select cas,name,smiles from aroma_chemicals"): rec = dict(row) if rec.get("name"): for key in alias_keys(rec["name"]): aliases.setdefault(key, rec) if rec.get("cas"): aliases.setdefault(norm(rec["cas"]), rec) con.close() resolution_records = list(records) for path in ANCHOR_RESOLUTION_DATASETS: if path.exists(): resolution_records.extend(load_jsonl(path)) for record in resolution_records: for comp in formula_components(record): if comp.get("name") and comp.get("cas"): for key in alias_keys(comp["name"]): aliases.setdefault(key, { "cas": comp.get("cas"), "name": comp.get("name"), "smiles": comp.get("smiles") or "", }) if comp.get("cas"): aliases.setdefault(norm(comp["cas"]), { "cas": comp.get("cas"), "name": comp.get("name") or "", "smiles": comp.get("smiles") or "", }) return aliases def resolve_anchor(term: str, alias_index: dict[str, dict[str, Any]]) -> dict[str, Any]: key = norm(term) if key in FAMILY_ANCHORS: return { "term": term, "status": "class-anchor-family", "provenance": "taxonomy-reference", "reason": "family_or_natural_complex_not_single_molecule", } lookup_key = norm(ANCHOR_ALIASES.get(key, key)) rec = alias_index.get(lookup_key) method = "exact-normalized-local-alias" if rec is None and key in EXACT_SINGLE_ANCHOR_FALLBACKS: rec = EXACT_SINGLE_ANCHOR_FALLBACKS[key] method = "exact-single-anchor-fallback" if rec is None: return { "term": term, "status": "unresolved-single-anchor", "provenance": "taxonomy-reference", "lookup_key": lookup_key, } smiles = rec.get("smiles") or "" ikey = inchi_key(smiles) return { "term": term, "status": "resolved-single-molecule" if ikey else "resolved-cas-only", "provenance": "taxonomy-reference", "resolution_method": method, "canonical": { "cas": rec.get("cas"), "name": rec.get("name"), "smiles": smiles or None, "inchikey": ikey, }, } def build_taxonomy(alias_index: dict[str, dict[str, Any]]) -> tuple[dict[str, Any], dict[str, list[dict[str, Any]]]]: by_class: dict[str, list[dict[str, Any]]] = {} classes = [] for row in TAXONOMY_ROWS: anchors = [resolve_anchor(term, alias_index) for term in row["reference_materials"]] by_class[row["code"]] = anchors classes.append({ **row, "is_carrier": row["code"] == "Z", "reference_material_anchors": anchors, }) taxonomy = { "metadata": { "pimt_version": "v8", "title": "Systematic Odour Classification Groups For Perfumery", "publisher": "Professional Perfumer's World / Perfumer's Bulletin", "source_location": "sample table p.35", "provenance": "structured-from-authoritative-source", "notes": [ "Reference terms are transcribed from the supplied taxonomy work order.", "Family or natural-complex anchors are retained as class anchors but excluded from single-molecule label counts.", ], }, "classes": classes, } return taxonomy, by_class def material_universe(records: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: universe: dict[str, dict[str, Any]] = {} for record in records: for comp in formula_components(record): cas = str(comp.get("cas") or "").strip() if not cas: continue entry = universe.setdefault(cas, { "cas": cas, "names": set(), "smiles": comp.get("smiles") or "", "formula_count": 0, }) if comp.get("name"): entry["names"].add(str(comp["name"])) if comp.get("smiles") and not entry.get("smiles"): entry["smiles"] = comp.get("smiles") entry["formula_count"] += 1 for entry in universe.values(): entry["names"] = sorted(entry["names"]) return universe def load_substantivity_targets() -> dict[str, dict[str, Any]]: targets: dict[str, dict[str, Any]] = {} if SUBSTANTIVITY_PATH.exists(): for row in load_jsonl(SUBSTANTIVITY_PATH): targets[str(row["cas"])] = row return targets def anchor_labels(by_class: dict[str, list[dict[str, Any]]]) -> dict[str, dict[str, Any]]: labels: dict[str, dict[str, Any]] = {} for code, anchors in by_class.items(): for anchor in anchors: if anchor["status"] not in {"resolved-single-molecule", "resolved-cas-only"}: continue cas = anchor["canonical"].get("cas") if not cas: continue labels[cas] = { "cas": cas, "class_code": code, "provenance": "taxonomy-reference", "source_term": anchor["term"], "source": "Systematic Odour Classification Groups For Perfumery sample table p.35", } return labels def registry_records() -> dict[str, dict[str, Any]]: rows: dict[str, dict[str, Any]] = {} db = ROOT / "src" / "pino" / "registry.db" con = sqlite3.connect(db) con.row_factory = sqlite3.Row for row in con.execute("select cas,name,smiles from aroma_chemicals"): rec = dict(row) if rec.get("cas"): rows[str(rec["cas"])] = rec con.close() return rows def enrich_universe( universe: dict[str, dict[str, Any]], subst: dict[str, dict[str, Any]], labels: dict[str, dict[str, Any]], ) -> dict[str, dict[str, Any]]: enriched = { cas: {**entry, "names": set(entry.get("names", []))} for cas, entry in universe.items() } registry = registry_records() for cas in set(subst) | set(labels) | set(registry): if cas not in set(subst) | set(labels) | set(universe): continue entry = enriched.setdefault(cas, {"cas": cas, "names": set(), "smiles": "", "formula_count": 0}) if cas in subst and subst[cas].get("name"): entry["names"].add(str(subst[cas]["name"])) if cas in labels and labels[cas].get("source_term"): entry["names"].add(str(labels[cas]["source_term"])) if cas in registry: if registry[cas].get("name"): entry["names"].add(str(registry[cas]["name"])) if registry[cas].get("smiles") and not entry.get("smiles"): entry["smiles"] = registry[cas].get("smiles") or "" for entry in enriched.values(): entry["names"] = sorted(entry["names"]) return enriched def load_arctander_records() -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] if not ARCTANDER_PATH.exists(): return out for line in ARCTANDER_PATH.read_text().splitlines(): if not line.strip(): continue outer = json.loads(line) rec = json.loads(outer["record"]) if isinstance(outer.get("record"), str) else outer out.append(rec) return out def arctander_aliases(rec: dict[str, Any]) -> set[str]: aliases = set() name = str(rec.get("name") or "") aliases.update(alias_keys(name)) raw_lines = [line.strip() for line in str(rec.get("raw_text") or "").splitlines()] title_lines: list[str] = [] for line in raw_lines[1:8]: if not line: continue alpha = re.sub(r"[^A-Za-z]+", "", line) if not alpha: break uppercase_ratio = sum(1 for ch in alpha if ch.isupper()) / max(1, len(alpha)) if "." in line or uppercase_ratio < 0.65: break title_lines.append(line) if len(title_lines) > 1: aliases.update(alias_keys(" ".join(title_lines))) # Synonyms are noisy OCR, so only accept short delimited aliases. This # remains exact matching after normalization and catches period/newline # synonym fragments such as "Anisic alcohol." without substring matching. synonyms = str(rec.get("synonyms") or "") for part in re.split(r"[,;/.\n]", synonyms): part = part.strip() if 3 <= len(part) <= 60 and not re.search(r"\d", part): aliases.update(alias_keys(part)) return {alias for alias in aliases if alias} def build_arctander_index(records: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: candidates: dict[str, list[dict[str, Any]]] = defaultdict(list) for rec in records: for alias in arctander_aliases(rec): candidates[alias].append(rec) return { alias: matches[0] for alias, matches in candidates.items() if len({m.get("monograph_number") for m in matches}) == 1 } def material_aliases(entry: dict[str, Any]) -> set[str]: aliases = set() for name in entry.get("names", []): aliases.update(alias_keys(name)) return aliases def source_odor_text(rec: dict[str, Any]) -> str: raw = str(rec.get("raw_text") or "") raw = re.sub(r"-\s*\n\s*", "", raw) raw = re.sub(r"\s*\n\s*", " ", raw) raw = re.sub(r"\s+", " ", raw).strip() sentences = [ s.strip() for s in re.split(r"(?<=[.!?])\s+", raw) if s.strip() ] odor_sentences = [ s for s in sentences if re.search(r"\b(?:odou?r|smell|aroma)\b", s, flags=re.IGNORECASE) ] text = " ".join(odor_sentences) desc = str(rec.get("odor_description") or "").strip() if desc and norm(desc) not in {"intermittent", "not available", "none"}: text = f"{desc}. {text}".strip() text = re.sub(r"\b[Tt]aste\b.*?(?=\.|$)", " ", text) text = re.sub(r"\bIt\s*blends\b[^.]*\bodou?r type\b", " ", text, flags=re.IGNORECASE) text = re.sub(r"\bIt\s*may find\b.*", " ", text, flags=re.IGNORECASE) text = re.sub(r"\bMuguet bases\b", "bases", text, flags=re.IGNORECASE) text = re.sub(r"\b([A-Za-z]+)\s+leaf oil\b", r"\1 oil", text, flags=re.IGNORECASE) text = re.sub(r"\s+", " ", text).strip() return text def crosswalk_patterns() -> dict[str, list[tuple[str, re.Pattern[str]]]]: return { code: [(term, word_pattern(term)) for term in terms] for code, terms in DESCRIPTOR_CROSSWALK.items() } def apply_crosswalk(text: str) -> tuple[list[str], dict[str, list[str]], list[str]]: text_norm = norm(text) hits_by_class: dict[str, set[str]] = defaultdict(set) for code, code_patterns in crosswalk_patterns().items(): for term, pat in code_patterns: if pat.search(text_norm): hits_by_class[code].add(term) ambiguous_hits = [ term for term in sorted(AMBIGUOUS_UNMAPPED_DESCRIPTORS) if word_pattern(term).search(text_norm) ] return ( sorted(hits_by_class), {code: sorted(terms) for code, terms in hits_by_class.items()}, ambiguous_hits, ) def arctander_labels( universe: dict[str, dict[str, Any]], existing: dict[str, dict[str, Any]], ) -> tuple[dict[str, dict[str, Any]], Counter[str], dict[str, Any]]: records = load_arctander_records() index = build_arctander_index(records) labels: dict[str, dict[str, Any]] = {} provenance_counts: Counter[str] = Counter() available_unlabelled = [] masked_with_source = [] for cas, entry in sorted(universe.items()): aliases = material_aliases(entry) matched_alias = next((alias for alias in sorted(aliases) if alias in index), None) if not matched_alias: continue rec = index[matched_alias] if cas not in existing: available_unlabelled.append(cas) if cas in existing: continue odor_text = source_odor_text(rec) classes, hits_by_class, ambiguous_hits = apply_crosswalk(odor_text) if not classes: masked_with_source.append(cas) provenance_counts["arctander_source_available_masked_no_clean_crosswalk_hit"] += 1 continue labels[cas] = { "cas": cas, "class_code": classes[0], "class_codes": classes, "provenance": "parsed-from-source", "source": "Arctander, Perfume and Flavor Chemicals, local monograph extract", "source_material_name": rec.get("name"), "matched_material_alias": matched_alias, "matched_source_descriptors_by_class": hits_by_class, "matched_source_descriptors": sorted({term for terms in hits_by_class.values() for term in terms}), "ambiguous_unmapped_descriptors_seen": ambiguous_hits, "source_text_excerpt": odor_text[:500], } provenance_counts["parsed_from_arctander"] += 1 if len(classes) > 1: provenance_counts["parsed_from_arctander_multihot"] += 1 coverage = { "arctander_monographs_total": len(records), "arctander_exact_aliases_indexed": len(index), "unlabelled_materials_with_arctander_text": len(set(available_unlabelled)), "unlabelled_materials_with_arctander_text_masked_no_clean_crosswalk_hit": len(set(masked_with_source)), "matched_unlabelled_cas_sample": sorted(set(available_unlabelled))[:50], } return labels, provenance_counts, coverage def write_labels(labels: dict[str, dict[str, Any]]) -> None: with LABELS_PATH.open("w") as f: for cas in sorted(labels): f.write(json.dumps(labels[cas], sort_keys=True) + "\n") def label_codes(label: dict[str, Any]) -> set[str]: codes = label.get("class_codes") if isinstance(codes, list): return {str(code) for code in codes} code = label.get("class_code") return {str(code)} if code else set() def is_z_label(label: dict[str, Any]) -> bool: codes = label_codes(label) return bool(codes) and codes <= {"Z"} def formula_character_sets(records: list[dict[str, Any]], labels: dict[str, dict[str, Any]]) -> dict[str, list[set[str]]]: by_genre: dict[str, list[set[str]]] = defaultdict(list) for record in records: if record.get("is_control"): continue genre = record.get("genre") or record.get("metadata", {}).get("generation_strategy") or "unknown" chars = set() for comp in formula_components(record): label = labels.get(str(comp.get("cas"))) if not label: continue chars.update(code for code in label_codes(label) if code != "Z") if chars: by_genre[genre].append(chars) return by_genre def discrimination_probe(records: list[dict[str, Any]], labels: dict[str, dict[str, Any]]) -> dict[str, Any]: by_genre = formula_character_sets(records, labels) genre_profiles: dict[str, set[str]] = {} threshold = 0.05 for genre, sets in sorted(by_genre.items()): if len(sets) < 10: continue counts = Counter(code for s in sets for code in s) genre_profiles[genre] = {code for code, count in counts.items() if count / len(sets) >= threshold} pairwise = [] genres = sorted(genre_profiles) for i, left in enumerate(genres): for right in genres[i + 1:]: a, b = genre_profiles[left], genre_profiles[right] pairwise.append({ "left": left, "right": right, "jaccard": len(a & b) / max(1, len(a | b)), "intersection": sorted(a & b), "union": sorted(a | b), }) mean_jaccard = sum(p["jaccard"] for p in pairwise) / len(pairwise) if pairwise else None return { "method": "per-genre formula-level set Jaccard over 26-class character labels; class active if present in >=5% of labelled formulas for genre", "formula_labelled_by_genre": {k: len(v) for k, v in by_genre.items()}, "genre_profiles": {k: sorted(v) for k, v in genre_profiles.items()}, "pairwise": pairwise, "mean_pairwise_jaccard": mean_jaccard, } def main() -> None: ARTIFACTS.mkdir(exist_ok=True) records = load_formula_records() universe = material_universe(records) alias_index = build_alias_index(records) taxonomy, by_class = build_taxonomy(alias_index) TAXONOMY_PATH.write_text(json.dumps(taxonomy, indent=2, sort_keys=True) + "\n") anchor_counts = { code: Counter(anchor["status"] for anchor in anchors) for code, anchors in by_class.items() } ANCHOR_REPORT_PATH.write_text(json.dumps({ "pimt_version": "v8", "taxonomy": str(TAXONOMY_PATH.relative_to(ROOT)), "per_class_anchor_counts": {k: dict(v) for k, v in anchor_counts.items()}, "totals": dict(sum(anchor_counts.values(), Counter())), }, indent=2, sort_keys=True) + "\n") crosswalk_artifact = { "pimt_version": "v8", "crosswalk_version": "v8.1", "mapping_policy": "Exact source descriptor token/phrase match to one 26-class code; ambiguous descriptor tokens remain masked. Multi-hot labels are allowed only when multiple explicit non-ambiguous descriptors appear in the source odour text.", "ambiguity_policy_verified": True, "source_scope": ["taxonomy-reference anchors", "Arctander local monograph extract"], "ambiguous_unmapped_descriptors": AMBIGUOUS_UNMAPPED_DESCRIPTORS, "descriptor_to_class": { term: code for code, terms in DESCRIPTOR_CROSSWALK.items() for term in terms }, } CROSSWALK_PATH.write_text(json.dumps(crosswalk_artifact, indent=2, sort_keys=True) + "\n") CROSSWALK_V81_PATH.write_text(json.dumps(crosswalk_artifact, indent=2, sort_keys=True) + "\n") CROSSWALK_CHANGELOG_PATH.write_text(json.dumps({ "pimt_version": "v8", "crosswalk_version": "v8.1", "changes": CROSSWALK_CHANGELOG, "removed_ambiguous_descriptors": sorted(AMBIGUOUS_UNMAPPED_DESCRIPTORS), "policy_check": { "ambiguous_terms_present_in_descriptor_to_class": sorted( set(AMBIGUOUS_UNMAPPED_DESCRIPTORS) & set(crosswalk_artifact["descriptor_to_class"]) ), "passes": not (set(AMBIGUOUS_UNMAPPED_DESCRIPTORS) & set(crosswalk_artifact["descriptor_to_class"])), }, }, indent=2, sort_keys=True) + "\n") labels = anchor_labels(by_class) subst = load_substantivity_targets() enriched_universe = enrich_universe(universe, subst, labels) arctander, arctander_counts, arctander_coverage = arctander_labels(enriched_universe, labels) labels.update(arctander) write_labels(labels) formula_universe_cas = set(universe) # The work order's substantivity row refers to the full model-ready Poucher # target pool (~170), not only targets that happen to occur in v8 formulas. universe_cas = formula_universe_cas | set(subst) | set(labels) pre_expansion_character_cas = { cas for cas, row in anchor_labels(by_class).items() if cas in universe_cas and any(code != "Z" for code in label_codes(row)) } pre_expansion_z_cas = { cas for cas, row in anchor_labels(by_class).items() if cas in universe_cas and is_z_label(row) } character_cas = {cas for cas, row in labels.items() if cas in universe_cas and any(code != "Z" for code in label_codes(row))} z_cas = {cas for cas, row in labels.items() if cas in universe_cas and is_z_label(row)} subst_cas = set(subst) & universe_cas class_counts: Counter[str] = Counter() for cas in character_cas: class_counts.update(code for code in label_codes(labels[cas]) if code != "Z") all_codes = [row["code"] for row in TAXONOMY_ROWS] per_class_counts = {code: int(class_counts.get(code, 0)) for code in all_codes} sparse_classes = { code: count for code, count in per_class_counts.items() if code != "Z" and count < 5 } classes_ge_5 = {code: count for code, count in per_class_counts.items() if code != "Z" and count >= 5} classes_ge_10 = {code: count for code, count in per_class_counts.items() if code != "Z" and count >= 10} probe = discrimination_probe(records, labels) poucher_coverage = { "source": "Poucher Vol II local odour-classification artifacts", "usable_for_26_class_character_labels": False, "reachable_unlabelled_materials": 0, "reason": "Local Poucher artifacts encode top/middle/base evaporation or profile tiers, not material-level 26-class odour character descriptors.", } source_coverage = { "pimt_version": "v8", "crosswalk_version": "v8.1", "unlabelled_materials_before_arctander_expansion_excluding_z": len(universe_cas - pre_expansion_character_cas - pre_expansion_z_cas), "unlabelled_materials_after_arctander_expansion_excluding_z": len(universe_cas - character_cas - z_cas), "arctander": arctander_coverage, "poucher": poucher_coverage, "supplier_tds": { "local_corpus_found": False, "reachable_unlabelled_materials": 0, "reason": "No local supplier TDS corpus was found in this repository pass.", }, } SOURCE_COVERAGE_PATH.write_text(json.dumps(source_coverage, indent=2, sort_keys=True) + "\n") census = { "pimt_version": "v8", "crosswalk_version": "v8.1", "dataset_scope": [str(p.relative_to(ROOT)) for p in FORMULA_DATASETS], "material_universe": { "definition": "unique CAS from empirical_dataset_v8 formula components plus model-ready Poucher substantivity targets plus resolved taxonomy anchors; solvent/carrier class reported separately", "unique_materials": len(universe_cas), "formula_visible_unique_materials": len(formula_universe_cas), "model_ready_poucher_substantivity_targets": len(subst), }, "coverage_table": { "materials_with_authoritative_character_label_taxonomy_reference_plus_parsed_from_source_excluding_z": len(character_cas), "materials_with_substantivity_measured_poucher_model_ready": len(subst_cas), "materials_with_both_axes_excluding_z": len(character_cas & subst_cas), "materials_with_neither_axis_excluding_z": len(universe_cas - character_cas - subst_cas - z_cas), "class_z_carriers_character_excluded": len(z_cas), "per_class_character_counts": per_class_counts, }, "effective_n": { "character_head_materials_excluding_z": len(character_cas), "substantivity_head_materials": len(subst_cas), "both_axes_materials_excluding_z": len(character_cas & subst_cas), }, "label_provenance_counts_in_universe": dict(Counter(labels[cas]["provenance"] for cas in labels if cas in universe_cas)), "masked_unlabelled_materials_excluding_z": sorted(universe_cas - character_cas - z_cas), "class_z_carriers": sorted(z_cas), "per_class_sparsity_lt5_examples": sparse_classes, "per_class_counts_ge5": classes_ge_5, "per_class_counts_ge10": classes_ge_10, "source_coverage_artifact": str(SOURCE_COVERAGE_PATH.relative_to(ROOT)), "arctander_parse_counts": dict(arctander_counts), "discrimination_precheck": probe, "stop": "STOP_AFTER_CENSUS_NO_DATASET_ASSEMBLY_NO_TRAINING_NO_HF_UPLOAD", "go_no_go_frame": { "green": "character coverage + per-class counts sufficient -> proceed to Phase 2.5 gate then Phase 3 training", "amber": "character axis too sparse, substantivity axis strong -> publish/ship substantivity forward result first and grow character coverage separately", "red": "discrimination probe high -> do not train", }, } CENSUS_PATH.write_text(json.dumps(census, indent=2, sort_keys=True) + "\n") lines = [ "# PIMT v8 odour taxonomy coverage census", "", "| Bucket | Count |", "|:--|--:|", f"| Materials with authoritative character label (excluding Z) | {len(character_cas)} |", f"| Materials with substantivity (measured Poucher, model-ready, in universe) | {len(subst_cas)} |", f"| Materials with both axes (excluding Z) | {len(character_cas & subst_cas)} |", f"| Materials with neither axis (excluding Z) | {len(universe_cas - character_cas - subst_cas - z_cas)} |", f"| Class Z carriers, character-excluded | {len(z_cas)} |", "", "## Per-class character counts", "", "| Code | Count |", "|:--:|--:|", ] lines.extend(f"| {code} | {per_class_counts[code]} |" for code in all_codes) lines.extend([ "", f"Non-Z classes >=5 labels: {len(classes_ge_5)}/25", f"Non-Z classes >=10 labels: {len(classes_ge_10)}/25", f"Both-axis material overlap: {len(character_cas & subst_cas)}", "", "## Discrimination pre-check", "", f"Mean pairwise Jaccard: {probe['mean_pairwise_jaccard'] if probe['mean_pairwise_jaccard'] is not None else 'n/a'}", "", "STOP: no empirical_dataset_v10.jsonl assembled; no training started; no HF upload.", ]) SUMMARY_PATH.write_text("\n".join(lines) + "\n") print(json.dumps({ "taxonomy": str(TAXONOMY_PATH.relative_to(ROOT)), "labels": str(LABELS_PATH.relative_to(ROOT)), "census": str(CENSUS_PATH.relative_to(ROOT)), "character_n": len(character_cas), "substantivity_n": len(subst_cas), "both_n": len(character_cas & subst_cas), "classes_ge5": len(classes_ge_5), "classes_ge10": len(classes_ge_10), "mean_discrimination_jaccard": probe["mean_pairwise_jaccard"], }, indent=2, sort_keys=True)) if __name__ == "__main__": main()