Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/schema_salad/jsonld_context.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 from __future__ import absolute_import
2
3 import logging
4 from typing import (
5 Any,
6 Dict,
7 Iterable,
8 List,
9 MutableMapping,
10 MutableSequence,
11 Optional,
12 Tuple,
13 Union,
14 cast,
15 )
16
17 import rdflib
18 import rdflib.namespace
19 import six
20 from rdflib import Graph, URIRef
21 from rdflib.namespace import RDF, RDFS
22 from six.moves import urllib
23 from typing_extensions import Text # pylint: disable=unused-import
24
25 from .exceptions import SchemaException
26 from .ref_resolver import ContextType # pylint: disable=unused-import
27 from .utils import aslist, json_dumps
28
29 # move to a regular typing import when Python 3.3-3.6 is no longer supported
30
31
32 _logger = logging.getLogger("salad")
33
34
35 def pred(
36 datatype, # type: MutableMapping[Text, Union[Dict[Text, Text], Text]]
37 field, # type: Optional[Dict[Text, Any]]
38 name, # type: str
39 context, # type: ContextType
40 defaultBase, # type: str
41 namespaces, # type: Dict[Text, rdflib.namespace.Namespace]
42 ): # type: (...) -> Union[Dict[Text, Text], Text]
43 split = urllib.parse.urlsplit(name)
44
45 vee = None # type: Optional[Text]
46
47 if split.scheme != "":
48 vee = name
49 (ns, ln) = rdflib.namespace.split_uri(six.text_type(vee))
50 name = ln
51 if ns[0:-1] in namespaces:
52 vee = six.text_type(namespaces[ns[0:-1]][ln])
53 _logger.debug("name, v %s %s", name, vee)
54
55 v = None # type: Optional[Dict[Text, Any]]
56
57 if field is not None and "jsonldPredicate" in field:
58 if isinstance(field["jsonldPredicate"], MutableMapping):
59 v = {}
60 for k, val in field["jsonldPredicate"].items():
61 v[("@" + k[1:] if k.startswith("_") else k)] = val
62 if "@id" not in v:
63 v["@id"] = vee
64 else:
65 v = field["jsonldPredicate"]
66 elif "jsonldPredicate" in datatype:
67 if isinstance(datatype["jsonldPredicate"], Iterable):
68 for d in datatype["jsonldPredicate"]:
69 if isinstance(d, MutableMapping):
70 if d["symbol"] == name:
71 v = d["predicate"]
72 else:
73 raise SchemaException(
74 "entries in the jsonldPredicate List must be " "Dictionaries"
75 )
76 else:
77 raise SchemaException("jsonldPredicate must be a List of Dictionaries.")
78
79 ret = v or vee
80
81 if not ret:
82 ret = defaultBase + name
83
84 if name in context:
85 if context[name] != ret:
86 raise SchemaException(
87 "Predicate collision on {}, '{}' != '{}'".format(
88 name, context[name], ret
89 )
90 )
91 else:
92 _logger.debug("Adding to context '%s' %s (%s)", name, ret, type(ret))
93 context[name] = ret
94
95 return ret
96
97
98 def process_type(
99 t, # type: MutableMapping[Text, Any]
100 g, # type: Graph
101 context, # type: ContextType
102 defaultBase, # type: str
103 namespaces, # type: Dict[Text, rdflib.namespace.Namespace]
104 defaultPrefix, # type: str
105 ): # type: (...) -> None
106 if t["type"] not in ("record", "enum"):
107 return
108
109 if "name" in t:
110 recordname = t["name"]
111
112 _logger.debug("Processing %s %s\n", t.get("type"), t)
113
114 classnode = URIRef(recordname)
115 g.add((classnode, RDF.type, RDFS.Class))
116
117 split = urllib.parse.urlsplit(recordname)
118 predicate = recordname
119 if t.get("inVocab", True):
120 if split.scheme:
121 (ns, ln) = rdflib.namespace.split_uri(six.text_type(recordname))
122 predicate = recordname
123 recordname = ln
124 else:
125 predicate = "{}:{}".format(defaultPrefix, recordname)
126
127 if context.get(recordname, predicate) != predicate:
128 raise SchemaException(
129 "Predicate collision on '{}', '{}' != '{}'".format(
130 recordname, context[recordname], predicate
131 )
132 )
133
134 if not recordname:
135 raise SchemaException("Unable to find/derive recordname for {}".format(t))
136
137 _logger.debug(
138 "Adding to context '%s' %s (%s)", recordname, predicate, type(predicate)
139 )
140 context[recordname] = predicate
141
142 if t["type"] == "record":
143 for i in t.get("fields", []):
144 fieldname = i["name"]
145
146 _logger.debug("Processing field %s", i)
147
148 v = pred(
149 t, i, fieldname, context, defaultPrefix, namespaces
150 ) # type: Union[Dict[Any, Any], Text, None]
151
152 if isinstance(v, six.string_types):
153 v = v if v[0] != "@" else None
154 elif v is not None:
155 v = v["_@id"] if v.get("_@id", "@")[0] != "@" else None
156
157 if bool(v):
158 (ns, ln) = rdflib.namespace.split_uri(six.text_type(v))
159 if ns[0:-1] in namespaces:
160 propnode = namespaces[ns[0:-1]][ln]
161 else:
162 propnode = URIRef(v)
163
164 g.add((propnode, RDF.type, RDF.Property))
165 g.add((propnode, RDFS.domain, classnode))
166
167 # TODO generate range from datatype.
168
169 if isinstance(i["type"], MutableMapping):
170 process_type(
171 i["type"], g, context, defaultBase, namespaces, defaultPrefix
172 )
173
174 if "extends" in t:
175 for e in aslist(t["extends"]):
176 g.add((classnode, RDFS.subClassOf, URIRef(e)))
177 elif t["type"] == "enum":
178 _logger.debug("Processing enum %s", t.get("name"))
179
180 for i in t["symbols"]:
181 pred(t, None, i, context, defaultBase, namespaces)
182
183
184 def salad_to_jsonld_context(
185 j, # type: Iterable[MutableMapping[Text, Any]]
186 schema_ctx, # type: MutableMapping[Text, Any]
187 ): # type: (...) -> Tuple[ContextType, Graph]
188 context = {} # type: ContextType
189 namespaces = {}
190 g = Graph()
191 defaultPrefix = ""
192
193 for k, v in schema_ctx.items():
194 context[k] = v
195 namespaces[k] = rdflib.namespace.Namespace(v)
196
197 if "@base" in context:
198 defaultBase = cast(str, context["@base"])
199 del context["@base"]
200 else:
201 defaultBase = ""
202
203 for k, v in namespaces.items():
204 g.bind(str(k), v)
205
206 for t in j:
207 process_type(t, g, context, defaultBase, namespaces, defaultPrefix)
208
209 return (context, g)
210
211
212 def fix_jsonld_ids(
213 obj, # type: Union[List[Dict[Text, Any]], MutableMapping[Text, Any]]
214 ids, # type: List[Text]
215 ): # type: (...) -> None
216 if isinstance(obj, MutableMapping):
217 for i in ids:
218 if i in obj:
219 obj["@id"] = obj[i]
220 for v in obj.values():
221 fix_jsonld_ids(v, ids)
222 if isinstance(obj, MutableSequence):
223 for entry in obj:
224 fix_jsonld_ids(entry, ids)
225
226
227 def makerdf(
228 workflow, # type: Text
229 wf, # type: Union[List[Dict[Text, Any]], MutableMapping[Text, Any]]
230 ctx, # type: ContextType
231 graph=None, # type: Optional[Graph]
232 ): # type: (...) -> Graph
233 prefixes = {}
234 idfields = []
235 for k, v in six.iteritems(ctx):
236 if isinstance(v, MutableMapping):
237 url = v["@id"]
238 else:
239 url = v
240 if url == "@id":
241 idfields.append(k)
242 doc_url, frg = urllib.parse.urldefrag(url)
243 if "/" in frg:
244 p = frg.split("/")[0]
245 prefixes[p] = u"{}#{}/".format(doc_url, p)
246
247 fix_jsonld_ids(wf, idfields)
248
249 if graph is None:
250 g = Graph()
251 else:
252 g = graph
253
254 if isinstance(wf, MutableSequence):
255 for w in wf:
256 w["@context"] = ctx
257 g.parse(data=json_dumps(w), format="json-ld", publicID=str(workflow))
258 else:
259 wf["@context"] = ctx
260 g.parse(data=json_dumps(wf), format="json-ld", publicID=str(workflow))
261
262 # Bug in json-ld loader causes @id fields to be added to the graph
263 for sub, pred, obj in g.triples((None, URIRef("@id"), None)):
264 g.remove((sub, pred, obj))
265
266 for k2, v2 in six.iteritems(prefixes):
267 g.namespace_manager.bind(k2, v2)
268
269 return g