| """ |
| Owlready2 wrapper for a small infrastructure ontology. |
| Provides classification and consistency checking. |
| """ |
| import logging |
| logger = logging.getLogger(__name__) |
|
|
| try: |
| from owlready2 import * |
| OWLREADY_AVAILABLE = True |
| except ImportError: |
| OWLREADY_AVAILABLE = False |
|
|
| class InfraOntology: |
| def __init__(self, path="infra.owl"): |
| self.available = OWLREADY_AVAILABLE |
| self.onto = None |
| if self.available: |
| try: |
| self.onto = get_ontology(path).load() |
| logger.info(f"Ontology loaded from {path}") |
| except Exception as e: |
| logger.warning(f"Failed to load ontology: {e}") |
| self.available = False |
| else: |
| logger.info("Owlready2 not installed, ontology disabled") |
| |
| def classify(self, component_type: str) -> dict: |
| """ |
| Run reasoner and return inferred classes for a given component type. |
| Returns dict with 'inferred' list and 'consistent' bool. |
| """ |
| if not self.available or self.onto is None: |
| return {"inferred": [], "consistent": True} |
| |
| try: |
| with self.onto: |
| sync_reasoner() |
| |
| inferred = [] |
| |
| if hasattr(self.onto, component_type.capitalize()): |
| cls = getattr(self.onto, component_type.capitalize()) |
| inferred = [str(i) for i in cls.instances()] |
| return {"inferred": inferred, "consistent": True} |
| except OSError as e: |
| |
| if "java" in str(e).lower(): |
| logger.warning("Java not available, using mock reasoning") |
| return {"inferred": [], "consistent": True, "note": "Java required for full reasoning, using mock"} |
| else: |
| logger.error(f"Reasoning error: {e}") |
| return {"inferred": [], "consistent": False, "error": str(e)} |
| except Exception as e: |
| logger.error(f"Reasoning error: {e}") |
| return {"inferred": [], "consistent": False, "error": str(e)} |