1
2
3
4
5
6
7
8
9
10
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
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
|
"""
A shallow interface to IVOA vocabularies.
"""
import json
import functools
import os
import time
from urllib import request
from twisted.web.template import tags as T
from gavo import base
from gavo import utils
from gavo.base import meta
IVOA_VOCABULARY_ROOT = "http://www.ivoa.net/rdf/"
def download_file(url, cache, show_progress, http_headers):
"""a partial implementation of astropy.utils.data.download_file
to be used until we can rely on astropy 4.
We actually ignore cache and show_progress...
"""
cache_name = os.path.join(
base.getConfig("cacheDir"),
utils.defuseFileName(url))
if not os.path.exists(cache_name) or cache=="update":
voc_req = request.Request(url,
headers=http_headers)
with request.urlopen(voc_req) as src:
payload = src.read()
# only write when we have all the data; we don't want to
# clobber a cache that's still halfway good.
with open(cache_name, "wb") as dest:
dest.write(payload)
return cache_name
@functools.lru_cache(30)
def get_vocabulary(voc_name, force_update=False):
"""returns an IVOA vocabulary in its "desise" form.
See Vocabularies in the VO 2 to see what is inside of this.
This will use a cache to avoid repeated updates, but it
will attempt to re-download if the cached copy is older than 6 months.
"""
# insert astropy.utils.data below once oldstable has astropy 4.
try:
src_name = download_file(
IVOA_VOCABULARY_ROOT+voc_name,
cache="update" if force_update else True,
show_progress=False,
http_headers={"accept": "application/x-desise+json"})
except Exception as msg:
raise base.ui.logOldExc(base.ReportableError(
"No such vocabulary: %s"%voc_name,
hint="Could not retrieve the vocabulary; failure: %s"%
str(msg)))
if time.time()-os.path.getmtime(src_name)>3600*60*150:
# attempt a re-retrieval, but ignore failure
try:
src_name = download_file(
IVOA_VOCABULARY_ROOT+voc_name,
cache="update", show_progress=False,
http_headers={"accept": "application/x-desise+json"})
except Exception as msg:
base.ui.notifyWarning("Updating cache for the vocabulary"
" {} failed: {}".format(voc_name, msg))
with open(src_name, "r", encoding="utf-8") as f:
return json.load(f)
def get_label(voc, term):
"""returns the label of term if it's in the desise vocabulary voc,
term capitalised otherwise.
"""
if term in voc["terms"]:
return voc["terms"][term]["label"]
else:
return term.capitalize()
@meta.forKeys("subject")
class SubjectMeta(meta.MetaValue):
"""A MetaValue translating UAT terms to their labels in HTML.
This follows the assumption that HTML is what humans look at, anything
else is either computers or nerds.
"""
def _getContentAsHTML(self, content):
try:
uat = get_vocabulary("uat")["terms"]
if content in uat:
return T.a(href="http://www.ivoa.net/rdf/uat#"+content)[
uat[content]["label"]]
except (base.ReportableError, KeyError):
# something went wrong fetching the UAT; just hand through stuff.
pass
return content
if __name__=="__main__":
print(get_vocabulary("datalink/core"))
# vi:et:sw=4:sta
|
