diff --git a/src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java b/src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java index 33434423..bd06a9f9 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java @@ -75,6 +75,15 @@ public class FilterFactory private final static Logger mLog = LoggerFactory.getLogger(FilterFactory.class); private static final double PERFECT_RECONSTRUCTION_GAIN_AT_BAND_EDGE = -6.020599842071533; //decibel(0.5, 0.0) private static final double MARGIN_OF_ERROR = 0.0003; + + /** + * Two-sided synthesizer filter cutoff, in channel bandwidths. The synthesizer filter is centered on the boundary + * between the two channels, but the desired signal can be located anywhere across that boundary, so the pass band + * has to extend beyond the +/- 0.5 channel bandwidth centers of the channels. It can't extend much further because + * the channelizer transition regions of the outer channel edges alias into the composite beyond about +/- 0.72 + * channel bandwidths from the boundary. + */ + private static final double SYNTHESIZER_CUTOFF_CHANNEL_BANDWIDTHS = 1.5; private static final Map REMEZ_LOW_PASS_FILTERS = new ConcurrentHashMap<>(); /** @@ -1330,9 +1339,8 @@ public class FilterFactory { int filterLength = (channels * tapsPerChannel) - 1; - double cutoff = (channelBandwidth * 1.10) / (channelSampleRate * (double)channels); + double cutoff = (channelBandwidth * SYNTHESIZER_CUTOFF_CHANNEL_BANDWIDTHS) / (channelSampleRate * (double)channels); - //Design the prototype synthesizer with 105% of the channel bandwidth produced by the channelizer. float[] taps = FilterFactory.getKaiserSinc(filterLength, cutoff, 80.0f); //This is an odd length filter - increase the length by 1 by pre-padding a zero coefficient diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java index a21bf25b..3263c11c 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java @@ -23,6 +23,8 @@ import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.complex.InterleavedComplexSamples; import io.github.dsheirer.source.ISourceEventListener; import io.github.dsheirer.source.SourceEvent; +import java.util.ArrayList; +import java.util.BitSet; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import org.slf4j.Logger; @@ -33,11 +35,11 @@ public abstract class AbstractComplexPolyphaseChannelizer implements Listener mSourceChangeBroadcaster = new Broadcaster(); private List mChannels = new CopyOnWriteArrayList<>(); + private List mDispatchChannels = List.of(); private double mSampleRate; private int mChannelCount; private int mSubChannelCount; private double mChannelSampleRate; - protected long mCurrentSamplesTimestamp; /** * Complex sample polyphase channelizer @@ -61,17 +63,6 @@ public abstract class AbstractComplexPolyphaseChannelizer implements Listener channelResultsList) + protected int[] prepareChannels() { - for(PolyphaseChannelSource channel : mChannels) + List channels = new ArrayList<>(mChannels); + BitSet indexes = new BitSet(getChannelCount()); + + for(PolyphaseChannelSource channel : channels) { - channel.receiveChannelResults(channelResultsList, mCurrentSamplesTimestamp); + try + { + channel.applyPendingOutputProcessorUpdate(); + + for(int index : channel.getOutputProcessorIndexes()) + { + indexes.set(index); + } + } + catch(Throwable t) + { + mLog.error("Error preparing polyphase channel source to receive channel results", t); + } + } + + mDispatchChannels = channels; + return indexes.stream().toArray(); + } + + /** + * Dispatches the processed channel samples to the channels identified by prepareChannels(). An error in one + * channel does not affect delivery to the other channels. + * + * @param channelResults batch of I/Q samples per channel. Channels must retain the batch to use it after returning. + */ + protected void dispatch(ChannelResultsBatch channelResults) + { + for(PolyphaseChannelSource channel : mDispatchChannels) + { + try + { + channel.receiveChannelResults(channelResults); + } + catch(Throwable t) + { + mLog.error("Error dispatching channel results to polyphase channel source", t); + } } } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java index 8636ec67..491339ad 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java @@ -31,6 +31,22 @@ public class ChannelCalculator private static final DecimalFormat FREQUENCY_FORMAT = new DecimalFormat("0.00000"); private final static Logger mLog = LoggerFactory.getLogger(ChannelCalculator.class); + /** + * Maximum distance of a tuner channel's band edge from the center of a single polyphase channel, as a fraction of + * the channel bandwidth, that is sourced from that one channel. The channelizer filter rolls off to -6 dB at 0.5 + * channel bandwidths and this limit keeps the band edge loss to about 0.5 dB. + */ + private static final double SINGLE_CHANNEL_MAXIMUM_EDGE_OFFSET = 0.35; + + /** + * Maximum distance of a tuner channel's center from the center of a single polyphase channel, as a fraction of the + * channel bandwidth, where a channel that exceeds the band edge limit is still sourced from that one channel. The + * synthesis filter is centered on the channel boundary and is flat to about 0.6 channel bandwidths, so a wide tuner + * channel centered closer than this to a polyphase channel center extends too far from either adjacent boundary + * and loses less of its band edge from the channelizer roll-off than it would from the synthesis filter. + */ + private static final double SINGLE_CHANNEL_MAXIMUM_CENTER_OFFSET = 0.125; + /** * Policy to determine desired corrective action for methods that calculate an index for a frequency or a frequency * from an index where the frequency may exist either at the overlapping boundary of two indexes, or in the middle @@ -218,7 +234,8 @@ public class ChannelCalculator * @param tunerChannel specifying the desired channel frequency and bandwidth * * @return an array of one or more polyphase channelizer index(es) to target in order to produce an output channel - * for the desired frequency and bandwidth. + * for the desired frequency and bandwidth. Tuner channels that are no wider than a polyphase channel use one or + * two indexes, whichever preserves the tuner channel's band edges best. * * @throws IllegalArgumentException if the tuner channel's minimum or maximum frequency is not contained * within the channelizer's minimum or maximum frequency range @@ -245,6 +262,11 @@ public class ChannelCalculator "within two channel bandwidths of the sample rate."); } + if(tunerChannel.getBandwidth() <= getChannelBandwidth()) + { + return getIndexesForNarrowChannel(tunerChannel); + } + List indexes = new ArrayList<>(); indexes.add(minIndex); @@ -283,6 +305,36 @@ public class ChannelCalculator return indexes; } + /** + * Identifies the single index, or the two adjacent indexes ordered low to high in frequency, that best preserve the + * band edges of a tuner channel that is no wider than a polyphase channel. + * + * @param tunerChannel specifying the desired channel frequency and bandwidth + * @return list of one or two indexes + */ + private List getIndexesForNarrowChannel(TunerChannel tunerChannel) + { + long frequency = tunerChannel.getFrequency(); + int index = getIndexForFrequency(frequency, IndexBoundaryPolicy.ADJUST_POSITIVE); + IndexBoundaryPolicy policy = frequency >= getCenterFrequency() ? IndexBoundaryPolicy.ADJUST_POSITIVE : + IndexBoundaryPolicy.ADJUST_NEGATIVE; + double centerOffset = (frequency - getIndexCenterFrequency(index, policy)) / getChannelBandwidth(); + double edgeOffset = FastMath.abs(centerOffset) + (tunerChannel.getBandwidth() / 2.0 / getChannelBandwidth()); + + if(edgeOffset <= SINGLE_CHANNEL_MAXIMUM_EDGE_OFFSET || + FastMath.abs(centerOffset) <= SINGLE_CHANNEL_MAXIMUM_CENTER_OFFSET) + { + return List.of(index); + } + + if(centerOffset < 0) + { + return List.of(normalize(index - 1), index); + } + + return List.of(index, normalize(index + 1)); + } + /** * Calculates the index for the specified frequency. Certain frequencies can exist as both the maximum frequency * of one channel and the minimum frequency of an adjacent channel. The index boundary policy identifies your diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFT.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFT.java new file mode 100644 index 00000000..4b0be6ca --- /dev/null +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFT.java @@ -0,0 +1,133 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +/** + * Direct Discrete Fourier Transform of selected polyphase channels, as an alternative to an IFFT of all channels when + * only a few channels are in use. Produces the same channel values as a scaled IFFT. + * + * Results are written in place to each channel's position in each block. All other positions in the block retain their + * input values and must not be used. + */ +public abstract class ChannelDFT +{ + private final int mChannelCount; + private final float[][] mChannelInphaseCoefficients; + private final float[][] mChannelQuadratureCoefficients; + private float[] mResults = new float[0]; + + /** + * Constructs an instance + * @param channelCount as the size of the transform + */ + protected ChannelDFT(int channelCount) + { + mChannelCount = channelCount; + mChannelInphaseCoefficients = new float[channelCount][]; + mChannelQuadratureCoefficients = new float[channelCount][]; + } + + /** + * Largest number of channels where this implementation is faster than an IFFT of all channels. + */ + public abstract int getMaximumChannels(); + + /** + * Calculates the inphase and quadrature values for a channel. + * + * @param samples containing the block + * @param offset to the start of the block + * @param length of the block in array elements + * @param inphaseCoefficients for the channel + * @param quadratureCoefficients for the channel + * @param results array to receive the inphase and quadrature values + * @param resultsOffset for the inphase value, followed by the quadrature value + */ + protected abstract void transform(float[] samples, int offset, int length, float[] inphaseCoefficients, + float[] quadratureCoefficients, float[] results, int resultsOffset); + + /** + * Transforms the specified channels in each block in the samples array. + * @param samples containing one or more contiguous blocks of interleaved I/Q values for each channel + * @param channels to transform + */ + public void process(float[] samples, int[] channels) + { + int blockLength = mChannelCount * 2; + float[][] inphaseCoefficients = new float[channels.length][]; + float[][] quadratureCoefficients = new float[channels.length][]; + + for(int x = 0; x < channels.length; x++) + { + updateCoefficients(channels[x]); + inphaseCoefficients[x] = mChannelInphaseCoefficients[channels[x]]; + quadratureCoefficients[x] = mChannelQuadratureCoefficients[channels[x]]; + } + + if(mResults.length < channels.length * 2) + { + mResults = new float[channels.length * 2]; + } + + float[] results = mResults; + + for(int offset = 0; offset < samples.length; offset += blockLength) + { + //Each channel needs all of the input values, so write the results after all channels are calculated + for(int x = 0; x < channels.length; x++) + { + transform(samples, offset, blockLength, inphaseCoefficients[x], quadratureCoefficients[x], results, x * 2); + } + + for(int x = 0; x < channels.length; x++) + { + samples[offset + channels[x] * 2] = results[x * 2]; + samples[offset + channels[x] * 2 + 1] = results[x * 2 + 1]; + } + } + } + + /** + * Creates the coefficients for a channel on first use. Coefficients are arranged to align with interleaved I/Q + * samples so that each output value is a dot product of the block and a coefficient array. + * @param channel to update + */ + private void updateCoefficients(int channel) + { + if(mChannelInphaseCoefficients[channel] == null) + { + float[] inphase = new float[mChannelCount * 2]; + float[] quadrature = new float[mChannelCount * 2]; + + for(int x = 0; x < mChannelCount; x++) + { + double angle = 2.0 * Math.PI * channel * x / mChannelCount; + float cosine = (float)(Math.cos(angle) / mChannelCount); + float sine = (float)(Math.sin(angle) / mChannelCount); + inphase[2 * x] = cosine; + inphase[2 * x + 1] = -sine; + quadrature[2 * x] = sine; + quadrature[2 * x + 1] = cosine; + } + + mChannelInphaseCoefficients[channel] = inphase; + mChannelQuadratureCoefficients[channel] = quadrature; + } + } +} diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFTFactory.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFTFactory.java new file mode 100644 index 00000000..9defc97b --- /dev/null +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFTFactory.java @@ -0,0 +1,49 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +import io.github.dsheirer.vector.calibrate.CalibrationManager; +import io.github.dsheirer.vector.calibrate.CalibrationType; +import io.github.dsheirer.vector.calibrate.Implementation; + +/** + * Factory for selecting the optimal implementation of the channel DFT for this hardware, as determined by the + * Calibration Manager. + */ +public class ChannelDFTFactory +{ + /** + * Creates the optimal channel DFT implementation. + * @param channelCount as the size of the transform + * @return channel DFT + */ + public static ChannelDFT getChannelDFT(int channelCount) + { + Implementation implementation = CalibrationManager.getInstance().getImplementation(CalibrationType.CHANNEL_DFT); + + switch(implementation) + { + case VECTOR_SIMD_PREFERRED: + return new VectorChannelDFT(channelCount); + case SCALAR: + default: + return new ScalarChannelDFT(channelCount); + } + } +} diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsBatch.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsBatch.java new file mode 100644 index 00000000..590ec8a2 --- /dev/null +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsBatch.java @@ -0,0 +1,107 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +import java.util.Queue; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Batch of polyphase channelizer output blocks stored in a single contiguous array, where each block is an interleaved + * I/Q pair for each channel. The sample at channel offset C of block B is at index (B * block length) + C. + * + * The sample array is returned to the channelizer's pool for reuse when the last reference is released. This batch + * starts with one reference that is owned by the creator. Each consumer that keeps the batch beyond the method that + * received it must invoke retain() and then invoke release() when finished, and the samples must not be accessed after + * that. A reference that is never released only means that the array is garbage collected instead of being reused. + */ +public class ChannelResultsBatch +{ + private final float[] mSamples; + private final int mBlockLength; + private final long mTimestamp; + private final Queue mPool; + private final AtomicInteger mReferenceCount = new AtomicInteger(1); + + /** + * Constructs an instance + * @param samples for this batch + * @param blockLength as the number of array elements per block + * @param timestamp of the sample buffer that completed this batch + * @param pool to receive the sample array when the last reference is released + */ + public ChannelResultsBatch(float[] samples, int blockLength, long timestamp, Queue pool) + { + mSamples = samples; + mBlockLength = blockLength; + mTimestamp = timestamp; + mPool = pool; + } + + /** + * Sample array containing all blocks. + */ + public float[] getSamples() + { + return mSamples; + } + + /** + * Number of array elements per block. + */ + public int getBlockLength() + { + return mBlockLength; + } + + /** + * Number of blocks in this batch. + */ + public int getBlockCount() + { + return mSamples.length / mBlockLength; + } + + /** + * Timestamp of the sample buffer that completed this batch. + * @return time in milliseconds + */ + public long getTimestamp() + { + return mTimestamp; + } + + /** + * Adds a reference to this batch. + */ + public void retain() + { + mReferenceCount.incrementAndGet(); + } + + /** + * Releases a reference to this batch. The sample array is returned to the pool after the last release. + */ + public void release() + { + if(mReferenceCount.decrementAndGet() == 0) + { + mPool.offer(mSamples); + } + } +} diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java index 5aa49683..707eea94 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java @@ -23,9 +23,9 @@ import io.github.dsheirer.dsp.filter.design.FilterDesignException; import io.github.dsheirer.sample.complex.InterleavedComplexSamples; import io.github.dsheirer.util.Dispatcher; import java.text.DecimalFormat; -import java.util.ArrayList; import java.util.Arrays; -import java.util.List; +import java.util.Queue; +import java.util.concurrent.ArrayBlockingQueue; import org.apache.commons.math3.util.FastMath; import org.jtransforms.fft.FloatFFT_1D; import org.slf4j.Logger; @@ -39,9 +39,9 @@ import org.slf4j.LoggerFactory; * Processing for Communications Systems, p230-233. * * Samples are loaded into this filter one block at a time (1/2 channel count) and a filtered output is calculated - * to produce an overall 2x oversampled channel sample rate. Each sample block load is preceded by a serpentine - * shift of the existing sample blocks. We use the java System.arrayCopy() method which is able to leverage - * native processor intrinsics for efficiency. + * to produce an overall 2x oversampled channel sample rate. Instead of shifting the existing sample blocks for each + * block load, the window of filtered samples moves toward the start of an oversized sample buffer, and the window is + * only copied back to the end of the buffer when it reaches the start. * * The prototype filter for the channelizer is rearranged to align with the structure of the sample buffer. * @@ -72,12 +72,30 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann */ private static final int PROCESSED_CHANNEL_RESULTS_THRESHOLD = 1024; + /** + * Maximum queued channel results batches awaiting IFFT, about one second of samples. A full batch is dropped when + * the IFFT processor falls behind, which preserves the even-sized batch block parity. + */ + private static final int MAXIMUM_QUEUED_BATCHES = 50; + + /** + * Maximum batch sample arrays held for reuse: the queued IFFT batches plus the batches queued for channel output. + */ + private static final int MAXIMUM_POOLED_BATCHES = 2 * MAXIMUM_QUEUED_BATCHES; + //Sized to process 40 times per second private IFFTProcessorDispatcher mIFFTProcessorDispatcher = new IFFTProcessorDispatcher(25); private FloatFFT_1D mFFT; + private ChannelDFT mChannelDFT; private float[] mInlineSamples; private float[] mInlineFilter; + /** + * Number of additional blocks of space in the sample buffer. The window of filtered samples moves through this + * space one block at a time, so the window is copied once per this many blocks, instead of being shifted per block. + */ + private static final int SAMPLE_BUFFER_SPARE_BLOCKS = 32; + /** * Reusable sub-channel accumulator. This is scratch space that is fully overwritten on each process() invocation * and never escapes the method, so it is allocated once here (sized to the sub-channel count) and reused rather @@ -90,9 +108,13 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann private int[] mTopBlockMap; private int[] mMiddleBlockMap; private int mSampleBufferPointer; + private int mWindowOffset; private int mSamplesPerBlock; private int mTapsPerChannel; - private List mProcessedChannelResultsList = new ArrayList<>(); + private Queue mBatchPool; + private float[] mBatchSamples; + private int mBatchBlockCount; + private long mLatestSamplesTimestamp; /** * Creates a NMDPFB channelizer instance. @@ -130,10 +152,11 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann { super(sampleRate, getChannelCount(sampleRate)); - mTapsPerChannel = tapsPerChannel; - float[] filterTaps = FilterFactory.getSincM2Channelizer(getChannelSampleRate(), getChannelCount(), - mTapsPerChannel, false); + tapsPerChannel, false); + + //The filter designer can increase the taps per channel in order to meet the band edge objective + mTapsPerChannel = filterTaps.length / getChannelCount(); init(filterTaps); } @@ -152,11 +175,14 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann public void stop() { mIFFTProcessorDispatcher.stop(); + mBatchPool.clear(); } /** * Calculates the multiple of two number of channels that can be channelized from the specified sample rate so that - * each channel has a minimum bandwidth of the default channel bandwidth (12.5 kHz). + * each channel has a minimum bandwidth of the default channel bandwidth (25 kHz). The channel count is the IFFT + * size, so it is reduced as needed to have only factors of 2, 3 and 5, which keeps the IFFT efficient. Other prime + * factors are many times slower, for example 244 channels (2*2*61) takes 10x as long as 240 channels. * @param sampleRate to channelize * @return number of multiple of two channels that can be channelized. */ @@ -169,42 +195,24 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann channels--; } + while(channels > 2 && !hasOnlyFactorsOf2And3And5(channels)) + { + channels -= 2; + } + mLog.info("Sample Rate [" + DECIMAL_FORMAT.format(sampleRate) + "] providing [" + channels + "] channels at [" + DECIMAL_FORMAT.format(sampleRate / (double)channels) + "] Hz each"); return channels; } - /** - * Updates this channelizer to use the new sample rate. This method creates a new filter suitable for the - * sample rate and reinitializes all internal data structures to prepare for processing the new sample rate. - * @param sampleRate in hertz - */ - @Override - public void setRates(double sampleRate, int channelCount) - { - try - { - super.setRates(sampleRate, channelCount); - float[] filterTaps = FilterFactory.getSincM2Channelizer(getChannelSampleRate(), getChannelCount(), - mTapsPerChannel, false); - - init(filterTaps); - } - catch(FilterDesignException fde) - { - throw new IllegalArgumentException("Cannot create a channelizer filter for the specified sample rate [" + - sampleRate + "]"); - } - } - /** * Receives the complex sample buffer and processes the results through the channelizer. */ @Override public void receive(InterleavedComplexSamples complexSamples) { - mCurrentSamplesTimestamp = complexSamples.timestamp(); + mLatestSamplesTimestamp = complexSamples.timestamp(); float[] samples = complexSamples.samples(); @@ -223,7 +231,7 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann samplesToCopy = samplesDiff; } - System.arraycopy(samples, samplesPointer, mInlineSamples, mSampleBufferPointer, samplesToCopy); + System.arraycopy(samples, samplesPointer, mInlineSamples, mWindowOffset + mSampleBufferPointer, samplesToCopy); mSampleBufferPointer += samplesToCopy; samplesPointer += samplesToCopy; @@ -232,21 +240,54 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann if(mSampleBufferPointer >= mSamplesPerBlock) { //Filter buffered samples and produce a single sample across each of the polyphase channels - mProcessedChannelResultsList.add(process()); + process(); - if(mProcessedChannelResultsList.size() >= PROCESSED_CHANNEL_RESULTS_THRESHOLD) + if(++mBatchBlockCount >= PROCESSED_CHANNEL_RESULTS_THRESHOLD) { - mIFFTProcessorDispatcher.receive(new ArrayList<>(mProcessedChannelResultsList)); - mProcessedChannelResultsList.clear(); + ChannelResultsBatch batch = new ChannelResultsBatch(mBatchSamples, getSubChannelCount(), + mLatestSamplesTimestamp, mBatchPool); + + if(!mIFFTProcessorDispatcher.offer(batch)) + { + batch.release(); + } + + mBatchSamples = getBatchSamples(); + mBatchBlockCount = 0; + } + + //Move the window back one block to make room for a new block of samples, or when the window is at the + //start of the buffer, copy the samples that remain in the window to the end of the buffer + mWindowOffset -= mSamplesPerBlock; + + if(mWindowOffset < 0) + { + mWindowOffset = mSamplesPerBlock * SAMPLE_BUFFER_SPARE_BLOCKS; + System.arraycopy(mInlineSamples, 0, mInlineSamples, mWindowOffset + mSamplesPerBlock, + mInlineFilter.length - mSamplesPerBlock); } - //Right-shift the samples in the buffer over to make room for a new block of samples - System.arraycopy(mInlineSamples, 0, mInlineSamples, mSamplesPerBlock, (mInlineSamples.length - mSamplesPerBlock)); mSampleBufferPointer = 0; } } } + /** + * Indicates if the value has no prime factors other than 2, 3 and 5. + */ + private static boolean hasOnlyFactorsOf2And3And5(int value) + { + for(int factor : new int[]{2, 3, 5}) + { + while(value % factor == 0) + { + value /= factor; + } + } + + return value == 1; + } + /** * Creates a top-block processing accumulator map that maps each interim filter and sample index product * to the corresponding final output index for the array that will feed the IFFT. @@ -344,23 +385,34 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann /** - * Processes the sample buffer for each new block of sample data that is loaded and distributes the results to any - * registered channel listeners. + * Provides a reused batch sample array when one is available, or a new array. Reused arrays hold old samples, which + * doesn't matter since each batch is fully overwritten. + */ + private float[] getBatchSamples() + { + float[] samples = mBatchPool.poll(); + return samples != null ? samples : new float[PROCESSED_CHANNEL_RESULTS_THRESHOLD * getSubChannelCount()]; + } + + /** + * Processes the sample buffer for each new block of sample data that is loaded and writes the results to the next + * block of the current batch. * - * Note: mInlineSamples and mInlineFilter are both exactly (subChannelCount * tapsPerChannel) in length (see init() - * and getAlignedFilter()), so the fused multiply-accumulate loop below indexes every element of each array exactly - * once. The multiply and per-sub-channel accumulation are fused into a single pass, which removes the previous + * Note: mInlineFilter and the window of mInlineSamples are both exactly (subChannelCount * tapsPerChannel) in length + * (see init() and getAlignedFilter()), so the fused multiply-accumulate loop below indexes every element of each + * exactly once. The multiply and per-sub-channel accumulation are fused into a single pass, which removes the previous * per-call interim product array entirely and preserves bit-for-bit identical output: for each sub-channel the * tap products are summed in the same ascending tap order as before, and the JVM does not contract the separate * float multiply/add into an FMA without an explicit Math.fma() call. */ - private float[] process() + private void process() { final int subChannelCount = getSubChannelCount(); final int tapsPerChannel = mTapsPerChannel; final float[] inlineSamples = mInlineSamples; final float[] inlineFilter = mInlineFilter; final float[] filterAccumulator = mFilterAccumulator; + final int windowOffset = mWindowOffset; //Reset the reused accumulator to zero before summing this block's tap products Arrays.fill(filterAccumulator, 0.0f); @@ -374,13 +426,12 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann for(int channel = 0; channel < subChannelCount; channel++) { final int index = tapOffset + channel; - filterAccumulator[channel] += inlineSamples[index] * inlineFilter[index]; + filterAccumulator[channel] += inlineSamples[windowOffset + index] * inlineFilter[index]; } } - //This output array is handed off to the IFFT dispatcher thread and therefore must be freshly allocated - //per block (it escapes this method and cannot be reused). - final float[] processed = new float[subChannelCount]; + final float[] batchSamples = mBatchSamples; + final int batchOffset = mBatchBlockCount * subChannelCount; //Rearrange the accumulated sub-channels into the order required by the IFFT, alternating between the //top-block and middle-block orderings on successive blocks. @@ -388,12 +439,10 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann for(int x = 0; x < subChannelCount; x++) { - processed[x] = filterAccumulator[blockMap[x]]; + batchSamples[batchOffset + x] = filterAccumulator[blockMap[x]]; } mTopBlockIndicator = !mTopBlockIndicator; - - return processed; } /** @@ -405,6 +454,7 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann { int channelCount = getChannelCount(); mFFT = new FloatFFT_1D(channelCount); + mChannelDFT = ChannelDFTFactory.getChannelDFT(channelCount); int bufferLength = getSubChannelCount() * mTapsPerChannel; mSamplesPerBlock = channelCount; //Same as subChannelCount / 2 mTopBlockMap = getTopBlockMap(channelCount); @@ -415,46 +465,60 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann mTopBlockIndicator = (mTapsPerChannel % 2 == 0); mMiddleBlockMap = getMiddleBlockMap(channelCount); mInlineFilter = getAlignedFilter(coefficients, channelCount, mTapsPerChannel); - mInlineSamples = new float[bufferLength]; + mWindowOffset = mSamplesPerBlock * SAMPLE_BUFFER_SPARE_BLOCKS; + mInlineSamples = new float[mWindowOffset + bufferLength]; mFilterAccumulator = new float[getSubChannelCount()]; + mBatchPool = new ArrayBlockingQueue<>(MAXIMUM_POOLED_BATCHES); + mBatchSamples = getBatchSamples(); } /** - * Separate threaded processor to receive and enqueue filtered channel results buffers, perform IFFT on each array - * as required to align the phase of each polyphase channel, and then dispatch the results to any registered - * sample consumer channels. + * Separate threaded processor to receive and enqueue filtered channel results batches, transform each block as + * required to align the phase of each polyphase channel, and then dispatch the results to any registered sample + * consumer channels. When few channels are in use, only those channels are transformed, and the results for all + * other channels are not valid. */ - public class IFFTProcessorDispatcher extends Dispatcher> + public class IFFTProcessorDispatcher extends Dispatcher { public IFFTProcessorDispatcher(long interval) { - super("sdrtrunk polyphase ifft processor", interval); + super("sdrtrunk polyphase ifft processor", interval, MAXIMUM_QUEUED_BATCHES); - //We create a listener interface to receive the batched channel results arrays from the scheduled thread pool + //We create a listener interface to receive the batched channel results from the scheduled thread pool //dispatcher thread that is part of this continuous buffer processor. We perform an IFFT on each - //channel results array contained in each results buffer and then dispatch the buffer - //so that it can be distributed to each channel listener. - setListener(list -> { + //block in the batch and then dispatch the batch so that it can be distributed to each channel listener. + setListener(batch -> { try { - List processedChannelResults = new ArrayList<>(); + float[] samples = batch.getSamples(); + int[] channels = prepareChannels(); - for(float[] channelResults: list) + if(channels.length > 0 && channels.length <= mChannelDFT.getMaximumChannels()) { - if(channelResults != null) + //Rotate only the channels that are in use to the correct phase + mChannelDFT.process(samples, channels); + } + else + { + int blockLength = batch.getBlockLength(); + + for(int offset = 0; offset < samples.length; offset += blockLength) { //Rotate each of the channels to the correct phase using the IFFT - mFFT.complexInverse(channelResults, true); - processedChannelResults.add(channelResults); + mFFT.complexInverse(samples, offset, true); } } - dispatch(processedChannelResults); + dispatch(batch); } catch(Throwable t) { mLog.error("Error during IFFT and dispatch of processed channel results", t); } + finally + { + batch.release(); + } }); } } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java index 44cf02b9..989fdb76 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java @@ -35,13 +35,12 @@ import io.github.dsheirer.source.tuner.channel.TunerChannelSource; import io.github.dsheirer.util.Dispatcher; import java.text.DecimalFormat; import java.util.ArrayList; -import java.util.HashMap; import java.util.Iterator; import java.util.List; -import java.util.Map; import java.util.SortedSet; import java.util.TreeSet; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.math3.util.FastMath; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -55,9 +54,9 @@ import org.slf4j.LoggerFactory; * notifications received from active DDC polyphase channel sources to adjust sample streams as required. * * Channel bandwidth and channel count are determined by the sample rate of the baseband buffer stream provider. This - * class is currently designed to provide channels each with a minimum usable bandwidth of 12.5 kHz and oversampled by - * 2.0 to a minimum of 25.0 kHz channel sample rate. If the baseband stream provider sample rate is not evenly - * divisible by 12.5 kHz channels for an even number of channels, the channel bandwidth will be increased. + * class is currently designed to provide channels each with a minimum bandwidth of 25 kHz and oversampled by 2.0 to a + * minimum of 50 kHz channel sample rate. If the baseband stream provider sample rate is not evenly divisible by 25 kHz + * channels for an even number of channels with an efficient IFFT size, the channel bandwidth will be increased. * * Note: add this channel manager as a source event listener to the complex buffer provider to ensure this manager * adapts to changes in source frequency and sample rate. @@ -67,9 +66,9 @@ public class PolyphaseChannelManager implements ISourceEventProcessor private static final DecimalFormat FREQUENCY_FORMAT = new DecimalFormat("0.00000"); private static final LoggingSuppressor LOGGING_SUPPRESSOR = new LoggingSuppressor(LoggerFactory.getLogger(PolyphaseChannelManager.class)); private final static Logger mLog = LoggerFactory.getLogger(PolyphaseChannelManager.class); - private static final double MINIMUM_CHANNEL_BANDWIDTH = 25000.0; private static final double CHANNEL_OVERSAMPLING = 2.0; private static final int POLYPHASE_CHANNELIZER_TAPS_PER_CHANNEL = 9; + private static final int MAXIMUM_QUEUED_NATIVE_BUFFERS = 500; private Broadcaster mSourceEventBroadcaster = new Broadcaster<>(); private TunerController mTunerController; @@ -80,7 +79,6 @@ public class PolyphaseChannelManager implements ISourceEventProcessor private ChannelSourceEventListener mChannelSourceEventListener = new ChannelSourceEventListener(); private NativeBufferReceiver mNativeBufferReceiver = new NativeBufferReceiver(); private Dispatcher mBufferDispatcher; - private Map mOutputProcessorFilters = new HashMap<>(); private boolean mRunning = true; /** @@ -97,17 +95,11 @@ public class PolyphaseChannelManager implements ISourceEventProcessor mTunerController = tunerController; - int channelCount = (int)(tunerController.getSampleRate() / MINIMUM_CHANNEL_BANDWIDTH); - - //Ensure channel count is an even integer since we're using a 2x oversampling polyphase channelizer - if(channelCount % 2 != 0) - { - channelCount--; - } + int channelCount = ComplexPolyphaseChannelizerM2.getChannelCount(tunerController.getSampleRate()); mChannelCalculator = new ChannelCalculator(tunerController.getSampleRate(), channelCount, tunerController.getFrequency(), CHANNEL_OVERSAMPLING); - mBufferDispatcher = new Dispatcher("sdrtrunk polyphase buffer processor", 10); + mBufferDispatcher = new Dispatcher("sdrtrunk polyphase buffer processor", 10, MAXIMUM_QUEUED_NATIVE_BUFFERS); mBufferDispatcher.setListener(mNativeBufferReceiver); } @@ -343,9 +335,6 @@ public class PolyphaseChannelManager implements ISourceEventProcessor { mLog.error("Could not create filter for polyphase channelizer for sample rate [" + tunerSampleRate + "]", fde); } - - //Clear any previous channel synthesis filters so they can be recreated for the new channel sample rate - mOutputProcessorFilters.clear(); } } @@ -358,16 +347,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor { for(PolyphaseChannelSource channelSource: mChannelSources) { - try - { - channelSource.updateOutputProcessor(mChannelCalculator, mFilterManager); - } - catch(IllegalArgumentException iae) - { - mLog.error("Error updating polyphase channel source output processor following tuner frequency or " + - "sample rate change"); - stopChannelSource(channelSource); - } + channelSource.updateOutputProcessor(mChannelCalculator, mFilterManager); } } @@ -460,13 +440,12 @@ public class PolyphaseChannelManager implements ISourceEventProcessor * Processes the incoming buffer stream from the provider and transfers the buffers to the polyphase channelizer. * * This monitor incorporates a source event handler that queues a center frequency update so that it can be - * handled on the buffer processing thread, avoiding having to lock on the output processor thread. Since we - * anticipate that these two threads will contend for access to this update required flag, we use an update lock - * to protect access to the flag. + * handled on the buffer processing thread, avoiding having to lock on the output processor thread. The update + * required flag is cleared before the update is applied so that a change that arrives during an update is not lost. */ public class NativeBufferReceiver implements Listener { - private boolean mOutputProcessorUpdateRequired = false; + private final AtomicBoolean mOutputProcessorUpdateRequired = new AtomicBoolean(); /** * Processes tuner center frequency change source events to flag when output processors need updating. @@ -480,7 +459,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor { //Update the channel calculator frequency so that it's ready when the output processor update occurs mChannelCalculator.setCenterFrequency(frequency); - mOutputProcessorUpdateRequired = true; + mOutputProcessorUpdateRequired.set(true); } } @@ -492,7 +471,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor @Override public void receive(INativeBuffer nativeBuffer) { - if(mOutputProcessorUpdateRequired) + if(mOutputProcessorUpdateRequired.compareAndSet(true, false)) { try { @@ -502,7 +481,6 @@ public class PolyphaseChannelManager implements ISourceEventProcessor { mLog.error("Error updating polyphase channel output processors"); } - mOutputProcessorUpdateRequired = false; } if(mPolyphaseChannelizer != null) diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java index 9503277a..1834f66b 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java @@ -28,8 +28,11 @@ import io.github.dsheirer.source.SourceEvent; import io.github.dsheirer.source.tuner.channel.TunerChannel; import io.github.dsheirer.source.tuner.channel.TunerChannelSource; import io.github.dsheirer.source.tuner.frequency.TunerFrequencyErrorManager; +import io.github.dsheirer.util.ThreadPool; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,9 +51,10 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen private List mOutputProcessorIndexes = new ArrayList<>(); private double mTunerSampleRate; private double mTunerCenterFrequency; - private long mFrequencyCorrection; - private boolean mPendingFrequencyCorrectionUpdate = false; - private PendingOutputProcessorUpdate mPendingOutputProcessorUpdate; + private volatile long mFrequencyCorrection; + private final AtomicBoolean mPendingFrequencyCorrectionUpdate = new AtomicBoolean(); + private final AtomicReference mPendingOutputProcessorUpdate = new AtomicReference<>(); + private boolean mOutputProcessorUpdateFailed = false; /** * Constructs an instance @@ -78,7 +82,7 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen public void setFrequencyCorrection(long correction) { mFrequencyCorrection = correction; - mPendingFrequencyCorrectionUpdate = true; + mPendingFrequencyCorrectionUpdate.set(true); } /** @@ -170,14 +174,14 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen * @param filterManager for designing and caching synthesis filters */ public void updateOutputProcessor(ChannelCalculator channelCalculator, SynthesisFilterManager filterManager) - throws IllegalArgumentException { - mPendingOutputProcessorUpdate = new PendingOutputProcessorUpdate(channelCalculator, filterManager); + mPendingOutputProcessorUpdate.set(new PendingOutputProcessorUpdate(channelCalculator, filterManager)); } /** - * Performs the output processor update. This method must be invoked on the channel results calling thread to - * ensure we don't have thread contention on the output processor. + * Performs the output processor update. This method must be invoked on the channel results calling thread so that + * replacing the output processor doesn't overlap with delivery of channel results to it. The output processor + * applies the configuration changes on its own dispatcher thread. * @param channelCalculator for updates * @param filterManager for updates * @throws IllegalArgumentException if there's an issue. @@ -280,31 +284,26 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen * queued for processing to extract the target channel samples, process them for frequency correction and/or * channel aggregation, and dispatch the results to the downstream sample listener/consumer. * - * @param channelResultsList containing a list of polyphase channelizer output arrays. - * @param currentSamplesTimestamp for the samples + * @param channelResults batch of polyphase channelizer output. */ - public void receiveChannelResults(List channelResultsList, long currentSamplesTimestamp) + public void receiveChannelResults(ChannelResultsBatch channelResults) { - if(mPendingOutputProcessorUpdate != null) + if(mOutputProcessorUpdateFailed) { - ChannelCalculator channelCalculator = mPendingOutputProcessorUpdate.getChannelCalculator(); - SynthesisFilterManager filterManager = mPendingOutputProcessorUpdate.getSynthesisFilterManager(); - mPendingOutputProcessorUpdate = null; - doUpdateOutputProcessor(channelCalculator, filterManager); + return; } //Apply channel decider requested frequency correction - if(mPendingFrequencyCorrectionUpdate) + if(mPendingFrequencyCorrectionUpdate.compareAndSet(true, false)) { mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); - mPendingFrequencyCorrectionUpdate = false; } try { if(mPolyphaseChannelOutputProcessor != null) { - mPolyphaseChannelOutputProcessor.receiveChannelResults(channelResultsList, currentSamplesTimestamp); + mPolyphaseChannelOutputProcessor.receiveChannelResults(channelResults); } } catch(Exception e) @@ -313,6 +312,43 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen } } + /** + * Applies a pending output processor update. The channelizer invokes this method before it transforms each batch, + * because the batch only contains valid results for the polyphase channel indexes that are in use at that time. + * + * If the update fails, for example when the tuner center frequency change means that this channel can no longer be + * sourced, this source stops processing results and signals an error to the consumer. + */ + public void applyPendingOutputProcessorUpdate() + { + if(mOutputProcessorUpdateFailed) + { + return; + } + + PendingOutputProcessorUpdate pendingUpdate = mPendingOutputProcessorUpdate.getAndSet(null); + + if(pendingUpdate != null) + { + ChannelCalculator channelCalculator = pendingUpdate.getChannelCalculator(); + SynthesisFilterManager filterManager = pendingUpdate.getSynthesisFilterManager(); + + try + { + doUpdateOutputProcessor(channelCalculator, filterManager); + } + catch(IllegalArgumentException iae) + { + mLog.error("Error updating polyphase channel output processor following tuner frequency or sample " + + "rate change - " + iae.getMessage()); + mOutputProcessorUpdateFailed = true; + + //Stopping the channel is a blocking operation, so don't do it on the channelizer's dispatch thread + ThreadPool.CACHED.execute(() -> setError(iae.getMessage())); + } + } + } + /** * Downstream channel sample rate * diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ScalarChannelDFT.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ScalarChannelDFT.java new file mode 100644 index 00000000..97720213 --- /dev/null +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ScalarChannelDFT.java @@ -0,0 +1,63 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +/** + * Scalar implementation of the channel DFT. + */ +public class ScalarChannelDFT extends ChannelDFT +{ + /** + * Break-even with a mixed-radix IFFT is about 3-6 channels across typical channel counts. + */ + private static final int MAXIMUM_CHANNELS = 3; + + /** + * Constructs an instance + * @param channelCount as the size of the transform + */ + public ScalarChannelDFT(int channelCount) + { + super(channelCount); + } + + @Override + public int getMaximumChannels() + { + return MAXIMUM_CHANNELS; + } + + @Override + protected void transform(float[] samples, int offset, int length, float[] inphaseCoefficients, + float[] quadratureCoefficients, float[] results, int resultsOffset) + { + float inphase = 0.0f; + float quadrature = 0.0f; + + for(int x = 0; x < length; x++) + { + float sample = samples[offset + x]; + inphase += sample * inphaseCoefficients[x]; + quadrature += sample * quadratureCoefficients[x]; + } + + results[resultsOffset] = inphase; + results[resultsOffset + 1] = quadrature; + } +} diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/SynthesisFilterManager.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/SynthesisFilterManager.java index 521ab4b8..a7c46b3b 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/SynthesisFilterManager.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/SynthesisFilterManager.java @@ -29,7 +29,7 @@ import java.util.Map; */ public class SynthesisFilterManager { - private static final int POLYPHASE_SYNTHESIZER_TAPS_PER_CHANNEL = 9; + private static final int POLYPHASE_SYNTHESIZER_TAPS_PER_CHANNEL = 24; private static final String SEPARATOR = "-"; private Map mFilterMap = new HashMap<>(); diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/VectorChannelDFT.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/VectorChannelDFT.java new file mode 100644 index 00000000..8bc9571b --- /dev/null +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/VectorChannelDFT.java @@ -0,0 +1,83 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +import jdk.incubator.vector.FloatVector; +import jdk.incubator.vector.VectorOperators; +import jdk.incubator.vector.VectorSpecies; + +/** + * SIMD implementation of the channel DFT using the preferred vector size for the hardware. + */ +public class VectorChannelDFT extends ChannelDFT +{ + private static final VectorSpecies VECTOR_SPECIES = FloatVector.SPECIES_PREFERRED; + + /** + * Break-even with an IFFT scales with the vector size and is about 20-40 channels with 16 lanes. Using the lane + * count stays below the break-even at each vector size. + */ + private static final int MAXIMUM_CHANNELS = VECTOR_SPECIES.length(); + + /** + * Constructs an instance + * @param channelCount as the size of the transform + */ + public VectorChannelDFT(int channelCount) + { + super(channelCount); + } + + @Override + public int getMaximumChannels() + { + return MAXIMUM_CHANNELS; + } + + @Override + protected void transform(float[] samples, int offset, int length, float[] inphaseCoefficients, + float[] quadratureCoefficients, float[] results, int resultsOffset) + { + FloatVector inphaseAccumulator = FloatVector.zero(VECTOR_SPECIES); + FloatVector quadratureAccumulator = FloatVector.zero(VECTOR_SPECIES); + int laneCount = VECTOR_SPECIES.length(); + int vectorLength = VECTOR_SPECIES.loopBound(length); + int x = 0; + + for(; x < vectorLength; x += laneCount) + { + FloatVector sample = FloatVector.fromArray(VECTOR_SPECIES, samples, offset + x); + inphaseAccumulator = sample.fma(FloatVector.fromArray(VECTOR_SPECIES, inphaseCoefficients, x), inphaseAccumulator); + quadratureAccumulator = sample.fma(FloatVector.fromArray(VECTOR_SPECIES, quadratureCoefficients, x), quadratureAccumulator); + } + + float inphase = inphaseAccumulator.reduceLanes(VectorOperators.ADD); + float quadrature = quadratureAccumulator.reduceLanes(VectorOperators.ADD); + + for(; x < length; x++) + { + float sample = samples[offset + x]; + inphase += sample * inphaseCoefficients[x]; + quadrature += sample * quadratureCoefficients[x]; + } + + results[resultsOffset] = inphase; + results[resultsOffset + 1] = quadrature; + } +} diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java index 6f721982..eb1848f3 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java @@ -18,11 +18,14 @@ */ package io.github.dsheirer.dsp.filter.channelizer.output; +import io.github.dsheirer.dsp.filter.channelizer.ChannelResultsBatch; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.complex.ComplexSamples; import io.github.dsheirer.source.heartbeat.HeartbeatManager; import io.github.dsheirer.util.Dispatcher; -import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicLong; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,9 +33,17 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP { private final static Logger mLog = LoggerFactory.getLogger(ChannelOutputProcessor.class); - private Dispatcher> mChannelResultsDispatcher; + /** + * Maximum queued channel results batches, about one second of samples. A full batch is dropped when this channel + * falls behind, which preserves the even-sized batch block parity that two-channel synthesis relies on. + */ + private static final int MAXIMUM_QUEUED_BATCHES = 50; + + private Dispatcher mChannelResultsDispatcher; + private final Queue mPendingUpdates = new ConcurrentLinkedQueue<>(); + private final AtomicLong mBatchSequence = new AtomicLong(); private HeartbeatManager mHeartbeatManager; - protected Listener mComplexSamplesListener; + protected volatile Listener mComplexSamplesListener; private int mInputChannelCount; private long mCurrentSampleTimestamp = System.currentTimeMillis(); @@ -49,19 +60,53 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP mInputChannelCount = inputChannelCount; //Process 1/10th of the sample rate per second at a rate of 20 times a second (200% of anticipated rate) mHeartbeatManager = heartbeatManager; - mChannelResultsDispatcher = new Dispatcher(threadName,50, mHeartbeatManager); - mChannelResultsDispatcher.setListener(floats -> { + mChannelResultsDispatcher = new Dispatcher(threadName,50, mHeartbeatManager, MAXIMUM_QUEUED_BATCHES); + mChannelResultsDispatcher.setListener(queued -> { try { - process(floats); + applyPendingUpdates(queued.sequence()); + mCurrentSampleTimestamp = queued.channelResults().getTimestamp(); + process(queued.channelResults()); } catch(Throwable t) { mLog.error("Error processing channel results", t); } + finally + { + queued.channelResults().release(); + } }); } + /** + * Queues a configuration change to apply on the dispatcher thread, so that configuration changes don't overlap with + * sample processing. The change takes effect at the first channel results batch received after this method is + * invoked and batches that are already queued continue to use the previous configuration. A change is not lost if + * that batch is dropped, since it then takes effect at the following batch. + * @param update to apply + */ + protected void enqueueUpdate(Runnable update) + { + mPendingUpdates.add(new PendingUpdate(update, mBatchSequence.get())); + } + + /** + * Applies queued configuration changes that take effect at or before the batch sequence number. This should only + * be invoked on the dispatcher thread. + * @param sequence of the batch about to be processed + */ + private void applyPendingUpdates(long sequence) + { + PendingUpdate update; + + while((update = mPendingUpdates.peek()) != null && update.lastSequenceBefore() < sequence) + { + mPendingUpdates.poll(); + update.update().run(); + } + } + /** * Timestamp for the current series of samples. * @return time in milliseconds to use with assembled complex sample buffers. @@ -103,21 +148,35 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP } @Override - public void receiveChannelResults(List channelResultsList, long timestamp) + public void receiveChannelResults(ChannelResultsBatch channelResults) { - mChannelResultsDispatcher.receive(channelResultsList); - mCurrentSampleTimestamp = timestamp; + channelResults.retain(); + + if(!mChannelResultsDispatcher.offer(new QueuedChannelResults(channelResults, mBatchSequence.incrementAndGet()))) + { + channelResults.release(); + } } /** - * Sub-class implementation to process one polyphase channelizer result array. + * Sub-class implementation to process one polyphase channelizer results batch. * @param channelResults to process */ - public abstract void process(List channelResults); + public abstract void process(ChannelResultsBatch channelResults); @Override public int getInputChannelCount() { return mInputChannelCount; } + + /** + * Channel results batch and its position in the sequence of received batches. + */ + private record QueuedChannelResults(ChannelResultsBatch channelResults, long sequence) {} + + /** + * Configuration change and the sequence number of the last batch received before the change was requested. + */ + private record PendingUpdate(Runnable update, long lastSequenceBefore) {} } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java index 6da0c0bf..f0f94066 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java @@ -18,6 +18,7 @@ */ package io.github.dsheirer.dsp.filter.channelizer.output; +import io.github.dsheirer.dsp.filter.channelizer.ChannelResultsBatch; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.complex.ComplexSamples; import java.util.List; @@ -41,9 +42,8 @@ public interface IPolyphaseChannelOutputProcessor /** * Receive and enqueue output results from the polyphase analysis channelizer * @param channelResults to enqueue - * @param timestamp for the first channel results buffer */ - void receiveChannelResults(List channelResults, long timestamp); + void receiveChannelResults(ChannelResultsBatch channelResults); /** * Listener to receive assembled complex samples buffers diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java index b4bf591a..521e03fe 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java @@ -18,6 +18,7 @@ */ package io.github.dsheirer.dsp.filter.channelizer.output; +import io.github.dsheirer.dsp.filter.channelizer.ChannelResultsBatch; import io.github.dsheirer.sample.complex.ComplexSamples; import io.github.dsheirer.source.heartbeat.HeartbeatManager; import java.util.List; @@ -80,27 +81,32 @@ public class OneChannelOutputProcessor extends ChannelOutputProcessor //Set the channelized output results offset to twice the channel index to account for each channel having //an I/Q pair - mChannelOffset = indexes.get(0) * 2; + int channelOffset = indexes.get(0) * 2; + enqueueUpdate(() -> mChannelOffset = channelOffset); } @Override public void setFrequencyOffset(long frequency) { - mMixerAssembler.getMixer().setFrequency(frequency); + enqueueUpdate(() -> mMixerAssembler.getMixer().setFrequency(frequency)); } /** * Extract the channel from the channel results array and pass to the assembler. The assembler will * apply frequency translation and gain and indicate when a buffer is fully assembled. * - * @param channelResultsList to process containing a list of a list of channel array of I/Q sample pairs (I0,Q0,I1,Q1...In,Qn) + * @param channelResults to process containing blocks of channel I/Q sample pairs (I0,Q0,I1,Q1...In,Qn) */ @Override - public void process(List channelResultsList) + public void process(ChannelResultsBatch channelResults) { - for(float[] channelResults: channelResultsList) + float[] samples = channelResults.getSamples(); + int blockLength = channelResults.getBlockLength(); + int channelOffset = mChannelOffset; + + for(int index = channelOffset; index < samples.length; index += blockLength) { - mMixerAssembler.receive(channelResults[mChannelOffset], channelResults[mChannelOffset + 1]); + mMixerAssembler.receive(samples[index], samples[index + 1]); if(mMixerAssembler.hasBuffer()) { diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java index dd6938cf..9e3820f0 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java @@ -18,6 +18,7 @@ */ package io.github.dsheirer.dsp.filter.channelizer.output; +import io.github.dsheirer.dsp.filter.channelizer.ChannelResultsBatch; import io.github.dsheirer.sample.complex.ComplexSamples; import io.github.dsheirer.source.heartbeat.HeartbeatManager; import java.util.List; @@ -51,7 +52,7 @@ public class TwoChannelOutputProcessor extends ChannelOutputProcessor setPolyphaseChannelIndices(channelIndexes); mMixerAssembler = new TwoChannelMixerAssembler(gain); mMixerAssembler.getMixer().setSampleRate(sampleRate); - setSynthesisFilter(filter); + mMixerAssembler.setSynthesisFilter(filter); } @Override @@ -72,13 +73,13 @@ public class TwoChannelOutputProcessor extends ChannelOutputProcessor @Override public void setFrequencyOffset(long frequencyOffset) { - mMixerAssembler.getMixer().setFrequency(frequencyOffset); + enqueueUpdate(() -> mMixerAssembler.getMixer().setFrequency(frequencyOffset)); } @Override public void setSynthesisFilter(float[] filter) { - mMixerAssembler.setSynthesisFilter(filter); + enqueueUpdate(() -> mMixerAssembler.setSynthesisFilter(filter)); } /** @@ -97,34 +98,45 @@ public class TwoChannelOutputProcessor extends ChannelOutputProcessor //Set the channelized output results offsets to twice the channel index to account for each channel having //an I/Q pair - mChannelOffset1 = indexes.get(0) * 2; - mChannelOffset2 = indexes.get(1) * 2; + int channelOffset1 = indexes.get(0) * 2; + int channelOffset2 = indexes.get(1) * 2; + + enqueueUpdate(() -> { + mChannelOffset1 = channelOffset1; + mChannelOffset2 = channelOffset2; + }); } /** * Extract the channel from the channel results array, apply frequency translation, and deliver the * extracted frequency-corrected channel I/Q sample set to the complex sample listener. * - * @param channelResultsList to process containing a list of a list of an array of channel I/Q sample pairs (I0,Q0,I1,Q1...In,Qn) + * @param channelResults to process containing blocks of channel I/Q sample pairs (I0,Q0,I1,Q1...In,Qn) */ @Override - public void process(List channelResultsList) + public void process(ChannelResultsBatch channelResults) { + float[] samples = channelResults.getSamples(); + int blockLength = channelResults.getBlockLength(); + int blockCount = channelResults.getBlockCount(); + int channelOffset1 = mChannelOffset1; + int channelOffset2 = mChannelOffset2; + //Translate channel 2 up by FS/2 (multiply by (-1)^n) so that it sits adjacent to channel 1 for synthesis. The //channelizer dispatches even-sized batches, so the index within the batch carries the absolute block parity. - for(int x = 0; x < channelResultsList.size(); x++) + for(int x = 0; x < blockCount; x++) { - float[] channelResults = channelResultsList.get(x); + int block = x * blockLength; if((x & 1) == 0) { - mMixerAssembler.receive(channelResults[mChannelOffset1], channelResults[mChannelOffset1 + 1], - channelResults[mChannelOffset2], channelResults[mChannelOffset2 + 1]); + mMixerAssembler.receive(samples[block + channelOffset1], samples[block + channelOffset1 + 1], + samples[block + channelOffset2], samples[block + channelOffset2 + 1]); } else { - mMixerAssembler.receive(channelResults[mChannelOffset1], channelResults[mChannelOffset1 + 1], - -channelResults[mChannelOffset2], -channelResults[mChannelOffset2 + 1]); + mMixerAssembler.receive(samples[block + channelOffset1], samples[block + channelOffset1 + 1], + -samples[block + channelOffset2], -samples[block + channelOffset2 + 1]); } if(mMixerAssembler.hasBuffer()) diff --git a/src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java b/src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java index 63d60c14..0df4335d 100644 --- a/src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java +++ b/src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java @@ -270,7 +270,7 @@ public class SynthesizerViewer extends JFrame { try { - float[] taps = FilterFactory.getSincM2Synthesizer( 25000.0, 12500.0, 2, 12); + float[] taps = FilterFactory.getSincM2Synthesizer( 25000.0, 12500.0, 2, 24); mSynthesizer = new TwoChannelSynthesizerM2(taps); } catch(FilterDesignException fde) diff --git a/src/main/java/io/github/dsheirer/util/Dispatcher.java b/src/main/java/io/github/dsheirer/util/Dispatcher.java index 3d0a53fe..4d17acfc 100644 --- a/src/main/java/io/github/dsheirer/util/Dispatcher.java +++ b/src/main/java/io/github/dsheirer/util/Dispatcher.java @@ -23,7 +23,9 @@ import io.github.dsheirer.sample.Listener; import io.github.dsheirer.source.heartbeat.HeartbeatManager; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; @@ -41,7 +43,8 @@ import org.slf4j.LoggerFactory; public class Dispatcher implements Listener { private final static Logger mLog = LoggerFactory.getLogger(Dispatcher.class); - private final LinkedTransferQueue mQueue = new LinkedTransferQueue<>(); + private static final long OVERFLOW_LOG_INTERVAL_MILLISECONDS = 10000; + private final BlockingQueue mQueue; private Listener mListener; private final AtomicBoolean mRunning = new AtomicBoolean(); private String mThreadName; @@ -49,6 +52,8 @@ public class Dispatcher implements Listener private ScheduledFuture mScheduledFuture; private final long mInterval; private HeartbeatManager mHeartbeatManager; + private long mDroppedCount = 0; + private long mLastOverflowLogTimestamp = 0; /** * Constructs an instance of a Dispatcher with integrated heartbeat support. @@ -63,7 +68,20 @@ public class Dispatcher implements Listener } /** - * Constructs an instance + * Constructs an instance of a bounded Dispatcher with integrated heartbeat support. + * @param threadName to name the dispatcher thread + * @param interval for processing each batch in milliseconds. + * @param heartbeatManager to receive a heartbeat command at each processing interval. + * @param maximumQueueSize for queued elements. Elements received while the queue is full are dropped. + */ + public Dispatcher(String threadName, long interval, HeartbeatManager heartbeatManager, int maximumQueueSize) + { + this(threadName, interval, maximumQueueSize); + mHeartbeatManager = heartbeatManager; + } + + /** + * Constructs an instance with an unbounded queue. * @param threadName to name the dispatcher thread * @param interval for processing each batch in milliseconds. */ @@ -71,6 +89,20 @@ public class Dispatcher implements Listener { mThreadName = threadName; mInterval = interval; + mQueue = new LinkedTransferQueue<>(); + } + + /** + * Constructs a bounded instance. + * @param threadName to name the dispatcher thread + * @param interval for processing each batch in milliseconds. + * @param maximumQueueSize for queued elements. Elements received while the queue is full are dropped. + */ + public Dispatcher(String threadName, long interval, int maximumQueueSize) + { + mThreadName = threadName; + mInterval = interval; + mQueue = new LinkedBlockingQueue<>(maximumQueueSize); } /** @@ -94,16 +126,46 @@ public class Dispatcher implements Listener /** * Primary input method for adding buffers to this processor. Note: incoming buffers will be ignored if this * processor is in a stopped state. You must invoke start() to allow incoming buffers and initiate buffer - * processing. + * processing. If this processor has a bounded queue and the queue is full, the incoming buffer is dropped. * * @param e to enqueue for distribution to a registered listener */ public void receive(E e) { - if(mRunning.get()) + offer(e); + } + + /** + * Adds the element to this processor's queue, for when the caller has to take action on an element that isn't + * accepted. + * + * @param e to enqueue for distribution to a registered listener + * @return true if the element was queued, or false if this processor is stopped or a bounded queue is full. + */ + public boolean offer(E e) + { + if(!mRunning.get()) { - mQueue.add(e); + return false; } + + boolean queued = mQueue.offer(e); + + if(!queued) + { + mDroppedCount++; + long now = System.currentTimeMillis(); + + if(now - mLastOverflowLogTimestamp >= OVERFLOW_LOG_INTERVAL_MILLISECONDS) + { + mLog.warn("Dispatcher [" + mThreadName + "] queue is full, consumer is not keeping up - dropped [" + + mDroppedCount + "] elements"); + mLastOverflowLogTimestamp = now; + mDroppedCount = 0; + } + } + + return queued; } /** diff --git a/src/main/java/io/github/dsheirer/vector/calibrate/CalibrationManager.java b/src/main/java/io/github/dsheirer/vector/calibrate/CalibrationManager.java index 1644e0fa..dabd85e9 100644 --- a/src/main/java/io/github/dsheirer/vector/calibrate/CalibrationManager.java +++ b/src/main/java/io/github/dsheirer/vector/calibrate/CalibrationManager.java @@ -22,6 +22,7 @@ package io.github.dsheirer.vector.calibrate; import io.github.dsheirer.dsp.am.AmDemodulatorCalibration; import io.github.dsheirer.preference.UserPreferences; import io.github.dsheirer.preference.calibration.VectorCalibrationPreference; +import io.github.dsheirer.vector.calibrate.channelizer.ChannelDFTCalibration; import io.github.dsheirer.vector.calibrate.demodulator.DifferentialDemodulatorCalibration; import io.github.dsheirer.vector.calibrate.demodulator.FmDemodulatorCalibration; import io.github.dsheirer.vector.calibrate.filter.FirFilterCalibration; @@ -106,6 +107,7 @@ public class CalibrationManager sInstance.add(new UnpackedSampleConverterCalibration()); sInstance.add(new UnpackedInterleavedSampleConverterCalibration()); sInstance.add(new AmDemodulatorCalibration()); + sInstance.add(new ChannelDFTCalibration()); sInstance.add(new ComplexGainCalibration()); sInstance.add(new ComplexGainControlCalibration()); sInstance.add(new ComplexOscillatorCalibration()); diff --git a/src/main/java/io/github/dsheirer/vector/calibrate/CalibrationType.java b/src/main/java/io/github/dsheirer/vector/calibrate/CalibrationType.java index cf557172..7bcc6b64 100644 --- a/src/main/java/io/github/dsheirer/vector/calibrate/CalibrationType.java +++ b/src/main/java/io/github/dsheirer/vector/calibrate/CalibrationType.java @@ -33,6 +33,7 @@ public enum CalibrationType GAIN_COMPLEX("Complex Gain", 1), AM_DEMODULATOR("AM Demodulator", 1), + CHANNEL_DFT("Polyphase Channel DFT", 1), DC_REMOVAL_REAL("Real DC Removal Filter", 1), DMR_SOFT_SYNC_DETECTOR("DMR Soft Sync Detector", 2), DIFFERENTIAL_DEMODULATOR("DQPSK Demodulator", 1), diff --git a/src/main/java/io/github/dsheirer/vector/calibrate/channelizer/ChannelDFTCalibration.java b/src/main/java/io/github/dsheirer/vector/calibrate/channelizer/ChannelDFTCalibration.java new file mode 100644 index 00000000..b80f1837 --- /dev/null +++ b/src/main/java/io/github/dsheirer/vector/calibrate/channelizer/ChannelDFTCalibration.java @@ -0,0 +1,103 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.vector.calibrate.channelizer; + +import io.github.dsheirer.dsp.filter.channelizer.ChannelDFT; +import io.github.dsheirer.dsp.filter.channelizer.ScalarChannelDFT; +import io.github.dsheirer.dsp.filter.channelizer.VectorChannelDFT; +import io.github.dsheirer.vector.calibrate.Calibration; +import io.github.dsheirer.vector.calibrate.CalibrationException; +import io.github.dsheirer.vector.calibrate.CalibrationType; +import io.github.dsheirer.vector.calibrate.Implementation; +import org.apache.commons.math3.stat.descriptive.moment.Mean; + +/** + * Determines the optimal scalar vs vector implementation of the polyphase channel DFT. + */ +public class ChannelDFTCalibration extends Calibration +{ + private static final int CHANNEL_COUNT = 96; + private static final int[] CHANNELS = {5, 6, 40, 90}; + private static final int ITERATION_DURATION_MS = 500; + private static final int WARMUP_ITERATIONS = 3; + private static final int TEST_ITERATIONS = 3; + + /** + * Constructs an instance + */ + public ChannelDFTCalibration() + { + super(CalibrationType.CHANNEL_DFT); + } + + @Override public void calibrate() throws CalibrationException + { + ChannelDFT scalar = new ScalarChannelDFT(CHANNEL_COUNT); + ChannelDFT vector = new VectorChannelDFT(CHANNEL_COUNT); + + for(int x = 0; x < WARMUP_ITERATIONS; x++) + { + test(scalar); + test(vector); + } + + Mean scalarMean = new Mean(); + Mean vectorMean = new Mean(); + + for(int x = 0; x < TEST_ITERATIONS; x++) + { + scalarMean.increment(test(scalar)); + vectorMean.increment(test(vector)); + } + + mLog.info("CHANNEL DFT - SCALAR:" + DECIMAL_FORMAT.format(scalarMean.getResult())); + mLog.info("CHANNEL DFT - VECTOR:" + DECIMAL_FORMAT.format(vectorMean.getResult())); + + if(scalarMean.getResult() > vectorMean.getResult()) + { + setImplementation(Implementation.SCALAR); + } + else + { + setImplementation(Implementation.VECTOR_SIMD_PREFERRED); + } + + mLog.info("CHANNEL DFT - SET IMPLEMENTATION TO:" + getImplementation()); + } + + private long test(ChannelDFT channelDFT) + { + float[] samples = getFloatSamples(CHANNEL_COUNT * 2); + float[] input = samples.clone(); + float accumulator = 0.0f; + long count = 0; + + long start = System.currentTimeMillis(); + + while((System.currentTimeMillis() - start) < ITERATION_DURATION_MS) + { + System.arraycopy(input, 0, samples, 0, samples.length); + channelDFT.process(samples, CHANNELS); + accumulator += samples[10]; + count++; + } + + return count + (long)(accumulator * 0); + } +} diff --git a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculatorTest.java b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculatorTest.java new file mode 100644 index 00000000..6aae288c --- /dev/null +++ b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculatorTest.java @@ -0,0 +1,141 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +import io.github.dsheirer.source.tuner.channel.TunerChannel; +import java.util.List; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies the selection of one or two polyphase channel indexes for a tuner channel. + */ +public class ChannelCalculatorTest +{ + private static final long CENTER = 100_000_000L; + private static final double SAMPLE_RATE = 2_400_000.0; + private static final int CHANNEL_COUNT = 96; + private static final int CHANNEL_BANDWIDTH = 25000; + + private static List indexes(int channelCenterOffset, int frequencyOffset, int bandwidth) + { + return calculator().getChannelIndexes(new TunerChannel(frequency(channelCenterOffset, frequencyOffset), bandwidth)); + } + + private static long frequency(int channelCenterOffset, int frequencyOffset) + { + return CENTER + (long)channelCenterOffset * CHANNEL_BANDWIDTH + frequencyOffset; + } + + private static ChannelCalculator calculator() + { + return new ChannelCalculator(SAMPLE_RATE, CHANNEL_COUNT, CENTER, 2.0); + } + + /** + * Test: a narrow channel stays on one index until its band edge reaches the channelizer roll-off. + */ + @Test + void narrowChannelMovesToTwoIndexesBeforeBandEdgeRollOff() + { + assertEquals(List.of(5), indexes(5, 2500, 12500)); + assertEquals(List.of(5, 6), indexes(5, 3500, 12500)); + assertEquals(List.of(4, 5), indexes(5, -3500, 12500)); + assertEquals(List.of(5), indexes(5, 6000, 5000)); + } + + /** + * Test: a wide channel near a polyphase channel center stays on one index, otherwise it uses the nearest boundary. + */ + @Test + void wideChannelUsesOneIndexNearChannelCenter() + { + assertEquals(List.of(5), indexes(5, 0, 25000)); + assertEquals(List.of(5), indexes(5, 1000, 25000)); + assertEquals(List.of(5, 6), indexes(5, 3500, 25000)); + assertEquals(List.of(4, 5), indexes(5, -3500, 25000)); + assertEquals(List.of(5, 6), indexes(5, 12500, 25000)); + } + + /** + * Test: index pairs are ordered low to high in frequency across the center and wrap around indexes. + */ + @Test + void indexPairsAreOrderedAcrossCenterAndWrapAround() + { + assertEquals(List.of(95, 0), indexes(0, -6000, 12500)); + assertEquals(List.of(0, 1), indexes(0, 6000, 12500)); + assertEquals(List.of(47, 48), indexes(47, 12500, 12500)); + assertEquals(List.of(48, 49), indexes(-47, -12500, 12500)); + } + + /** + * Test: every sourceable channel across the spectrum uses at most two adjacent indexes and the synthesized center + * frequency is the boundary nearest to the channel. + */ + @Test + void allChannelsUseAtMostTwoIndexesAroundNearestBoundary() + { + ChannelCalculator calculator = calculator(); + + for(int bandwidth : new int[]{3000, 6250, 12500, 15000, 25000}) + { + long start = CENTER - (long)(SAMPLE_RATE / 2) + bandwidth; + long stop = CENTER + (long)(SAMPLE_RATE / 2) - bandwidth; + + for(long frequency = start; frequency <= stop; frequency += 730) + { + TunerChannel channel = new TunerChannel(frequency, bandwidth); + List indexes; + + try + { + indexes = calculator.getChannelIndexes(channel); + } + catch(IllegalArgumentException iae) + { + continue; //Within the wrap around channel, which can't be sourced + } + + assertTrue(indexes.size() <= 2, "Too many indexes " + indexes + " for " + channel); + + long error = Math.abs(calculator.getCenterFrequencyForIndexes(indexes) - frequency); + assertTrue(error <= CHANNEL_BANDWIDTH / 2 + 1, "Center " + error + " Hz from " + channel); + + if(indexes.size() == 2) + { + assertEquals((indexes.get(0) + 1) % CHANNEL_COUNT, indexes.get(1), "Not adjacent: " + indexes); + } + } + } + } + + /** + * Test: channels outside of the tuner bandwidth are still rejected. + */ + @Test + void channelBeyondTunerBandwidthIsRejected() + { + assertThrows(IllegalArgumentException.class, () -> calculator().getChannelIndexes( + new TunerChannel(CENTER + (long)(SAMPLE_RATE / 2), 12500))); + } +} diff --git a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFTTest.java b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFTTest.java new file mode 100644 index 00000000..fa17108a --- /dev/null +++ b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFTTest.java @@ -0,0 +1,289 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +import io.github.dsheirer.sample.complex.InterleavedComplexSamples; +import io.github.dsheirer.source.tuner.channel.TunerChannel; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Random; +import java.util.concurrent.CopyOnWriteArrayList; +import org.jtransforms.fft.FloatFFT_1D; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that the channel DFT produces the same channel values as the IFFT, both directly and through the threaded + * channelizer, including across channel changes. + */ +public class ChannelDFTTest +{ + private static final float TOLERANCE = 0.00001f; + private static final int BLOCKS_PER_BATCH = 1024; + private static final int TAPS_PER_CHANNEL = 9; + private static final int[] CHANNEL_COUNTS = {2, 8, 72, 96, 100, 128, 400}; + + private static float[] randomSamples(int length, Random random) + { + float[] samples = new float[length]; + + for(int x = 0; x < samples.length; x++) + { + samples[x] = random.nextFloat() * 2.0f - 1.0f; + } + + return samples; + } + + private static void assertMatchesIFFT(ChannelDFT channelDFT, int channelCount, int[] channels, Random random) + { + int blocks = 3; + int blockLength = channelCount * 2; + float[] input = randomSamples(blocks * blockLength, random); + float[] expected = input.clone(); + FloatFFT_1D fft = new FloatFFT_1D(channelCount); + + for(int block = 0; block < blocks; block++) + { + fft.complexInverse(expected, block * blockLength, true); + } + + float[] actual = input.clone(); + channelDFT.process(actual, channels); + + for(int block = 0; block < blocks; block++) + { + for(int channel : channels) + { + int index = block * blockLength + channel * 2; + assertEquals(expected[index], actual[index], TOLERANCE, "I channel " + channel + " of " + channelCount); + assertEquals(expected[index + 1], actual[index + 1], TOLERANCE, "Q channel " + channel + " of " + channelCount); + } + } + } + + /** + * Test: both implementations match the IFFT for each single channel and for sets of channels. + */ + @Test + void implementationsMatchIFFT() + { + Random random = new Random(0xC0FFEEL); + + for(int channelCount : CHANNEL_COUNTS) + { + for(ChannelDFT channelDFT : List.of(new ScalarChannelDFT(channelCount), new VectorChannelDFT(channelCount))) + { + for(int channel = 0; channel < channelCount; channel++) + { + assertMatchesIFFT(channelDFT, channelCount, new int[]{channel}, random); + } + + //All channels at once verifies that results don't overwrite the input of the other channels + int[] all = new int[channelCount]; + Arrays.setAll(all, x -> x); + assertMatchesIFFT(channelDFT, channelCount, all, random); + } + } + } + + /** + * Test: each implementation limits use to channel counts that it can process faster than the IFFT. + */ + @Test + void maximumChannelsAreAboveZero() + { + assertTrue(new ScalarChannelDFT(96).getMaximumChannels() > 0); + assertTrue(new VectorChannelDFT(96).getMaximumChannels() > 0); + } + + /** + * Test: channels in use receive the same values as the IFFT through the threaded channelizer, on one index and on + * two adjacent indexes. Channelizer outputs include the alternating block phase handling, so this also verifies + * that both transforms see the same block order. + */ + @Test + void channelizerMatchesIFFTForChannelsInUse() throws Exception + { + int channelCount = 96; + float[] taps = taps(channelCount); + Random random = new Random(1234); + + for(long frequency : new long[]{100_125_000L, 100_137_500L}) + { + ChannelCalculator calculator = new ChannelCalculator(2_400_000.0, channelCount, 100_000_000L, 2.0); + List sparse = new CopyOnWriteArrayList<>(); + List full = new CopyOnWriteArrayList<>(); + ComplexPolyphaseChannelizerM2 sparseChannelizer = channelizer(taps, channelCount); + ComplexPolyphaseChannelizerM2 fullChannelizer = channelizer(taps, channelCount); + CapturingSource source = new CapturingSource(frequency, calculator, sparse); + sparseChannelizer.addChannel(source); + fullChannelizer.addChannel(new CapturingSource(frequency, calculator, full, channelCount)); + sparseChannelizer.start(); + fullChannelizer.start(); + + float[] samples = randomSamples(BLOCKS_PER_BATCH * channelCount * 3, random); + sparseChannelizer.receive(new InterleavedComplexSamples(samples, 0)); + fullChannelizer.receive(new InterleavedComplexSamples(samples, 0)); + await(sparse, 3); + await(full, 3); + sparseChannelizer.stop(); + fullChannelizer.stop(); + + assertChannelsMatch(sparse, full, source.getOutputProcessorIndexes(), channelCount); + } + } + + /** + * Test: when an update moves a channel to a different index, the first batch after the update has valid results for + * the new index. + */ + @Test + void channelUpdateAppliesBeforeTransform() throws Exception + { + int channelCount = 96; + float[] taps = taps(channelCount); + long frequency = 100_125_000L; + List sparse = new CopyOnWriteArrayList<>(); + List full = new CopyOnWriteArrayList<>(); + ComplexPolyphaseChannelizerM2 sparseChannelizer = channelizer(taps, channelCount); + ComplexPolyphaseChannelizerM2 fullChannelizer = channelizer(taps, channelCount); + ChannelCalculator calculator = new ChannelCalculator(2_400_000.0, channelCount, 100_000_000L, 2.0); + CapturingSource source = new CapturingSource(frequency, calculator, sparse); + sparseChannelizer.addChannel(source); + fullChannelizer.addChannel(new CapturingSource(frequency, calculator, full, channelCount)); + sparseChannelizer.start(); + fullChannelizer.start(); + + ChannelCalculator retuned = new ChannelCalculator(2_400_000.0, channelCount, 100_050_000L, 2.0); + source.updateOutputProcessor(retuned, new SynthesisFilterManager()); + + float[] samples = randomSamples(BLOCKS_PER_BATCH * channelCount, new Random(5678)); + sparseChannelizer.receive(new InterleavedComplexSamples(samples, 0)); + fullChannelizer.receive(new InterleavedComplexSamples(samples, 0)); + await(sparse, 1); + await(full, 1); + sparseChannelizer.stop(); + fullChannelizer.stop(); + + assertEquals(List.of(3), source.getOutputProcessorIndexes()); + assertChannelsMatch(sparse, full, source.getOutputProcessorIndexes(), channelCount); + } + + private static void assertChannelsMatch(List actual, List expected, List indexes, int channelCount) + { + int blockLength = channelCount * 2; + + for(int batch = 0; batch < expected.size(); batch++) + { + for(int index : indexes) + { + float[] expectedChannel = extract(expected.get(batch), index, blockLength); + float[] actualChannel = extract(actual.get(batch), index, blockLength); + assertArrayEquals(expectedChannel, actualChannel, TOLERANCE, "Channel " + index + " batch " + batch); + } + } + } + + private static float[] extract(float[] samples, int index, int blockLength) + { + float[] channel = new float[samples.length / blockLength * 2]; + + for(int block = 0; block < channel.length / 2; block++) + { + channel[block * 2] = samples[block * blockLength + index * 2]; + channel[block * 2 + 1] = samples[block * blockLength + index * 2 + 1]; + } + + return channel; + } + + private static float[] taps(int channelCount) + { + float[] taps = randomSamples(channelCount * TAPS_PER_CHANNEL, new Random(99)); + + for(int x = 0; x < taps.length; x++) + { + taps[x] /= channelCount; + } + + return taps; + } + + private static ComplexPolyphaseChannelizerM2 channelizer(float[] taps, int channelCount) + { + return new ComplexPolyphaseChannelizerM2(taps, 2_400_000, channelCount); + } + + private static void await(List list, int size) throws Exception + { + long timeout = System.currentTimeMillis() + 5000; + + while(list.size() < size && System.currentTimeMillis() < timeout) + { + Thread.sleep(2); + } + + assertEquals(size, list.size(), "Batches received"); + } + + /** + * Channel source that captures a copy of each batch. A source can also claim more indexes than its channel, which + * forces the channelizer to use the IFFT. + */ + private static class CapturingSource extends PolyphaseChannelSource + { + private final List mCaptured; + private final List mForcedIndexes; + + CapturingSource(long frequency, ChannelCalculator calculator, List captured) + { + this(frequency, calculator, captured, 0); + } + + CapturingSource(long frequency, ChannelCalculator calculator, List captured, int forcedIndexCount) + { + super(new TunerChannel(frequency, 12500), calculator, new SynthesisFilterManager(), sourceEvent -> {}, + "test", null); + mCaptured = captured; + mForcedIndexes = new ArrayList<>(); + + for(int x = 0; x < forcedIndexCount; x++) + { + mForcedIndexes.add(x); + } + } + + @Override + public List getOutputProcessorIndexes() + { + return mForcedIndexes.isEmpty() ? super.getOutputProcessorIndexes() : mForcedIndexes; + } + + @Override + public void receiveChannelResults(ChannelResultsBatch channelResults) + { + mCaptured.add(channelResults.getSamples().clone()); + } + } +} diff --git a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsBatchTest.java b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsBatchTest.java new file mode 100644 index 00000000..b60b1e6b --- /dev/null +++ b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsBatchTest.java @@ -0,0 +1,80 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +import io.github.dsheirer.dsp.filter.channelizer.output.OneChannelOutputProcessor; +import java.util.ArrayDeque; +import java.util.List; +import java.util.Queue; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies the shared ownership of channel results batch sample arrays. + */ +public class ChannelResultsBatchTest +{ + /** + * Test: the sample array only returns to the pool after every reference is released. + */ + @Test + void samplesReturnToPoolAfterLastRelease() + { + Queue pool = new ArrayDeque<>(); + float[] samples = new float[8]; + ChannelResultsBatch batch = new ChannelResultsBatch(samples, 4, 0, pool); + batch.retain(); + batch.retain(); + + batch.release(); + batch.release(); + assertTrue(pool.isEmpty(), "Returned to pool with an outstanding reference"); + + batch.release(); + assertSame(samples, pool.poll()); + } + + /** + * Test: blocks are derived from the array and block lengths. + */ + @Test + void blockCountIsDerivedFromLengths() + { + assertEquals(2, new ChannelResultsBatch(new float[8], 4, 0, new ArrayDeque<>()).getBlockCount()); + } + + /** + * Test: an output processor that doesn't accept a batch because it isn't running doesn't keep a reference. + */ + @Test + void stoppedOutputProcessorReleasesBatch() + { + Queue pool = new ArrayDeque<>(); + ChannelResultsBatch batch = new ChannelResultsBatch(new float[8], 4, 0, pool); + OneChannelOutputProcessor processor = new OneChannelOutputProcessor(50000.0, List.of(0), 1.0f, null, "test"); + + processor.receiveChannelResults(batch); + batch.release(); + + assertEquals(1, pool.size()); + } +} diff --git a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsTimestampTest.java b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsTimestampTest.java new file mode 100644 index 00000000..aa1b0389 --- /dev/null +++ b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsTimestampTest.java @@ -0,0 +1,124 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +import io.github.dsheirer.dsp.filter.channelizer.output.OneChannelOutputProcessor; +import io.github.dsheirer.sample.complex.ComplexSamples; +import io.github.dsheirer.sample.complex.InterleavedComplexSamples; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Verifies that channel results batches keep their own timestamp and configuration through the threaded dispatchers. + */ +public class ChannelResultsTimestampTest +{ + private static final int BATCH_SIZE = 2048; + private static final int CHANNEL_COUNT = 8; + private static final int SAMPLES_PER_BATCH = 1024 * CHANNEL_COUNT / 2; + + private static void awaitSize(List list, int size) throws Exception + { + long timeout = System.currentTimeMillis() + 5000; + + while(list.size() < size && System.currentTimeMillis() < timeout) + { + Thread.sleep(5); + } + } + + private static ChannelResultsBatch batch(float i, float q, long timestamp) + { + float[] samples = new float[BATCH_SIZE * 4]; + + for(int x = 0; x < samples.length; x += 4) + { + samples[x] = i; + samples[x + 1] = q; + samples[x + 2] = -i; + samples[x + 3] = -q; + } + + return new ChannelResultsBatch(samples, 4, timestamp, new ArrayDeque<>()); + } + + /** + * Test: each batch dispatched by the channelizer has the timestamp of the sample buffer that completed the batch, + * even when a later sample buffer arrives before the batch is dispatched. + */ + @Test + void channelizerDispatchesEachBatchWithItsOwnTimestamp() throws Exception + { + List timestamps = new CopyOnWriteArrayList<>(); + ComplexPolyphaseChannelizerM2 channelizer = new ComplexPolyphaseChannelizerM2(new float[CHANNEL_COUNT * 9], + 25000 * CHANNEL_COUNT, CHANNEL_COUNT) + { + @Override + protected void dispatch(ChannelResultsBatch channelResults) + { + timestamps.add(channelResults.getTimestamp()); + } + }; + + channelizer.start(); + + for(long timestamp : new long[]{10, 20, 30}) + { + channelizer.receive(new InterleavedComplexSamples(new float[SAMPLES_PER_BATCH * 2], timestamp)); + } + + awaitSize(timestamps, 3); + channelizer.stop(); + + assertEquals(List.of(10L, 20L, 30L), timestamps); + } + + /** + * Test: assembled sample buffers have the timestamp of their batch, and a configuration change requested between + * two batches takes effect at the start of the second batch, even when the first batch hasn't been processed yet. + */ + @Test + void outputProcessorAppliesTimestampAndConfigurationPerBatch() throws Exception + { + List output = new CopyOnWriteArrayList<>(); + OneChannelOutputProcessor processor = new OneChannelOutputProcessor(50000.0, List.of(0), 1.0f, null, "test"); + processor.setListener(output::add); + + processor.start(); + processor.receiveChannelResults(batch(1.0f, 2.0f, 100)); + processor.setPolyphaseChannelIndices(List.of(1)); + processor.receiveChannelResults(batch(1.0f, 2.0f, 200)); + awaitSize(output, 2); + processor.stop(); + + float[] expected = new float[BATCH_SIZE]; + assertEquals(100, output.get(0).timestamp()); + Arrays.fill(expected, 1.0f); + assertArrayEquals(expected, output.get(0).i()); + assertEquals(200, output.get(1).timestamp()); + Arrays.fill(expected, -1.0f); + assertArrayEquals(expected, output.get(1).i()); + } +} diff --git a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2Test.java b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2Test.java index a92b15de..92472897 100644 --- a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2Test.java +++ b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2Test.java @@ -19,8 +19,12 @@ package io.github.dsheirer.dsp.filter.channelizer; +import io.github.dsheirer.sample.complex.InterleavedComplexSamples; import java.util.Arrays; +import java.util.List; import java.util.Random; +import java.util.concurrent.CopyOnWriteArrayList; +import org.jtransforms.fft.FloatFFT_1D; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -86,6 +90,94 @@ public class ComplexPolyphaseChannelizerM2Test } } + /** + * Test: the channelizer's output is bit-for-bit identical to a reference that shifts all of its samples for each + * block, across several wraps of the sample buffer window and with buffers that end part way through a block. + */ + @Test + void slidingWindowMatchesShiftingReference() throws Exception + { + Random rng = new Random(0xC0FFEEL); + + for(int channelCount : new int[]{2, 8, 96, 100, 400}) + { + Harness harness = new Harness(channelCount, TAPS_PER_CHANNEL, rng); + List batches = new CopyOnWriteArrayList<>(); + ComplexPolyphaseChannelizerM2 channelizer = new ComplexPolyphaseChannelizerM2(harness.getCoefficients(), + 25000 * channelCount, channelCount) + { + @Override + protected void dispatch(ChannelResultsBatch channelResults) + { + batches.add(channelResults.getSamples().clone()); + } + }; + + int batchCount = 3; + int blocksPerBatch = 1024; + float[] samples = new float[batchCount * blocksPerBatch * channelCount]; + + for(int x = 0; x < samples.length; x++) + { + samples[x] = rng.nextFloat() * 2f - 1f; + } + + channelizer.start(); + int pointer = 0; + + while(pointer < samples.length) + { + int length = Math.min(2 * (1 + rng.nextInt(channelCount * 3)), samples.length - pointer); + channelizer.receive(new InterleavedComplexSamples(Arrays.copyOfRange(samples, pointer, pointer + length), 0)); + pointer += length; + } + + long timeout = System.currentTimeMillis() + 5000; + + while(batches.size() < batchCount && System.currentTimeMillis() < timeout) + { + Thread.sleep(2); + } + + channelizer.stop(); + assertEquals(batchCount, batches.size(), "Batches (channelCount=" + channelCount + ")"); + + FloatFFT_1D fft = new FloatFFT_1D(channelCount); + int blockLength = channelCount * 2; + boolean topBlock = TAPS_PER_CHANNEL % 2 == 0; + + for(int block = 0; block < batchCount * blocksPerBatch; block++) + { + harness.load(samples, block * channelCount, channelCount); + float[] expected = harness.processFused(topBlock); + topBlock = !topBlock; + harness.shift(channelCount); + fft.complexInverse(expected, true); + + float[] batch = batches.get(block / blocksPerBatch); + int offset = (block % blocksPerBatch) * blockLength; + assertBitIdentical(expected, Arrays.copyOfRange(batch, offset, offset + blockLength), channelCount, 0, block); + } + } + } + + /** + * Test: channel counts are even, have only factors of 2, 3 and 5, and are as large as possible for the minimum + * channel bandwidth. + */ + @Test + void channelCountIsAnEfficientIFFTSize() + { + assertEquals(96, ComplexPolyphaseChannelizerM2.getChannelCount(2_400_000.0)); + assertEquals(128, ComplexPolyphaseChannelizerM2.getChannelCount(3_200_000.0)); + assertEquals(400, ComplexPolyphaseChannelizerM2.getChannelCount(10_000_000.0)); + assertEquals(72, ComplexPolyphaseChannelizerM2.getChannelCount(1_920_000.0)); + assertEquals(100, ComplexPolyphaseChannelizerM2.getChannelCount(2_560_000.0)); + assertEquals(108, ComplexPolyphaseChannelizerM2.getChannelCount(2_880_000.0)); + assertEquals(240, ComplexPolyphaseChannelizerM2.getChannelCount(6_100_000.0)); + assertEquals(2, ComplexPolyphaseChannelizerM2.getChannelCount(50_000.0)); + } + private static void assertBitIdentical(float[] reference, float[] fused, int channelCount, int mode, int iteration) { assertEquals(reference.length, fused.length, @@ -116,6 +208,7 @@ public class ComplexPolyphaseChannelizerM2Test private final int[] mTopBlockMap; // length mSubChannelCount private final int[] mMiddleBlockMap; // length mSubChannelCount private final float[] mReusableAccumulator; + private final float[] mCoefficients; private Harness(int channelCount, int tapsPerChannel, Random rng) { @@ -129,6 +222,7 @@ public class ComplexPolyphaseChannelizerM2Test coefficients[i] = rng.nextFloat() * 2f - 1f; } + mCoefficients = coefficients; mInlineFilter = getAlignedFilter(coefficients, channelCount, tapsPerChannel); mInlineSamples = new float[mBufferLength]; mTopBlockMap = getTopBlockMap(channelCount); @@ -136,6 +230,23 @@ public class ComplexPolyphaseChannelizerM2Test mReusableAccumulator = new float[mSubChannelCount]; } + private float[] getCoefficients() + { + return mCoefficients; + } + + //Loads a block of samples to the start of the sample buffer + private void load(float[] samples, int offset, int blockLength) + { + System.arraycopy(samples, offset, mInlineSamples, 0, blockLength); + } + + //Shifts all of the samples by one block + private void shift(int blockLength) + { + System.arraycopy(mInlineSamples, 0, mInlineSamples, blockLength, mBufferLength - blockLength); + } + //Reference: multiply the whole array into an interim buffer, then accumulate across taps (prior behavior). private float[] processReference(boolean topBlock) { diff --git a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSourceTest.java b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSourceTest.java new file mode 100644 index 00000000..9ebcc14f --- /dev/null +++ b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSourceTest.java @@ -0,0 +1,139 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.dsp.filter.channelizer; + +import io.github.dsheirer.sample.Listener; +import io.github.dsheirer.source.SourceEvent; +import io.github.dsheirer.source.tuner.channel.TunerChannel; +import java.util.ArrayDeque; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that a failure in one polyphase channel source is contained to that source and is reported to its consumer. + */ +public class PolyphaseChannelSourceTest +{ + private static final long CENTER = 100_000_000L; + private static final double SAMPLE_RATE = 2_400_000.0; + private static final int CHANNEL_COUNT = 96; + + private static ChannelCalculator calculator(long center) + { + return new ChannelCalculator(SAMPLE_RATE, CHANNEL_COUNT, center, 2.0); + } + + private static ChannelResultsBatch batch() + { + return new ChannelResultsBatch(new float[CHANNEL_COUNT * 2], CHANNEL_COUNT * 2, 0, new ArrayDeque<>()); + } + + private static PolyphaseChannelSource source() + { + return new PolyphaseChannelSource(new TunerChannel(CENTER + 125000, 12500), calculator(CENTER), + new SynthesisFilterManager(), sourceEvent -> {}, "test", null); + } + + /** + * Test: an output processor update failure doesn't throw to the channelizer, signals one error to the consumer and + * stops this source from processing further results. + */ + @Test + void failedOutputProcessorUpdateSignalsErrorOnce() throws Exception + { + CountDownLatch errorLatch = new CountDownLatch(1); + AtomicInteger errorCount = new AtomicInteger(); + Listener consumer = sourceEvent -> { + if(sourceEvent.getEvent() == SourceEvent.Event.NOTIFICATION_ERROR_STATE) + { + errorCount.incrementAndGet(); + errorLatch.countDown(); + } + }; + + PolyphaseChannelSource source = source(); + source.setSourceEventListener(consumer); + + //Retune so that the channel is outside of the tuner bandwidth + source.updateOutputProcessor(calculator(CENTER + 10_000_000L), new SynthesisFilterManager()); + + source.applyPendingOutputProcessorUpdate(); + source.receiveChannelResults(batch()); + source.applyPendingOutputProcessorUpdate(); + source.receiveChannelResults(batch()); + + assertTrue(errorLatch.await(5, TimeUnit.SECONDS), "Error was not signalled to the consumer"); + Thread.sleep(50); + assertEquals(1, errorCount.get()); + } + + /** + * Test: an exception from one channel doesn't stop delivery of the results to the following channels. + */ + @Test + void channelFailureDoesNotAffectFollowingChannels() + { + AtomicInteger received = new AtomicInteger(); + + PolyphaseChannelSource failing = new TestSource() + { + @Override + public void receiveChannelResults(ChannelResultsBatch channelResults) + { + throw new IllegalStateException("Test failure"); + } + }; + + PolyphaseChannelSource following = new TestSource() + { + @Override + public void receiveChannelResults(ChannelResultsBatch channelResults) + { + received.incrementAndGet(); + } + }; + + ComplexPolyphaseChannelizerM2 channelizer = new ComplexPolyphaseChannelizerM2(new float[CHANNEL_COUNT * 9], + (int)SAMPLE_RATE, CHANNEL_COUNT); + channelizer.addChannel(failing); + channelizer.addChannel(following); + + channelizer.prepareChannels(); + channelizer.dispatch(batch()); + + assertEquals(1, received.get()); + } + + /** + * Polyphase channel source for overriding the channel results processing. + */ + private static class TestSource extends PolyphaseChannelSource + { + TestSource() + { + super(new TunerChannel(CENTER + 125000, 12500), calculator(CENTER), new SynthesisFilterManager(), + sourceEvent -> {}, "test", null); + } + } +} diff --git a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/TwoChannelSynthesizerM2Test.java b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/TwoChannelSynthesizerM2Test.java index ec2195f7..1447745c 100644 --- a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/TwoChannelSynthesizerM2Test.java +++ b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/TwoChannelSynthesizerM2Test.java @@ -48,7 +48,9 @@ public class TwoChannelSynthesizerM2Test /** * Test: tones at mirrored offsets from the boundary between two channels produce equal output levels, and a tone - * exactly on the boundary produces the same level as one near the center of the composite. + * exactly on the boundary produces the same level as one near the center of the composite. The pass band must + * extend past the center of each channel so that a signal offset from the boundary doesn't lose its outer band edge, + * and tones beyond the alias-free region of the composite must be rejected. */ @Test void boundaryResponseIsFlatAndSymmetric() throws Exception @@ -58,19 +60,17 @@ public class TwoChannelSynthesizerM2Test double center = gain(0.0, tapsPerChannel); assertEquals(1.0, center, 0.01, "Unity gain at channel boundary (taps per channel=" + tapsPerChannel + ")"); - for(double offset : new double[]{0.1, 0.2, 0.3, 0.4}) + for(double offset : new double[]{0.1, 0.2, 0.3, 0.4, 0.5, 0.6}) { double below = gain(-offset, tapsPerChannel); double above = gain(offset, tapsPerChannel); assertEquals(below, above, 0.01, "Asymmetric response at +/-" + offset + " channel bandwidths from boundary (taps per channel=" + tapsPerChannel + ")"); - - //Synthesis filter cutoff is 0.55 channel bandwidths, so only check flatness for the inner region - if(offset <= 0.25) - { - assertTrue(below > 0.98, "Passband droop at " + offset + " channel bandwidths: " + below); - } + assertTrue(below > 0.98, "Passband droop at " + offset + " channel bandwidths: " + below); } + + double stopBand = gain(0.9, tapsPerChannel); + assertTrue(stopBand < 0.001, "Stop band leakage at 0.9 channel bandwidths: " + stopBand); } } @@ -88,9 +88,15 @@ public class TwoChannelSynthesizerM2Test CHANNEL_COUNT) { @Override - protected void dispatch(List channelResultsList) + protected void dispatch(ChannelResultsBatch channelResults) { - results.addAll(channelResultsList); + float[] samples = channelResults.getSamples(); + int blockLength = channelResults.getBlockLength(); + + for(int offset = 0; offset < samples.length; offset += blockLength) + { + results.add(Arrays.copyOfRange(samples, offset, offset + blockLength)); + } } }; @@ -126,7 +132,7 @@ public class TwoChannelSynthesizerM2Test q2[x] = sign * result[2 * (LOWER_CHANNEL + 1) + 1]; } - float[] filter = FilterFactory.getSincM2Synthesizer(2.0 * CHANNEL_BANDWIDTH, CHANNEL_BANDWIDTH, 2, 9); + float[] filter = new SynthesisFilterManager().getFilter(2.0 * CHANNEL_BANDWIDTH, CHANNEL_BANDWIDTH, 2); ComplexSamples output = new TwoChannelSynthesizerM2(filter).process(new ComplexSamples(i1, q1, 0), new ComplexSamples(i2, q2, 0)); diff --git a/src/test/java/io/github/dsheirer/util/DispatcherTest.java b/src/test/java/io/github/dsheirer/util/DispatcherTest.java new file mode 100644 index 00000000..50389fb9 --- /dev/null +++ b/src/test/java/io/github/dsheirer/util/DispatcherTest.java @@ -0,0 +1,128 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * **************************************************************************** + */ +package io.github.dsheirer.util; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies dispatcher queue limits. + */ +public class DispatcherTest +{ + private static final int MAXIMUM_QUEUE_SIZE = 4; + + /** + * Sends the count of elements while the dispatcher's listener is blocked on the first element. + * @return elements received by the listener after it is unblocked + */ + private static List dispatchWhileBlocked(Dispatcher dispatcher, int count) throws Exception + { + List received = new CopyOnWriteArrayList<>(); + CountDownLatch blocked = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + + dispatcher.setListener(element -> { + received.add(element); + blocked.countDown(); + + try + { + release.await(); + } + catch(InterruptedException ie) + { + Thread.currentThread().interrupt(); + } + }); + + dispatcher.start(); + dispatcher.receive(0); + assertTrue(blocked.await(5, TimeUnit.SECONDS), "Listener did not receive the first element"); + + for(int x = 1; x <= count; x++) + { + dispatcher.receive(x); + } + + release.countDown(); + return received; + } + + private static void awaitSize(List list, int size) throws Exception + { + long timeout = System.currentTimeMillis() + 5000; + + while(list.size() < size && System.currentTimeMillis() < timeout) + { + Thread.sleep(5); + } + } + + /** + * Test: elements received while a bounded queue is full are dropped and the queued elements stay in order. + */ + @Test + void boundedQueueDropsNewestElementsWhenFull() throws Exception + { + Dispatcher dispatcher = new Dispatcher<>("test", 1, MAXIMUM_QUEUE_SIZE); + List received = dispatchWhileBlocked(dispatcher, MAXIMUM_QUEUE_SIZE + 10); + awaitSize(received, MAXIMUM_QUEUE_SIZE + 1); + Thread.sleep(50); + dispatcher.stop(); + + assertEquals(List.of(0, 1, 2, 3, 4), received); + } + + /** + * Test: a bounded dispatcher resumes delivery after an overflow. + */ + @Test + void boundedQueueRecoversAfterOverflow() throws Exception + { + Dispatcher dispatcher = new Dispatcher<>("test", 1, MAXIMUM_QUEUE_SIZE); + List received = dispatchWhileBlocked(dispatcher, MAXIMUM_QUEUE_SIZE + 10); + awaitSize(received, MAXIMUM_QUEUE_SIZE + 1); + dispatcher.receive(99); + awaitSize(received, MAXIMUM_QUEUE_SIZE + 2); + dispatcher.stop(); + + assertEquals(99, received.get(received.size() - 1)); + } + + /** + * Test: an unbounded dispatcher doesn't drop elements. + */ + @Test + void unboundedQueueKeepsAllElements() throws Exception + { + Dispatcher dispatcher = new Dispatcher<>("test", 1); + List received = dispatchWhileBlocked(dispatcher, 100); + awaitSize(received, 101); + dispatcher.stop(); + + assertEquals(101, received.size()); + } +}