Skip to content

FnSoundDocs.kt

src/jvmMain/kotlin/afield/audio/synth/FnSoundDocs.kt · 341 lines

@file:afield.runtime.completion.CompletionCategory("Synthesis")
package afield.audio.synth

import afield.audio.engine.WavFloatArrayLoader
import afield.audio.engine.WavFloatArrayWriter
import afield.math.functions.Functions
import afield.math.functions.certifiedRange
import java.io.File
import kotlin.math.abs
import kotlin.math.ln
import kotlin.math.sqrt

FIELD_TEST id: synth.fn.sound.roundtrip runner: jvm-main entrypoint: afield.audio.synth.FnSoundDocs tags: [smoke, synth, functions, audio, wav, analysis, cpu] timeoutSec: 300 expect: exitCode: 0 stdoutContains: - "fn sound roundtrip measured"

One Functions.Fn taken through sound and back, and the distance between what went in and what came out — the harness behind public/docs/chain_fn_sound_roundtrip.md.

The route is four parts of Field that do not know about each other. contour is built from formula ops only, so afield.math.functions.FnInterval can certify its range on any interval. It enters a Dough graph either per sample, through FnSignal, or per block, through Param { … }. The graph is rendered offline by doughRenderMono, written to a 16-bit WAV without normalisation — so the amplitude in the file is the amplitude the function asked for — and read back from that file. The analysis is the in-process pair PatOnsetDetection and PatPitchDetection already provide: an RMS envelope (patEnergyEnvelope) for loudness, and YIN (patF0Frames) for pitch.

Two encodings of one function

  • Encoding.LOUDNESS: a CARRIER_HZ sine whose amplitude is LOUDNESS_GAIN · f(t). Recovered as √2 · RMS / LOUDNESS_GAIN. The carrier is chosen so one period is exactly 128 samples at SR and the RMS window, RMS_WINDOW, holds two whole periods: then a steady sine's RMS is exactly A/√2 and the window adds no ripple of its own.
  • Encoding.PITCH: a sine at PITCH_BASE_HZ · 2^f(t) and constant level, so f spans a fraction of an octave. Recovered as log2(hz / PITCH_BASE_HZ) from voiced YIN frames only.

What is measured, per encoding and route

Every analysis frame is one sample of the population: its recovered value against f at the frame centre (RoundTrip.maxAbsError, RoundTrip.rmsError); the delay that best aligns them (RoundTrip.bestLagSamples); and whether the recovered value lies inside the certified range of f over the samples the frame actually read (RoundTrip.insideCertified) — the only interval a windowed measurement can be expected to land in. Blip is the narrow feature: a bump with a 3 ms sigma, shorter than both analysis windows, and how much of its height comes back.

No engine, no device, no Field window. The renders are deterministic, so the committed .wavs regenerate byte-stably; as a FIELD_TEST everything is written to a temp dir.

object FnSoundDocs

Renders and measures the Fn sound round trip; see the file overview.

object FnSoundDocs {

const val SR: Int = 48_000

Sample rate of every render and of both published clips.

const val BLOCK: Int = 128

Frames per doughRenderMono block — the granularity of a Param { … } supplier.

const val SECONDS: Double = 3.0

const val CARRIER_HZ: Double = 375.0

128 samples per period at SR, so RMS_WINDOW holds exactly two.

const val LOUDNESS_GAIN: Double = 0.8

const val PITCH_BASE_HZ: Double = 220.0

const val PITCH_LEVEL: Double = 0.5

const val RMS_WINDOW: Int = 256

const val RMS_HOP: Int = 32

const val PITCH_MIN_HZ: Double = 200.0

The register YIN is told about: contour stays inside 0.2..0.71, so about 253..360 Hz.

const val PITCH_MAX_HZ: Double = 450.0

const val BLIP_AT: Double = 2.6

Where the narrow bump sits, its sigma, and its height above what is under it.

const val BLIP_SIGMA: Double = 0.003

const val BLIP_HEIGHT: Double = 0.25

val DOMAIN: Functions.Domain = Functions.Domain(0.0, SECONDS)

fun contour(): Functions.Fn

The function taken on the trip: a resting level, a broad swell, a step up to a plateau that wobbles at 6 Hz, and one bump narrower than either analysis window. Formula ops only, so it certifies.

    fun contour(): Functions.Fn = with(Functions) {
        val d = DOMAIN
        (c(0.2, d) +
            gaussian(0.8, 0.2, 0.5, d) +
            smoothStep(1.4, 2.0, d) * (c(0.3, d) + sin(6.0, amplitude = 0.08)) +
            gaussian(BLIP_AT, BLIP_SIGMA, BLIP_HEIGHT, d)).withName("f")
    }

enum class Encoding { LOUDNESS, PITCH }

enum class Route { PER_SAMPLE, PER_BLOCK }

How the function enters the graph: FnSignal at every sample, or Param { … } once a block.

fun pitchHz(fn: Functions.Fn): Functions.Fn

The frequency contour the PITCH encoding plays, PITCH_BASE_HZ · 2^f, still an Fn.

    fun pitchHz(fn: Functions.Fn): Functions.Fn = with(Functions) {
        fn.scaleY(ln(2.0)).exp().scaleY(PITCH_BASE_HZ)
    }

fun signal(fn: Functions.Fn, encoding: Encoding, route: Route): Signal

The Dough graph for one encoding and route.

    fun signal(fn: Functions.Fn, encoding: Encoding, route: Route): Signal = when (encoding) {
        Encoding.LOUDNESS -> {
            val gain = with(Functions) { fn.scaleY(LOUDNESS_GAIN) }
            val amplitude: Signal = when (route) {
                Route.PER_SAMPLE -> gain.toSignal()
                Route.PER_BLOCK -> Param { gain(time.toDouble()).toFloat() }
            }
            SineOsc(Param(CARRIER_HZ.toFloat())) * amplitude
        }
        Encoding.PITCH -> {
            val hz = pitchHz(fn)
            val freq = when (route) {
                Route.PER_SAMPLE -> Param(0f).modulate(hz.toSignal())
                Route.PER_BLOCK -> Param { hz(time.toDouble()).toFloat() }
            }
            SineOsc(freq) * PITCH_LEVEL
        }
    }

fun render(encoding: Encoding, route: Route, fn: Functions.Fn = contour()): FloatArray

Offline-render one encoding and route; no normalisation anywhere.

    fun render(encoding: Encoding, route: Route, fn: Functions.Fn = contour()): FloatArray =
        doughRenderMono(signal(fn, encoding, route), SECONDS.toFloat(), SR)

data class Frame

One analysis frame: the samples it read, [firstSample, lastSample], the time the analysis attributes its value to, and what it recovered — NaN for a YIN frame that was not voiced. A YIN frame's time is PatF0Frame.measuredCenterSample, the centre of the samples it compared.

    data class Frame(
        val firstSample: Int,
        val lastSample: Int,
        val centerSeconds: Double,
        val recovered: Double,
    )

fun analyse(samples: FloatArray, encoding: Encoding): List<Frame>

Analyse samples that were read back from a .wav.

    fun analyse(samples: FloatArray, encoding: Encoding): List<Frame> = when (encoding) {
        Encoding.LOUDNESS -> patEnergyEnvelope(samples, RMS_HOP, RMS_WINDOW).mapIndexed { i, rms ->
            val first = i * RMS_HOP
            Frame(first, first + RMS_WINDOW - 1, (first + (RMS_WINDOW - 1) / 2.0) / SR, sqrt(2.0) * rms / LOUDNESS_GAIN)
        }
        Encoding.PITCH -> {
            patF0Frames(samples, SR, PITCH_MIN_HZ, PITCH_MAX_HZ).map { f ->
                val voiced = f.voiced && f.hz > 0.0
                val value = if (voiced) ln(f.hz / PITCH_BASE_HZ) / ln(2.0) else Double.NaN
                Frame(f.startSample, f.startSample + f.spanSamples - 1, f.measuredCenterSample.toDouble() / SR, value)
            }
        }
    }

data class Blip(val truePeak: Double, val base: Double, val recoveredPeak: Double)

How much of the narrow bump came back: its true height over the base, and the recovered one.

    data class Blip(val truePeak: Double, val base: Double, val recoveredPeak: Double) {

val survival: Double get() = (recoveredPeak - base) / (truePeak - base)

Recovered height above the base as a fraction of the true height; 1.0 is all of it.

class RoundTrip

Everything one trip measured. wav is the file the analysis read.

    class RoundTrip(
        val encoding: Encoding,
        val route: Route,
        val fn: Functions.Fn,
        val wav: File,
        val frames: List<Frame>,
    ) {

val measured: List<Frame> = frames.filter { it.recovered.isFinite() }

The frames with a recovered value: every RMS frame, and the voiced YIN frames.

val errors: DoubleArray

recovered − f(centre) per measured frame.

        val errors: DoubleArray = DoubleArray(measured.size) { measured[it].recovered - fn(measured[it].centerSeconds) }

val maxAbsError: Double = errors.maxOfOrNull { abs(it) } ?: Double.NaN

val rmsError: Double = rms(errors)

val bestLagSamples: Int

The delay, in samples, that minimises the RMS error when f is read lag samples earlier than each frame centre, over -BLOCK*2 .. BLOCK*2. Positive means the sound is late.

val rmsErrorAtBestLag: Double

init

        init {
            val (lag, r) = bestLag { it.centerSeconds }
            bestLagSamples = lag
            rmsErrorAtBestLag = r
        }

        private fun bestLag(time: (Frame) -> Double): Pair<Int, Double> {
            var best = 0
            var bestRms = Double.POSITIVE_INFINITY
            for (lag in -2 * BLOCK..2 * BLOCK) {
                var sum = 0.0
                for (f in measured) {
                    val e = f.recovered - fn(time(f) - lag.toDouble() / SR)
                    sum += e * e
                }
                val r = sqrt(sum / measured.size.coerceAtLeast(1))
                if (r < bestRms) { bestRms = r; best = lag }
            }
            return best to bestRms
        }

val maxErrorAtSeconds: Double

Where the largest error is, in seconds.

        val maxErrorAtSeconds: Double = errors.indices.maxByOrNull { abs(errors[it]) }?.let { measured[it].centerSeconds } ?: Double.NaN

Per measured frame, whether its value lies inside the certified range of f over the samples that frame read. Bounds are certifiedRange's proven ones, not sampled ones.

        private val outside = DoubleArray(measured.size)

val insideCertified: BooleanArray

        val insideCertified: BooleanArray = BooleanArray(measured.size) { i ->
            val f = measured[i]
            // A bound is proven whether or not the search converged; convergence only says how
            // close it came to the reached value. A refusal (no bounds at all) is an error.
            val range = fn.certifiedRange(f.firstSample.toDouble() / SR, f.lastSample.toDouble() / SR, tolerance = 1e-6, maxEnclosures = 4_000)
            val lo = checkNotNull(range.min) { "no certified bound over frame $i: ${range.refusal}" }
            val hi = checkNotNull(range.max) { "no certified bound over frame $i: ${range.refusal}" }
            outside[i] = maxOf(0.0, lo.bound - f.recovered, f.recovered - hi.bound)
            outside[i] == 0.0
        }

val insideCount: Int = insideCertified.count { it }

val worstOutside: Double get() = outside.maxOrNull() ?: 0.0

The furthest any frame's value lies outside its certified range; 0.0 when none does.

val blip: Blip

        val blip: Blip = run {
            val near = measured.filter { abs(it.centerSeconds - BLIP_AT) <= 0.05 }
            val truth = fn.certifiedRange(BLIP_AT - 0.05, BLIP_AT + 0.05, tolerance = 1e-9)
            Blip(
                truePeak = truth.max!!.value,
                base = fn(BLIP_AT) - BLIP_HEIGHT,
                recoveredPeak = near.maxOfOrNull { it.recovered } ?: Double.NaN,
            )
        }

fun recoveredAt(t: Double): Double

The recovered curve between frame centres, linearly; NaN across an unvoiced frame.

        fun recoveredAt(t: Double): Double {
            if (frames.isEmpty()) return Double.NaN
            var lo = 0
            var hi = frames.size - 1
            if (t <= frames[lo].centerSeconds) return frames[lo].recovered
            if (t >= frames[hi].centerSeconds) return frames[hi].recovered
            while (hi - lo > 1) {
                val mid = (lo + hi) ushr 1
                if (frames[mid].centerSeconds <= t) lo = mid else hi = mid
            }
            val a = frames[lo]
            val b = frames[hi]
            val u = (t - a.centerSeconds) / (b.centerSeconds - a.centerSeconds)
            return a.recovered + (b.recovered - a.recovered) * u
        }

fun errorAt(t: Double): Double = recoveredAt(t) - fn(t)

recoveredAt(t) − f(t).

fun describe(): String

        fun describe(): String =
            "${encoding.name.lowercase().padEnd(8)} ${route.name.lowercase().padEnd(10)} " +
                "frames=${measured.size}/${frames.size} max|e|=${"%.5f".format(maxAbsError)} " +
                "rms=${"%.5f".format(rmsError)} at=${"%.4f".format(maxErrorAtSeconds)} lag=$bestLagSamples " +
                "rms@lag=${"%.5f".format(rmsErrorAtBestLag)} " +
                "inside=$insideCount/${measured.size} worstOutside=${"%.2e".format(worstOutside)} " +
                "blip=${"%.3f".format(blip.survival)}"
    }

fun roundTrip

Render, write the .wav into dir, read it back from the file, and analyse what was read.

    fun roundTrip(
        encoding: Encoding,
        route: Route = Route.PER_SAMPLE,
        dir: File = File(System.getProperty("java.io.tmpdir"), "fn-sound-roundtrip"),
        fn: Functions.Fn = contour(),
    ): RoundTrip {
        dir.mkdirs()
        val wav = File(dir, "${encoding.name.lowercase()}-${route.name.lowercase()}.wav")
        WavFloatArrayWriter.writeMono16(wav, render(encoding, route, fn), SR, normalizePeak = null)
        val readBack = WavFloatArrayLoader.loadWavToFloatArray(wav.path)
        check(readBack.sampleRate == SR) { "read back ${readBack.sampleRate} Hz from $wav, wrote $SR" }
        return RoundTrip(encoding, route, fn, wav, analyse(readBack.samples, encoding))
    }

    private val SCENES: Map<String, Encoding> = mapOf("loudness" to Encoding.LOUDNESS, "pitch" to Encoding.PITCH)

fun renderWav(name: String, outPath: String)

Offline-render scene loudness or pitch (the per-sample route) to a mono 16-bit WAV.

    fun renderWav(name: String, outPath: String) {
        val encoding = SCENES[name] ?: error("unknown FnSoundDocs scene '$name'; known: ${SCENES.keys.sorted()}")
        val samples = render(encoding, Route.PER_SAMPLE)
        require(samples.any { it != 0f }) { "scene '$name' rendered silence" }
        WavFloatArrayWriter.writeMono16(outPath, samples, SR, normalizePeak = null)
    }

    private fun rms(values: DoubleArray): Double =
        if (values.isEmpty()) Double.NaN else sqrt(values.sumOf { it * it } / values.size)

FIELD_TEST entry point

fun main(args: Array<String>)

    @JvmStatic
    fun main(args: Array<String>) {
        val dir = File(args.getOrNull(0) ?: File(System.getProperty("java.io.tmpdir"), "fn-sound-roundtrip").path)
        dir.mkdirs()
        val fn = contour()
        val whole = fn.certifiedRange(0.0, SECONDS, tolerance = 1e-9)
        println("f over [0, $SECONDS]: certified min ${whole.min!!.value} .. max ${whole.max!!.value}")

        val trips = Encoding.values().flatMap { e -> Route.values().map { r -> roundTrip(e, r, dir, fn) } }
        trips.forEach { println(it.describe()) }
        fun trip(e: Encoding, r: Route) = trips.single { it.encoding == e && it.route == r }

        // The structural claims the page makes, checked here so the smoke fails when one stops holding.
        check(trip(Encoding.LOUDNESS, Route.PER_SAMPLE).bestLagSamples == 0) { "FnSignal is no longer on time" }
        check(trip(Encoding.LOUDNESS, Route.PER_BLOCK).bestLagSamples == BLOCK) { "Param { } is no longer one block late" }
        check(abs(trip(Encoding.PITCH, Route.PER_SAMPLE).bestLagSamples) <= 3) {
            "YIN frames are no longer reported at the instant they measured"
        }
        check(trips.all { it.measured.isNotEmpty() && it.insideCount > 0 }) { "a trip measured nothing" }

        // The published clips are the analysed files: the fence renderer writes the same bytes.
        SCENES.forEach { (name, encoding) ->
            val published = File(dir, "published-$name.wav")
            renderWav(name, published.path)
            val analysed = trips.single { it.encoding == encoding && it.route == Route.PER_SAMPLE }.wav
            check(published.readBytes().contentEquals(analysed.readBytes())) {
                "renderWav('$name') and the analysed $analysed differ"
            }
        }
        println("published clips are byte-identical to the analysed files")
        println("fn sound roundtrip measured")
    }
}