Datasets (Taxonomies)

src/examples/example4.py

@@ -0,0 +1,9 @@
+from openmisp.datasets import Taxonomies
+
+pap_tag = Taxonomies.PAP.RED.tag
+confidence_tag = Taxonomies.MISP.CONFIDENCE_LEVEL.USUALLY_CONFIDENT.tag
+
+print(pap_tag)
+print(confidence_tag)
+
+pass

src/openmisp/__init__.py

@@ -23,6 +23,7 @@
 from .sharing_groups import SharingGroupCriteria
 from .sightings import SightingCriteria, SightingType
 from .tags import TagCriteria
+from . import datasets
 
 __all__ = [
     "MISP",
@@ -53,5 +54,6 @@
     "SightingCriteria",
     "SightingType",
     "TagCriteria",
+    "datasets",
     "ThreatLevel",
 ]

src/openmisp/datasets/__init__.py

@@ -0,0 +1,3 @@
+from .taxonomies import Taxonomies
+
+__all__ = ["Taxonomies"]

src/openmisp/datasets/taxonomies/__init__.py

@@ -0,0 +1,7 @@
+import pathlib
+
+from .loader import Loader
+
+Taxonomies = Loader(pathlib.Path(__file__).parent / "data" / "misp-taxonomies")
+
+__all__ = ["Taxonomies"]

src/openmisp/datasets/taxonomies/loader.py

@@ -0,0 +1,98 @@
+import json
+from typing import Optional
+
+from .models import Category, Entry, MachineTag, Taxonomy
+
+FILENAME = "machinetag.json"
+
+
+def normalize_key(key: str) -> str:
+    return key.replace("-", "_").replace(":", "_").upper()
+
+
+class Loader:
+    def __init__(self, base_path: str):
+        self._taxonomies = {}
+        self._load(base_path)
+
+    def _load(self, base_path: str):
+        for path in base_path.iterdir():
+            file = path / FILENAME
+
+            if not file.is_file():
+                continue
+
+            with open(file, "r") as fp:
+                data = json.load(fp)
+                data = MachineTag(**data)
+
+            taxonomy = self._load_taxonomy(data)
+            codename = normalize_key(taxonomy.namespace)
+            self._taxonomies[codename] = taxonomy
+
+    def _load_taxonomy(self, data: dict) -> Taxonomy:
+        categories = self._load_categories(data)
+
+        entries = self._load_entries(
+            data,
+            namespace=data.namespace,
+            lookup="predicates",
+        )
+
+        return Taxonomy(
+            namespace=data.namespace,
+            description=data.description,
+            uuid=data.uuid,
+            version=data.version,
+            categories=categories,
+            entries=entries,
+        )
+
+    def _load_categories(self, data: dict) -> dict[str, Category]:
+        categories = {}
+
+        for item in data.values:
+            entries = self._load_entries(
+                item,
+                namespace=data.namespace,
+                category=item.predicate,
+                lookup="entry",
+            )
+
+            for predicate in data.predicates:
+                if predicate.value != item.predicate:
+                    continue
+
+                codename = normalize_key(item.predicate)
+
+                categories[codename] = Category(
+                    name=predicate.value,
+                    exclusive=predicate.exclusive,
+                    entries=entries,
+                )
+
+        return categories
+
+    def _load_entries(
+        self,
+        data: dict,
+        namespace: str,
+        category: Optional[str] = None,
+        lookup: str = "entry",
+    ) -> dict[str, Entry]:
+        return {
+            normalize_key(item.value): Entry(
+                namespace=namespace,
+                category=category,
+                **item.dict(),
+            )
+            for item in getattr(data, lookup, [])
+        }
+
+    def __getattr__(self, item: str) -> Taxonomy:
+        if item in self._taxonomies:
+            return self._taxonomies[item]
+        raise AttributeError(f"No taxonomy named '{item}'")
+
+    def __dir__(self):
+        return list(self._taxonomies.keys()) + super().__dir__()

src/openmisp/datasets/taxonomies/models/__init__.py

@@ -0,0 +1,4 @@
+from .machinetag import MachineTag
+from .taxonomy import Category, Entry, Taxonomy
+
+__all__ = ["Category", "Entry", "MachineTag", "Taxonomy"]

src/openmisp/datasets/taxonomies/models/machinetag.py

@@ -0,0 +1,30 @@
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class Predicate(BaseModel):
+    uuid: str
+    value: str
+    expanded: Optional[str] = None
+    exclusive: Optional[bool] = False
+
+
+class Entry(BaseModel):
+    uuid: str
+    value: str
+    expanded: Optional[str] = None
+
+
+class Value(BaseModel):
+    predicate: str
+    entry: list[Entry]
+
+
+class MachineTag(BaseModel):
+    uuid: str
+    namespace: str
+    description: str
+    version: int
+    predicates: list[Predicate]
+    values: Optional[list[Value]] = Field(default_factory=list)

src/openmisp/datasets/taxonomies/models/taxonomy.py

@@ -0,0 +1,67 @@
+from typing import Dict, Optional
+
+from pydantic import BaseModel, Field
+
+
+class Entry(BaseModel):
+    uuid: str
+    value: str
+    namespace: str
+    category: Optional[str] = None
+    expanded: Optional[str] = None
+    description: Optional[str] = None
+    numerical_value: Optional[float] = None
+
+    def __str__(self):
+        return self.value
+
+    @property
+    def tag(self) -> str:
+        if self.category:
+            return f"{self.namespace}:{self.category}:{self.value}"
+        return f"{self.namespace}:{self.value}"
+
+
+class _Base(BaseModel):
+    entries: Dict[str, Entry]
+
+    def __getattr__(self, item: str) -> Entry:
+        if item in self.entries:
+            return self.entries[item]
+        raise AttributeError(f"{item} not found")
+
+    def __dir__(self):
+        return list(self.entries.keys()) + super().__dir__()
+
+    def __iter__(self):
+        return iter(self.entries.values())
+
+    def keys(self):
+        return self.entries.keys()
+
+    def values(self):
+        return self.entries.values()
+
+    def items(self):
+        return self.entries.items()
+
+
+class Category(_Base):
+    name: str
+    exclusive: Optional[bool] = False
+
+
+class Taxonomy(_Base):
+    namespace: str
+    description: Optional[str] = ""
+    uuid: str
+    version: int
+    categories: Optional[Dict[str, Category]] = Field(default_factory=dict)
+
+    def __getattr__(self, item: str) -> Category | Entry:
+        if item in self.categories:
+            return self.categories[item]
+        return super().__getattr__(item)
+
+    def __dir__(self):
+        return super().__dir__() + list(self.categories.keys())