Mercurial > code > home > repos > homeauto
view service/mqtt_to_rdf/rdf_debug.py @ 1726:7d3797ed6681
rough port to starlette and reactivex
author | drewp@bigasterisk.com |
---|---|
date | Tue, 20 Jun 2023 23:14:28 -0700 |
parents | 20474ad4968e |
children |
line wrap: on
line source
import logging from typing import List, Union, cast from rdflib.graph import Graph from rdflib.namespace import Namespace from inference_types import Triple log = logging.getLogger('infer') ROOM = Namespace("http://projects.bigasterisk.com/room/") def graphDump(g: Union[Graph, List[Triple]], oneLine=True): # this is very slow- debug only! if not log.isEnabledFor(logging.DEBUG): return "(skipped dump)" try: if not isinstance(g, Graph): g2 = Graph() g2 += g g = g2 g.bind('', ROOM) g.bind('ex', Namespace('http://example.com/')) lines = g.serialize(format='n3').splitlines() lines = [line for line in lines if not line.startswith('@prefix')] if oneLine: lines = [line.strip() for line in lines] return ' '.join(lines) except TypeError: return repr(g)