/*******************************************************************************
 * sdrtrunk
 * Copyright (C) 2014-2017 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 source.tuner;

import dsp.filter.FilterFactory;
import dsp.filter.Window.WindowType;
import dsp.filter.cic.ComplexPrimeCICDecimate;
import dsp.mixer.Oscillator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sample.Buffer;
import sample.Listener;
import sample.OverflowableTransferQueue;
import sample.complex.Complex;
import sample.complex.ComplexBuffer;
import sample.real.IOverflowListener;
import source.ComplexSource;
import source.SourceException;
import source.tuner.frequency.FrequencyChangeEvent;
import source.tuner.frequency.FrequencyChangeEvent.Event;
import source.tuner.frequency.IFrequencyChangeProcessor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class TunerChannelSource extends ComplexSource implements IFrequencyChangeProcessor, Listener<ComplexBuffer>
{
    private final static Logger mLog = LoggerFactory.getLogger(TunerChannelSource.class);

    //Maximum number of filled buffers for the blocking queue
    private static final int BUFFER_MAX_CAPACITY = 300;

    //Threshold for resetting buffer overflow condition
    private static final int BUFFER_OVERFLOW_RESET_THRESHOLD = 100;

    private static int CHANNEL_RATE = 48000;
    private static int CHANNEL_PASS_FREQUENCY = 12000;

    private OverflowableTransferQueue<ComplexBuffer> mBuffer;

    private Tuner mTuner;
    private TunerChannel mTunerChannel;
    private Oscillator mMixer;
    private ComplexPrimeCICDecimate mDecimationFilter;
    private Listener<ComplexBuffer> mListener;
    private IFrequencyChangeProcessor mFrequencyChangeProcessor;
    private ScheduledFuture<?> mTaskHandle;

    private long mTunerFrequency = 0;
    private int mTunerSampleRate;
    private int mChannelFrequencyCorrection = 0;

    private DecimationProcessor mDecimationProcessor = new DecimationProcessor();

    private AtomicBoolean mRunning = new AtomicBoolean();
    private boolean mExpended = false;
    private int mCount;

    /**
     * Provides a Digital Drop Channel (DDC) to decimate the IQ output from a
     * tuner down to a 48 kHz IQ channel rate.
     *
     * Note: this class can only be used once (started and stopped) and a new
     * tuner channel source must be requested from the tuner once this object
     * has been stopped.  This is because channels are managed dynamically and
     * center tuned frequency may have changed since this source was obtained
     * and thus the tuner might no longer be able to source this channel once it
     * has been stopped.
     *
     * @param tuner to obtain wideband IQ samples from
     * @param tunerChannel specifying the center frequency for the DDC
     * @throws RejectedExecutionException if the thread pool manager cannot
     *                                    accept the decimation processing task
     * @throws SourceException            if the tuner has an issue providing IQ samples
     */
    public TunerChannelSource(Tuner tuner, TunerChannel tunerChannel) throws RejectedExecutionException, SourceException
    {
        mTuner = tuner;
        mTunerChannel = tunerChannel;
        mTuner.getTunerController().addListener((IFrequencyChangeProcessor) this);
        mTunerFrequency = mTuner.getTunerController().getFrequency();

        mBuffer = new OverflowableTransferQueue<>(BUFFER_MAX_CAPACITY, BUFFER_OVERFLOW_RESET_THRESHOLD);

	    /* Setup the frequency translator to the current source frequency */
        long frequencyOffset = mTunerFrequency - mTunerChannel.getFrequency();

        mMixer = new Oscillator(frequencyOffset, mTuner.getTunerController().getSampleRate());

		/* Fire a sample rate change event to setup the decimation chain */
        frequencyChanged(new FrequencyChangeEvent(Event.NOTIFICATION_SAMPLE_RATE_CHANGE, mTuner.getTunerController().getSampleRate()));
    }

    /**
     * Overrides the default source overflow listener management to delegate responsibility to the overflow buffer
     */
    @Override
    public void setOverflowListener(IOverflowListener listener)
    {
        mBuffer.setOverflowListener(listener);
    }

    public void start(ScheduledExecutorService executor)
    {
        if(mExpended)
        {
            throw new IllegalStateException("Attempt to re-start an expended tuner channel source.  TunerChannelSource" +
                " objects can only be used once. ");
        }

        if(mRunning.compareAndSet(false, true))
        {
            //Schedule the decimation task to run every 9 ms (111 iterations/second), an odd periodicity relative
            //to the inbound periodicity of 20 ms, to attempt to avoid thread queue contention
            mTaskHandle = executor.scheduleAtFixedRate(mDecimationProcessor, 0, 9, TimeUnit.MILLISECONDS);

		    /* Finally, register to receive samples from the tuner */
            mTuner.addListener((Listener<ComplexBuffer>) this);
        }
        else
        {
            mLog.warn("Attempt to start() an already running tuner channel source was ignored");
        }
    }

    @Override
    public void reset()
    {
    }

    @Override
    public void stop()
    {
        if(mRunning.compareAndSet(true, false))
        {
            mTuner.releaseChannel(this);
            mDecimationProcessor.shutdown();

            if(mTaskHandle != null)
            {
                mTaskHandle.cancel(true);
                mTaskHandle = null;
            }

            mBuffer.clear();

            mExpended = true;
        }
        else
        {
            mLog.warn("Attempt to stop() an already stopped tuner channel source was ignored");
        }
    }


    @Override
    public void dispose()
    {
        if(!mRunning.get())
        {
	    	/* Tell the tuner to release/unregister our resources */
            mTuner.getTunerController().removeListener(this);
        }
    }

    public Tuner getTuner()
    {
        return mTuner;
    }

    public TunerChannel getTunerChannel()
    {
        return mTunerChannel;
    }

    @Override
    public void receive(ComplexBuffer buffer)
    {
        if(mRunning.get())
        {
            mBuffer.offer(buffer);
        }
    }

    public void setFrequencyChangeListener(IFrequencyChangeProcessor processor)
    {
        mFrequencyChangeProcessor = processor;
    }

    @Override
    public void setListener(Listener<ComplexBuffer> listener)
    {
		/* Save a pointer to the listener so that if we have to change the 
		 * decimation filter, we can re-add the listener */
        mListener = listener;

        mDecimationFilter.setListener(listener);
    }

    @Override
    public void removeListener(Listener<ComplexBuffer> listener)
    {
        mDecimationFilter.removeListener();
    }

    /**
     * Handler for frequency change events received from the tuner and channel
     * frequency correction events received from the channel consumer/listener
     */
    @Override
    public void frequencyChanged(FrequencyChangeEvent event) throws SourceException
    {
        // Echo the event to the registered event listener
        if(mFrequencyChangeProcessor != null)
        {
            mFrequencyChangeProcessor.frequencyChanged(event);
        }

        switch(event.getEvent())
        {
            case NOTIFICATION_FREQUENCY_CHANGE:
                mTunerFrequency = event.getValue().longValue();
                updateMixerFrequencyOffset();
                break;
            case REQUEST_CHANNEL_FREQUENCY_CORRECTION_CHANGE:
                mChannelFrequencyCorrection = event.getValue().intValue();

                updateMixerFrequencyOffset();

                if(mFrequencyChangeProcessor != null)
                {
                    mFrequencyChangeProcessor.frequencyChanged(new FrequencyChangeEvent(
                        Event.NOTIFICATION_CHANNEL_FREQUENCY_CORRECTION_CHANGE, mChannelFrequencyCorrection));
                }
                break;
            case NOTIFICATION_SAMPLE_RATE_CHANGE:
                int sampleRate = event.getValue().intValue();

                if(mTunerSampleRate != sampleRate)
                {
                    mMixer.setSampleRate(sampleRate);

					/* Get new decimation filter */
                    mDecimationFilter = FilterFactory.getDecimationFilter(sampleRate, CHANNEL_RATE, 1,
                        CHANNEL_PASS_FREQUENCY, 60, WindowType.HAMMING);
					
					/* re-add the original output listener */
                    mDecimationFilter.setListener(mListener);

                    mTunerSampleRate = sampleRate;
                }
                break;
            default:
                break;
        }
    }

    /**
     * Calculates the local mixer frequency offset from the tuned frequency,
     * channel's requested frequency, and channel frequency correction.
     */
    private void updateMixerFrequencyOffset()
    {
        long offset = mTunerFrequency -
            mTunerChannel.getFrequency() -
            mChannelFrequencyCorrection;

        mMixer.setFrequency(offset);
    }

    public int getSampleRate() throws SourceException
    {
        return CHANNEL_RATE;
    }

    public long getFrequency() throws SourceException
    {
        return mTunerChannel.getFrequency();
    }

    /**
     * Decimates an inbound buffer of I/Q samples from the source down to the
     * standard 48000 channel sample rate
     */
    public class DecimationProcessor implements Runnable
    {
        private boolean mProcessing = true;
        private List<ComplexBuffer> mSampleBuffers = new ArrayList<ComplexBuffer>();

        public void shutdown()
        {
            mProcessing = false;
        }

        @Override
        public void run()
        {
			/* General exception handler so that any errors won't kill the
			 * decimation thread and cause the input buffers to fill up and
			 * run the program out of memory */
            try
            {
                if(mProcessing)
                {
                    mBuffer.drainTo(mSampleBuffers, 20);

                    for(Buffer buffer : mSampleBuffers)
                    {
							/* Check to see if we've been shutdown */
                        if(!mProcessing)
                        {
                            mBuffer.clear();
                            return;
                        }
                        else
                        {
                            float[] samples = buffer.getSamples();

								/* We make a copy of the buffer so that we don't affect
								 * anyone else that is using the same buffer, like other
								 * channels or the spectral display */
                            float[] translated = new float[samples.length];

								/* Perform frequency translation */
                            for(int x = 0; x < samples.length; x += 2)
                            {
                                mMixer.rotate();

                                translated[x] = Complex.multiplyInphase(
                                    samples[x], samples[x + 1], mMixer.inphase(), mMixer.quadrature());

                                translated[x + 1] = Complex.multiplyQuadrature(
                                    samples[x], samples[x + 1], mMixer.inphase(), mMixer.quadrature());
                            }

                            if(mProcessing)
                            {
                                final ComplexPrimeCICDecimate filter = mDecimationFilter;
                                filter.receive(new ComplexBuffer(translated));
                            }
                        }
                    }

                    mSampleBuffers.clear();
                }
            }
            catch(Exception e)
            {
				/* Only log the stack trace if we're still processing */
                if(mProcessing)
                {
                    mLog.error("Error encountered during decimation process", e);
                }
            }
            catch(Throwable throwable)
            {
                mLog.error("Code error encountered during decimation process - channel thread will probably die", throwable);
            }

			/* Check to see if we've been shutdown */
            if(!mProcessing)
            {
                mBuffer.clear();
                mSampleBuffers.clear();
            }
        }
    }
}