Mercurial > code > home > repos > homeauto
view service/reasoning/inference.py @ 1059:faf1f8677a91
pi device support for motionStart edge trigger
Ignore-this: 1eda3d0bed8def055573fdc1db3cd7c3
darcs-hash:b77ef0bc873415b7319e8dff11d484d48f4bd10e
author | drewp <drewp@bigasterisk.com> |
---|---|
date | Mon, 21 Mar 2016 04:18:45 -0700 |
parents | 2fa7daf06f34 |
children | 9728288c7f2f |
line wrap: on
line source
""" see ./reasoning for usage """ import sys try: from rdflib import StringInputSource from rdflib.Graph import Graph except ImportError: from rdflib.parser import StringInputSource from rdflib import Graph sys.path.append("/my/proj/room/fuxi/build/lib.linux-x86_64-2.6") from FuXi.Rete.Util import generateTokenSet from FuXi.Rete import ReteNetwork from rdflib import plugin from rdflib.store import Store def infer(graph, rules): """ returns new graph of inferred statements """ # based on fuxi/tools/rdfpipe.py store = plugin.get('IOMemory',Store)() store.open('') target = Graph() tokenSet = generateTokenSet(graph) network = ReteNetwork(rules, inferredTarget=target) network.feedFactsToAdd(tokenSet) store.rollback() return target import time, logging log = logging.getLogger() def logTime(func): def inner(*args, **kw): t1 = time.time() try: ret = func(*args, **kw) finally: log.info("Call to %s took %.1f ms" % ( func.__name__, 1000 * (time.time() - t1))) return ret return inner