refactor: added display names

cognite/neat/_rules/_shared.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Any, Generic, TypeAlias, TypeVar
+from typing import Any, Generic, TypeAlias, TypeVar, get_args
 
 from cognite.neat._rules.models import (
     DMSRules,
@@ -22,6 +22,16 @@ class ReadRules(Generic[T_InputRules]):
     rules: T_InputRules | None
     read_context: dict[str, Any]
 
+    @classmethod
+    def display_type_name(cls) -> str:
+        return get_args(cls)[0].display_type_name()
+
+    @property
+    def display_name(self):
+        if self.rules is None:
+            return "FailedRead"
+        return self.rules.display_name
+
 
 ReadInputRules: TypeAlias = ReadRules[DMSInputRules] | ReadRules[InformationInputRules]
 T_ReadInputRules = TypeVar("T_ReadInputRules", bound=ReadInputRules)

cognite/neat/_rules/models/_base_rules.py

@@ -50,6 +50,7 @@
 )
 from cognite.neat._rules.models.data_types import DataType
 from cognite.neat._rules.models.entities import EdgeEntity, ReverseConnectionEntity, ViewEntity
+from cognite.neat._utils.rdf_ import uri_display_name
 
 if sys.version_info >= (3, 11):
     from enum import StrEnum
@@ -285,6 +286,10 @@ def headers_by_sheet(cls, by_alias: bool = False) -> dict[str, list[str]]:
             ]
         return headers_by_sheet
 
+    @property
+    def display_name(self):
+        return uri_display_name(self.metadata.identifier)
+
     def dump(
         self,
         entities_exclude_defaults: bool = True,

cognite/neat/_rules/models/dms/_rules.py

@@ -447,6 +447,10 @@ def as_schema(self, instance_space: str | None = None, remove_cdf_spaces: bool =
 
         return _DMSExporter(self, instance_space, remove_cdf_spaces=remove_cdf_spaces).to_schema()
 
+    @classmethod
+    def display_type_name(cls) -> str:
+        return "VerifiedDMSModel"
+
     def _repr_html_(self) -> str:
         summary = {
             "aspect": self.metadata.aspect,

cognite/neat/_rules/models/dms/_rules_input.py

@@ -21,6 +21,7 @@
     load_connection,
     load_dms_value_type,
 )
+from cognite.neat._utils.rdf_ import uri_display_name
 
 from ._rules import _DEFAULT_VERSION, DMSContainer, DMSEnum, DMSMetadata, DMSNode, DMSProperty, DMSRules, DMSView
 
@@ -289,6 +290,14 @@ def dump(self) -> dict[str, Any]:
             "Nodes": [node_type.dump(default_space) for node_type in self.nodes or []] or None,
         }
 
+    @classmethod
+    def display_type_name(cls) -> str:
+        return "UnverifiedDMSModel"
+
+    @property
+    def display_name(self):
+        return uri_display_name(self.metadata.identifier)
+
     def _repr_html_(self) -> str:
         summary = {
             "type": "Physical Data Model",

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

@@ -272,6 +272,10 @@ def as_dms_rules(self) -> "DMSRules":
 
         return _InformationRulesConverter(self).as_dms_rules()
 
+    @classmethod
+    def display_type_name(cls) -> str:
+        return "VerifiedInformationModel"
+
     def _repr_html_(self) -> str:
         summary = {
             "type": "Logical Data Model",

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

@@ -14,6 +14,7 @@
     UnknownEntity,
     load_value_type,
 )
+from cognite.neat._utils.rdf_ import uri_display_name
 
 from ._rules import (
     InformationClass,
@@ -150,6 +151,14 @@ def dump(self) -> dict[str, Any]:
             Prefixes=self.prefixes,
         )
 
+    @classmethod
+    def display_type_name(cls) -> str:
+        return "UnverifiedInformationModel"
+
+    @property
+    def display_name(self):
+        return uri_display_name(self.metadata.identifier)
+
     def _repr_html_(self) -> str:
         summary = {
             "type": "Logical Data Model",

cognite/neat/_session/_show.py

@@ -5,7 +5,6 @@
 import networkx as nx
 from IPython.display import HTML, display
 from pyvis.network import Network as PyVisNetwork  # type: ignore
-from rdflib import URIRef
 
 from cognite.neat._constants import IN_NOTEBOOK, IN_PYODIDE
 from cognite.neat._rules._constants import EntityTypes
@@ -14,7 +13,7 @@
 from cognite.neat._rules.models.information._rules import InformationRules
 from cognite.neat._session.exceptions import NeatSessionError
 from cognite.neat._utils.io_ import to_directory_compatible
-from cognite.neat._utils.rdf_ import remove_namespace_from_uri
+from cognite.neat._utils.rdf_ import remove_namespace_from_uri, uri_display_name
 
 from ._state import SessionState
 from .exceptions import session_class_wrapper
@@ -282,9 +281,9 @@ def _generate_dm_provenance_di_graph_and_types(self) -> nx.DiGraph:
         hex_colored_types = _generate_hex_color_per_type(["Agent", "Entity", "Activity", "Export", "Pruned"])
 
         for change in self._state.rule_store.provenance:
-            source = self._shorten_id(change.source_entity.id_)
-            target = self._shorten_id(change.target_entity.id_)
-            agent = self._shorten_id(change.agent.id_)
+            source = uri_display_name(change.source_entity.id_)
+            target = uri_display_name(change.target_entity.id_)
+            agent = uri_display_name(change.agent.id_)
 
             di_graph.add_node(
                 source,
@@ -314,9 +313,9 @@ def _generate_dm_provenance_di_graph_and_types(self) -> nx.DiGraph:
             di_graph.add_edge(agent, target, label="generated", color="grey")
 
         for source_id, exports in self._state.rule_store.exports_by_source_entity_id.items():
-            source_shorten = self._shorten_id(source_id)
+            source_shorten = uri_display_name(source_id)
             for export in exports:
-                export_id = self._shorten_id(export.target_entity.id_)
+                export_id = uri_display_name(export.target_entity.id_)
                 di_graph.add_node(
                     export_id,
                     label=export_id,
@@ -329,8 +328,8 @@ def _generate_dm_provenance_di_graph_and_types(self) -> nx.DiGraph:
         for pruned_lists in self._state.rule_store.pruned_by_source_entity_id.values():
             for prune_path in pruned_lists:
                 for change in prune_path:
-                    source = self._shorten_id(change.source_entity.id_)
-                    target = self._shorten_id(change.target_entity.id_)
+                    source = uri_display_name(change.source_entity.id_)
+                    target = uri_display_name(change.target_entity.id_)
                     di_graph.add_node(
                         target,
                         label=target,
@@ -342,14 +341,6 @@ def _generate_dm_provenance_di_graph_and_types(self) -> nx.DiGraph:
 
         return di_graph
 
-    @staticmethod
-    def _shorten_id(thing: URIRef) -> str:
-        if "https://cognitedata.com/dms/data-model/" in thing:
-            return "DMS(" + ",".join(thing.replace("https://cognitedata.com/dms/data-model/", "").split("/")) + ")"
-        elif "http://purl.org/cognite/neat/data-model/" in thing:
-            return "NEAT(" + ",".join(thing.replace("http://purl.org/cognite/neat/data-model/", "").split("/")) + ")"
-        return remove_namespace_from_uri(thing)
-
 
 @session_class_wrapper
 class ShowInstanceAPI(ShowBaseAPI):

cognite/neat/_utils/rdf_.py

@@ -257,3 +257,11 @@ def check_commit(force_commit: bool = False):
         check_commit()
 
     check_commit(force_commit=True)
+
+
+def uri_display_name(thing: URIRef) -> str:
+    if "https://cognitedata.com/dms/data-model/" in thing:
+        return "DMS(" + ",".join(thing.replace("https://cognitedata.com/dms/data-model/", "").split("/")) + ")"
+    elif "http://purl.org/cognite/neat/data-model/" in thing:
+        return "NEAT(" + ",".join(thing.replace("http://purl.org/cognite/neat/data-model/", "").split("/")) + ")"
+    return remove_namespace_from_uri(thing)