Mercurial > code > home > repos > homeauto
view service/reasoning/graphop.py @ 696:c52b172c0824
add publish to ON/OFF messages. split up the main statement handler
Ignore-this: b1ec515e6873a6841b1c31c3fb7a2a36
author | drewp@bigasterisk.com |
---|---|
date | Fri, 31 Jan 2020 23:55:27 -0800 |
parents | 0f2ac014d9ea |
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