|
1 | 1 | from logging import warning |
2 | | -from typing import Optional, Union, Dict |
| 2 | +from typing import Optional, Union, Dict, List |
3 | 3 |
|
4 | 4 | from rdflib import Namespace, URIRef |
| 5 | +from curies import Converter, Record |
| 6 | + |
| 7 | +class CurieNamespaceCatalog(object): |
| 8 | + catalog: Dict[str, "CurieNamespace"] |
| 9 | + def __init__(self) -> None: |
| 10 | + self.namespaces = [] |
| 11 | + self._converter: Optional[Converter] = None |
| 12 | + |
| 13 | + @property |
| 14 | + def converter(self): |
| 15 | + if not self._converter: |
| 16 | + self._converter = self._buildConverter() |
| 17 | + return self._converter |
| 18 | + |
| 19 | + def _buildConverter(self): |
| 20 | + records = [] |
| 21 | + namespaces_to_treat = self.namespaces[:] |
| 22 | + while len(namespaces_to_treat) > 0: |
| 23 | + ns = namespaces_to_treat.pop(0) |
| 24 | + prefix = ns.prefix |
| 25 | + uri = str(ns) |
| 26 | + all_prefixes = [prefix] |
| 27 | + all_uris = [uri] |
| 28 | + iteration_needed = True |
| 29 | + while iteration_needed: |
| 30 | + iteration_needed = False |
| 31 | + for possible_synonym in namespaces_to_treat[:]: |
| 32 | + if possible_synonym.prefix in all_prefixes: |
| 33 | + all_uris.append(str(possible_synonym)) |
| 34 | + namespaces_to_treat.remove(possible_synonym) |
| 35 | + iteration_needed = True |
| 36 | + if str(possible_synonym) in all_uris: |
| 37 | + all_prefixes.append(possible_synonym.prefix) |
| 38 | + namespaces_to_treat.remove(possible_synonym) |
| 39 | + iteration_needed = True |
| 40 | + records.append(Record(prefix, uri , [x for x in all_prefixes if not x == prefix], [x for x in all_uris if not x == uri])) |
| 41 | + return Converter(records=records) |
5 | 42 |
|
6 | 43 |
|
7 | | -class CurieNamespace(Namespace): |
8 | | - # We would prefer to use curies.Converter here, but there doesn't appear to be any way to build it incrementally |
9 | | - catalog: Dict[str, "CurieNamespace"] = dict() |
10 | 44 |
|
11 | | - @classmethod |
12 | | - def to_curie(cls, uri: Union[str, URIRef]) -> str: |
13 | | - uri = str(uri) |
14 | | - candidate_ns = "" |
15 | | - for prefix, ns in cls.catalog.items(): |
16 | | - if uri.startswith(ns) and len(ns) > len(candidate_ns): |
17 | | - candidate_ns = ns |
18 | | - if candidate_ns: |
19 | | - return candidate_ns.curie(uri[len(candidate_ns):]) |
20 | | - return None |
21 | 45 |
|
| 46 | + def to_curie(self, uri: Union[str, URIRef]) -> str: |
| 47 | + return self.converter.compress(uri) |
| 48 | + |
| 49 | + def to_uri(self, curie: str) -> Optional[URIRef]: |
| 50 | + expanded = self.converter.expand(curie) |
| 51 | + return None if expanded is None else URIRef(expanded) |
| 52 | + |
| 53 | + def add_namespace(self,ns: "CurieNamespace"): |
| 54 | + self.namespaces.append(ns) |
| 55 | + self._converter = None |
| 56 | + |
22 | 57 | @classmethod |
23 | | - def to_uri(cls, curie: str) -> Optional[URIRef]: |
24 | | - prefix, localname = curie.split(':', 1) |
25 | | - ns = CurieNamespace.catalog.get(prefix, None) |
26 | | - return ns[localname] if ns else None |
| 58 | + def create(cls, *namespaces: List["CurieNamespace"]): |
| 59 | + cat = CurieNamespaceCatalog() |
| 60 | + [cat.add_namespace(x) for x in namespaces] |
| 61 | + return cat |
| 62 | + |
| 63 | + def clear(self): |
| 64 | + self.catalog = dict() |
| 65 | + |
| 66 | + def as_dict(self): |
| 67 | + return self.catalog.copy() |
| 68 | + |
27 | 69 |
|
28 | | - def __new__(cls, prefix: str, ns: Union[str, bytes, URIRef]) -> "CurieNamespace": |
| 70 | +class CurieNamespace(Namespace): |
| 71 | + def __new__(cls, prefix: str, ns: Union[str, URIRef]): |
29 | 72 | rt = Namespace.__new__(cls, str(ns) if not isinstance(ns, bytes) else ns) |
30 | 73 | rt.prefix = prefix |
31 | | - if prefix in CurieNamespace.catalog: |
32 | | - if CurieNamespace.catalog[prefix] != str(rt): |
33 | | - # prefix is bound to a different namespace |
34 | | - warning(f"Prefix: {prefix} already references {CurieNamespace.catalog[prefix]} - not updated to {rt}") |
35 | | - else: |
36 | | - CurieNamespace.catalog[prefix] = rt |
37 | 74 | return rt |
38 | 75 |
|
39 | 76 | def curie(self, reference: Optional[str] = '') -> str: |
40 | 77 | return self.prefix + ':' + reference |
| 78 | + |
| 79 | + def addTo(self, catalog: CurieNamespaceCatalog) -> "CurieNamespace": |
| 80 | + catalog.add_namespace(self) |
| 81 | + return self |
0 commit comments