Pyling notebook 04
Measurements that explain how to normalize themselves¶
A stream may carry decibels, octaves, pH values, and other logarithmic quantities. Hard-coding every conversion in the consuming Python service couples transport code to domain knowledge and makes provenance difficult to preserve.
This notebook takes a different route. An RDF Message log keeps each reading as a distinct payload, while the maintained QUDT/CDT normalization runtime identifies the unit and performs the conversion. The Python code orchestrates the process; it contains no conversion formulas.
1. Why an RDF Message log?¶
An RDF Message log is an ordered stream of RDF payloads in one document. VERSION "1.2-messages" selects the format and each @message . directive ends one payload.
Those boundaries are semantic, not cosmetic. Every payload is reasoned over independently, blank nodes remain scoped to their message, and an application can consume results incrementally instead of pretending the stream is one ever-growing graph.
pyling.reason_message_stream yields one result per payload. Internally it gives each message an envelope and a named payload formula; later we will bridge that formula to the QUDT rules for the current message only.
2. Bring maintained rules and a real example to the notebook¶
We use the repository's logarithmic QUDT example, including its message log and SHACL shapes. Printing the input makes the four independent readings visible before any reasoning happens.
from urllib.request import Request, urlopen
from rdflib import Graph, Namespace, URIRef
from pyling import reason_message_stream
RAW_ROOT = "https://raw.githubusercontent.com/pietercolpaert/rdfjs-inference-engine/refs/heads/main"
QUDT_RUNTIME_URL = f"{RAW_ROOT}/rules/qudt/qudt-cdt-normalization.runtime.n3"
EXAMPLE_ROOT = f"{RAW_ROOT}/examples/qudt-logarithmic"
def load_text(url):
request = Request(url, headers={"User-Agent": "pyling-notebook"})
with urlopen(request, timeout=30) as response:
return response.read().decode("utf-8")
runtime = load_text(QUDT_RUNTIME_URL)
message_log = load_text(f"{EXAMPLE_ROOT}/input.messages.trig")
shapes_in = load_text(f"{EXAMPLE_ROOT}/shapes-in.n3")
shapes_out = load_text(f"{EXAMPLE_ROOT}/shapes-out.n3")
print(f"Downloaded {len(runtime):,} characters of rules and QUDT data")
print(message_log)
Downloaded 1,625,802 characters of rules and QUDT data @prefix ex: <https://example.org/qudt-logarithmic#> . @prefix cdt: <https://w3id.org/cdt/> . VERSION "1.2-messages" ex:m1 ex:decibelLevel "20 dB"^^cdt:ucum . @message . ex:m2 ex:decibelLevel "10 dB"^^cdt:ucum . @message . ex:m3 ex:octaveSpan "3 octave"^^cdt:ucum . @message . ex:m4 ex:phValue "7 pH"^^cdt:ucum . @message .
3. Let the shapes focus a large knowledge base¶
The runtime contains metadata for thousands of QUDT units. The upstream engine uses the example's SHACL input and output shapes to specialize that projection before reasoning.
We apply the same unit-level preparation here: retain the complete normalization rule kernel, but keep QUDT facts only for units named by sh:unit and sh:hasValue. This is a performance choice, not a change to the conversion semantics.
SH = Namespace("http://www.w3.org/ns/shacl#")
UNIT_ROOT = "http://qudt.org/vocab/unit/"
shape_graph = Graph()
shape_graph.parse(data=shapes_in, format="turtle")
shape_graph.parse(data=shapes_out, format="turtle")
required_units = {
str(obj)
for _, predicate, obj in shape_graph
if predicate in {SH.unit, SH.hasValue}
and isinstance(obj, URIRef)
and str(obj).startswith(UNIT_ROOT)
}
def specialize_runtime(source, units):
marker = "# Precomputed background facts and closure"
kernel, background = source.split(marker, 1)
unit_subject = f"<{UNIT_ROOT}"
def keep(line):
if not line.startswith(unit_subject):
return True
subject = line[1 : line.index(">")]
return subject in units
selected = "\n".join(line for line in background.splitlines() if keep(line))
return f"{kernel}{marker}\n{selected}\n"
qudt_rules = specialize_runtime(runtime, required_units)
unit_names = sorted(unit.removeprefix(UNIT_ROOT) for unit in required_units)
print("Units selected by SHACL:", ", ".join(unit_names))
print(f"Focused runtime: {len(qudt_rules):,} of {len(runtime):,} characters")
assert unit_names == ["DeciB", "OCT", "PH", "UNITLESS"]
Units selected by SHACL: DeciB, OCT, PH, UNITLESS Focused runtime: 73,233 of 1,625,802 characters
4. Expose one payload at a time¶
The bridge rule reads the current message envelope, follows its payload graph, and copies that graph's triples into the reasoning context. Because reason_message_stream starts a separate run for every payload, no measurement can accidentally borrow facts from a neighboring message.
message_bridge = """
@prefix log: <http://www.w3.org/2000/10/swap/log#> .
@prefix eymsg: <https://eyereasoner.github.io/eyeling/vocab/message#> .
{
?envelope eymsg:payloadGraph ?payload .
?payload log:nameOf ?graph .
?graph log:includes { ?subject ?predicate ?object . } .
} => {
?subject ?predicate ?object .
} .
"""
results = list(reason_message_stream({"sources": [qudt_rules, message_bridge, message_log]}))
print(f"Processed {len(results)} RDF messages independently")
assert len(results) == 4
Processed 4 RDF messages independently
5. Recover normalized values as ordinary RDF¶
Every result contains a QUDT QuantityValue, its source literal, numeric value, and normalized unit. The table is generated from those RDF statements, not from positional assumptions about the input text.
QCR = Namespace("https://www.pieter.pm/rdfjs-inference-engine/ns/qudt-inference#")
QUDT = Namespace("http://qudt.org/schema/qudt/")
rows = []
for index, result in enumerate(results, start=1):
graph = result.as_rdflib_graph()
subject, quantity = next(graph.subject_objects(QCR.normalizedQuantity))
source = graph.value(quantity, QCR.sourceLiteral)
value = graph.value(quantity, QUDT.numericValue)
unit = graph.value(quantity, QUDT.unit)
rows.append((index, source, float(value), str(unit).removeprefix(UNIT_ROOT)))
print(f"{'message':<9} {'source':<18} {'normalized':>12} unit")
for index, source, value, unit in rows:
print(f"{index:<9} {str(source):<18} {value:>12g} {unit}")
expected = [100.0, 10.0, 8.0, 1e-7]
for row, wanted in zip(rows, expected):
assert abs(row[2] - wanted) <= max(abs(wanted) * 1e-10, 1e-12)
message source normalized unit 1 20 dB 100 UNITLESS 2 10 dB 10 UNITLESS 3 3 octave 8 UNITLESS 4 7 pH 1e-07 UNITLESS
What the stream now tells us¶
The N3 profile resolved each unit token and applied its logarithmic convention: 20 dB became a power ratio of 100, 10 dB became 10, three octaves became a frequency ratio of 8, and pH 7 became hydrogen-ion activity 1e-7. Every result is a QUDT QuantityValue in unit:UNITLESS and also carries a normalized cdt:ucum literal.
The next question is no longer "what follows from these measurements?" but "is an intended action allowed under a policy and the world as it is now?" Notebook 05 builds that decision as an auditable ODRL FORCE compliance report.