GitHub/sdrtrunk
1
0
Fork 0
mirror of https://github.com/DSheirer/sdrtrunk.git synced 2026-11-12 08:37:25 -05:00
Code Issues Packages Projects Releases 2 Wiki Activity

#2533 Polyphase channelizer enhancements for correctness and efficiency. (#2534)

Browse Source 
Co-authored-by: denny <dennis.sheirer@gmail.com>
This commit is contained in:
Denny Sheirer
2026-10-30 06:47:50 -04:00
committed by GitHub
parent 00c0418017
commit 9ebbf957a2
29 changed files with 2062 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/io/github/dsheirer/dsp/filter/FilterFactory.java
View File

@@ -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<LowPassSpecification,float[]> 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

66
src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java
View File

@@ -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<In
private final static Logger mLog = LoggerFactory.getLogger(AbstractComplexPolyphaseChannelizer.class);
private Broadcaster<SourceEvent> mSourceChangeBroadcaster = new Broadcaster();
private List<PolyphaseChannelSource> mChannels = new CopyOnWriteArrayList<>();
private List<PolyphaseChannelSource> 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<In
return mSampleRate;
}
/**
* Sets the input sample rate for for this channelizer
* @param sampleRate in hertz
*/
public void setRates(double sampleRate, int channelCount)
{
mSampleRate = sampleRate;
mChannelCount = channelCount;
mChannelSampleRate = mSampleRate / (double)mChannelCount;
}
/**
* Number of channels being processed by this channelizer.
*/
@@ -98,15 +89,56 @@ public abstract class AbstractComplexPolyphaseChannelizer implements Listener<In
}
/**
* Dispatches the processed channel samples to any registered polyphase channel outputs.
* Prepares the registered channels to receive the next channel results batch by applying any pending channel
* updates, and identifies the polyphase channel indexes that they need. The next invocation of dispatch() delivers
* to these same channels, so a channel registered after this method is invoked starts with the following batch.
*
* @param channelResultsList a list of arrays of I/Q samples per channel
* @return sorted polyphase channel indexes in use, or an empty array if there are no registered channels.
*/
protected void dispatch(List<float[]> channelResultsList)
protected int[] prepareChannels()
{
for(PolyphaseChannelSource channel : mChannels)
List<PolyphaseChannelSource> 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);
}
}
}

54
src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelCalculator.java
View File

@@ -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<Integer> 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<Integer> 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

133
src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFT.java Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>
* ****************************************************************************
*/
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;
}
}
}

49
src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelDFTFactory.java Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>
* ****************************************************************************
*/
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);
}
}
}

107
src/main/java/io/github/dsheirer/dsp/filter/channelizer/ChannelResultsBatch.java Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>
* ****************************************************************************
*/
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<float[]> 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<float[]> 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);
}
}
}

204
src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java
View File

@@ -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<float[]> mProcessedChannelResultsList = new ArrayList<>();
private Queue<float[]> 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<List<float[]>>
public class IFFTProcessorDispatcher extends Dispatcher<ChannelResultsBatch>
{
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<float[]> 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();
}
});
}
}

48
src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java
View File

@@ -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<SourceEvent> 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<Integer,float[]> 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<INativeBuffer>
{
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)

74
src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java
View File

@@ -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<Integer> 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<PendingOutputProcessorUpdate> 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<float[]> 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
*

63
src/main/java/io/github/dsheirer/dsp/filter/channelizer/ScalarChannelDFT.java Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>
* ****************************************************************************
*/
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;
}
}

2
src/main/java/io/github/dsheirer/dsp/filter/channelizer/SynthesisFilterManager.java
View File

@@ -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<String,float[]> mFilterMap = new HashMap<>();

83
src/main/java/io/github/dsheirer/dsp/filter/channelizer/VectorChannelDFT.java Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>
* ****************************************************************************
*/
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<Float> 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;
}
}

81
src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java
View File

@@ -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<List<float[]>> 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<QueuedChannelResults> mChannelResultsDispatcher;
private final Queue<PendingUpdate> mPendingUpdates = new ConcurrentLinkedQueue<>();
private final AtomicLong mBatchSequence = new AtomicLong();
private HeartbeatManager mHeartbeatManager;
protected Listener<ComplexSamples> mComplexSamplesListener;
protected volatile Listener<ComplexSamples> 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<float[]> 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<float[]> 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) {}
}

4
src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java
View File

@@ -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<float[]> channelResults, long timestamp);
void receiveChannelResults(ChannelResultsBatch channelResults);
/**
* Listener to receive assembled complex samples buffers

18
src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java
View File

@@ -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<float[]> 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())
{

38
src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java
View File

@@ -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<float[]> 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())

2
src/main/java/io/github/dsheirer/gui/channelizer/SynthesizerViewer.java
View File

@@ -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)

72
src/main/java/io/github/dsheirer/util/Dispatcher.java
View File

@@ -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<E> implements Listener<E>
{
private final static Logger mLog = LoggerFactory.getLogger(Dispatcher.class);
private final LinkedTransferQueue<E> mQueue = new LinkedTransferQueue<>();
private static final long OVERFLOW_LOG_INTERVAL_MILLISECONDS = 10000;
private final BlockingQueue<E> mQueue;
private Listener<E> mListener;
private final AtomicBoolean mRunning = new AtomicBoolean();
private String mThreadName;
@@ -49,6 +52,8 @@ public class Dispatcher<E> implements Listener<E>
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<E> implements Listener<E>
}
/**
* 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<E> implements Listener<E>
{
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<E> implements Listener<E>
/**
* 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;
}
/**

2
src/main/java/io/github/dsheirer/vector/calibrate/CalibrationManager.java
View File

@@ -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());

1
src/main/java/io/github/dsheirer/vector/calibrate/CalibrationType.java
View File

@@ -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),

103
src/main/java/io/github/dsheirer/vector/calibrate/channelizer/ChannelDFTCalibration.java Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>
* ****************************************************************************
*/
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);
}
}