#438 Adds feature for duplicate call audio detection and suppression for playback, recording and streaming. (#835)

Co-authored-by: Denny <denny@denny-desktop>

src/main/java/io/github/dsheirer/audio/AbstractAudioModule.java

@@ -81,6 +81,7 @@ public abstract class AbstractAudioModule extends Module implements IAudioSegmen
             {
                 mAudioSegment.completeProperty().set(true);
                 mIdentifierUpdateNotificationBroadcaster.removeListener(mAudioSegment);
+                mAudioSegment.decrementConsumerCount();
                 mAudioSegment = null;
             }
         }
@@ -103,6 +104,7 @@ public abstract class AbstractAudioModule extends Module implements IAudioSegmen
             if(mAudioSegment == null)
             {
                 mAudioSegment = new AudioSegment(mAliasList, getTimeslot());
+                mAudioSegment.incrementConsumerCount();
                 mAudioSegment.addIdentifiers(mIdentifierCollection.getIdentifiers());
                 mIdentifierUpdateNotificationBroadcaster.addListener(mAudioSegment);
 

src/main/java/io/github/dsheirer/audio/AudioSegment.java

@@ -65,6 +65,7 @@ public class AudioSegment implements Listener<IdentifierUpdateNotification>
 {
     private final static Logger mLog = LoggerFactory.getLogger(AudioSegment.class);
     private BooleanProperty mComplete = new SimpleBooleanProperty(false);
+    private BooleanProperty mDuplicate = new SimpleBooleanProperty(false);
     private BooleanProperty mRecordAudio = new SimpleBooleanProperty(false);
     private IntegerProperty mMonitorPriority = new SimpleIntegerProperty(Priority.DEFAULT_PRIORITY);
     private ObservableSet<BroadcastChannel> mBroadcastChannels = FXCollections.observableSet(new HashSet<>());
@@ -74,6 +75,7 @@ public class AudioSegment implements Listener<IdentifierUpdateNotification>
     private AtomicInteger mConsumerCount = new AtomicInteger();
     private AliasList mAliasList;
     private long mStartTimestamp = System.currentTimeMillis();
+    private long mSampleCount = 0;
     private boolean mDisposing = false;
     private AudioSegment mLinkedAudioSegment;
     private int mTimeslot;
@@ -107,6 +109,24 @@ public class AudioSegment implements Listener<IdentifierUpdateNotification>
         return mStartTimestamp;
     }
 
+    /**
+     * End timestamp as calculated from start timestamp and current sample count
+     */
+    public long getEndTimestamp()
+    {
+        return mStartTimestamp + getDuration();
+    }
+
+    /**
+     * Current duration of this audio segment.  Note: this is a dynamic value until the complete property is set to true.
+     *
+     * @return duration in milliseconds
+     */
+    public long getDuration()
+    {
+        return (mSampleCount / 8); //8 kHz audio generates 8 samples per millisecond
+    }
+
     /**
      * The complete property is used by the audio segment producer to signal that the segment is complete and no
      * additional audio or identifiers will be added to the segment.
@@ -118,6 +138,15 @@ public class AudioSegment implements Listener<IdentifierUpdateNotification>
         return mComplete;
     }
 
+    /**
+     * Duplicate call audio property.  This flag is set to true whenever a duplicate call detection function detects
+     * an audio segment is a duplicate.
+     */
+    public BooleanProperty duplicateProperty()
+    {
+        return mDuplicate;
+    }
+
     /**
      * An observable list of broadcast channels for this audio segment.  Broadcast channels are added to this segment
      * across the life-cycle of the segment.  As each new alias identifier is added to the segment, any broadcast
@@ -347,6 +376,7 @@ public class AudioSegment implements Listener<IdentifierUpdateNotification>
         }
 
         mAudioBuffers.add(audioBuffer);
+        mSampleCount += audioBuffer.length;
     }
 
     /**
@@ -396,6 +426,23 @@ public class AudioSegment implements Listener<IdentifierUpdateNotification>
         }
     }
 
+    /**
+     * Indicates if this audio segment has been flagged as a duplicate audio call
+     */
+    public boolean isDuplicate()
+    {
+        return mDuplicate.get();
+    }
+
+    /**
+     * Sets the duplicate audio call flag for this audio segment
+     * @param duplicate true if this is a duplicate audio segment
+     */
+    public void setDuplicate(boolean duplicate)
+    {
+        mDuplicate.set(duplicate);
+    }
+
     /**
      * Processes identifier update notifications and updates audio segment properties using the associated aliases
      * from the alias list.

src/main/java/io/github/dsheirer/audio/DuplicateCallDetector.java

@@ -0,0 +1,310 @@
+/*
+ * *****************************************************************************
+ *  Copyright (C) 2014-2020 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.audio;
+
+import io.github.dsheirer.identifier.Form;
+import io.github.dsheirer.identifier.Identifier;
+import io.github.dsheirer.identifier.IdentifierClass;
+import io.github.dsheirer.identifier.Role;
+import io.github.dsheirer.identifier.configuration.SystemConfigurationIdentifier;
+import io.github.dsheirer.identifier.patch.PatchGroupIdentifier;
+import io.github.dsheirer.identifier.radio.RadioIdentifier;
+import io.github.dsheirer.identifier.talkgroup.TalkgroupIdentifier;
+import io.github.dsheirer.preference.UserPreferences;
+import io.github.dsheirer.preference.duplicate.DuplicateCallDetectionPreference;
+import io.github.dsheirer.sample.Listener;
+import io.github.dsheirer.util.ThreadPool;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.LinkedTransferQueue;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Detects duplicate calls that occur within the same system.  This detector is thread safe for the receive() method.
+ *
+ * Note: system in this context refers to the system name value that is used in channel configurations.  All decoder
+ * channels must share the same system name for call duplication detection.
+ */
+public class DuplicateCallDetector implements Listener<AudioSegment>
+{
+    private final static Logger mLog = LoggerFactory.getLogger(DuplicateCallDetector.class);
+    private DuplicateCallDetectionPreference mDuplicateCallDetectionPreference;
+    private Map<String,SystemDuplicateCallDetector> mDetectorMap = new HashMap();
+
+    public DuplicateCallDetector(UserPreferences userPreferences)
+    {
+        mDuplicateCallDetectionPreference = userPreferences.getDuplicateCallDetectionPreference();
+    }
+
+    @Override
+    public void receive(AudioSegment audioSegment)
+    {
+        if(mDuplicateCallDetectionPreference.isDuplicateCallDetectionEnabled())
+        {
+            Identifier identifier = audioSegment.getIdentifierCollection()
+                .getIdentifier(IdentifierClass.CONFIGURATION, Form.SYSTEM, Role.ANY);
+
+            if(identifier instanceof SystemConfigurationIdentifier)
+            {
+                String system = ((SystemConfigurationIdentifier)identifier).getValue();
+
+                synchronized(mDetectorMap)
+                {
+                    SystemDuplicateCallDetector detector = mDetectorMap.get(system);
+
+                    if(detector == null)
+                    {
+                        detector = new SystemDuplicateCallDetector();
+                        mDetectorMap.put(system, detector);
+                    }
+
+                    detector.add(audioSegment);
+                }
+            }
+        }
+    }
+
+    public class SystemDuplicateCallDetector
+    {
+        private LinkedTransferQueue<AudioSegment> mAudioSegmentQueue = new LinkedTransferQueue<>();
+        private List<AudioSegment> mAudioSegments = new ArrayList<>();
+        private AtomicBoolean mMonitoring = new AtomicBoolean();
+        private ScheduledFuture<?> mProcessorFuture;
+
+        public SystemDuplicateCallDetector()
+        {
+        }
+
+        public void add(AudioSegment audioSegment)
+        {
+            //Block on audio segment queue so that we don't interfere with monitoring shutdown
+            synchronized(mAudioSegmentQueue)
+            {
+                mAudioSegmentQueue.add(audioSegment);
+                startMonitoring();
+            }
+        }
+
+        private void startMonitoring()
+        {
+            if(mMonitoring.compareAndSet(false, true))
+            {
+                mProcessorFuture = ThreadPool.SCHEDULED.scheduleAtFixedRate(() -> process(),
+                    0, 25, TimeUnit.MILLISECONDS);
+            }
+        }
+
+        private void stopMonitoring()
+        {
+            if(mMonitoring.compareAndSet(true, false))
+            {
+                if(mProcessorFuture != null)
+                {
+                    mProcessorFuture.cancel(true);
+                }
+            }
+        }
+
+        /**
+         * Checks if either the TO or FROM integer identifiers for the audio segments are the same
+         * @param segment1 to compare
+         * @param segment2 to compare
+         * @return true if integer identifiers are the same
+         */
+        private boolean isDuplicate(AudioSegment segment1, AudioSegment segment2)
+        {
+            if(mDuplicateCallDetectionPreference.isDuplicateCallDetectionByTalkgroupEnabled())
+            {
+                //Step 1 check for duplicate TO values
+                List<Identifier> to1 = segment1.getIdentifierCollection().getIdentifiers(Role.TO);
+                List<Identifier> to2 = segment2.getIdentifierCollection().getIdentifiers(Role.TO);
+
+                if(isDuplicate(to1, to2))
+                {
+                    return true;
+                }
+            }
+
+            if(mDuplicateCallDetectionPreference.isDuplicateCallDetectionByRadioEnabled())
+            {
+                //Step 2 check for duplicate FROM values
+                List<Identifier> from1 = segment1.getIdentifierCollection().getIdentifiers(Role.FROM);
+                List<Identifier> from2 = segment2.getIdentifierCollection().getIdentifiers(Role.FROM);
+
+                return isDuplicate(from1, from2);
+            }
+
+            return false;
+        }
+
+        /**
+         * Checks both lists of identifiers to determine if there are talkgroups or radio identifiers that are
+         * the same in both lists.  Note: talkgroup check compares both talkgroups and patch groups.
+         *
+         * @param identifiers1
+         * @param identifiers2
+         * @return
+         */
+        private boolean isDuplicate(List<Identifier> identifiers1, List<Identifier> identifiers2)
+        {
+            for(Identifier identifier1: identifiers1)
+            {
+                if(identifier1 instanceof TalkgroupIdentifier)
+                {
+                    int talkgroup1 = ((TalkgroupIdentifier)identifier1).getValue();
+
+                    for(Identifier identifier2: identifiers2)
+                    {
+                        if(identifier2 instanceof TalkgroupIdentifier &&
+                           ((TalkgroupIdentifier)identifier2).getValue() == talkgroup1)
+                        {
+                            return true;
+                        }
+                        else if(identifier2 instanceof PatchGroupIdentifier &&
+                            ((PatchGroupIdentifier)identifier2).getValue().getPatchGroup().getValue() == talkgroup1)
+                        {
+                            return true;
+                        }
+                    }
+                }
+                else if(identifier1 instanceof PatchGroupIdentifier)
+                {
+                    int talkgroup1 = ((PatchGroupIdentifier)identifier1).getValue().getPatchGroup().getValue();
+
+                    for(Identifier identifier2: identifiers2)
+                    {
+                        if(identifier2 instanceof TalkgroupIdentifier &&
+                            ((TalkgroupIdentifier)identifier2).getValue() == talkgroup1)
+                        {
+                            return true;
+                        }
+                        else if(identifier2 instanceof PatchGroupIdentifier &&
+                            ((PatchGroupIdentifier)identifier2).getValue().getPatchGroup().getValue() == talkgroup1)
+                        {
+                            return true;
+                        }
+                    }
+                }
+                else if(identifier1 instanceof RadioIdentifier)
+                {
+                    int radio1 = ((RadioIdentifier)identifier1).getValue();
+
+                    for(Identifier identifier2: identifiers2)
+                    {
+                        if(identifier2 instanceof RadioIdentifier && ((RadioIdentifier)identifier2).getValue() == radio1)
+                        {
+                            return true;
+                        }
+                    }
+                }
+            }
+
+            return false;
+        }
+
+        /**
+         * Processes audio segments to detect duplicates
+         */
+        private void process()
+        {
+            //Transfer in newly arrived audio segments
+            mAudioSegmentQueue.drainTo(mAudioSegments);
+
+            //Remove any completed audio segments.
+            mAudioSegments.removeIf(audioSegment -> {
+                boolean complete = audioSegment.completeProperty().get();
+
+                if(complete)
+                {
+                    audioSegment.decrementConsumerCount();
+                }
+
+                return complete;
+            });
+
+            //Only check for duplicates if there is more than one call
+            if(mAudioSegments.size() > 1)
+            {
+                List<AudioSegment> duplicates = new ArrayList<>();
+
+                int currentIndex = 0;
+                while(currentIndex < mAudioSegments.size() - 1)
+                {
+                    AudioSegment current = mAudioSegments.get(currentIndex);
+
+                    if(!current.isDuplicate())
+                    {
+                        int checkIndex = currentIndex + 1;
+
+                        while(checkIndex < mAudioSegments.size())
+                        {
+                            AudioSegment toCheck = mAudioSegments.get(checkIndex);
+
+                            if(!toCheck.isDuplicate())
+                            {
+                                if(isDuplicate(current, toCheck))
+                                {
+                                    toCheck.setDuplicate(true);
+                                    toCheck.decrementConsumerCount();
+                                    duplicates.add(toCheck);
+                                }
+                            }
+
+                            checkIndex++;
+                        }
+                    }
+
+                    currentIndex++;
+                }
+
+                mAudioSegments.removeAll(duplicates);
+            }
+
+            //Finally, if the audio segment queue is empty, shutdown montitoring until a new segment arrives
+            if(mAudioSegments.isEmpty())
+            {
+                //Block on the audio segment queue so that we can shutdown before any new segments are added, and
+                //allow the add(segment) to restart monitoring as soon as needed.
+                synchronized(mAudioSegmentQueue)
+                {
+                    if(mAudioSegmentQueue.isEmpty())
+                    {
+                        try
+                        {
+                            stopMonitoring();
+                        }
+                        catch(Exception e)
+                        {
+                            mLog.error("Unexpected error during duplicate audio segment monitoring shutdown", e);
+                            //Do nothing, we got interrupted
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

src/main/java/io/github/dsheirer/audio/broadcast/AudioStreamingManager.java

@@ -127,7 +127,12 @@ public class AudioStreamingManager implements Listener<AudioSegment>
         {
             audioSegment = it.next();
 
-            if(audioSegment.completeProperty().get())
+            if(audioSegment.isDuplicate() && mUserPreferences.getDuplicateCallDetectionPreference().isDuplicateStreamingSuppressionEnabled())
+            {
+                it.remove();
+                audioSegment.decrementConsumerCount();
+            }
+            else if(audioSegment.completeProperty().get())
             {
                 it.remove();
 

src/main/java/io/github/dsheirer/audio/playback/AudioOutput.java

@@ -381,8 +381,9 @@ public abstract class AudioOutput implements LineListener, Listener<IdentifierUp
 
             try
             {
-                //Remove an assigned audio segment that's subsequently flagged as do not monitor before playback
-                if(mNextAudioSegment.isDoNotMonitor())
+                //Remove an assigned segment that's subsequently flagged as duplicate or do not monitor before playback
+                if(mNextAudioSegment.isDoNotMonitor() || (mNextAudioSegment.isDuplicate() &&
+                    mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled()))
                 {
                     mNextAudioSegment = null;
 
@@ -456,9 +457,11 @@ public abstract class AudioOutput implements LineListener, Listener<IdentifierUp
 
         if(mCurrentAudioSegment != null)
         {
-            //Check for completed audio segment or a segment flagged as Do Not Monitor
-            if(mCurrentAudioSegment.isDoNotMonitor() || (mCurrentAudioSegment.completeProperty().get() &&
-               mCurrentBufferIndex >= mCurrentAudioSegment.getAudioBufferCount()))
+            //Check for completed audio segment or a segment flagged as duplicate or Do Not Monitor
+            if(mCurrentAudioSegment.isDoNotMonitor() ||
+               (mCurrentAudioSegment.isDuplicate() &&
+                mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled()) ||
+                (mCurrentAudioSegment.completeProperty().get() && mCurrentBufferIndex >= mCurrentAudioSegment.getAudioBufferCount()))
             {
                 if(mCurrentAudioSegment.isDoNotMonitor())
                 {
@@ -490,7 +493,8 @@ public abstract class AudioOutput implements LineListener, Listener<IdentifierUp
             //segment is a linked segment, ignore it so that we can close out the current segment.
             while(mCurrentAudioSegment != null && (mNextAudioSegment == null || mNextAudioSegment.isLinked()) &&
                    mCurrentBufferIndex < mCurrentAudioSegment.getAudioBufferCount() &&
-                   !mCurrentAudioSegment.isDoNotMonitor())
+                   !mCurrentAudioSegment.isDoNotMonitor() && !(mCurrentAudioSegment.isDuplicate() &&
+                mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled()))
             {
                 float[] audioBuffer = mCurrentAudioSegment.getAudioBuffers().get(mCurrentBufferIndex++);
 

src/main/java/io/github/dsheirer/audio/playback/AudioPlaybackManager.java

@@ -27,20 +27,15 @@ import io.github.dsheirer.audio.IAudioController;
 import io.github.dsheirer.eventbus.MyEventBus;
 import io.github.dsheirer.preference.PreferenceType;
 import io.github.dsheirer.preference.UserPreferences;
-import io.github.dsheirer.properties.SystemProperties;
 import io.github.dsheirer.sample.Broadcaster;
 import io.github.dsheirer.sample.Listener;
 import io.github.dsheirer.source.mixer.MixerChannel;
 import io.github.dsheirer.source.mixer.MixerChannelConfiguration;
-import io.github.dsheirer.source.mixer.MixerManager;
 import io.github.dsheirer.util.ThreadPool;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.sound.sampled.AudioSystem;
-import javax.sound.sampled.Mixer;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.Comparator;
 import java.util.Iterator;
 import java.util.List;
@@ -123,7 +118,12 @@ public class AudioPlaybackManager implements Listener<AudioSegment>, IAudioContr
 
         while(newSegment != null)
         {
-            if(newSegment.hasAudio())
+            if(newSegment.isDuplicate() &&
+               mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled())
+            {
+                newSegment.decrementConsumerCount();
+            }
+            else if(newSegment.hasAudio())
             {
                 mAudioSegments.add(newSegment);
             }
@@ -146,7 +146,13 @@ public class AudioPlaybackManager implements Listener<AudioSegment>, IAudioContr
             {
                 audioSegment = it.next();
 
-                if(audioSegment.hasAudio())
+                if(audioSegment.isDuplicate() &&
+                   mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled())
+                {
+                    it.remove();
+                    audioSegment.decrementConsumerCount();
+                }
+                else if(audioSegment.hasAudio())
                 {
                     //Queue it up for replay
                     it.remove();
@@ -174,7 +180,8 @@ public class AudioPlaybackManager implements Listener<AudioSegment>, IAudioContr
             {
                 audioSegment = it.next();
 
-                if(audioSegment.isDoNotMonitor())
+                if(audioSegment.isDoNotMonitor() || (audioSegment.isDuplicate() &&
+                   mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled()))
                 {
                     it.remove();
                     audioSegment.decrementConsumerCount();
@@ -235,7 +242,8 @@ public class AudioPlaybackManager implements Listener<AudioSegment>, IAudioContr
             {
                 audioSegment = it.next();
 
-                if(audioSegment.completeProperty().get())
+                if(audioSegment.completeProperty().get() || (audioSegment.isDuplicate() &&
+                   mUserPreferences.getDuplicateCallDetectionPreference().isDuplicatePlaybackSuppressionEnabled()))
                 {
                     it.remove();
                     audioSegment.decrementConsumerCount();