run_generation.py

from __future__ import annotations

"""
End-to-end generative fragrance design script for PINO.

This script demonstrates a complete inverse-design workflow:
    1. Read a high-level design brief from a JSON file.
    2. Run the CMA-ES composer (src/pino/optimizer.py) to evolve a formula.
    3. Verify the winning formula through the physical/IFRA pipeline.
    4. Save the generated recipe and its 8-hour dry-down trajectory to
       data/generated_recipe_output.json.

It is intended as a reference execution profile for downstream agents and
manufacturing workflows. Run it as:

    cd /home/hermes/pino
    source .venv/bin/activate
    PYTHONPATH=src python run_generation.py --brief data/design_brief.json

The design brief JSON has the following shape:

    {
      "name": "Citrus Summer Cologne",
      "brief": "citrus summer masculine",
      "max_iterations": 50,
      "palette_size": 15,
      "allow_synthetic": false
    }
"""

import argparse
import json
import logging
from pathlib import Path
from typing import Any

from pino.ingest_formulas import (
    load_literature_manifest,
    normalize_and_unpack_recipe_from_dict,
)
from pino.optimizer import CompositionTarget, FormulationOptimizer, OptimizerConfig
from pino.verifier import FragrancePipelineVerifier

logger = logging.getLogger("run_generation")


def load_brief(path: str) -> dict[str, Any]:
    """Load a design brief from JSON."""
    with Path(path).open("r", encoding="utf-8") as f:
        return json.load(f)


def save_output(output: dict[str, Any], path: str) -> None:
    """Persist the generated recipe and trajectory to JSON."""
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    with Path(path).open("w", encoding="utf-8") as f:
        json.dump(output, f, indent=2, ensure_ascii=False)
    logger.info("Saved generated recipe to %s", path)


def run(brief_path: str, output_path: str, literature_formulas: str = "data/literature_formulas.json") -> dict[str, Any]:
    """Execute the full generative design profile."""
    brief = load_brief(brief_path)
    target = CompositionTarget.from_text(brief.get("brief", ""))
    target.name = brief.get("name", target.name)

    config = OptimizerConfig(
        max_iterations=brief.get("max_iterations", 50),
        allow_synthetic=brief.get("allow_synthetic", True),
        seed=brief.get("seed", 2026),
        verbose=1,
    )

    logger.info(
        "Starting generation: brief=%r, iterations=%d, palette=%d, synthetic=%s",
        target.name,
        config.max_iterations,
        brief.get("palette_size", 20),
        config.allow_synthetic,
    )

    seed_recipe = None
    seed_id = brief.get("seed_id")
    if seed_id:
        recipes = load_literature_manifest(literature_formulas)
        recipe = next((r for r in recipes if r.get("formula_id") == seed_id), None)
        if recipe is None:
            raise ValueError(f"Seed recipe {seed_id} not found in {literature_formulas}")
        seed_recipe = normalize_and_unpack_recipe_from_dict(recipe)
        logger.info("Loaded literature seed recipe %s with %d components", seed_id, len(seed_recipe["components"]))

    optimizer = FormulationOptimizer(config)
    best = optimizer.optimize(
        target,
        palette_size=brief.get("palette_size", 20),
        seed_recipe=seed_recipe,
    )

    # Re-verify the winning formula to capture a full 8-hour trajectory.
    formula_records = best.to_formula_dict()
    verifier = FragrancePipelineVerifier()
    verification = verifier.run_sim(
        formula_records,
        duration_seconds=8 * 3600.0,
        interval_seconds=600.0,
    )

    output = {
        "design_brief": brief,
        "generated_formula": formula_records,
        "weight_fractions": {
            item["name"]: item["weight_fraction"] for item in formula_records
        },
        "status": best.status,
        "fitness": best.fitness,
        "ifra_passed": best.ifra_report.get("passed", False),
        "ifra_report": best.ifra_report,
        "prediction": best.prediction,
        "trajectory_8h": verification.get("trajectory", []),
        "message": best.message,
    }

    save_output(output, output_path)
    return output


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )

    parser = argparse.ArgumentParser(description="Generate a fragrance recipe from a design brief")
    parser.add_argument("--brief", default="data/design_brief.json", help="Path to design brief JSON")
    parser.add_argument("--output", default="data/generated_recipe_output.json", help="Output JSON path")
    parser.add_argument("--literature-formulas", default="data/literature_formulas.json", help="Path to literature formula manifest")
    args = parser.parse_args()

    result = run(args.brief, args.output, literature_formulas=args.literature_formulas)
    print(json.dumps(result["generated_formula"], indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()