from typing import Callable, Dict
import rdflib
from buildingmotif.namespaces import BACNET, BRICK, PARAM, RDF, REF
from buildingmotif.utils import new_temporary_graph
RULES: Dict[str, Callable] = {}
"""
Notes:
- maybe more useful for authoring than for distribution
- "compile" into template definitions that then get distributed
- standardize the "subject" of the template: 'name' param
- have rules generate dependencies rather than introducing additional subjects?
- give the rules/macros access to the library and/or the template they
are associated with?:
- new signature: def foo(library, template, *args, **kwargs) -> rdflib.Graph
- don't hardcode template names into these rules/macros --- pass in from user
"""
[docs]def rule(name: str):
"""This decorator is applied to a function in order to register that
function as a rule that can be executed during template compilation.
:param name: name
:type name: str
"""
def wrapper(func: Callable):
RULES[name] = func
return func
return wrapper
[docs]@rule("body")
def body(*args, **kwargs) -> rdflib.Graph:
"""Returns the first value of the `body` argument as the template body.
:return: graph of first value of `body` argument
:rtype: rdflib.Graph
"""
G = new_temporary_graph()
G.parse(data=args[0], format="turtle")
return G
[docs]@rule("type")
def hasType(*args, **kwargs) -> rdflib.Graph:
"""_summary_
:return: _description_
:rtype: rdflib.Graph
"""
G = new_temporary_graph()
for arg in args:
arg = rdflib.URIRef(arg)
G.add((PARAM["name"], RDF["type"], arg))
return G
[docs]@rule("hasPoint")
def points(*args, **kwargs) -> rdflib.Graph:
"""_summary_
:return: _description_
:rtype: rdflib.Graph
"""
G = new_temporary_graph()
for param, pointtype in kwargs.items():
pointtype = rdflib.URIRef(pointtype)
G.add(
(PARAM[param], RDF["type"], pointtype)
) # should really be a dependency on that type
G.add((PARAM["name"], BRICK.hasPoint, PARAM[param]))
return G
[docs]@rule("hasPart")
def parts(*args, **kwargs) -> rdflib.Graph:
"""_summary_
:return: _description_
:rtype: rdflib.Graph
"""
G = new_temporary_graph()
for param, pointtype in kwargs.items():
pointtype = rdflib.URIRef(pointtype)
G.add(
(PARAM[param], RDF["type"], pointtype)
) # should really be a dependency on that type
G.add((PARAM["name"], BRICK.hasPart, PARAM[param]))
return G
[docs]@rule("upstream")
def isFedBy(*args, **kwargs) -> rdflib.Graph:
"""_summary_
:return: _description_
:rtype: rdflib.Graph
"""
G = new_temporary_graph()
assert kwargs is not None
for param, equiptype in kwargs.items():
equiptype = rdflib.URIRef(equiptype)
G.add((PARAM[param], RDF["type"], equiptype))
G.add((PARAM["name"], BRICK.isFedBy, PARAM[param]))
return G
[docs]@rule("downstream")
def feeds(*args, **kwargs) -> rdflib.Graph:
"""_summary_
:return: _description_
:rtype: rdflib.Graph
"""
G = new_temporary_graph()
assert kwargs is not None
for param, equiptype in kwargs.items():
equiptype = rdflib.URIRef(equiptype)
G.add((PARAM[param], RDF["type"], equiptype))
G.add((PARAM["name"], BRICK.feeds, PARAM[param]))
return G
[docs]@rule("bacnet")
def bacnet(*args, **kwargs) -> rdflib.Graph:
"""Adds template fields for each of the param names in `args` to ensure
they have BACnet references.
:return: a graph with BACnet references added
:rtype: rdflib.Graph
"""
G = new_temporary_graph()
G.add((PARAM["BACnet_device"], RDF["type"], BACNET.BACnetDevice))
for param in args:
ref_name = rdflib.BNode()
object_name = PARAM[param + "_object"]
G.add((PARAM[param], REF.hasExternalReference, ref_name))
G.add((ref_name, BACNET["object-identifier"], object_name))
G.add((ref_name, BACNET["objectOf"], PARAM["BACnet_device"]))
return G
[docs]def compile_template_spec(spec: Dict) -> Dict:
"""Compiles a template specification into the body of a template by
applying rules defined in this file and others.
:param spec: the Python dictionary that is the result of decoding the
YAML structure (usually a dictionary or list) defining the
template
:type spec: Dict
:raises ValueError: if unknown rule
:return: the template spec containing the compiled body
:rtype: Dict
"""
# extract metadata that doesn't correspond to rule executions
deps = spec.pop("dependencies", [])
optionals = spec.pop("optional", [])
# compile the template's body from the rules
body = new_temporary_graph()
existing_body = spec.pop("body", None)
if existing_body is not None:
body.parse(data=existing_body, format="turtle")
for rule_name in tuple(spec.keys()):
rule_args = spec.pop(rule_name)
if rule_name not in RULES:
raise ValueError(f"Unknown rule {rule_name}")
if isinstance(rule_args, dict):
G = RULES[rule_name](**rule_args)
elif isinstance(rule_args, list):
G = RULES[rule_name](*rule_args)
else:
G = RULES[rule_name](*[rule_args])
body += G
# put the metadata back
spec["dependencies"] = deps
spec["body"] = body
spec["optional"] = optionals
return spec
