Add decision_rule_generation/src/decision_rule_generation/cli.py

decision_rule_generation/src/decision_rule_generation/cli.py

@@ -0,0 +1,70 @@
+import argparse
+import json
+import sys
+from pathlib import Path
+import logging
+import pandas as pd
+
+from decision_rule_generation import core
+
+
+def _validate_input_path(input_path: Path) -> Path:
+    if not input_path.exists():
+        raise FileNotFoundError(f"Input file not found: {input_path}")
+    if not input_path.is_file():
+        raise ValueError(f"Input path is not a file: {input_path}")
+    return input_path
+
+
+def _validate_output_path(output_path: Path) -> Path:
+    output_dir = output_path.parent
+    output_dir.mkdir(parents=True, exist_ok=True)
+    return output_path
+
+
+def main() -> None:
+    """CLI entry point for generating decision rules from grid_results.csv."""
+    parser = argparse.ArgumentParser(description="Generate decision rules from grid search results.")
+    parser.add_argument("--input", required=True, help="Pfad zur grid_results.csv")
+    parser.add_argument("--output", required=True, help="Pfad zur Ausgabe-Datei mit Entscheidungsregeln")
+    parser.add_argument(
+        "--policy-constants",
+        required=False,
+        default="config/policy_constants.json",
+        help="Pfad zur policy_constants.json Datei, falls Aktualisierung gewünscht",
+    )
+
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    try:
+        input_path = _validate_input_path(Path(args.input))
+        output_path = _validate_output_path(Path(args.output))
+        policy_constants_path = Path(args.policy_constants)
+
+        logging.info(f"Reading grid results from {input_path}")
+        grid_results = pd.read_csv(input_path)
+
+        logging.info("Generating decision rules ...")
+        decision_rules = core.generate_decision_rules(grid_results)
+
+        logging.info(f"Writing decision rules to {output_path}")
+        with open(output_path, "w", encoding="utf-8") as f:
+            json.dump(decision_rules, f, indent=2)
+
+        if policy_constants_path.exists():
+            logging.info(f"Updating policy constants at {policy_constants_path}")
+            core.update_policy_constants(decision_rules, str(policy_constants_path))
+        else:
+            logging.warning(f"Policy constants file not found at {policy_constants_path}, skipping update.")
+
+        logging.info("Decision rule generation completed successfully.")
+
+    except Exception as e:
+        logging.error(f"Error during execution: {e}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()