Skip to content

ontocast.tool.agg.promoter

URI promotion from chunk-level to document-level.

This module handles the promotion of chunk-level entity URIs to document-level URIs, while preserving ontology entity URIs unchanged.

URIPromoter

Promotes chunk-level URIs to document-level URIs.

This class is responsible for converting entities from chunk namespaces to the document namespace, while keeping ontology entities unchanged.

Source code in ontocast/tool/agg/promoter.py 
 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
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
class URIPromoter:
    """Promotes chunk-level URIs to document-level URIs.

    This class is responsible for converting entities from chunk namespaces
    to the document namespace, while keeping ontology entities unchanged.
    """

    def __init__(
        self,
        doc_namespace: str,
        chunk_namespaces: set[str],
        ontology_namespaces: set[str],
    ):
        """Initialize the URI promoter.

        Args:
            doc_namespace: Document namespace for promoted URIs
            chunk_namespaces: Set of chunk namespace URIs
            ontology_namespaces: Set of ontology namespace URIs (preserved as-is)
        """
        self.doc_namespace = self._normalize_namespace(doc_namespace)
        self.chunk_namespaces = chunk_namespaces
        self.ontology_namespaces = ontology_namespaces
        self._used_uris: set[str] = set()

    def _normalize_namespace(self, namespace: str) -> str:
        """Ensure namespace ends with appropriate separator."""
        return namespace if namespace.endswith(("/", "#")) else namespace + "/"

    def _clean_local_name(self, name: str) -> str:
        """Clean a name for use as URI local part.

        Args:
            name: Name to clean

        Returns:
            Cleaned name suitable for URI
        """
        # Replace invalid URI characters with underscores
        cleaned = re.sub(r"[^\w\-.]", "_", name)
        # Remove consecutive underscores
        cleaned = re.sub(r"_+", "_", cleaned)
        # Remove leading/trailing underscores
        cleaned = cleaned.strip("_")
        return cleaned or "entity"

    def _ensure_unique_uri(self, uri: str) -> str:
        """Ensure URI is unique by appending counter if needed.

        Args:
            uri: Proposed URI

        Returns:
            Unique URI
        """
        if uri not in self._used_uris:
            self._used_uris.add(uri)
            return uri

        # URI already used, append counter
        base_uri = uri
        counter = 1

        while uri in self._used_uris:
            # Extract local name and add counter
            if "#" in base_uri:
                namespace, local = base_uri.rsplit("#", 1)
                uri = f"{namespace}#{local}_{counter}"
            else:
                namespace = base_uri.rstrip("/").rsplit("/", 1)[0]
                local = (
                    base_uri.rstrip("/").rsplit("/", 1)[1]
                    if "/" in base_uri.rstrip("/")
                    else base_uri
                )
                uri = f"{namespace}/{local}_{counter}"
            counter += 1

        self._used_uris.add(uri)
        return uri

    def should_promote(self, entity: URIRef) -> bool:
        """Check if an entity should be promoted to document namespace.

        Args:
            entity: Entity URI to check

        Returns:
            True if entity should be promoted (is from chunk namespace)
        """
        entity_str = str(entity)

        # Don't promote ontology entities
        if any(entity_str.startswith(ns) for ns in self.ontology_namespaces):
            return False

        # Promote chunk entities
        if any(entity_str.startswith(ns) for ns in self.chunk_namespaces):
            return True

        # Unknown namespace - be conservative, don't promote
        return False

    def promote_entity(
        self, entity: URIRef, representation: EntityRepresentation
    ) -> URIRef:
        """Promote a chunk entity to document namespace.

        Args:
            entity: Original entity URI
            representation: Entity representation with metadata

        Returns:
            Promoted entity URI
        """
        if not self.should_promote(entity):
            # Keep ontology entities unchanged
            return entity

        # Create new URI in document namespace
        # Use normalized form as local name
        local_name = self._clean_local_name(representation.normal_form)

        # Construct promoted URI
        promoted_uri_str = f"{self.doc_namespace}{local_name}"

        # Ensure uniqueness
        unique_uri_str = self._ensure_unique_uri(promoted_uri_str)

        return URIRef(unique_uri_str)

    def create_promotion_mapping(
        self,
        entities: list[URIRef],
        representations: dict[URIRef, EntityRepresentation],
    ) -> dict[URIRef, URIRef]:
        """Create mapping from original to promoted URIs.

        Args:
            entities: List of entity URIs to promote
            representations: Dictionary mapping entities to their representations

        Returns:
            Dictionary mapping original URIs to promoted URIs (e -> promoted(e))
        """
        mapping = {}
        promoted_count = 0
        preserved_count = 0

        for entity in entities:
            representation = representations.get(entity)
            if representation is None:
                logger.warning(f"No representation found for entity {entity}")
                mapping[entity] = entity
                continue

            promoted = self.promote_entity(entity, representation)
            mapping[entity] = promoted

            if promoted != entity:
                promoted_count += 1
            else:
                preserved_count += 1

        logger.info(
            f"Created promotion mapping: "
            f"{promoted_count} promoted, "
            f"{preserved_count} preserved (ontology entities)"
        )

        return mapping

    def compose_mappings(
        self,
        clustering_mapping: dict[URIRef, URIRef],
        promotion_mapping: dict[URIRef, URIRef],
    ) -> dict[URIRef, URIRef]:
        """Compose clustering and promotion mappings.

        First, entities are mapped to their cluster representatives (clustering_mapping).
        Then, representatives are promoted to document namespace (promotion_mapping).

        The composed mapping is: e -> promoted(representative(e))

        Args:
            clustering_mapping: Map from entity to cluster representative (e -> e_rep)
            promotion_mapping: Map from representative to promoted URI (e_rep -> e')

        Returns:
            Composed mapping (e -> e')
        """
        composed = {}

        for original, representative in clustering_mapping.items():
            # Look up the promoted version of the representative
            promoted = promotion_mapping.get(representative, representative)
            composed[original] = promoted

        logger.info(
            f"Composed mapping: {len(composed)} entities mapped to "
            f"{len(set(composed.values()))} final URIs"
        )

        return composed

__init__(doc_namespace, chunk_namespaces, ontology_namespaces)

Initialize the URI promoter.

Parameters:

Name Type Description Default
doc_namespace str

Document namespace for promoted URIs

 required
chunk_namespaces set[str]

Set of chunk namespace URIs

 required
ontology_namespaces set[str]

Set of ontology namespace URIs (preserved as-is)

 required
Source code in ontocast/tool/agg/promoter.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    doc_namespace: str,
    chunk_namespaces: set[str],
    ontology_namespaces: set[str],
):
    """Initialize the URI promoter.

    Args:
        doc_namespace: Document namespace for promoted URIs
        chunk_namespaces: Set of chunk namespace URIs
        ontology_namespaces: Set of ontology namespace URIs (preserved as-is)
    """
    self.doc_namespace = self._normalize_namespace(doc_namespace)
    self.chunk_namespaces = chunk_namespaces
    self.ontology_namespaces = ontology_namespaces
    self._used_uris: set[str] = set()

compose_mappings(clustering_mapping, promotion_mapping)

Compose clustering and promotion mappings.

First, entities are mapped to their cluster representatives (clustering_mapping). Then, representatives are promoted to document namespace (promotion_mapping).

The composed mapping is: e -> promoted(representative(e))

Parameters:

Name Type Description Default
clustering_mapping dict[URIRef, URIRef]

Map from entity to cluster representative (e -> e_rep)

 required
promotion_mapping dict[URIRef, URIRef]

Map from representative to promoted URI (e_rep -> e')

 required

Returns:

Type Description
dict[URIRef, URIRef]

Composed mapping (e -> e')
Source code in ontocast/tool/agg/promoter.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def compose_mappings(
    self,
    clustering_mapping: dict[URIRef, URIRef],
    promotion_mapping: dict[URIRef, URIRef],
) -> dict[URIRef, URIRef]:
    """Compose clustering and promotion mappings.

    First, entities are mapped to their cluster representatives (clustering_mapping).
    Then, representatives are promoted to document namespace (promotion_mapping).

    The composed mapping is: e -> promoted(representative(e))

    Args:
        clustering_mapping: Map from entity to cluster representative (e -> e_rep)
        promotion_mapping: Map from representative to promoted URI (e_rep -> e')

    Returns:
        Composed mapping (e -> e')
    """
    composed = {}

    for original, representative in clustering_mapping.items():
        # Look up the promoted version of the representative
        promoted = promotion_mapping.get(representative, representative)
        composed[original] = promoted

    logger.info(
        f"Composed mapping: {len(composed)} entities mapped to "
        f"{len(set(composed.values()))} final URIs"
    )

    return composed

create_promotion_mapping(entities, representations)

Create mapping from original to promoted URIs.

Parameters:

Name Type Description Default
entities list[URIRef]

List of entity URIs to promote

 required
representations dict[URIRef, EntityRepresentation]

Dictionary mapping entities to their representations

 required

Returns:

Type Description
dict[URIRef, URIRef]

Dictionary mapping original URIs to promoted URIs (e -> promoted(e))
Source code in ontocast/tool/agg/promoter.py 
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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def create_promotion_mapping(
    self,
    entities: list[URIRef],
    representations: dict[URIRef, EntityRepresentation],
) -> dict[URIRef, URIRef]:
    """Create mapping from original to promoted URIs.

    Args:
        entities: List of entity URIs to promote
        representations: Dictionary mapping entities to their representations

    Returns:
        Dictionary mapping original URIs to promoted URIs (e -> promoted(e))
    """
    mapping = {}
    promoted_count = 0
    preserved_count = 0

    for entity in entities:
        representation = representations.get(entity)
        if representation is None:
            logger.warning(f"No representation found for entity {entity}")
            mapping[entity] = entity
            continue

        promoted = self.promote_entity(entity, representation)
        mapping[entity] = promoted

        if promoted != entity:
            promoted_count += 1
        else:
            preserved_count += 1

    logger.info(
        f"Created promotion mapping: "
        f"{promoted_count} promoted, "
        f"{preserved_count} preserved (ontology entities)"
    )

    return mapping

promote_entity(entity, representation)

Promote a chunk entity to document namespace.

Parameters:

Name Type Description Default
entity URIRef

Original entity URI

 required
representation EntityRepresentation

Entity representation with metadata

 required

Returns:

Type Description
URIRef

Promoted entity URI
Source code in ontocast/tool/agg/promoter.py 
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
def promote_entity(
    self, entity: URIRef, representation: EntityRepresentation
) -> URIRef:
    """Promote a chunk entity to document namespace.

    Args:
        entity: Original entity URI
        representation: Entity representation with metadata

    Returns:
        Promoted entity URI
    """
    if not self.should_promote(entity):
        # Keep ontology entities unchanged
        return entity

    # Create new URI in document namespace
    # Use normalized form as local name
    local_name = self._clean_local_name(representation.normal_form)

    # Construct promoted URI
    promoted_uri_str = f"{self.doc_namespace}{local_name}"

    # Ensure uniqueness
    unique_uri_str = self._ensure_unique_uri(promoted_uri_str)

    return URIRef(unique_uri_str)

should_promote(entity)

Check if an entity should be promoted to document namespace.

Parameters:

Name Type Description Default
entity URIRef

Entity URI to check

 required

Returns:

Type Description
bool

True if entity should be promoted (is from chunk namespace)
Source code in ontocast/tool/agg/promoter.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def should_promote(self, entity: URIRef) -> bool:
    """Check if an entity should be promoted to document namespace.

    Args:
        entity: Entity URI to check

    Returns:
        True if entity should be promoted (is from chunk namespace)
    """
    entity_str = str(entity)

    # Don't promote ontology entities
    if any(entity_str.startswith(ns) for ns in self.ontology_namespaces):
        return False

    # Promote chunk entities
    if any(entity_str.startswith(ns) for ns in self.chunk_namespaces):
        return True

    # Unknown namespace - be conservative, don't promote
    return False