Neat 333 query build asset centric construct query

Makefile

@@ -1,5 +1,5 @@
 .PHONY: run-explorer run-tests run-linters build-ui build-python build-docker run-docker compose-up
-version="0.84.0"
+version="0.84.1"
 run-explorer:
 	@echo "Running explorer API server..."
 	# open "http://localhost:8000/static/index.html" || true

cognite/neat/_version.py

@@ -1 +1 @@
-__version__ = "0.84.0"
+__version__ = "0.84.1"

cognite/neat/graph/loaders/_rdf2asset.py

@@ -0,0 +1,123 @@
+from collections.abc import Sequence
+from dataclasses import dataclass, fields
+
+from cognite.client import CogniteClient
+from cognite.client.data_classes import AssetWrite
+
+from cognite.neat.graph._tracking.base import Tracker
+from cognite.neat.graph._tracking.log import LogTracker
+from cognite.neat.graph.stores import NeatGraphStore
+from cognite.neat.issues import NeatIssue, NeatIssueList
+from cognite.neat.rules.models import AssetRules
+
+from ._base import CDFLoader
+
+
+@dataclass(frozen=True)
+class AssetLoaderMetadataKeys:
+    """Class holding mapping between NEAT metadata key names and their desired names
+    in CDF Asset metadata
+
+    Args:
+        start_time: Start time key name
+        end_time: End time key name
+        update_time: Update time key name
+        resurrection_time: Resurrection time key name
+        identifier: Identifier key name
+        active: Active key name
+        type: Type key name
+    """
+
+    start_time: str = "start_time"
+    end_time: str = "end_time"
+    update_time: str = "update_time"
+    resurrection_time: str = "resurrection_time"
+    identifier: str = "identifier"
+    active: str = "active"
+    type: str = "type"
+
+    def as_aliases(self) -> dict[str, str]:
+        return {str(field.default): getattr(self, field.name) for field in fields(self)}
+
+
+class AssetLoader(CDFLoader[AssetWrite]):
+    def __init__(
+        self,
+        rules: AssetRules,
+        graph_store: NeatGraphStore,
+        data_set_id: int,
+        use_orphanage: bool = False,
+        use_labels: bool = False,
+        asset_external_id_prefix: str | None = None,
+        metadata_keys: AssetLoaderMetadataKeys | None = None,
+        create_issues: Sequence[NeatIssue] | None = None,
+        tracker: type[Tracker] | None = None,
+    ):
+        super().__init__(graph_store)
+
+        self.rules = rules
+        self.data_set_id = data_set_id
+        self.use_labels = use_labels
+        self.use_orphanage = use_orphanage
+
+        self.orphanage_external_id = (
+            f"{asset_external_id_prefix or ''}orphanage-{data_set_id}" if use_orphanage else None
+        )
+
+        self.asset_external_id_prefix = asset_external_id_prefix
+        self.metadata_keys = metadata_keys or AssetLoaderMetadataKeys()
+
+        self._issues = NeatIssueList[NeatIssue](create_issues or [])
+        self._tracker: type[Tracker] = tracker or LogTracker
+
+    @classmethod
+    def from_rules(
+        cls,
+        rules: AssetRules,
+        graph_store: NeatGraphStore,
+        data_set_id: int,
+        use_orphanage: bool = False,
+        use_labels: bool = False,
+        asset_external_id_prefix: str | None = None,
+        metadata_keys: AssetLoaderMetadataKeys | None = None,
+    ) -> "AssetLoader":
+        issues: list[NeatIssue] = []
+
+        return cls(
+            rules, graph_store, data_set_id, use_orphanage, use_labels, asset_external_id_prefix, metadata_keys, issues
+        )
+
+    def _create_validation_classes(self) -> None:
+        # need to get back class-property pairs where are definition of
+        # asset implementations, extend InformationRulesAnalysis make it generic
+
+        # by default if there is not explicitly stated external_id
+        # use rdf:type and drop the prefix
+
+        # based on those create pydantic model AssetDefinition
+        # which will have .to_asset_write()
+
+        raise NotImplementedError("Not implemented yet, this is placeholder")
+
+    def categorize_assets(self, client: CogniteClient) -> None:
+        """Categorize assets to those to be created, updated, decommissioned, or resurrected"""
+
+        raise NotImplementedError("Not implemented yet, this is placeholder")
+
+    def load_to_cdf(self, client: CogniteClient, dry_run: bool = False) -> Sequence[AssetWrite]:
+        # generate assets
+        # check for circular asset hierarchy
+        # check for orphaned assets
+        # batch upsert of assets to CDF (otherwise we will hit the API rate limit)
+
+        raise NotImplementedError("Not implemented yet, this is placeholder")
+
+    @classmethod
+    def _check_for_circular_asset_hierarchy(cls, assets: list[AssetWrite]) -> None:
+        """Check for circular references in the asset rules"""
+        raise NotImplementedError("Not implemented yet, this is placeholder")
+
+    @classmethod
+    def _check_for_orphaned_assets(cls, assets: list[AssetWrite]) -> None:
+        """Check for circular references in the asset rules"""
+        raise NotImplementedError("Not implemented yet, this is placeholder")

cognite/neat/graph/loaders/_rdf2dms.py

@@ -34,14 +34,14 @@ def __init__(
         data_model: dm.DataModel[dm.View] | None,
         instance_space: str,
         class_by_view_id: dict[ViewId, str] | None = None,
-        creat_issues: Sequence[NeatIssue] | None = None,
+        create_issues: Sequence[NeatIssue] | None = None,
         tracker: type[Tracker] | None = None,
     ):
         super().__init__(graph_store)
         self.data_model = data_model
         self.instance_space = instance_space
         self.class_by_view_id = class_by_view_id or {}
-        self._issues = NeatIssueList[NeatIssue](creat_issues or [])
+        self._issues = NeatIssueList[NeatIssue](create_issues or [])
         self._tracker: type[Tracker] = tracker or LogTracker
 
     @classmethod

cognite/neat/graph/stores/_base.py

@@ -14,8 +14,8 @@
 from cognite.neat.graph.models import Triple
 from cognite.neat.graph.queries import Queries
 from cognite.neat.graph.transformers import Transformers
+from cognite.neat.rules.models import InformationRules
 from cognite.neat.rules.models.entities import ClassEntity
-from cognite.neat.rules.models.information import InformationRules
 from cognite.neat.utils.auxiliary import local_import
 
 from ._provenance import Change, Provenance
@@ -66,7 +66,8 @@ def __init__(
 
     def add_rules(self, rules: InformationRules) -> None:
         """This method is used to add rules to the graph store and it is the only correct
-        way to add rules to the graph store, after the graph store has been initialized."""
+        way to add rules to the graph store, after the graph store has been initialized.
+        """
 
         self.rules = rules
         self.base_namespace = self.rules.metadata.namespace
@@ -169,7 +170,10 @@ def read(self, class_: str) -> list[tuple[str, str, str]]:
         # not yet developed
 
         if not self.rules:
-            warnings.warn("No rules found for the graph store, returning empty list.", stacklevel=2)
+            warnings.warn(
+                "No rules found for the graph store, returning empty list.",
+                stacklevel=2,
+            )
             return []
 
         class_entity = ClassEntity(prefix=self.rules.metadata.prefix, suffix=class_)

cognite/neat/rules/importers/_base.py

@@ -11,7 +11,7 @@
 
 from cognite.neat.rules._shared import Rules
 from cognite.neat.rules.issues.base import IssueList, NeatValidationError, ValidationWarning
-from cognite.neat.rules.models import DMSRules, InformationRules, RoleTypes
+from cognite.neat.rules.models import AssetRules, DMSRules, InformationRules, RoleTypes
 
 
 class BaseImporter(ABC):
@@ -48,9 +48,9 @@ def _to_output(
 
         if rules.metadata.role is role or role is None:
             output = rules
-        elif isinstance(rules, DMSRules) and role is RoleTypes.information_architect:
+        elif isinstance(rules, DMSRules) or isinstance(rules, AssetRules) and role is RoleTypes.information_architect:
             output = rules.as_information_architect_rules()
-        elif isinstance(rules, InformationRules) and role is RoleTypes.dms_architect:
+        elif isinstance(rules, InformationRules) or isinstance(rules, AssetRules) and role is RoleTypes.dms_architect:
             output = rules.as_dms_architect_rules()
         else:
             raise NotImplementedError(f"Role {role} is not supported for {type(rules).__name__} rules")

cognite/neat/rules/models/_rdfpath.py

@@ -6,7 +6,7 @@
 from functools import total_ordering
 from typing import ClassVar, Literal
 
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, field_validator, model_serializer
 
 from cognite.neat.rules import exceptions
 
@@ -82,6 +82,7 @@ class EntityTypes(StrEnum):
 
 StepDirection = Literal["source", "target", "origin"]
 _direction_by_symbol: dict[str, StepDirection] = {"->": "target", "<-": "source"}
+_symbol_by_direction: dict[StepDirection, str] = {"source": "<-", "target": "->"}
 
 Undefined = type(object())
 Unknown = type(object())
@@ -196,10 +197,29 @@ def from_string(cls, raw: str, **kwargs) -> Self:
         msg += " ->prefix:suffix, <-prefix:suffix, ->prefix:suffix(prefix:suffix) or <-prefix:suffix(prefix:suffix)"
         raise ValueError(msg)
 
+    def __str__(self) -> str:
+        if self.property:
+            return f"{self.class_}({self.property})"
+        else:
+            return f"{_symbol_by_direction[self.direction]}{self.class_}"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
 
 class Traversal(BaseModel):
     class_: Entity
 
+    def __str__(self) -> str:
+        return f"{self.class_}"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
+    @model_serializer(when_used="unless-none", return_type=str)
+    def as_str(self) -> str:
+        return str(self)
+
 
 class SingleProperty(Traversal):
     property: Entity
@@ -208,6 +228,9 @@ class SingleProperty(Traversal):
     def from_string(cls, class_: str, property_: str) -> Self:
         return cls(class_=Entity.from_string(class_), property=Entity.from_string(property_))
 
+    def __str__(self) -> str:
+        return f"{self.class_}({self.property})"
+
 
 class AllReferences(Traversal):
     @classmethod
@@ -220,6 +243,9 @@ class AllProperties(Traversal):
     def from_string(cls, class_: str) -> Self:
         return cls(class_=Entity.from_string(class_))
 
+    def __str__(self) -> str:
+        return f"{self.class_}(*)"
+
 
 class Origin(BaseModel):
     class_: Entity
@@ -245,6 +271,9 @@ def from_string(cls, class_: str, traversal: str | list[Step]) -> Self:
             ),
         )
 
+    def __str__(self) -> str:
+        return f"{self.class_}{''.join([str(step) for step in self.traversal])}"
+
 
 class TableLookup(BaseModel):
     name: str
@@ -261,7 +290,17 @@ class Query(BaseModel):
 
 
 class RDFPath(Rule):
-    traversal: Traversal | Query
+    traversal: SingleProperty | AllProperties | AllReferences | Hop
+
+    def __str__(self) -> str:
+        return f"{self.traversal}"
+
+    def __repr__(self) -> str:
+        return self.__str__()
+
+    @model_serializer(when_used="unless-none", return_type=str)
+    def as_str(self) -> str:
+        return str(self)
 
 
 class RawLookup(RDFPath):

cognite/neat/rules/models/asset/_rules.py

@@ -148,9 +148,12 @@ def dump(
     def as_domain_rules(self) -> DomainRules:
         from ._converter import _AssetRulesConverter
 
-        return _AssetRulesConverter(cast(InformationRules, self)).as_domain_rules()
+        return _AssetRulesConverter(self.as_information_architect_rules()).as_domain_rules()
 
     def as_dms_architect_rules(self) -> "DMSRules":
         from ._converter import _AssetRulesConverter
 
-        return _AssetRulesConverter(cast(InformationRules, self)).as_dms_architect_rules()
+        return _AssetRulesConverter(self.as_information_architect_rules()).as_dms_architect_rules()
+
+    def as_information_architect_rules(self) -> InformationRules:
+        return InformationRules.model_validate(self.model_dump())

cognite/neat/rules/models/information/_converter.py

@@ -16,11 +16,15 @@
 from cognite.neat.rules.models.data_types import DataType
 from cognite.neat.rules.models.domain import DomainRules
 from cognite.neat.rules.models.entities import (
+    AssetEntity,
+    AssetFields,
     ClassEntity,
     ContainerEntity,
     DMSUnknownEntity,
+    EntityTypes,
     MultiValueTypeInfo,
     ReferenceEntity,
+    RelationshipEntity,
     UnknownEntity,
     ViewEntity,
     ViewPropertyEntity,
@@ -29,6 +33,7 @@
 from ._rules import InformationClass, InformationMetadata, InformationProperty, InformationRules
 
 if TYPE_CHECKING:
+    from cognite.neat.rules.models.asset._rules import AssetRules
     from cognite.neat.rules.models.dms._rules import DMSMetadata, DMSProperty, DMSRules
 
 
@@ -52,6 +57,28 @@ def __init__(self, information: InformationRules):
     def as_domain_rules(self) -> DomainRules:
         raise NotImplementedError("DomainRules not implemented yet")
 
+    def as_asset_architect_rules(self) -> "AssetRules":
+        from cognite.neat.rules.models.asset._rules import AssetClass, AssetMetadata, AssetProperty, AssetRules
+
+        classes: SheetList[AssetClass] = SheetList[AssetClass](
+            data=[AssetClass(**class_.model_dump()) for class_ in self.rules.classes]
+        )
+        properties: SheetList[AssetProperty] = SheetList[AssetProperty]()
+        for prop_ in self.rules.properties:
+            if prop_.type_ == EntityTypes.data_property:
+                properties.append(
+                    AssetProperty(**prop_.model_dump(), implementation=[AssetEntity(property=AssetFields.metadata)])
+                )
+            elif prop_.type_ == EntityTypes.object_property:
+                properties.append(AssetProperty(**prop_.model_dump(), implementation=[RelationshipEntity()]))
+
+        return AssetRules(
+            metadata=AssetMetadata(**self.rules.metadata.model_dump()),
+            properties=properties,
+            classes=classes,
+            prefixes=self.rules.prefixes,
+        )
+
     def as_dms_architect_rules(self) -> "DMSRules":
         from cognite.neat.rules.models.dms._rules import (
             DMSContainer,

cognite/neat/rules/models/information/_rules.py

@@ -49,7 +49,7 @@
 )
 
 if TYPE_CHECKING:
-    from cognite.neat.rules.models.dms._rules import DMSRules
+    from cognite.neat.rules.models import AssetRules, DMSRules
 
 
 if sys.version_info >= (3, 11):
@@ -341,6 +341,11 @@ def as_domain_rules(self) -> DomainRules:
 
         return _InformationRulesConverter(self).as_domain_rules()
 
+    def as_asset_architect_rules(self) -> "AssetRules":
+        from ._converter import _InformationRulesConverter
+
+        return _InformationRulesConverter(self).as_asset_architect_rules()
+
     def as_dms_architect_rules(self) -> "DMSRules":
         from ._converter import _InformationRulesConverter