Analyze independent signal spectra

// CPU comparison: ordinary sequential loop, one worker, and multiple workers.
// Run: aner run examples/parallel/cpu_signal_spectrum.aner --max-steps 500000000
// Standalone timings are one trial; run_cpu_cases.py repeats and varies execution order.
// Machine-condition monitoring: a spectrum for each independent vibration window.
// Synthetic acceleration in m/s^2, sampled at 1024 Hz; no recorded sensor files.
import aner.parallel
import aner.math

fn analyze_signal(parameters: List, iteration: Int64, seed: Int64) -> List {
    let samples = cast<Int64>(parameters.get(0))
    let bins = cast<Int64>(parameters.get(1))
    let sample_rate_hz = cast<Float64>(parameters.get(2))
    let tau = 6.283185307179586
    let windowed = Array<Float64>(samples, 0.0)
    // Each identity represents another motor/window. Frequencies land on DFT bins.
    let fundamental_bin = 5 + iteration % 7
    let fundamental_hz = Float64(fundamental_bin) * sample_rate_hz / Float64(samples)
    let amplitude = 1.0 + 0.05 * Float64(iteration % 9)
    var square_total = 0.0
    for sample in Range(samples) {
        let time_s = Float64(sample) / sample_rate_hz
        let value = amplitude * math.sin(tau * fundamental_hz * time_s) +
            0.25 * math.sin(tau * 2.0 * fundamental_hz * time_s + 0.3) +
            0.05 * math.sin(tau * 29.0 * sample_rate_hz / Float64(samples) * time_s + Float64(iteration))
        square_total = square_total + value * value
        // Hann taper reduces leakage at the finite recording boundary.
        let window = 0.5 - 0.5 * math.cos(tau * Float64(sample) / Float64(samples - 1))
        windowed[sample] = value * window
    }
    var strongest_power = -1.0
    var strongest_bin = 0
    var band_power = 0.0
    // Direct DFT is intentionally transparent for teaching, with O(samples*bins) work.
    // Production spectral pipelines commonly use an FFT; that optimization is separate.
    for bin in Range(1, bins + 1) {
        var real_part = 0.0
        var imaginary_part = 0.0
        for sample in Range(samples) {
            let angle = tau * Float64(bin) * Float64(sample) / Float64(samples)
            real_part = real_part + windowed[sample] * math.cos(angle)
            imaginary_part = imaginary_part - windowed[sample] * math.sin(angle)
        }
        let power = (real_part * real_part + imaginary_part * imaginary_part) /
            (Float64(samples) * Float64(samples))
        band_power = band_power + power
        if power > strongest_power { strongest_power = power; strongest_bin = bin }
    }
    // Seed is unused: the fixture is deterministic for its logical window identity.
    // Outputs: dominant Hz, untapered acceleration RMS (m/s^2), uncalibrated band score.
    return List(Float64(strongest_bin) * sample_rate_hz / Float64(samples),
        math.sqrt(square_total / Float64(samples)), band_power)
}

// Change only these settings to try a larger batch or another CPU allocation.
let jobs = 64
let columns = 3
// 0 = automatic admitted CPU allocation; 1 = serial control.
let requested_workers = 0
// Runner rotates 0/1/2 to vary which method executes first.
let benchmark_order = 0
let parameters = List(256, 32, 1024.0)

fn require(condition: Bool, message: String) -> Unit {
    if !condition {
        let failure: Result<Bool, String> = Result.err(message)
        print(failure.unwrap())
    }
}

// NO PARALLELISM: finish one job before calling the same function for the next.
fn measure_sequential(parameters: List, jobs: Int64, columns: Int64) -> List {
    let started = Duration.monotonic()
    let values = Array<Float64>(jobs * columns, 0.0)
    for iteration in Range(jobs) {
        let row = analyze_signal(parameters, iteration, 2026)
        for column in Range(columns) {
            values[iteration * columns + column] = cast<Float64>(row.get(column))
        }
    }
    let seconds = Duration.monotonic().subtract(started).total_seconds()
    return List(values, seconds)
}

// PARALLELISM: the task body is unchanged. The executor collects ordered rows.
// workers=1 is the control; workers=0 selects the machine's admitted allocation.
fn measure_repeat(parameters: List, jobs: Int64, workers: Int64) -> List {
    let started = Duration.monotonic()
    let results = parallel.repeat(iterations: jobs, task: analyze_signal,
        parameters: parameters, execution: parallel.local(workers: workers), seed: 2026)
    let seconds = Duration.monotonic().subtract(started).total_seconds()
    return List(results, seconds)
}

// One untimed task exercises the model before the comparison.
let warmup = analyze_signal(parameters, 0, 2026)
var sequential_run = List()
var one_worker_run = List()
var parallel_run = List()
for phase in Range(3) {
    let mode = (benchmark_order + phase) % 3
    if mode == 0 { sequential_run = measure_sequential(parameters, jobs, columns) }
    if mode == 1 { one_worker_run = measure_repeat(parameters, jobs, 1) }
    if mode == 2 { parallel_run = measure_repeat(parameters, jobs, requested_workers) }
}
let sequential_values = cast<Array<Float64>>(sequential_run.get(0))
let one_worker = cast<SimulationResults>(one_worker_run.get(0))
let concurrent = cast<SimulationResults>(parallel_run.get(0))
let sequential_seconds = cast<Float64>(sequential_run.get(1))
let one_worker_seconds = cast<Float64>(one_worker_run.get(1))
let parallel_seconds = cast<Float64>(parallel_run.get(1))

// Verification and printing happen after all timers stop.
require(one_worker.rows() == jobs && concurrent.rows() == jobs &&
    one_worker.cols() == columns && concurrent.cols() == columns &&
    one_worker.workers() == 1, "Incomplete results or incorrect serial allocation")
require(sequential_seconds > 0.0 && one_worker_seconds > 0.0 && parallel_seconds > 0.0,
    "Clock resolution is insufficient for this batch")
for iteration in Range(jobs) {
    require(one_worker.seed(iteration) == concurrent.seed(iteration), "Task seed mismatch")
    for column in Range(columns) {
        let expected = sequential_values[iteration * columns + column]
        require(expected == one_worker.value(iteration, column) &&
            expected == concurrent.value(iteration, column), "Output differs across execution methods")
    }
    require(concurrent.value(iteration, 0) == Float64(5 + iteration % 7) * cast<Float64>(parameters.get(2)) / Float64(cast<Int64>(parameters.get(0))) &&
        concurrent.value(iteration, 1) > 0.0 && concurrent.value(iteration, 2) > 0.0, "Incorrect synthetic signal spectrum")
}
print("case=cpu_signal_spectrum")
print("jobs=" + String(jobs))
print("columns=" + String(columns))
print("workers=" + String(concurrent.workers()))
print("verified=true")
// Caller elapsed times include allocation, CPU context startup and result collection.
print("sequential_seconds=" + String(sequential_seconds))
print("one_worker_seconds=" + String(one_worker_seconds))
print("parallel_seconds=" + String(parallel_seconds))
print("speedup=" + String(sequential_seconds / parallel_seconds))
print("executor_speedup=" + String(one_worker.seconds() / concurrent.seconds()))
for column in Range(columns) {
    print("output_" + String(column) + "_mean=" + String(concurrent.mean(column)))
}
print("summary_json=" + concurrent.summary())
print("one_worker_summary_json=" + one_worker.summary())