modm_data.owl.store

 1# Copyright 2022, Niklas Hauser
 2# SPDX-License-Identifier: MPL-2.0
 3
 4from ..utils import ext_path
 5import owlready2 as owl
 6import lxml.etree as ET
 7from pathlib import Path
 8
 9
10XSLT_SORT = r"""
11<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform" version="1.0"
12                xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
13                xmlns:xsd="http://www.w3.org/2001/XMLSchema#"
14                xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#"
15                xmlns:owl="http://www.w3.org/2002/07/owl#">
16
17  <xsl:template match="/">
18    <xsl:apply-templates/>
19  </xsl:template>
20
21  <xsl:template match="owl:Ontology">
22    <xsl:copy>
23      <!-- Sort the attributes by name. -->
24      <xsl:for-each select="Package">
25        <xsl:sort select="@rdf:resource(.)"/>
26        <xsl:copy/>
27      </xsl:for-each>
28      <xsl:apply-templates/>
29    </xsl:copy>
30  </xsl:template>
31
32  <xsl:template match="text()|comment()|processing-instruction()">
33    <xsl:copy/>
34  </xsl:template>
35
36</xsl:stylesheet>
37"""
38
39
40class Store:
41    def __init__(self, vendor, device):
42        self.base_url = "https://data.modm.io/kg"
43        self.vendor = vendor
44        self.device = device
45        self._path = ext_path(f"{vendor}/owl")
46        # Add the directory to the search path
47        owl.onto_path.append(self._path)
48        self.ontology = owl.get_ontology(f"{self.base_url}/{vendor}/{device}")
49
50    # def set_device_iri(self, device):
51    #     self.ontology.set_base_iri(f"{self.base_url}/{self.vendor}/{device}", rename_entities=False)
52
53    def namespace(self, name):
54        return self.ontology.get_namespace(f"{self.vendor}/{name}")
55
56    def load(self, name=None):
57        if name is None: name = "ontology"
58        fileobj = open(self._path / f"{name}.owl", "rb")
59        self.ontology.load(only_local=True, fileobj=fileobj, reload=True)
60
61    def save(self, name=None):
62        self._path.mkdir(exist_ok=True, parents=True)
63        if name is None: name = "ontology"
64        file = str(self._path / f"{name}.owl")
65        self.ontology.save(file=file)
66
67        # dom = ET.parse(file)
68        # xslt = ET.XML(XSLT_SORT)
69        # transform = ET.XSLT(xslt)
70        # newdom = transform(dom)
71        # Path(file).write_bytes(ET.tostring(newdom, pretty_print=True))
72
73        # owl.default_world.set_backend(filename=str(self._path / f"{name}.sqlite3"))
74        # owl.default_world.save()
75
76    def clear(self):
77        # self.ontology._destroy_cached_entities()
78        self.ontology.world._destroy_cached_entities()
79        # self.ontology.destroy()
80        # self.ontology = owl.get_ontology(f"{self.base_url}/{self.vendor}")
81
82    def __repr__(self) -> str:
83        return f"Store({self.vendor}/{self.device})"
XSLT_SORT = '\n<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform" version="1.0"\n xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"\n xmlns:xsd="http://www.w3.org/2001/XMLSchema#"\n xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#"\n xmlns:owl="http://www.w3.org/2002/07/owl#">\n\n <xsl:template match="/">\n <xsl:apply-templates/>\n </xsl:template>\n\n <xsl:template match="owl:Ontology">\n <xsl:copy>\n <!-- Sort the attributes by name. -->\n <xsl:for-each select="Package">\n <xsl:sort select="@rdf:resource(.)"/>\n <xsl:copy/>\n </xsl:for-each>\n <xsl:apply-templates/>\n </xsl:copy>\n </xsl:template>\n\n <xsl:template match="text()|comment()|processing-instruction()">\n <xsl:copy/>\n </xsl:template>\n\n</xsl:stylesheet>\n'
class Store:
41class Store:
42    def __init__(self, vendor, device):
43        self.base_url = "https://data.modm.io/kg"
44        self.vendor = vendor
45        self.device = device
46        self._path = ext_path(f"{vendor}/owl")
47        # Add the directory to the search path
48        owl.onto_path.append(self._path)
49        self.ontology = owl.get_ontology(f"{self.base_url}/{vendor}/{device}")
50
51    # def set_device_iri(self, device):
52    #     self.ontology.set_base_iri(f"{self.base_url}/{self.vendor}/{device}", rename_entities=False)
53
54    def namespace(self, name):
55        return self.ontology.get_namespace(f"{self.vendor}/{name}")
56
57    def load(self, name=None):
58        if name is None: name = "ontology"
59        fileobj = open(self._path / f"{name}.owl", "rb")
60        self.ontology.load(only_local=True, fileobj=fileobj, reload=True)
61
62    def save(self, name=None):
63        self._path.mkdir(exist_ok=True, parents=True)
64        if name is None: name = "ontology"
65        file = str(self._path / f"{name}.owl")
66        self.ontology.save(file=file)
67
68        # dom = ET.parse(file)
69        # xslt = ET.XML(XSLT_SORT)
70        # transform = ET.XSLT(xslt)
71        # newdom = transform(dom)
72        # Path(file).write_bytes(ET.tostring(newdom, pretty_print=True))
73
74        # owl.default_world.set_backend(filename=str(self._path / f"{name}.sqlite3"))
75        # owl.default_world.save()
76
77    def clear(self):
78        # self.ontology._destroy_cached_entities()
79        self.ontology.world._destroy_cached_entities()
80        # self.ontology.destroy()
81        # self.ontology = owl.get_ontology(f"{self.base_url}/{self.vendor}")
82
83    def __repr__(self) -> str:
84        return f"Store({self.vendor}/{self.device})"
Store(vendor, device)
42    def __init__(self, vendor, device):
43        self.base_url = "https://data.modm.io/kg"
44        self.vendor = vendor
45        self.device = device
46        self._path = ext_path(f"{vendor}/owl")
47        # Add the directory to the search path
48        owl.onto_path.append(self._path)
49        self.ontology = owl.get_ontology(f"{self.base_url}/{vendor}/{device}")
base_url
vendor
device
ontology
def namespace(self, name):
54    def namespace(self, name):
55        return self.ontology.get_namespace(f"{self.vendor}/{name}")
def load(self, name=None):
57    def load(self, name=None):
58        if name is None: name = "ontology"
59        fileobj = open(self._path / f"{name}.owl", "rb")
60        self.ontology.load(only_local=True, fileobj=fileobj, reload=True)
def save(self, name=None):
62    def save(self, name=None):
63        self._path.mkdir(exist_ok=True, parents=True)
64        if name is None: name = "ontology"
65        file = str(self._path / f"{name}.owl")
66        self.ontology.save(file=file)
67
68        # dom = ET.parse(file)
69        # xslt = ET.XML(XSLT_SORT)
70        # transform = ET.XSLT(xslt)
71        # newdom = transform(dom)
72        # Path(file).write_bytes(ET.tostring(newdom, pretty_print=True))
73
74        # owl.default_world.set_backend(filename=str(self._path / f"{name}.sqlite3"))
75        # owl.default_world.save()
def clear(self):
77    def clear(self):
78        # self.ontology._destroy_cached_entities()
79        self.ontology.world._destroy_cached_entities()
80        # self.ontology.destroy()
81        # self.ontology = owl.get_ontology(f"{self.base_url}/{self.vendor}")