pipeline to check class constraints

.gitignore

@@ -3,3 +3,4 @@ node_modules
 *.tgz
 *.d.ts
 *.js.map
+temp

packages/cube/lib/classConstraint.js

@@ -0,0 +1,60 @@
+import { Readable, Duplex } from 'stream'
+import { sort, compareOn, createStore } from 'external-merge-sort'
+import through2 from 'through2'
+
+// generic enough to be moved to a separate utils package
+const compositeComparer = (...comparers) => {
+  return (x, y) => {
+    for (const comparer of comparers) {
+      const result = comparer(x, y)
+      if (result !== 0) {
+        return result
+      }
+    }
+    return 0
+  }
+}
+
+export function choose(quad) {
+  const predicate = this.env.namedNode(this.variables.get('path'))
+  const classNode = this.env.namedNode(this.variables.get('class'))
+  if (predicate.equals(quad.predicate)) {
+    return this.env.quad(quad.subject, quad.predicate, quad.object, quad.object)
+  }
+  if (classNode.equals(quad.object) && this.env.ns.rdf.type.equals(quad.predicate)) {
+    return this.env.quad(quad.subject, quad.predicate, quad.object, quad.subject)
+  }
+}
+
+export function sortByGraph(sortChunkSize) {
+  const write = async (chunk, filename) => {
+    await this.env.toFile(Readable.from(chunk), filename)
+    return this.env.fromFile(filename)
+  }
+
+  const comparer = compositeComparer(
+    compareOn(quad => quad.graph.value),
+    compareOn(quad => this.env.ns.rdf.type.equals(quad.predicate) ? 0 : 1))
+
+  const store = createStore(write, '.nt')
+  const maxSize = Number(sortChunkSize)
+
+  const stream = Duplex.from(iterable => sort(iterable, { comparer, store, maxSize }))
+  stream.on('finish', store.dispose)
+  stream.on('error', store.dispose)
+  return stream
+}
+
+export function check() {
+  const a = this.env.ns.rdf.type
+  const t = through2.obj(function (chunk, _encoding, callback) {
+    if (chunk.predicate.equals(a)) {
+      t.current = chunk
+    } else if (!chunk.object.equals(t.current?.subject)) {
+      this.push(chunk)
+    }
+    callback()
+  })
+
+  return t
+}

packages/cube/manifest.ttl

@@ -48,3 +48,10 @@
   rdfs:label "Validate input observations against cube constraint" ;
   b59:source "barnard59-cube/pipeline/check-observations.ttl" ;
 .
+
+<command/cube/check-class>
+  a b59:CliCommand ;
+  b59:command "check-class" ;
+  rdfs:label "Validate input observations against a class constraint" ;
+  b59:source "barnard59-cube/pipeline/check-class.ttl" ;
+.

packages/cube/pipeline/check-class.ttl

@@ -0,0 +1,70 @@
+@prefix code: <https://code.described.at/> .
+@prefix p: <https://pipeline.described.at/> .
+@prefix shacl: <https://barnard59.zazuko.com/operations/shacl/> .
+@prefix base: <https://barnard59.zazuko.com/operations/base/> .
+@prefix getDataset: <https://barnard59.zazuko.com/operations/rdf/getDataset> .
+@prefix splitDataset: <https://barnard59.zazuko.com/operations/rdf/splitDataset/> .
+@prefix n3: <https://barnard59.zazuko.com/operations/formats/n3/> .
+@prefix ntriples: <https://barnard59.zazuko.com/operations/formats/ntriples/> .
+@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
+@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
+
+@base <http://barnard59.zazuko.com/pipeline/cube-validation/> .
+
+_:path a p:Variable ;
+  p:name "path" ;
+  rdfs:label "property whose range is class".
+
+_:class a p:Variable ;
+  p:name "class" ;
+  rdfs:label "range class of path".
+
+_:maxViolations a p:Variable ;
+  p:name "maxViolations" ;
+  rdfs:label "max number of violations" ;
+  p:value 500 ;
+.
+
+_:sortChunkSize a p:Variable ;
+  p:name "sortChunkSize" ;
+  rdfs:label "sort chunk size" ;
+  p:value 100000 ;
+.
+
+<check-class> a p:Pipeline , p:Readable;
+  p:variables [ p:variable _:path, _:class, _:maxViolations, _:sortChunkSize ] ;
+  p:steps
+    [
+      p:stepList
+      (
+        [ base:stdin () ]
+        [ n3:parse () ]
+        _:choose
+        _:sortByGraph
+        _:check
+        [ base:limit ("maxViolations"^^p:VariableName)]
+        [ ntriples:serialize () ]
+      )
+    ]
+.
+
+_:choose base:map (
+  [
+    a code:EcmaScriptModule ;
+    code:link <file:../lib/classConstraint.js#choose>
+  ] 
+) .
+
+_:sortByGraph a p:Step ;
+  code:implementedBy [ a code:EcmaScriptModule ;
+    code:link <file:../lib/classConstraint.js#sortByGraph>
+  ] ;
+  code:arguments ("sortChunkSize"^^p:VariableName)
+.
+
+_:check a p:Step ;
+  code:implementedBy [ a code:EcmaScriptModule ;
+    code:link <file:../lib/classConstraint.js#check>
+  ] 
+.
+

packages/shacl/report.js

@@ -5,6 +5,25 @@ import SHACLValidator from 'rdf-validate-shacl'
 async function * validate(ds, maxViolations, iterable) {
   let totalViolations = 0
 
+  const removeClassConstraints = ds => {
+    const properties = this.env.clownface({ dataset: ds, term: this.env.ns.cube.Constraint })
+      .in(this.env.ns.rdf.type)
+      .out(this.env.ns.sh.property)
+    const classConstraints = properties.has(this.env.ns.sh.class)
+    if (classConstraints.terms.length > 0) {
+      const cmds = classConstraints.map(c => {
+        const path = c.out(this.env.ns.sh.path).term.value
+        const classTerm = c.out(this.env.ns.sh.class).term.value
+        return `\n\tbarnard59 cube check-class --path ${path} --class ${classTerm}`
+      })
+      this.logger.warn(`Class constraints are not supported. Use the check-class command instead: ${cmds.join('')}`)
+    }
+
+    properties.deleteOut(this.env.ns.sh.class)
+  }
+  // consider removing class constraints only if batch size is > 0
+  removeClassConstraints(ds)
+
   for await (const chunk of iterable) {
     if (maxViolations && totalViolations > maxViolations) {
       this.logger.warn('Exceeded max violations. Aborting')