Mercurial > code > home > repos > homeauto
view service/reasoning/graphop.py @ 1111:aa70001ea0c9
sse_collector stats page
Ignore-this: 73b6ac46c87bbcca85e7d1ca9127a61c
darcs-hash:102cd65f76f723fb0f0d1b74a035b809daac6895
author | drewp <drewp@bigasterisk.com> |
---|---|
date | Fri, 16 Sep 2016 01:21:19 -0700 |
parents | 5da9200418db |
children | f3f667769aef |
line wrap: on
line source
import logging from rdflib import URIRef, ConjunctiveGraph log = logging.getLogger() def graphWithoutMetadata(g, ignorePredicates=[]): """ graph filter that removes any statements whose subjects are contexts in the graph and also any statements with the given predicates """ ctxs = [ctx.identifier for ctx in g.contexts()] out = ConjunctiveGraph() for stmt in g.quads((None, None, None)): if stmt[0] not in ctxs and stmt[1] not in ignorePredicates: out.addN([stmt]) return out def graphEqual(a, b, ignorePredicates=[]): """ compare graphs, omitting any metadata statements about contexts (especially modification times) and also any statements using the given predicates """ stmtsA = set(graphWithoutMetadata(a, ignorePredicates)) stmtsB = set(graphWithoutMetadata(b, ignorePredicates)) if stmtsA == stmtsB: return True if log.getEffectiveLevel() <= logging.INFO: lost = stmtsA - stmtsB if lost: log.info("graph A only:") for s in lost: log.info(" %s", s) new = stmtsB - stmtsA if new: log.info("graph B only:") for s in new: log.info(" %s", s) return False