N3 step

[tpluscode]

Current prototype of such a step from museumplus

• Batching prevents high CPU when eye runs too often

import { Readable } from 'stream'
import type { Quad } from '@rdfjs/types'
import { n3reasoner } from 'eyereasoner'
import through2 from 'through2'
import getStream from 'get-stream'
import $rdf from 'rdf-ext'
import type { Context } from 'barnard59-core'
import { Parser } from 'n3'

const parser = new Parser()

export async function deriveRules(this: Context, { rules, batchSize = 20000 }: { rules: Readable; batchSize?: number }) {
  const { logger } = this
  const ruleQuads = await getStream.array<Quad>(rules)
  let batchedQuads = 0
  const batch: Quad[][] = []

  async function runInferences(chunks: Quad[][]) {
    logger.debug(`Running n3 rules. Batch size ${batchedQuads}`)
    const input = chunks.flatMap(arr => [...arr])

    const inferred = await n3reasoner([...input, ...ruleQuads], undefined, { outputType: 'string' })

    return $rdf.dataset(parser.parse(inferred))
  }

  return through2.obj(async function (chunk: Quad[], _, next) {
    batch.push(chunk)
    batchedQuads += chunk.length

    if (batchedQuads >= batchSize) {
      this.push(await runInferences(batch.splice(0)))
      batchedQuads = 0
    }

    next()
  }, async function (done) {
    if (batch.length > 0) {
      this.push(await runInferences(batch))
    }
    done()
  })
}