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


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