Babol/micarrsim
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: harden realtime localization pipeline

Browse Source
This commit is contained in:
Codex
2026-08-28 18:58:48 +09:00
parent 35c33e360e
commit 256558e601
13 changed files with 765 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/acoustics-core/realtime-localizer.js
+168 -33
View File
@@ -1,13 +1,18 @@
import { computeDasMap, createDasAnalysisPlan } from "./das.js";
import { findLocalPeaks } from "./evaluation.js";
import {
DEFAULT_MAX_COMPENSATION_DB,
MAX_FREQUENCY_RESPONSE_HZ,
MAX_FREQUENCY_RESPONSE_POINTS,
requireMaxCompensationDb,
} from "./frequency-response.js";
export const REALTIME_LOCALIZATION_PLAN_KIND = "micarrsim.realtime-localization-plan/1";
export const DEFAULT_STREAM_RESULTS_PER_CALL = 1;
export const MAX_STREAM_RESULTS_PER_CALL = 32;
export const DEFAULT_MAXIMUM_STREAM_INPUT_BYTES = 256 * 1024 * 1024;
export const DEFAULT_REALTIME_POINT_BATCH_SIZE = 256;
export const DEFAULT_REALTIME_STFT_TRANSFORM_BATCH_SIZE = 128;
function requirePositiveFinite(value, label) {
if (!Number.isFinite(value) || value <= 0) {
@@ -31,6 +36,10 @@ function requirePowerOfTwo(value, label) {
return value;
}
function monotonicNow() {
return globalThis.performance?.now?.() ?? Date.now();
}
function freezePosition(position, label) {
if (!Array.isArray(position) || position.length !== 3 || !position.every(Number.isFinite)) {
throw new TypeError(`${label} must be a finite [x, y, z] array.`);
@@ -155,6 +164,28 @@ function copyBands(bands, sampleRate) {
}));
}
function copyPeakDetection(value, scan) {
const options = value ?? {};
if (!options || typeof options !== "object" || Array.isArray(options)) {
throw new TypeError("peakDetection must be an object.");
}
const thresholdRatio = options.thresholdRatio ?? 0.5;
if (!Number.isFinite(thresholdRatio) || thresholdRatio < 0 || thresholdRatio > 1) {
throw new RangeError("peakDetection.thresholdRatio must lie inside 0..1.");
}
const xStep = scan.columns > 1 ? (scan.xMax - scan.xMin) / (scan.columns - 1) : 0;
const yStep = scan.rows > 1 ? (scan.yMax - scan.yMin) / (scan.rows - 1) : 0;
const minimumPeakDistanceM = options.minimumPeakDistanceM ?? Math.hypot(xStep, yStep);
if (!Number.isFinite(minimumPeakDistanceM) || minimumPeakDistanceM < 0) {
throw new RangeError("peakDetection.minimumPeakDistanceM must be a non-negative finite number.");
}
const maximumDetections = options.maximumDetections ?? 8;
if (!Number.isInteger(maximumDetections) || maximumDetections < 1 || maximumDetections > 128) {
throw new RangeError("peakDetection.maximumDetections must be an integer from 1 through 128.");
}
return Object.freeze({ thresholdRatio, minimumPeakDistanceM, maximumDetections });
}
function validateChannels(channels, expectedChannelCount) {
if (!Array.isArray(channels) || channels.length !== expectedChannelCount) {
throw new Error(
@@ -182,13 +213,42 @@ function throwIfAborted(signal) {
if (signal?.aborted) throw createAbortError(signal);
}
function locationsFromBands(bands) {
return bands.map((band) => ({
bandId: band.id,
index: band.peak.index,
position: [...band.peak.position],
rawPower: band.peak.rawPower,
}));
export function detectBandPeaks(rawMap, scan, options = {}) {
const peakDetection = copyPeakDetection(options, scan);
const localPeaks = findLocalPeaks(rawMap, scan);
const threshold = (localPeaks[0]?.value ?? 0) * peakDetection.thresholdRatio;
const detections = [];
for (const peak of localPeaks) {
if (peak.value < threshold) continue;
if (
peakDetection.minimumPeakDistanceM > 0 &&
detections.some((selected) =>
Math.hypot(
selected.position[0] - peak.position[0],
selected.position[1] - peak.position[1],
) <= peakDetection.minimumPeakDistanceM
)
) {
continue;
}
detections.push({
rank: detections.length + 1,
index: peak.index,
position: [...peak.position],
rawPower: peak.value,
});
if (detections.length === peakDetection.maximumDetections) break;
}
return detections;
}
function locationsFromBands(bands, scan, peakDetection) {
return bands.flatMap((band) =>
detectBandPeaks(band.rawMap, scan, peakDetection).map((peak) => ({
bandId: band.id,
...peak,
})),
);
}
/**
@@ -202,6 +262,7 @@ export function createLocalizationPlan(config) {
const sampleRate = requirePositiveFinite(config.sampleRate, "sampleRate");
const arrayDefinition = copyArrayDefinition(config.array ?? config.arrayDefinition);
const scan = copyScan(config.scan);
const peakDetection = copyPeakDetection(config.peakDetection, scan);
const frequencyBands = copyBands(config.bands ?? config.frequencyBands, sampleRate);
const fftSize = requirePowerOfTwo(config.stft?.fftSize ?? config.fftSize ?? 512, "fftSize");
const hopSize = requirePositiveInteger(
@@ -221,13 +282,22 @@ export function createLocalizationPlan(config) {
if (typeof compensateFrequencyResponse !== "boolean") {
throw new TypeError("compensateFrequencyResponse must be boolean.");
}
const maxCompensationDb = requireMaxCompensationDb(
config.maxCompensationDb ??
config.beamforming?.maxCompensationDb ??
DEFAULT_MAX_COMPENSATION_DB,
);
const scene = Object.freeze({ speedOfSound });
const experiment = Object.freeze({
stft: Object.freeze({ fftSize, hopSize, window: "hann" }),
scan,
frequencyBands,
beamforming: Object.freeze({ normalization, compensateFrequencyResponse }),
beamforming: Object.freeze({
normalization,
compensateFrequencyResponse,
maxCompensationDb,
}),
});
const dasPlan = createDasAnalysisPlan({
sampleRate,
@@ -245,13 +315,19 @@ export function createLocalizationPlan(config) {
arrayDefinition,
scan,
bands: frequencyBands,
peakDetection,
stft: experiment.stft,
speedOfSound,
normalization,
maxCompensationDb,
scene,
experiment,
dasPlan,
computeOptions: Object.freeze({ ...(config.computeOptions ?? {}) }),
computeOptions: Object.freeze({
pointBatchSize: DEFAULT_REALTIME_POINT_BATCH_SIZE,
stftTransformBatchSize: DEFAULT_REALTIME_STFT_TRANSFORM_BATCH_SIZE,
...(config.computeOptions ?? {}),
}),
});
}
@@ -273,6 +349,7 @@ export async function localizeMultichannel(options) {
const sampleCount = validateChannels(options.channels, plan.channelCount);
if (sampleCount === 0) throw new RangeError("channels must contain at least one sample.");
throwIfAborted(options.signal);
const processingStartedAt = monotonicNow();
const das = await computeDasMap({
...plan.computeOptions,
...(options.computeOptions ?? {}),
@@ -285,12 +362,15 @@ export async function localizeMultichannel(options) {
signal: options.signal,
onProgress: options.onProgress,
});
const locations = locationsFromBands(das.bands, plan.scan, plan.peakDetection);
const processingTimeMs = Math.max(0, monotonicNow() - processingStartedAt);
return {
...das,
sampleRate: plan.sampleRate,
sampleCount,
duration: sampleCount / plan.sampleRate,
locations: locationsFromBands(das.bands),
processingTimeMs,
locations,
};
}
@@ -298,8 +378,9 @@ export async function localizeMultichannel(options) {
* A fixed-capacity chunk buffer for real microphone adapters.
*
* append() is synchronous and remains safe while an analysis is awaiting CPU
* batches. analyzeAvailable() copies one window before yielding, consumes its
* hop, and returns no more than maximumResultsPerCall maps.
* batches. analyzeAvailable() copies one window before yielding, commits its
* hop only after successful localization, and returns no more than
* maximumResultsPerCall maps.
*/
export class StreamingDasLocalizer {
constructor(config) {
@@ -319,6 +400,7 @@ export class StreamingDasLocalizer {
if (this.analysisHopSamples > this.analysisWindowSamples) {
throw new RangeError("analysisHopSamples must not exceed analysisWindowSamples.");
}
this.analysisDeadlineMs = this.analysisHopSamples / this.plan.sampleRate * 1000;
this.maximumBufferedSamples = requirePositiveInteger(
config.maximumBufferedSamples ?? this.analysisWindowSamples * 4,
"maximumBufferedSamples",
@@ -370,15 +452,37 @@ export class StreamingDasLocalizer {
this.analysisAttempts = 0;
this.completedAnalyses = 0;
this.failedAnalyses = 0;
this.lastProcessingTimeMs = null;
this.maximumProcessingTimeMs = 0;
this.deadlineMisses = 0;
this.nextSequence = 0;
this.pendingAnalysis = null;
this.busy = false;
}
availableAnalysisCount() {
if (this.bufferedSamples < this.analysisWindowSamples) return 0;
return 1 + Math.floor(
(this.bufferedSamples - this.analysisWindowSamples) / this.analysisHopSamples,
const bufferEndSample = this.bufferStartSample + this.bufferedSamples;
if (!this.pendingAnalysis) {
if (this.bufferedSamples < this.analysisWindowSamples) return 0;
return 1 + Math.floor(
(this.bufferedSamples - this.analysisWindowSamples) / this.analysisHopSamples,
);
}
// The failed window remains in analysisChannels even if capture overflow
// advances the ring while localization is running. Count it once, then
// count windows that will remain after its successful hop commit.
const nextRingStart = Math.max(
this.bufferStartSample,
this.pendingAnalysis.startSample + this.analysisHopSamples,
);
const futureSamples = bufferEndSample - nextRingStart;
const futureCount = futureSamples < this.analysisWindowSamples
? 0
: 1 + Math.floor(
(futureSamples - this.analysisWindowSamples) / this.analysisHopSamples,
);
return 1 + futureCount;
}
getState() {
@@ -387,6 +491,7 @@ export class StreamingDasLocalizer {
channelCount: this.plan.channelCount,
analysisWindowSamples: this.analysisWindowSamples,
analysisHopSamples: this.analysisHopSamples,
analysisDeadlineMs: this.analysisDeadlineMs,
maximumBufferedSamples: this.maximumBufferedSamples,
maximumInputBufferBytes: this.maximumInputBufferBytes,
maximumResultsPerCall: this.maximumResultsPerCall,
@@ -398,6 +503,9 @@ export class StreamingDasLocalizer {
analysisAttempts: this.analysisAttempts,
completedAnalyses: this.completedAnalyses,
failedAnalyses: this.failedAnalyses,
lastProcessingTimeMs: this.lastProcessingTimeMs,
maximumProcessingTimeMs: this.maximumProcessingTimeMs,
deadlineMisses: this.deadlineMisses,
nextSequence: this.nextSequence,
busy: this.busy,
retainedInputBytes: this.retainedInputBytes,
@@ -481,39 +589,62 @@ export class StreamingDasLocalizer {
try {
while (
results.length < maximumResults &&
this.bufferedSamples >= this.analysisWindowSamples
(this.pendingAnalysis || this.bufferedSamples >= this.analysisWindowSamples)
) {
throwIfAborted(options.signal);
const startSample = this.bufferStartSample;
const retry = this.pendingAnalysis;
const startSample = retry?.startSample ?? this.bufferStartSample;
const endSample = startSample + this.analysisWindowSamples;
this.copyAnalysisWindow();
this.consume(this.analysisHopSamples, false);
if (!retry) this.copyAnalysisWindow();
this.analysisAttempts += 1;
let localized;
try {
const localized = await localizeMultichannel({
localized = await localizeMultichannel({
plan: this.plan,
channels: this.analysisChannels,
signal: options.signal,
onProgress: options.onProgress,
computeOptions: options.computeOptions,
});
const result = {
sequence: this.nextSequence,
startSample,
endSample,
startTime: startSample / this.plan.sampleRate,
endTime: endSample / this.plan.sampleRate,
centerTime: (startSample + endSample) / (2 * this.plan.sampleRate),
...localized,
};
this.nextSequence += 1;
this.completedAnalyses += 1;
results.push(result);
options.onResult?.(result);
} catch (error) {
this.pendingAnalysis = { startSample, endSample };
this.failedAnalyses += 1;
throw error;
}
const result = {
sequence: this.nextSequence,
startSample,
endSample,
startTime: startSample / this.plan.sampleRate,
endTime: endSample / this.plan.sampleRate,
centerTime: (startSample + endSample) / (2 * this.plan.sampleRate),
...localized,
deadlineMs: this.analysisDeadlineMs,
deadlineMet: localized.processingTimeMs <= this.analysisDeadlineMs,
};
this.pendingAnalysis = null;
const commitThroughSample = startSample + this.analysisHopSamples;
this.consume(
Math.min(
this.bufferedSamples,
Math.max(0, commitThroughSample - this.bufferStartSample),
),
false,
);
this.nextSequence += 1;
this.completedAnalyses += 1;
this.lastProcessingTimeMs = localized.processingTimeMs;
this.maximumProcessingTimeMs = Math.max(
this.maximumProcessingTimeMs,
localized.processingTimeMs,
);
if (!result.deadlineMet) this.deadlineMisses += 1;
results.push(result);
// Consumer failures must not turn a successfully committed analysis
// into a localization failure or make its window eligible for retry.
options.onResult?.(result);
}
return results;
} finally {
@@ -536,7 +667,11 @@ export class StreamingDasLocalizer {
this.analysisAttempts = 0;
this.completedAnalyses = 0;
this.failedAnalyses = 0;
this.lastProcessingTimeMs = null;
this.maximumProcessingTimeMs = 0;
this.deadlineMisses = 0;
this.nextSequence = 0;
this.pendingAnalysis = null;
}
}