From 2044c2a2debebd27481f5eb82c8183d7ea7f91a9 Mon Sep 17 00:00:00 2001 From: Denny Date: Fri, 17 Aug 2018 07:27:04 -0400 Subject: [PATCH] WIP - Updated P25 C4FM and LSM decoders to assemble dibits and push reusable byte buffers out to the processing chain. This enables the processing chain to broadcast these buffers back to the decoder for message assembly and to any running binary recorder. Updated P25Decoder to be a reusable byte buffer listener. The decoder will now pass reusable byte buffers down to the message framer for message processing. This should make the code more efficient since it can process byte arrays versus handling each individual dibit. The P25MessageFramer has been updated to be a listener of reusable byte buffers instead of Dibits, but the actual processing of byte arrays has to be coded. Added max recording size to the binary recorder so that it cycles at 500 kB max file size. --- .../symbol/DibitToByteBufferAssembler.java | 125 ++++++++++++++++++ .../module/decode/DecoderFactory.java | 5 +- .../module/decode/p25/P25Decoder.java | 45 ++++++- .../module/decode/p25/P25DecoderC4FM.java | 11 +- .../p25/P25DecoderC4FMInstrumented.java | 2 +- .../module/decode/p25/P25DecoderLSM.java | 10 +- .../decode/p25/P25DecoderLSMInstrumented.java | 2 +- .../module/decode/p25/P25MessageFramer.java | 39 +++--- .../decode/p25/P25MessageProcessor.java | 18 ++- .../record/binary/BinaryRecorder.java | 67 ++++++++-- 10 files changed, 286 insertions(+), 38 deletions(-) create mode 100644 src/main/java/io/github/dsheirer/dsp/symbol/DibitToByteBufferAssembler.java diff --git a/src/main/java/io/github/dsheirer/dsp/symbol/DibitToByteBufferAssembler.java b/src/main/java/io/github/dsheirer/dsp/symbol/DibitToByteBufferAssembler.java new file mode 100644 index 00000000..edc34e28 --- /dev/null +++ b/src/main/java/io/github/dsheirer/dsp/symbol/DibitToByteBufferAssembler.java @@ -0,0 +1,125 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.symbol; + +import io.github.dsheirer.sample.Listener; +import io.github.dsheirer.sample.buffer.IReusableByteBufferProvider; +import io.github.dsheirer.sample.buffer.ReusableByteBuffer; +import io.github.dsheirer.sample.buffer.ReusableByteBufferQueue; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Assembles reusable byte buffers from an incoming stream of Dibits. + */ +public class DibitToByteBufferAssembler implements Listener, IReusableByteBufferProvider +{ + private final static Logger mLog = LoggerFactory.getLogger(DibitToByteBufferAssembler.class); + + private ReusableByteBufferQueue mBufferQueue = new ReusableByteBufferQueue("DibitToByteBufferAssembler"); + private ReusableByteBuffer mCurrentBuffer; + private int mBufferPointer; + private int mBufferSize; + private byte mCurrentByte; + private int mDibitCount; + private Listener mBufferListener; + + /** + * Constructs an assembler to produce reusable byte buffers of the specified size + * + * @param bufferSize + */ + public DibitToByteBufferAssembler(int bufferSize) + { + mBufferSize = bufferSize; + getNextBuffer(); + } + + /** + * Broadcasts the current buffer to the registered listener and creates a new buffer, resetting + * the buffer pointer to zero so that new dibits can be loaded. + */ + private void getNextBuffer() + { + if(mCurrentBuffer != null && mBufferListener != null) + { + mBufferListener.receive(mCurrentBuffer); + } + + mCurrentBuffer = mBufferQueue.getBuffer(mBufferSize); + mBufferPointer = 0; + } + + @Override + public void receive(Dibit dibit) + { + mCurrentByte <<= 2; + + switch(dibit) + { + case D01_PLUS_3: + mCurrentByte |= 0x01; + break; + case D00_PLUS_1: + //no-op - value is already 00 + break; + case D10_MINUS_1: + mCurrentByte |= 0x02; + break; + case D11_MINUS_3: + mCurrentByte |= 0x03; + break; + } + + mDibitCount++; + + if(mDibitCount >= 4) + { + mCurrentBuffer.getBytes()[mBufferPointer++] = mCurrentByte; + mCurrentByte = 0x00; + mDibitCount = 0; + + if(mBufferPointer >= mBufferSize) + { + getNextBuffer(); + } + } + } + + /** + * Registers the listener to receive fully assembled byte buffers from this assembler. + */ + @Override + public void setBufferListener(Listener listener) + { + mBufferListener = listener; + } + + /** + * Removes the listener from receiving buffers from this assembler + */ + @Override + public void removeBufferListener(Listener listener) + { + mBufferListener = null; + } + + @Override + public boolean hasBufferListeners() + { + return mBufferListener != null; + } +} diff --git a/src/main/java/io/github/dsheirer/module/decode/DecoderFactory.java b/src/main/java/io/github/dsheirer/module/decode/DecoderFactory.java index 0218aed5..7a53ce4d 100644 --- a/src/main/java/io/github/dsheirer/module/decode/DecoderFactory.java +++ b/src/main/java/io/github/dsheirer/module/decode/DecoderFactory.java @@ -476,9 +476,6 @@ public class DecoderFactory */ public static EnumSet getBitstreamDecoders() { -// return EnumSet.of(DecoderType.P25_PHASE1); - - //None of the decoders currently support bitstream output. - return EnumSet.noneOf(DecoderType.class); + return EnumSet.of(DecoderType.P25_PHASE1); } } \ No newline at end of file diff --git a/src/main/java/io/github/dsheirer/module/decode/p25/P25Decoder.java b/src/main/java/io/github/dsheirer/module/decode/p25/P25Decoder.java index 0fe9f4a1..95a6e592 100644 --- a/src/main/java/io/github/dsheirer/module/decode/p25/P25Decoder.java +++ b/src/main/java/io/github/dsheirer/module/decode/p25/P25Decoder.java @@ -16,19 +16,25 @@ package io.github.dsheirer.module.decode.p25; import io.github.dsheirer.alias.AliasList; +import io.github.dsheirer.dsp.symbol.DibitToByteBufferAssembler; import io.github.dsheirer.module.decode.Decoder; import io.github.dsheirer.module.decode.DecoderType; import io.github.dsheirer.sample.Listener; +import io.github.dsheirer.sample.buffer.IReusableByteBufferListener; +import io.github.dsheirer.sample.buffer.IReusableByteBufferProvider; import io.github.dsheirer.sample.buffer.IReusableComplexBufferListener; +import io.github.dsheirer.sample.buffer.ReusableByteBuffer; import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; import io.github.dsheirer.source.ISourceEventListener; import io.github.dsheirer.source.ISourceEventProvider; import io.github.dsheirer.source.SourceEvent; -public abstract class P25Decoder extends Decoder implements ISourceEventListener, ISourceEventProvider, - IReusableComplexBufferListener, Listener +public abstract class P25Decoder extends Decoder + implements ISourceEventListener, ISourceEventProvider, IReusableComplexBufferListener, Listener, + IReusableByteBufferProvider, IReusableByteBufferListener { private double mSampleRate; + private DibitToByteBufferAssembler mByteBufferAssembler = new DibitToByteBufferAssembler(300); private P25MessageProcessor mMessageProcessor; private AliasList mAliasList; private Listener mSourceEventListener; @@ -42,6 +48,41 @@ public abstract class P25Decoder extends Decoder implements ISourceEventListener mMessageProcessor.setMessageListener(getMessageListener()); } + /** + * Assembler for packaging Dibit stream into reusable byte buffers. + */ + protected DibitToByteBufferAssembler getByteBufferAssembler() + { + return mByteBufferAssembler; + } + + /** + * Implements the IByteBufferProvider interface - delegates to the byte buffer assembler + */ + @Override + public void setBufferListener(Listener listener) + { + getByteBufferAssembler().setBufferListener(listener); + } + + /** + * Implements the IByteBufferProvider interface - delegates to the byte buffer assembler + */ + @Override + public void removeBufferListener(Listener listener) + { + getByteBufferAssembler().removeBufferListener(listener); + } + + /** + * Implements the IByteBufferProvider interface - delegates to the byte buffer assembler + */ + @Override + public boolean hasBufferListeners() + { + return getByteBufferAssembler().hasBufferListeners(); + } + protected double getSymbolRate() { return mSymbolRate; diff --git a/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderC4FM.java b/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderC4FM.java index a67eeed5..c2623dac 100644 --- a/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderC4FM.java +++ b/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderC4FM.java @@ -25,6 +25,7 @@ import io.github.dsheirer.dsp.psk.DQPSKDecisionDirectedDemodulator; import io.github.dsheirer.dsp.psk.InterpolatingSampleBuffer; import io.github.dsheirer.dsp.psk.pll.AdaptivePLLGainMonitor; import io.github.dsheirer.dsp.psk.pll.CostasLoop; +import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; import io.github.dsheirer.source.SourceEvent; import org.slf4j.Logger; @@ -76,7 +77,15 @@ public class P25DecoderC4FM extends P25Decoder mMessageFramer = new P25MessageFramer(getAliasList(), mCostasLoop, mPLLGainMonitor); mMessageFramer.setListener(getMessageProcessor()); mMessageFramer.setSampleRate(sampleRate); - mQPSKDemodulator.setSymbolListener(mMessageFramer); + +// mQPSKDemodulator.setSymbolListener(mMessageFramer); + mQPSKDemodulator.setSymbolListener(getByteBufferAssembler()); + } + + @Override + public Listener getReusableByteBufferListener() + { + return mMessageFramer; } /** diff --git a/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderC4FMInstrumented.java b/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderC4FMInstrumented.java index c69a44e6..e7dbb7aa 100644 --- a/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderC4FMInstrumented.java +++ b/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderC4FMInstrumented.java @@ -78,7 +78,7 @@ public class P25DecoderC4FMInstrumented extends P25DecoderC4FM instrumented.setPLLFrequencyListener(mPLLFrequencyListener); instrumented.setSymbolDecisionDataListener(mSymbolDecisionDataListener); instrumented.setSamplesPerSymbolListener(mSamplesPerSymbolListener); - instrumented.setSymbolListener(mMessageFramer); + instrumented.setSymbolListener(getByteBufferAssembler()); } public void setComplexSymbolListener(Listener listener) diff --git a/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderLSM.java b/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderLSM.java index 4f2356dd..580aeb45 100644 --- a/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderLSM.java +++ b/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderLSM.java @@ -24,6 +24,7 @@ import io.github.dsheirer.dsp.psk.DQPSKGardnerDemodulator; import io.github.dsheirer.dsp.psk.InterpolatingSampleBuffer; import io.github.dsheirer.dsp.psk.pll.AdaptivePLLGainMonitor; import io.github.dsheirer.dsp.psk.pll.CostasLoop; +import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; import io.github.dsheirer.source.SourceEvent; @@ -78,7 +79,14 @@ public class P25DecoderLSM extends P25Decoder mMessageFramer = new P25MessageFramer(getAliasList(), mCostasLoop, mPLLGainMonitor); mMessageFramer.setListener(getMessageProcessor()); mMessageFramer.setSampleRate(sampleRate); - mQPSKDemodulator.setSymbolListener(mMessageFramer); +// mQPSKDemodulator.setSymbolListener(mMessageFramer); + mQPSKDemodulator.setSymbolListener(getByteBufferAssembler()); + } + + @Override + public Listener getReusableByteBufferListener() + { + return mMessageFramer; } /** diff --git a/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderLSMInstrumented.java b/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderLSMInstrumented.java index 7997ae9f..366851b4 100644 --- a/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderLSMInstrumented.java +++ b/src/main/java/io/github/dsheirer/module/decode/p25/P25DecoderLSMInstrumented.java @@ -79,7 +79,7 @@ public class P25DecoderLSMInstrumented extends P25DecoderLSM instrumented.setPLLFrequencyListener(mPLLFrequencyListener); instrumented.setSymbolDecisionDataListener(mSymbolDecisionDataListener); instrumented.setSamplesPerSymbolListener(mSamplesPerSymbolListener); - instrumented.setSymbolListener(mMessageFramer); + instrumented.setSymbolListener(getByteBufferAssembler()); } public void setComplexSymbolListener(Listener listener) diff --git a/src/main/java/io/github/dsheirer/module/decode/p25/P25MessageFramer.java b/src/main/java/io/github/dsheirer/module/decode/p25/P25MessageFramer.java index 4f6579b1..14db0e8c 100644 --- a/src/main/java/io/github/dsheirer/module/decode/p25/P25MessageFramer.java +++ b/src/main/java/io/github/dsheirer/module/decode/p25/P25MessageFramer.java @@ -32,13 +32,14 @@ import io.github.dsheirer.module.decode.p25.message.vselp.VSELP1Message; import io.github.dsheirer.module.decode.p25.message.vselp.VSELP2Message; import io.github.dsheirer.module.decode.p25.reference.DataUnitID; import io.github.dsheirer.sample.Listener; +import io.github.dsheirer.sample.buffer.ReusableByteBuffer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.List; -public class P25MessageFramer implements Listener +public class P25MessageFramer implements Listener { private final static Logger mLog = LoggerFactory.getLogger(P25MessageFramer.class); @@ -199,24 +200,30 @@ public class P25MessageFramer implements Listener } @Override - public void receive(Dibit symbol) + public void receive(ReusableByteBuffer reusableByteBuffer) { - for(P25MessageAssembler assembler : mAssemblers) - { - if(assembler.isActive()) - { - assembler.receive(symbol); - - if(assembler.complete()) - { - assembler.reset(); - } - } - } - - mMatcher.receive(symbol.getBit1(), symbol.getBit2()); + //process the buffer received } + // @Override +// public void receive(Dibit symbol) +// { +// for(P25MessageAssembler assembler : mAssemblers) +// { +// if(assembler.isActive()) +// { +// assembler.receive(symbol); +// +// if(assembler.complete()) +// { +// assembler.reset(); +// } +// } +// } +// +// mMatcher.receive(symbol.getBit1(), symbol.getBit2()); +// } + public void setListener(Listener listener) { mListener = listener; diff --git a/src/main/java/io/github/dsheirer/module/decode/p25/P25MessageProcessor.java b/src/main/java/io/github/dsheirer/module/decode/p25/P25MessageProcessor.java index 51d3978d..c6e5c6be 100644 --- a/src/main/java/io/github/dsheirer/module/decode/p25/P25MessageProcessor.java +++ b/src/main/java/io/github/dsheirer/module/decode/p25/P25MessageProcessor.java @@ -1,3 +1,18 @@ +/******************************************************************************* + * sdr-trunk + * Copyright (C) 2014-2018 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.module.decode.p25; import io.github.dsheirer.alias.AliasList; @@ -13,8 +28,7 @@ import java.util.HashMap; public class P25MessageProcessor implements Listener { - private final static Logger mLog = - LoggerFactory.getLogger( P25MessageProcessor.class ); + private final static Logger mLog = LoggerFactory.getLogger( P25MessageProcessor.class ); private Listener mMessageListener; diff --git a/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java b/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java index a1f000a7..697a85c5 100644 --- a/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java +++ b/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java @@ -45,6 +45,7 @@ import java.util.concurrent.atomic.AtomicBoolean; public class BinaryRecorder extends Module implements IReusableByteBufferListener { private final static Logger mLog = LoggerFactory.getLogger(BinaryRecorder.class); + private static final int MAX_RECORDING_BYTE_SIZE = 524288; //500 kB private ContinuousReusableBufferProcessor mBufferProcessor = new ContinuousReusableBufferProcessor<>(500, 50); @@ -53,6 +54,7 @@ public class BinaryRecorder extends Module implements IReusableByteBufferListene private Path mBaseRecordingPath; private String mRecordingIdentifier; private BinaryWriter mBinaryWriter = new BinaryWriter(); + private int mBytesRecordedCounter; /** * Constructs a binary recorder. @@ -115,7 +117,7 @@ public class BinaryRecorder extends Module implements IReusableByteBufferListene StringBuilder sb = new StringBuilder(); sb.append(TimeStamp.getTimeStamp("_")); sb.append("_"); - sb.append(mRecordingIdentifier); + sb.append(mRecordingIdentifier.trim()); sb.append(".bits"); return mBaseRecordingPath.resolve(sb.toString()); @@ -148,22 +150,62 @@ public class BinaryRecorder extends Module implements IReusableByteBufferListene public void start(Path path) throws IOException { - mCurrentPath = path; - mWritableByteChannel = Files.newByteChannel(path, - EnumSet.of(StandardOpenOption.CREATE, StandardOpenOption.WRITE)); + synchronized(this) + { + mCurrentPath = path; + mWritableByteChannel = Files.newByteChannel(path, + EnumSet.of(StandardOpenOption.CREATE, StandardOpenOption.WRITE)); + mLog.info("Binary (bitstream) recording started: " + mCurrentPath.toString()); + } } public void stop() throws IOException { - if(mWritableByteChannel != null) + synchronized(this) { - mWritableByteChannel.close(); - } + if(mWritableByteChannel != null) + { + mWritableByteChannel.close(); + } - mWritableByteChannel = null; - mCurrentPath = null; + mWritableByteChannel = null; + mCurrentPath = null; + } } + /** + * Closes the current recording and starts a new recording. This method is normally used + * when the current recording size has reached the maximum threshold. + */ + private void cycleRecording() + { + synchronized(this) + { + try + { + if(mWritableByteChannel != null) + { + mWritableByteChannel.close(); + } + + mCurrentPath = getRecordingPath(); + mWritableByteChannel = Files.newByteChannel(mCurrentPath, + EnumSet.of(StandardOpenOption.CREATE, StandardOpenOption.WRITE)); + mLog.info("Binary (bitstream) recording started: " + mCurrentPath.toString()); + } + catch(IOException ioe) + { + mLog.error("Error while cycling a max-size bit stream recorder", ioe); + } + + mBytesRecordedCounter = 0; + } + } + + /** + * Primary receive method for incoming byte buffers + * @param reusableComplexBuffers to record + */ @Override public void receive(List reusableComplexBuffers) { @@ -174,7 +216,12 @@ public class BinaryRecorder extends Module implements IReusableByteBufferListene try { ByteBuffer byteBuffer = ByteBuffer.wrap(buffer.getSamplesCopy()); - mWritableByteChannel.write(byteBuffer); + mBytesRecordedCounter += mWritableByteChannel.write(byteBuffer); + + if(mBytesRecordedCounter > MAX_RECORDING_BYTE_SIZE) + { + cycleRecording(); + } } catch(IOException ioe) {