from __future__ import annotations import argparse import json import logging import sys from pathlib import Path from typing import Any from .feedback import Critique, FeedbackLoopManager from .inference_cloud import predict_cloud from .optimizer import OptimizerConfig, FormulationOptimizer, CompositionTarget logger = logging.getLogger("pino.cli") def _load_formula(path: str) -> dict[str, Any]: with Path(path).open("r", encoding="utf-8") as f: return json.load(f) def _predict(args: argparse.Namespace) -> int: try: formula = _load_formula(args.file) except Exception as e: logger.error("Failed to load formula file: %s", e) return 2 try: result = predict_cloud(formula, token=args.token) except Exception as e: logger.error("Prediction failed: %s", e) return 2 if args.json_mode: print(json.dumps(result, separators=(",", ":"), ensure_ascii=False)) else: print("Predicted note pyramid and psychometrics:") print(json.dumps(result, indent=2, ensure_ascii=False)) return 0 def _feedback(args: argparse.Namespace) -> int: try: critique = Critique( formula_id=args.formula_id, correct_gender=args.correct_gender, correct_notes=json.loads(args.correct_notes) if args.correct_notes else {}, notes=args.notes or "", ) manager = FeedbackLoopManager() manager.append(critique) except Exception as e: logger.error("Failed to log feedback: %s", e) return 2 if args.json_mode: print(json.dumps({"status": "logged", "formula_id": args.formula_id}, separators=(",", ":"))) else: print(f"Feedback logged for {args.formula_id}") return 0 def _sync_flywheel(args: argparse.Namespace) -> int: try: manager = FeedbackLoopManager() result = manager.sync_flywheel(job_config_path=args.job_config) except Exception as e: logger.error("Sync flywheel failed: %s", e) return 2 if args.json_mode: print(json.dumps(result, separators=(",", ":"), ensure_ascii=False)) else: print(json.dumps(result, indent=2, ensure_ascii=False)) return 0 if result.get("action") != "none" or args.allow_empty else 0 def _compose(args: argparse.Namespace) -> int: """Run the generative composer (CMA-ES) to evolve a formula against a target.""" try: target = CompositionTarget.from_json(args.target_json) except Exception as e: logger.error("Failed to load target brief: %s", e) return 2 config = OptimizerConfig( max_iterations=args.max_iterations, allow_synthetic=args.allow_synthetic, seed=args.seed, token=args.token, verbose=1 if args.json_mode else 1, ) def progress(iteration: int, fitness: float, best: Any) -> None: # Real-time progress to stderr, never to stdout when --json is set. msg = f"[{iteration}/{args.max_iterations}] best fitness (MSE) = {fitness:.6f}" if args.json_mode: print(msg, file=sys.stderr) else: print(msg, file=sys.stderr) seed_recipe = None if args.seed_id: from .ingest_formulas import load_literature_manifest, normalize_and_unpack_recipe_from_dict try: recipes = load_literature_manifest(args.literature_formulas) recipe = next((r for r in recipes if r.get("formula_id") == args.seed_id), None) if recipe is None: logger.error("Seed recipe %s not found in %s", args.seed_id, args.literature_formulas) return 2 seed_recipe = normalize_and_unpack_recipe_from_dict(recipe) logger.info("Loaded literature seed recipe %s with %d components", args.seed_id, len(seed_recipe["components"])) except Exception as e: logger.error("Failed to load seed recipe: %s", e) return 2 try: optimizer = FormulationOptimizer(config) best = optimizer.optimize( target, palette_size=args.palette_size, progress_callback=progress, seed_recipe=seed_recipe, ) except Exception as e: logger.error("Composition failed: %s", e) return 2 output = { "target": target.name, "formula": best.to_formula_dict(), "status": best.status, "fitness": best.fitness, "ifra_passed": best.ifra_report.get("passed", False), "message": best.message, } if args.json_mode: print(json.dumps(output, separators=(",", ":"), ensure_ascii=False)) else: print(json.dumps(output, indent=2, ensure_ascii=False)) return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="pino-critic") parser.add_argument("--token", default=None, help="Hugging Face access token") subparsers = parser.add_subparsers(dest="command", required=True) predict_parser = subparsers.add_parser("predict", help="Run cloud inference on a formula") predict_parser.add_argument("--file", required=True, help="Path to formula JSON file") predict_parser.add_argument("--json", action="store_true", dest="json_mode", help="Emit raw parseable JSON only") predict_parser.set_defaults(func=_predict) feedback_parser = subparsers.add_parser("feedback", help="Log a correction for a recipe") feedback_parser.add_argument("--formula_id", required=True) feedback_parser.add_argument("--correct_gender", type=float, default=None, help="-1.0 to +1.0") feedback_parser.add_argument("--correct_notes", default="{}", help="JSON dict of index -> value") feedback_parser.add_argument("--notes", default="", help="Free-form correction notes") feedback_parser.add_argument("--json", action="store_true", dest="json_mode", help="Emit raw parseable JSON only") feedback_parser.set_defaults(func=_feedback) sync_parser = subparsers.add_parser("sync-flywheel", help="Merge feedback, push dataset, retrain") sync_parser.add_argument("--job-config", default="hf_job.yaml") sync_parser.add_argument("--allow-empty", action="store_true", help="Return 0 even if no feedback") sync_parser.add_argument("--json", action="store_true", dest="json_mode", help="Emit raw parseable JSON only") sync_parser.set_defaults(func=_sync_flywheel) compose_parser = subparsers.add_parser("compose", help="Evolve a formula against a sensory target") compose_parser.add_argument("--target-json", required=True, help="Path to JSON brief defining desired sensory characteristics") compose_parser.add_argument("--max-iterations", type=int, default=500, help="Maximum generation cycles") compose_parser.add_argument("--palette-size", type=int, default=20, help="Number of palette ingredients") compose_parser.add_argument("--allow-synthetic", action="store_true", default=True, help="Allow non-renewable ingredients") compose_parser.add_argument("--no-synthetic", action="store_false", dest="allow_synthetic", help="Restrict to renewable ingredients") compose_parser.add_argument("--seed", type=int, default=None, help="Random seed") compose_parser.add_argument("--seed-id", default=None, help="Literature recipe ID to seed the first generation") compose_parser.add_argument("--literature-formulas", default="data/literature_formulas.json", help="Path to literature formula manifest") compose_parser.add_argument("--json", action="store_true", dest="json_mode", help="Emit raw parseable JSON only") compose_parser.set_defaults(func=_compose) args = parser.parse_args(argv) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) try: return args.func(args) except Exception as e: logger.exception("Unhandled runtime exception") return 2 if __name__ == "__main__": sys.exit(main())