Skip to content

filter

Filter that removes all but specified instances in the collection.

DLiteFilterConfig

Bases: FilterConfig

DLite generate strategy config.

Source code in oteapi_dlite/strategies/filter.py 
82
83
84
85
86
87
88
class DLiteFilterConfig(FilterConfig):
    """DLite generate strategy config."""

    configuration: Annotated[
        DLiteQueryConfig,
        Field(description="DLite filter strategy-specific configuration."),
    ]

configuration: Annotated[DLiteQueryConfig, Field(description='DLite filter strategy-specific configuration.')] instance-attribute

DLiteFilterStrategy

Filter that removes all but specified instances in the collection.

The query configuration should be a regular expression matching labels to keep in the collection. All other labels will be removed.

Registers strategies:

  • ("filterType", "dlite/filter")
Source code in oteapi_dlite/strategies/filter.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@dataclass
class DLiteFilterStrategy:
    """Filter that removes all but specified instances in the collection.

    The `query` configuration should be a regular expression matching labels
    to keep in the collection.  All other labels will be removed.

    **Registers strategies**:

    - `("filterType", "dlite/filter")`

    """

    filter_config: DLiteFilterConfig

    def initialize(
        self,
        session: Optional[dict[str, Any]] = None,
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy."""
        config = self.filter_config.configuration

        # Alias for query configuration
        keep_label = (
            config.keep_label if config.keep_label else self.filter_config.query
        )

        instdict = {}  # Map instance labels to [uuid, metaURI]
        coll = get_collection(session)
        for s, _, o in coll.get_relations(p="_has-uuid"):
            instdict[s] = [o]
        for s, _, o in coll.get_relations(p="_has-meta"):
            instdict[s].append(o)

        removal = set()  # Labels marked for removal

        # 1: remove_label, remove_datamodel
        if config.remove_label or config.remove_datamodel:
            for label, (_, metauri) in instdict.items():
                if config.remove_label and re.match(config.remove_label, label):
                    removal.add(label)

                if config.remove_datamodel and re.match(
                    config.remove_datamodel, metauri
                ):
                    removal.add(label)
        else:
            removal.update(instdict.keys())

        # 2: keep_label, keep_datamodel
        for label in set(removal):
            if keep_label and re.match(keep_label, label):
                removal.remove(label)

            _, metauri = instdict[label]
            if config.keep_datamodel and re.match(
                config.keep_datamodel, metauri
            ):
                removal.remove(label)

        # 3: keep_referred
        if config.keep_referred:
            labels = {uuid: label for label, (uuid, _) in instdict.items()}
            kept = set(instdict.keys()).difference(removal)
            for label in kept:
                removal.difference_update(
                    labels[inst.uuid]
                    for inst in get_referred_instances(coll.get(label))
                    if inst.uuid in labels
                )

        # 4: remove from collection
        for label in removal:
            coll.remove(label)

        update_collection(coll)
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

filter_config: DLiteFilterConfig instance-attribute

get(session=None)

Execute the strategy.

Source code in oteapi_dlite/strategies/filter.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy."""
    config = self.filter_config.configuration

    # Alias for query configuration
    keep_label = (
        config.keep_label if config.keep_label else self.filter_config.query
    )

    instdict = {}  # Map instance labels to [uuid, metaURI]
    coll = get_collection(session)
    for s, _, o in coll.get_relations(p="_has-uuid"):
        instdict[s] = [o]
    for s, _, o in coll.get_relations(p="_has-meta"):
        instdict[s].append(o)

    removal = set()  # Labels marked for removal

    # 1: remove_label, remove_datamodel
    if config.remove_label or config.remove_datamodel:
        for label, (_, metauri) in instdict.items():
            if config.remove_label and re.match(config.remove_label, label):
                removal.add(label)

            if config.remove_datamodel and re.match(
                config.remove_datamodel, metauri
            ):
                removal.add(label)
    else:
        removal.update(instdict.keys())

    # 2: keep_label, keep_datamodel
    for label in set(removal):
        if keep_label and re.match(keep_label, label):
            removal.remove(label)

        _, metauri = instdict[label]
        if config.keep_datamodel and re.match(
            config.keep_datamodel, metauri
        ):
            removal.remove(label)

    # 3: keep_referred
    if config.keep_referred:
        labels = {uuid: label for label, (uuid, _) in instdict.items()}
        kept = set(instdict.keys()).difference(removal)
        for label in kept:
            removal.difference_update(
                labels[inst.uuid]
                for inst in get_referred_instances(coll.get(label))
                if inst.uuid in labels
            )

    # 4: remove from collection
    for label in removal:
        coll.remove(label)

    update_collection(coll)
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/filter.py 
106
107
108
109
110
111
def initialize(
    self,
    session: Optional[dict[str, Any]] = None,
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

DLiteQueryConfig

Bases: AttrDict

Configuration for the DLite filter strategy.

First the remove_label and remove_datamodel configurations are used to mark matching instances for removal. If neither remove_label or remove_datamodel are given, all instances are marked for removal.

Then instances matching keep_label and keep_datamodel are unmarked for removal.

If keep_referred is true, any instance that is referred to by an instance not marked for removal is also unmarked for removal.

Finally, the instances that are still marked for removal are removed from the collection.

Source code in oteapi_dlite/strategies/filter.py 
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
72
73
74
75
76
77
78
79
class DLiteQueryConfig(AttrDict):
    """Configuration for the DLite filter strategy.

    First the `remove_label` and `remove_datamodel` configurations are
    used to mark matching instances for removal.  If neither
    `remove_label` or `remove_datamodel` are given, all instances are
    marked for removal.

    Then instances matching `keep_label` and `keep_datamodel` are unmarked
    for removal.

    If `keep_referred` is true, any instance that is referred to by
    an instance not marked for removal is also unmarked for removal.

    Finally, the instances that are still marked for removal are removed
    from the collection.
    """

    remove_label: Annotated[
        Optional[str],
        Field(description="Regular expression matching labels to remove."),
    ] = None
    remove_datamodel: Annotated[
        Optional[str],
        Field(
            description="Regular expression matching datamodel URIs to remove.",
        ),
    ] = None
    keep_label: Annotated[
        Optional[str],
        Field(
            description=(
                "Regular expression matching labels to keep. This "
                "configuration overrides `remove_label` and "
                "`remove_datamodel`. Alias for the FilterStrategy `query` "
                "configuration, that is inherited from the oteapi-core Filter "
                "data model."
            ),
        ),
    ] = None
    keep_datamodel: Annotated[
        Optional[str],
        Field(
            description=(
                "Regular expression matching datamodel URIs to keep in "
                "collection. This configuration overrides `remove_label` and "
                "`remove_datamodel`."
            ),
        ),
    ] = None
    keep_referred: Annotated[
        bool,
        Field(
            description=(
                "Whether to keep all instances in the collection that are "
                "directly or indirectly referred to (via ref-types or "
                "collections) by kept instances."
            ),
        ),
    ] = True

keep_datamodel: Annotated[Optional[str], Field(description='Regular expression matching datamodel URIs to keep in collection. This configuration overrides `remove_label` and `remove_datamodel`.')] = None class-attribute instance-attribute

keep_label: Annotated[Optional[str], Field(description='Regular expression matching labels to keep. This configuration overrides `remove_label` and `remove_datamodel`. Alias for the FilterStrategy `query` configuration, that is inherited from the oteapi-core Filter data model.')] = None class-attribute instance-attribute

keep_referred: Annotated[bool, Field(description='Whether to keep all instances in the collection that are directly or indirectly referred to (via ref-types or collections) by kept instances.')] = True class-attribute instance-attribute

remove_datamodel: Annotated[Optional[str], Field(description='Regular expression matching datamodel URIs to remove.')] = None class-attribute instance-attribute

remove_label: Annotated[Optional[str], Field(description='Regular expression matching labels to remove.')] = None class-attribute instance-attribute