Add decision_table_generator/src/decision_table_generator/cli.py

decision_table_generator/src/decision_table_generator/cli.py

@@ -0,0 +1,92 @@
+import argparse
+import json
+import sys
+import logging
+from pathlib import Path
+from typing import Any, Dict, List
+import pandas as pd
+
+# Lokaler Import aus decision_table_generator.core
+from decision_table_generator.core import generate_decision_table
+
+
+class DecisionConfigError(Exception):
+    """Custom exception for invalid decision configuration."""
+
+
+def _load_config(config_path: Path) -> Dict[str, Any]:
+    """Lädt DecisionConfig aus einer JSON-Datei mit strenger Validierung."""
+    if not config_path.exists():
+        raise DecisionConfigError(f"Config file not found: {config_path}")
+
+    try:
+        with config_path.open('r', encoding='utf-8') as f:
+            config_data = json.load(f)
+    except json.JSONDecodeError as e:
+        raise DecisionConfigError(f"Invalid JSON file: {e}")
+
+    required_keys = {"N_values", "warn_threshold", "rerun_options", "unknown_handling"}
+    missing = required_keys - config_data.keys()
+    if missing:
+        raise DecisionConfigError(f"Missing config fields: {', '.join(missing)}")
+
+    if not isinstance(config_data["N_values"], list) or not all(isinstance(n, int) for n in config_data["N_values"]):
+        raise DecisionConfigError("N_values must be a list of integers.")
+
+    if not isinstance(config_data["warn_threshold"], (int, float)):
+        raise DecisionConfigError("warn_threshold must be numeric.")
+
+    if not isinstance(config_data["rerun_options"], list) or not all(isinstance(x, str) for x in config_data["rerun_options"]):
+        raise DecisionConfigError("rerun_options must be a list of strings.")
+
+    if not isinstance(config_data["unknown_handling"], str):
+        raise DecisionConfigError("unknown_handling must be a string.")
+
+    return config_data
+
+
+def _save_to_csv(entries: List[Dict[str, Any]], output_path: Path) -> None:
+    """Speichert die Entscheidungstabelle als CSV."""
+    if not entries:
+        logging.warning("No entries to save. The decision table is empty.")
+        return
+    
+    df = pd.DataFrame(entries)
+    df.to_csv(output_path, index=False)
+
+
+def main(argv: List[str] | None = None) -> None:
+    """CLI entrypoint to generate a decision table from configuration JSON."""
+    parser = argparse.ArgumentParser(description="Generate decision table from config JSON.")
+    parser.add_argument('--config', required=True, help='Pfad zur JSON-Konfigurationsdatei.')
+    parser.add_argument('--output', required=True, help='Pfad zur Ausgabe-CSV-Datei.')
+
+    args = parser.parse_args(argv)
+
+    logging.basicConfig(level=logging.INFO, format='[%(levelname)s] %(message)s')
+
+    config_path = Path(args.config)
+    output_path = Path(args.output)
+
+    try:
+        config_data = _load_config(config_path)
+        logging.info(f"Loaded config from {config_path}.")
+
+        table_entries = generate_decision_table(config_data)
+        assert isinstance(table_entries, list), "generate_decision_table must return a list."
+        _save_to_csv(table_entries, output_path)
+
+        logging.info(f"Decision table successfully saved to {output_path}.")
+    except DecisionConfigError as e:
+        logging.error(f"Configuration error: {e}")
+        sys.exit(1)
+    except AssertionError as e:
+        logging.error(f"Assertion failed: {e}")
+        sys.exit(2)
+    except Exception as e:
+        logging.exception(f"Unexpected error: {e}")
+        sys.exit(99)
+
+
+if __name__ == '__main__':
+    main()