Source code for gavo.protocols.vocabularies

"""
A shallow interface to IVOA vocabularies.
"""

#c Copyright 2008-2023, the GAVO project <gavo@ari.uni-heidelberg.de>
#c
#c This program is free software, covered by the GNU GPL.  See the
#c COPYING file in the source distribution.


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/"


[docs]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
[docs]@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)
[docs]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()
[docs]@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