cube package

.changeset/silver-humans-joke.md

@@ -0,0 +1,6 @@
+---
+"barnard59-cube": major
+"barnard59-rdf": major
+---
+
+Move cube operations from rdf package to the new cube package

codecov.yml

@@ -6,6 +6,7 @@ flag_management:
     - name: barnard59-base
     - name: barnard59-core
     - name: barnard59-csvw
+    - name: barnard59-cube
     - name: barnard59-formats
     - name: barnard59-ftp
     - name: barnard59-graph-store

packages/cli/lib/discoverManifests.js

@@ -10,6 +10,7 @@ const require = module.createRequire(import.meta.url)
 export default async function * () {
   const packages = findPlugins({
     includeDev: true,
+    includePeer: true,
     filter({ pkg }) {
       return packagePattern.test(pkg.name) && hasManifest(pkg.name)
     },
@@ -19,7 +20,7 @@ export default async function * () {
   if (hasManifest(dir)) {
     const { name, version } = require(`${dir}/package.json`)
     yield {
-      name,
+      name: packagePattern.test(name) ? name.match(packagePattern)[1] : name,
       manifest: rdf.clownface({ dataset: await rdf.dataset().import(rdf.fromFile(`${dir}/manifest.ttl`)) }),
       version,
     }

packages/cli/lib/pipeline.js

@@ -34,7 +34,7 @@ export const desugar = async (dataset, { logger, knownOperations } = {}) => {
       const [quad] = step.dataset.match(step.term)
       const knownStep = knownOperations.get(quad?.predicate)
       if (!knownStep) {
-        logger?.warn(`Operation <${quad?.predicate.value}> not found in known manifests. Have you added the right \`branard59-*\` package as dependency?`)
+        logger?.warn(`Operation <${quad?.predicate.value}> not found in known manifests. Have you added the right \`barnard59-*\` package as dependency?`)
         continue
       }
 

packages/cube/README.md

@@ -0,0 +1,90 @@
+# barnard59-cube
+
+This package provides operations and commands for RDF cubes in Barnard59 Linked Data pipelines.
+The `manifest.ttl` file contains a full list of all operations included in this package. 
+
+## Operations
+
+### `cube/buildCubeShape`
+
+TBD
+
+### `cube/toObservation`
+
+TBD
+
+
+## Commands
+
+## Cube validation
+
+`cube-validation.ttl` contains pipelines to retrieve and validate cube observations and their constraints.
+
+### fetch cube constraint
+
+Pipeline `fetch-cube-constraint` queries a given SPARQL endpoint (default is https://lindas.admin.ch/query) to retrieve 
+a [concise bounded description](https://docs.stardog.com/query-stardog/#describe-queries) of the `cube:Constraint` part of a given cube.
+
+```bash
+npx barnard59 run ./pipeline/cube-validation.ttl \
+    --pipeline http://barnard59.zazuko.com/pipeline/cube-validation/fetch-cube-constraint \
+    --variable cube=https://agriculture.ld.admin.ch/agroscope/PRIFm8t15/2 \
+    --variable endpoint=https://int.lindas.admin.ch/query 
+```
+
+Taking advantage of [package-specific commands](https://data-centric.zazuko.com/docs/workflows/reference/cli/#package-specific-commands), we can express the same as:
+
+```bash
+npx barnard59 cube fetch-constraint \
+  --cube https://agriculture.ld.admin.ch/agroscope/PRIFm8t15/2 \
+  --endpoint https://int.lindas.admin.ch/query
+```
+
+
+This pipeline is useful mainly for cubes published with [cube creator](https://github.com/zazuko/cube-creator) (if the cube definition is manually crafted, likely it's already available as a local file).
+
+
+### check cube constraint
+
+Pipeline `check-cube-constraint` validates the input constraint against the shapes provided with the `profile` variable (the default profile is https://cube.link/latest/shape/standalone-constraint-constraint).
+
+The pipeline reads the constraint from `stdin`, allowing input from a local file (as in the following example) as well as from the output of the `fetch-cube-constraint` pipeline (in most cases it's useful to have the constraint in a local file because it's needed also for the `check-cube-observations` pipeline).
+
+```bash
+cat myConstraint.ttl \
+| npx barnard59 cube check-constraint \
+    --profile https://cube.link/v0.1.0/shape/standalone-constraint-constraint
+```
+SHACL reports for violations are written to `stdout`.
+
+
+### fetch cube observations
+
+Pipeline `fetch-cube-observations` queries a given SPARQL endpoint (default is https://lindas.admin.ch/query) to retrieve the observations of a given cube.
+
+```bash
+npx barnard59 cube fetch-observations \
+    --cube https://agriculture.ld.admin.ch/agroscope/PRIFm8t15/2 \
+    --endpoint https://int.lindas.admin.ch/query
+```
+Results are written to `stdout`.
+
+### check cube observations
+
+Pipeline `check-cube-observations` validates the input observations against the shapes provided with the `constraint` variable.
+
+The pipeline reads the observations from `stdin`, allowing input from a local file (as in the following example) as well as from the output of the `fetch-cube-observations` pipeline.
+
+```bash
+cat myObservations.ttl \
+| npx barnard59 cube check-observations \
+    --constraint myConstraint.ttl
+```
+
+To enable validation, the pipeline adds to the constraint a `sh:targetClass` property with value `cube:Observation` (assuming that each observation has an explicit property `rdf:type` with value `cube:Observation`).
+
+To leverage streaming, input is split and validated in little batches of adjustable size (the default is 50 and likely it's appropriate in most cases). This allows the validation of very big cubes because observations are not loaded in memory all at once. To ensure triples for the same observation are adjacent (hence processed in the same batch), the input is sorted by subject (and in case the input is large the sorting step relies on temporary local files).
+
+SHACL reports for violations are written to `stdout`.
+
+To limit the output size, there is also a `maxViolations` option to stop validation when the given number of violations is reached.

packages/cube/lib/batch.js

@@ -0,0 +1,30 @@
+import { Duplex } from 'stream'
+import rdf from '@zazuko/env-node'
+
+// Iterable<X> => Iterable<X[]>
+export async function * chunkObjectsBySize(size, iterable) {
+  let chunk = []
+  for await (const item of iterable) {
+    chunk.push(item)
+    if (chunk.length === size) {
+      yield chunk
+      chunk = []
+    }
+  }
+  if (chunk.length > 0) {
+    yield chunk
+  }
+}
+
+// Iterable<Dataset> => Iterable<Dataset>
+export async function * chunkBySize(size, iterable) {
+  for await (const array of chunkObjectsBySize(size, iterable)) {
+    const batch = rdf.dataset()
+    for (const dataset of array) {
+      batch.addAll(dataset)
+    }
+    yield batch
+  }
+}
+
+export const batch = size => Duplex.from(iterable => chunkBySize(Number(size), iterable))

packages/rdf/lib/cbdCopy.js → packages/cube/lib/cbdCopy.js

@@ -1,4 +1,4 @@
-import rdf from '@zazuko/env'
+import rdf from '@zazuko/env-node'
 
 function cbdCopy(source, target, { ignore = rdf.termSet() } = {}) {
   for (const quad of source.dataset.match(source.term)) {

packages/rdf/lib/cube/buildCubeShape/Cube.js → packages/cube/lib/cube/buildCubeShape/Cube.js

@@ -1,4 +1,4 @@
-import rdf from '@zazuko/env'
+import rdf from '@zazuko/env-node'
 import addAll from 'rdf-dataset-ext/addAll.js'
 import cbdCopy from '../../cbdCopy.js'
 import * as ns from '../../namespaces.js'

packages/rdf/lib/cube/buildCubeShape/Dimension.js → packages/cube/lib/cube/buildCubeShape/Dimension.js

@@ -1,4 +1,4 @@
-import rdf from '@zazuko/env'
+import rdf from '@zazuko/env-node'
 import { fromRdf } from 'rdf-literal'
 import cbdCopy from '../../cbdCopy.js'
 import * as ns from '../../namespaces.js'

packages/rdf/lib/cube/buildCubeShape/index.js → packages/cube/lib/cube/buildCubeShape/index.js

@@ -1,5 +1,5 @@
 import once from 'lodash/once.js'
-import $rdf from '@zazuko/env'
+import $rdf from '@zazuko/env-node'
 import { Transform } from 'readable-stream'
 import fromStream from 'rdf-dataset-ext/fromStream.js'
 import * as ns from '../../namespaces.js'

packages/rdf/lib/cube/toObservation.js → packages/cube/lib/cube/toObservation.js

@@ -1,5 +1,5 @@
 import { URL } from 'url'
-import $rdf from '@zazuko/env'
+import $rdf from '@zazuko/env-node'
 import { Transform } from 'readable-stream'
 import dateToId from '../dateToId.js'
 import urlJoin from '../urlJoin.js'

packages/cube/lib/namespaces.js

@@ -0,0 +1,6 @@
+import $rdf from '@zazuko/env-node'
+
+const cube = $rdf.namespace('https://cube.link/')
+const { rdf, rdfs, sh, xsd, _void, dcat, schema, dcterms } = $rdf.ns
+
+export { cube, rdf, rdfs, sh, xsd, _void, dcat, schema, dcterms }

packages/cube/lib/sort.js

@@ -0,0 +1,15 @@
+import { Readable, Duplex } from 'stream'
+import { sort, compareOn, createStore } from 'external-merge-sort'
+import rdf from '@zazuko/env-node'
+
+async function write(chunk, filename) {
+  await rdf.toFile(Readable.from(chunk), filename)
+  return rdf.fromFile(filename)
+}
+
+export const sortRDF = key => {
+  const comparer = compareOn(key)
+  const store = createStore(write, '.nt')
+
+  return Duplex.from(iterable => sort(iterable, { comparer, store, maxSize: 100000 }))
+}