|
|
import yaml |
|
|
import logging |
|
|
from typing import Dict, Any, List |
|
|
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
class RuleEngine: |
|
|
def __init__(self): |
|
|
pass |
|
|
|
|
|
def load_rules(self, rules_yaml: str) -> Dict[str, Any]: |
|
|
"""Loads rules from a YAML string.""" |
|
|
try: |
|
|
return yaml.safe_load(rules_yaml) |
|
|
except yaml.YAMLError as e: |
|
|
logger.error(f"Error parsing rules YAML: {e}") |
|
|
raise ValueError(f"Invalid rules format: {e}") |
|
|
|
|
|
def evaluate(self, metrics: Dict[str, Any], rules: Dict[str, Any]) -> Dict[str, Any]: |
|
|
""" |
|
|
Evaluates metrics against a set of rules. |
|
|
|
|
|
Args: |
|
|
metrics: Dictionary of extracted metrics (e.g., {'building_height': 15.0}). |
|
|
rules: Dictionary of rules (e.g., {'max_height': 12.0}). |
|
|
|
|
|
Returns: |
|
|
Dictionary with pass/fail results and details. |
|
|
""" |
|
|
results = { |
|
|
"passed": True, |
|
|
"checks": [] |
|
|
} |
|
|
|
|
|
|
|
|
if "max_height" in rules: |
|
|
limit = rules["max_height"] |
|
|
actual = metrics.get("building_height", 0) |
|
|
passed = actual <= limit |
|
|
results["checks"].append({ |
|
|
"rule": "Max Height", |
|
|
"limit": limit, |
|
|
"actual": actual, |
|
|
"passed": passed, |
|
|
"message": "Height within limits" if passed else f"Height {actual} exceeds limit {limit}" |
|
|
}) |
|
|
if not passed: |
|
|
results["passed"] = False |
|
|
|
|
|
|
|
|
if "min_area" in rules: |
|
|
limit = rules["min_area"] |
|
|
actual = metrics.get("total_floor_area", 0) |
|
|
passed = actual >= limit |
|
|
results["checks"].append({ |
|
|
"rule": "Min Floor Area", |
|
|
"limit": limit, |
|
|
"actual": actual, |
|
|
"passed": passed, |
|
|
"message": "Area meets minimum" if passed else f"Area {actual} is below minimum {limit}" |
|
|
}) |
|
|
if not passed: |
|
|
results["passed"] = False |
|
|
|
|
|
return results |
|
|
|
|
|
|
|
|
engine = RuleEngine() |
|
|
|