Skip to content

Transformers

cognite.neat.graph.transformers #

AddAssetDepth #

Bases: BaseTransformer

Source code in cognite/neat/graph/transformers/_classic_cdf.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class AddAssetDepth(BaseTransformer):
    description: str = "Adds depth of asset in the asset hierarchy to the graph"
    _use_only_once: bool = True
    _need_changes = frozenset({str(extractors.AssetsExtractor.__name__)})

    _parent_template: str = """SELECT ?child ?parent WHERE {{
                              <{asset_id}> <{parent_prop}> ?child .
                              OPTIONAL{{?child <{parent_prop}>+ ?parent .}}}}"""

    _root_template: str = """SELECT ?root WHERE {{
                             <{asset_id}> <{root_prop}> ?root .}}"""

    def __init__(
        self,
        asset_type: URIRef | None = None,
        root_prop: URIRef | None = None,
        parent_prop: URIRef | None = None,
        depth_typing: dict[int, str] | None = None,
    ):
        self.asset_type = asset_type or DEFAULT_NAMESPACE.Asset
        self.root_prop = root_prop or DEFAULT_NAMESPACE.root
        self.parent_prop = parent_prop or DEFAULT_NAMESPACE.parent
        self.depth_typing = depth_typing

    def transform(self, graph: Graph) -> None:
        """Adds depth of asset in the asset hierarchy to the graph."""
        for result in graph.query(f"SELECT DISTINCT ?asset_id WHERE {{?asset_id a <{self.asset_type}>}}"):
            asset_id = cast(tuple, result)[0]
            if depth := self.get_depth(graph, asset_id, self.root_prop, self.parent_prop):
                graph.add((asset_id, DEFAULT_NAMESPACE.depth, Literal(depth)))

                if self.depth_typing and (type_ := self.depth_typing.get(depth, None)):
                    # remove existing type
                    graph.remove((asset_id, RDF.type, None))

                    # add new type
                    graph.add((asset_id, RDF.type, DEFAULT_NAMESPACE[type_]))

    @classmethod
    def get_depth(
        cls,
        graph: Graph,
        asset_id: URIRef,
        root_prop: URIRef,
        parent_prop: URIRef,
    ) -> int | None:
        """Get asset depth in the asset hierarchy."""

        # Handles non-root assets
        if result := list(graph.query(cls._parent_template.format(asset_id=asset_id, parent_prop=parent_prop))):
            return len(cast(list[tuple], result)) + 2 if cast(list[tuple], result)[0][1] else 2

        # Handles root assets
        elif (
            (result := list(graph.query(cls._root_template.format(asset_id=asset_id, root_prop=root_prop))))
            and len(cast(list[tuple], result)) == 1
            and cast(list[tuple], result)[0][0] == asset_id
        ):
            return 1
        else:
            return None

transform(graph) #

Adds depth of asset in the asset hierarchy to the graph.

Source code in cognite/neat/graph/transformers/_classic_cdf.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
def transform(self, graph: Graph) -> None:
    """Adds depth of asset in the asset hierarchy to the graph."""
    for result in graph.query(f"SELECT DISTINCT ?asset_id WHERE {{?asset_id a <{self.asset_type}>}}"):
        asset_id = cast(tuple, result)[0]
        if depth := self.get_depth(graph, asset_id, self.root_prop, self.parent_prop):
            graph.add((asset_id, DEFAULT_NAMESPACE.depth, Literal(depth)))

            if self.depth_typing and (type_ := self.depth_typing.get(depth, None)):
                # remove existing type
                graph.remove((asset_id, RDF.type, None))

                # add new type
                graph.add((asset_id, RDF.type, DEFAULT_NAMESPACE[type_]))

get_depth(graph, asset_id, root_prop, parent_prop) classmethod #

Get asset depth in the asset hierarchy.

Source code in cognite/neat/graph/transformers/_classic_cdf.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def get_depth(
    cls,
    graph: Graph,
    asset_id: URIRef,
    root_prop: URIRef,
    parent_prop: URIRef,
) -> int | None:
    """Get asset depth in the asset hierarchy."""

    # Handles non-root assets
    if result := list(graph.query(cls._parent_template.format(asset_id=asset_id, parent_prop=parent_prop))):
        return len(cast(list[tuple], result)) + 2 if cast(list[tuple], result)[0][1] else 2

    # Handles root assets
    elif (
        (result := list(graph.query(cls._root_template.format(asset_id=asset_id, root_prop=root_prop))))
        and len(cast(list[tuple], result)) == 1
        and cast(list[tuple], result)[0][0] == asset_id
    ):
        return 1
    else:
        return None