view lib/stategraph.py @ 1421:c451aad69f65

output-widgets.html path Ignore-this: 3e3625215c69ee5600c20d42d3afaabc darcs-hash:06ed2ea6a38e6fa7a40a36f3bfc93b22522a49c8
author drewp <drewp@bigasterisk.com>
date Sun, 04 Aug 2019 13:03:04 -0700
parents b9c6b8724d43
children
line wrap: on
line source

import datetime, os, inspect
from dateutil.tz import tzlocal
from rdflib import Graph, Namespace, Literal
DCTERMS = Namespace("http://purl.org/dc/terms/")

class StateGraph(object):
    """
    helper to create a graph with some of the current state of the world
    """
    def __init__(self, ctx):
        """
        note that we put the time of the __init__ call into the graph
        as its dcterms:modified time.
        """
        self.g = Graph()
        self.ctx = ctx

        try:
            requestingFile = inspect.stack()[1][1]
            self.g.add((ctx, DCTERMS['creator'], 
                        Literal(os.path.abspath(requestingFile))))
        except IndexError:
            pass
        self.g.add((ctx, DCTERMS['modified'],
               Literal(datetime.datetime.now(tzlocal()))))

    def add(self, *args, **kw):
        self.g.add(*args, **kw)

    def ntLines(self):
        nt = self.g.serialize(format='nt')
        # this canonical order is just for debugging, so the lines are
        # stable when you refresh the file repeatedly
        return sorted(filter(None, nt.splitlines()))
        
    def asTrig(self):
        return "%s {\n%s\n}\n" % (self.ctx.n3(), '\n'.join(self.ntLines()))

    def asJsonLd(self):
        return self.g.serialize(format='json-ld')

    def asAccepted(self, acceptHeader):
        if acceptHeader == 'application/nquads':
            return 'application/nquads', '\n'.join(
                line.strip().rstrip('.') + '%s .' % self.ctx.n3()
                for line in self.ntLines())
        else:
            return 'application/x-trig', self.asTrig()