Distributed Spectral Analysis with FFTW-WASM (Node.js)

This tutorial distributes a real signal-processing workload across the DCP network: a set of independent sensor readings, each analyzed with a real Fast Fourier Transform to find its dominant frequency, using FFTW compiled to WebAssembly as the transform engine inside each worker.

It assumes you’ve already seen the basic DCP job tutorial; this one reuses the same shape (input set → work function → compute.for → events → results) and adds one new ingredient: a compiled WASM module, running inside a sandboxed worker.

Note

This tutorial uses your own WASM module, not one already published to the DCP package manager – contrast with the DuckDB-WASM tutorial, which consumes duckdbwasm, a package someone already published. There’s no publishing step here: job.requires() with a local, relative path works directly, and DCP handles bundling it for you. See Building WASM modules for DCP jobs for how the module itself gets built – this tutorial picks up once you already have it.

Prerequisites

• Node.js (LTS)

• npm i dcp-client

• Keystore files in ~/.dcp (see the basic tutorial if you don’t have these yet)

• fftw-wasm.js and fftw-module.js, built as described in Building WASM modules for DCP jobs, sitting next to your job script

The scenario

A predictive-maintenance system reads short vibration snapshots from several sensors on different pieces of machinery. For each sensor, the question is the same: what frequency dominates this snapshot, and is its magnitude large enough to flag for review? Each sensor’s reading is independent of every other sensor’s – exactly the shape distributed compute wants: one sensor’s snapshot is one unit of work, and a DCP worker can answer the question completely on its own with nothing but the samples it was handed and the anomaly threshold every worker shares.

Building it, concept by concept

Input set

One element per sensor. Each element carries that sensor’s raw samples – nothing shared, nothing another sensor’s worker needs to know about:

const SAMPLE_RATE_HZ = 1000;

const inputSet = [
  { sensor: 'bearing-1', samples: makeSignal(1.0, 3) },
  { sensor: 'bearing-2', samples: makeSignal(3.0, 5) },
  { sensor: 'gearbox-1', samples: makeSignal(0.2, 1) },
];

makeSignal just generates a synthetic sine wave so this tutorial has real numbers to work with – swap it for however your own sensors actually get read:

function makeSignal(amplitude, bin, n = 16) {
  return Array.from(
    { length: n },
    (_, i) => amplitude * Math.sin((2 * Math.PI * bin * i) / n),
  );
}

A real deployment would pull each sensor’s snapshot from wherever your telemetry pipeline stores it. The FFT logic below doesn’t change if samples becomes “16 real values read off a message queue” instead of a generated array.

Static arguments

The sample rate and the anomaly threshold are shared facts, not per-sensor data – every worker needs the same values to turn an FFT bin index into a real frequency and to decide what counts as large. compute.for’s third argument delivers values like this once, to every invocation, without repeating them inside every input-set element:

const ANOMALY_MAGNITUDE_THRESHOLD = 15;

async function workFunction(
  { sensor, samples },
  sampleRateHz,
  anomalyThreshold,
) {
  /* ... */
}

const job = compute.for(inputSet, workFunction, [
  SAMPLE_RATE_HZ,
  ANOMALY_MAGNITUDE_THRESHOLD,
]);

Work function

This is where the actual analysis happens, once per sensor, inside a sandboxed DCP worker. It runs a real FFT (fftw.r2c1d, a real-to-complex transform – the natural choice for real-valued sensor samples) over that sensor’s snapshot, finds the bin with the largest magnitude, and converts it to a frequency in Hz using the shared sample rate.

async function workFunction(
  { sensor, samples },
  sampleRateHz,
  anomalyThreshold,
) {
  progress();

  const fftw = require('fftw-wasm.js'); // your own module, via job.requires() below
  const { re, im } = await fftw.r2c1d(samples);

  // Bin 0 is the DC component (the signal's average level) - skip it,
  // only oscillating content is interesting here.
  let peakBin = 0;
  let peakMagnitude = 0;
  for (let bin = 1; bin < re.length; bin++) {
    const magnitude = Math.sqrt(re[bin] * re[bin] + im[bin] * im[bin]);
    if (magnitude > peakMagnitude) {
      peakMagnitude = magnitude;
      peakBin = bin;
    }
  }

  const dominantFrequencyHz = (peakBin * sampleRateHz) / samples.length;
  const isAnomaly = peakMagnitude >= anomalyThreshold;

  return { sensor, dominantFrequencyHz, peakMagnitude, isAnomaly };
}

fftw.r2c1d(samples) returns the non-redundant half of the spectrum (Math.floor(n / 2) + 1 bins) as parallel re/im arrays – magnitude at each bin is the usual sqrt(re² + im²).

Job creation

const job = compute.for(inputSet, workFunction, [
  SAMPLE_RATE_HZ,
  ANOMALY_MAGNITUDE_THRESHOLD,
]);

Job configuration

job.requires() is what makes require('fftw-wasm.js') resolve inside the worker; without it, that line throws. Note the path: a local, relative path to your own file, not a package name – this is the local-module form, different from job.requires(['duckdbwasm/duckdb-wasm.js']) in the DuckDB tutorial. DCP bundles fftw-wasm.js and everything it require()s (here, fftw-module.js) automatically; you don’t list transitive dependencies yourself.

job.requires(['./fftw-wasm']);

job.public = {
  name: 'sensorSpectralAnalysis',
  description: 'Per-sensor dominant-frequency analysis computed with FFTW-wasm',
  link: 'https://distributive.network',
};

job.computeGroups = [{ joinKey: 'public' }];

Events

job.on('readystatechange', (ev) => console.log(`Ready state: ${ev}`));
job.on('accepted', () =>
  console.log(`  Job id: ${job.id}\n  Awaiting results...`),
);
job.on('result', (ev) => console.log(`  ${ev.result.sensor} done`));
job.on('error', (error) => console.error('  Job error:', error));
job.on('nofunds', (ev) => console.log(ev));

Execution

const results = await job.exec();

Result post-processing

Each sensor came back as its own small verdict. Flagging anomalies across the whole fleet of sensors is just a filter over the results:

for (const r of results) {
  console.log(
    `${r.sensor.padEnd(10)} dominant=${r.dominantFrequencyHz}Hz  magnitude=${r.peakMagnitude}  ${r.isAnomaly ? '** ANOMALY **' : ''}`,
  );
}

const anomalies = results.filter((r) => r.isAnomaly);
console.log(
  `\n${anomalies.length} sensor(s) flagged for review: ${anomalies.map((r) => r.sensor).join(', ') || 'none'}`,
);

Full listing

async function main() {
  const compute = require('dcp/compute');

  /* INPUT SET */
  function makeSignal(amplitude, bin, n = 16) {
    return Array.from(
      { length: n },
      (_, i) => amplitude * Math.sin((2 * Math.PI * bin * i) / n),
    );
  }

  const inputSet = [
    { sensor: 'bearing-1', samples: makeSignal(1.0, 3) },
    { sensor: 'bearing-2', samples: makeSignal(3.0, 5) },
    { sensor: 'gearbox-1', samples: makeSignal(0.2, 1) },
  ];

  /* STATIC ARGUMENTS -- shared by every worker, not part of the input set */
  const SAMPLE_RATE_HZ = 1000;
  const ANOMALY_MAGNITUDE_THRESHOLD = 15;

  /* WORK FUNCTION */
  async function workFunction(
    { sensor, samples },
    sampleRateHz,
    anomalyThreshold,
  ) {
    progress();

    const fftw = require('fftw-wasm.js');
    const { re, im } = await fftw.r2c1d(samples);

    let peakBin = 0;
    let peakMagnitude = 0;
    for (let bin = 1; bin < re.length; bin++) {
      const magnitude = Math.sqrt(re[bin] * re[bin] + im[bin] * im[bin]);
      if (magnitude > peakMagnitude) {
        peakMagnitude = magnitude;
        peakBin = bin;
      }
    }

    const dominantFrequencyHz = (peakBin * sampleRateHz) / samples.length;
    const isAnomaly = peakMagnitude >= anomalyThreshold;

    return { sensor, dominantFrequencyHz, peakMagnitude, isAnomaly };
  }

  /* JOB CREATION */
  const job = compute.for(inputSet, workFunction, [
    SAMPLE_RATE_HZ,
    ANOMALY_MAGNITUDE_THRESHOLD,
  ]);

  /* JOB CONFIGURATION */
  job.requires(['./fftw-wasm']);
  job.public = {
    name: 'sensorSpectralAnalysis',
    description:
      'Per-sensor dominant-frequency analysis computed with FFTW-wasm',
    link: 'https://distributive.network',
  };
  job.computeGroups = [{ joinKey: 'public' }];

  /* EVENTS */
  job.on('readystatechange', (ev) => console.log(`Ready state: ${ev}`));
  job.on('accepted', () =>
    console.log(`  Job id: ${job.id}\n  Awaiting results...`),
  );
  job.on('result', (ev) => console.log(`  ${ev.result.sensor} done`));
  job.on('error', (error) => console.error('  Job error:', error));
  job.on('nofunds', (ev) => console.log(ev));

  /* EXECUTION */
  const results = await job.exec();

  /* RESULT POST-PROCESSING */
  for (const r of results) {
    console.log(
      `${r.sensor.padEnd(10)} dominant=${r.dominantFrequencyHz}Hz  magnitude=${r.peakMagnitude}  ${r.isAnomaly ? '** ANOMALY **' : ''}`,
    );
  }

  const anomalies = results.filter((r) => r.isAnomaly);
  console.log(
    `\n${anomalies.length} sensor(s) flagged for review: ${anomalies.map((r) => r.sensor).join(', ') || 'none'}`,
  );
}
require('dcp-client')
  .init('https://scheduler.distributed.computer')
  .then(main)
  .catch(console.error);

Running it

node sensor-spectral-analysis.js

Expected output:

Ready state: ...
  Job id: ...
  Awaiting results...
  bearing-1 done
  bearing-2 done
  gearbox-1 done

bearing-1  dominant=187.5Hz  magnitude=8
bearing-2  dominant=312.5Hz  magnitude=24  ** ANOMALY **
gearbox-1  dominant=62.5Hz  magnitude=1.6

1 sensor(s) flagged for review: bearing-2

Going further

• Bigger, real data: swap makeSignal’s generated arrays for real telemetry – an FFT this size doesn’t care whether its 16 samples came from a formula or a sensor.

• Longer snapshots: r2c1d doesn’t require a power-of-two length, but FFTW is fastest for one; a real deployment would size its snapshot length accordingly.

• More static arguments: the same pattern from SAMPLE_RATE_HZ/ANOMALY_MAGNITUDE_THRESHOLD extends to anything every worker needs identically – a per-machine baseline spectrum to compare against, a set of frequency bands to ignore, whatever your real anomaly rule needs.

• Publishing it: if this module turns out to be useful across more than one job, it’s a candidate for actually publishing to the DCP package manager – see the “Using a published package” note in the WASM build guide, and email dan@dcp.dev.