<filename>tests/unit/media/test_silent_player.py
"""
Tests for the silent audio driver.
"""
from __future__ import division
from builtins import object

from tests import mock
import unittest

from pyglet.media.drivers.silent import (EventBuffer, SilentAudioBuffer, SilentAudioPacket,
        SilentAudioPlayerPacketConsumer)
from pyglet.media.events import MediaEvent
from pyglet.media.sources import AudioData, AudioFormat

class SilentAudioPacketTest(unittest.TestCase):
    def test_partial_consume(self):
        packet = SilentAudioPacket(0., 1.)

        dt = .4
        consumed = packet.consume(dt)

        self.assertAlmostEqual(.4, consumed)
        self.assertAlmostEqual(.4, packet.timestamp)
        self.assertAlmostEqual(.6, packet.duration)
        self.assertFalse(packet.is_empty())

    def test_exact_consume(self):
        packet = SilentAudioPacket(0., 1.)

        dt = 1.
        consumed = packet.consume(dt)

        self.assertAlmostEqual(1., consumed)
        self.assertAlmostEqual(1., packet.timestamp)
        self.assertAlmostEqual(0., packet.duration)
        self.assertTrue(packet.is_empty())

    def test_over_consume(self):
        packet = SilentAudioPacket(0., 1.)

        dt = 2.
        consumed = packet.consume(dt)

        self.assertAlmostEqual(1., consumed)
        self.assertAlmostEqual(1., packet.timestamp)
        self.assertAlmostEqual(0., packet.duration)
        self.assertTrue(packet.is_empty())


class SilentAudioBufferTest(unittest.TestCase):
    def test_add_audio_data(self):
        buf = SilentAudioBuffer()
        self.assertTrue(buf.is_empty())
        self.assertAlmostEqual(0., buf.duration)

        data1 = AudioData('', 0, 0., 1., [])
        buf.add_audio_data(data1)
        self.assertFalse(buf.is_empty())
        self.assertAlmostEqual(1., buf.duration)

        data2 = AudioData('', 0, 1., 2., [])
        buf.add_audio_data(data2)
        self.assertFalse(buf.is_empty())
        self.assertAlmostEqual(3., buf.duration)

    def test_consume_audio_data(self):
        buf = SilentAudioBuffer()
        buf.add_audio_data(AudioData('', 0, 0., 1., []))
        buf.add_audio_data(AudioData('', 0, 1., 2., []))
        self.assertFalse(buf.is_empty())
        self.assertAlmostEqual(3., buf.duration)
        self.assertAlmostEqual(0., buf.get_current_timestamp())

        buf.consume_audio_data(0.8)
        self.assertFalse(buf.is_empty())
        self.assertAlmostEqual(2.2, buf.duration)
        self.assertAlmostEqual(0.8, buf.get_current_timestamp())

        buf.consume_audio_data(0.8)
        self.assertFalse(buf.is_empty())
        self.assertAlmostEqual(1.4, buf.duration)
        self.assertAlmostEqual(1.6, buf.get_current_timestamp())

        buf.consume_audio_data(1.4)
        self.assertTrue(buf.is_empty())
        self.assertAlmostEqual(0., buf.duration)
        self.assertAlmostEqual(3., buf.get_current_timestamp())

    def test_consume_too_much(self):
        buf = SilentAudioBuffer()
        buf.add_audio_data(AudioData('', 0, 0., 1., []))
        buf.add_audio_data(AudioData('', 0, 1., 2., []))

        buf.consume_audio_data(4.)
        self.assertTrue(buf.is_empty())
        self.assertAlmostEqual(0., buf.duration)
        self.assertAlmostEqual(3., buf.get_current_timestamp())

    def test_time_to_next_update(self):
        buf = SilentAudioBuffer()
        self.assertIsNone(buf.get_time_to_next_update())

        buf.add_audio_data(AudioData('', 0, 0., 1., []))
        buf.add_audio_data(AudioData('', 0, 1., 2., []))
        self.assertAlmostEqual(1., buf.get_time_to_next_update())

        buf.consume_audio_data(0.5)
        self.assertAlmostEqual(0.5, buf.get_time_to_next_update())

        buf.consume_audio_data(1.0)
        self.assertAlmostEqual(1.5, buf.get_time_to_next_update())

        buf.consume_audio_data(1.5)
        self.assertIsNone(buf.get_time_to_next_update())

    def test_current_timestamp(self):
        buf = SilentAudioBuffer()
        self.assertAlmostEqual(0., buf.get_current_timestamp())

        buf.add_audio_data(AudioData('', 0, 2., 1., []))
        buf.add_audio_data(AudioData('', 0, 1., 2., []))
        self.assertAlmostEqual(2., buf.get_current_timestamp())

        buf.consume_audio_data(0.2)
        self.assertAlmostEqual(2.2, buf.get_current_timestamp())

        buf.consume_audio_data(1.)
        self.assertAlmostEqual(1.2, buf.get_current_timestamp())

        buf.consume_audio_data(2.)
        self.assertAlmostEqual(3., buf.get_current_timestamp())


class EventBufferTest(unittest.TestCase):
    def test_add_events(self):
        buf = EventBuffer()
        self.assertIsNone(buf.get_next_event_timestamp())

        event1 = MediaEvent(.1, 'Event1')
        event2 = MediaEvent(.5, 'Event2')
        data = AudioData('', 0, 0., 1., [event1, event2])
        buf.add_events(data)

        self.assertAlmostEqual(.1, buf.get_next_event_timestamp())

    def test_get_expired_events(self):
        buf = EventBuffer()
        self.assertIsNone(buf.get_next_event_timestamp())

        event1 = MediaEvent(.1, 'Event1')
        event2 = MediaEvent(.5, 'Event2')
        data = AudioData('', 0, 0., 1., [event1, event2])
        buf.add_events(data)

        expired_events = buf.get_expired_events(0.)
        self.assertListEqual([], expired_events)

        expired_events = buf.get_expired_events(.1)
        self.assertListEqual([event1], expired_events)

        expired_events = buf.get_expired_events(.1)
        self.assertListEqual([], expired_events)

        expired_events = buf.get_expired_events(.6)
        self.assertListEqual([event2], expired_events)

        expired_events = buf.get_expired_events(.6)
        self.assertListEqual([], expired_events)

    def test_get_multiple_events(self):
        buf = EventBuffer()
        self.assertIsNone(buf.get_next_event_timestamp())

        event1 = MediaEvent(.2, 'Event1')
        event2 = MediaEvent(.2, 'Event2')
        data1 = AudioData('', 0, 0., 1., [event1, event2])
        buf.add_events(data1)

        event3 = MediaEvent(.3, 'Event3')
        event4 = MediaEvent(.4, 'Event4')
        data2 = AudioData('', 0, 1., 1., [event3, event4])
        buf.add_events(data2)

        expired_events = buf.get_expired_events(0.)
        self.assertListEqual([], expired_events)

        expired_events = buf.get_expired_events(.2)
        self.assertListEqual([event1, event2], expired_events)

        expired_events = buf.get_expired_events(1.6)
        self.assertListEqual([event3, event4], expired_events)

    def test_get_next_event_timestamp(self):
        buf = EventBuffer()
        self.assertIsNone(buf.get_next_event_timestamp())

        event1 = MediaEvent(.2, 'Event1')
        event2 = MediaEvent(.2, 'Event2')
        data1 = AudioData('', 0, 0., 1., [event1, event2])
        buf.add_events(data1)

        event3 = MediaEvent(.3, 'Event3')
        event4 = MediaEvent(.4, 'Event4')
        data2 = AudioData('', 0, 1., 1., [event3, event4])
        buf.add_events(data2)

        self.assertAlmostEqual(.2, buf.get_next_event_timestamp())

        buf.get_expired_events(.2)
        self.assertAlmostEqual(1.3, buf.get_next_event_timestamp())

        buf.get_expired_events(1.3)
        self.assertAlmostEqual(1.4, buf.get_next_event_timestamp())

        buf.get_expired_events(1.4)
        self.assertIsNone(buf.get_next_event_timestamp())

    def test_get_time_to_next_event(self):
        buf = EventBuffer()
        self.assertIsNone(buf.get_next_event_timestamp())

        event1 = MediaEvent(.2, 'Event1')
        event2 = MediaEvent(.2, 'Event2')
        data1 = AudioData('', 0, 0., 1., [event1, event2])
        buf.add_events(data1)

        event3 = MediaEvent(.3, 'Event3')
        event4 = MediaEvent(.4, 'Event4')
        data2 = AudioData('', 0, 1., 1., [event3, event4])
        buf.add_events(data2)

        self.assertAlmostEqual(.2, buf.get_time_to_next_event(0.))
        self.assertAlmostEqual(.1, buf.get_time_to_next_event(.1))

        buf.get_expired_events(.2)
        self.assertAlmostEqual(1.1, buf.get_time_to_next_event(.2))
        self.assertAlmostEqual(.1, buf.get_time_to_next_event(1.2))


class MockSourceGroup(object):
    audio_format = AudioFormat(1, 8, 44100)

    def __init__(self, duration, timestamp=0.):
        self.mock = mock.MagicMock()
        type(self.mock).audio_format = mock.PropertyMock(return_value=self.audio_format)
        self.mock.get_audio_data.side_effect = self._get_audio_data

        self.timestamp = timestamp
        self.duration = duration

        self.seconds_buffered = 0.
        self.bytes_buffered = 0

    def _get_audio_data(self, length):
        secs = length / self.audio_format.bytes_per_second
        if secs > self.duration:
            secs = self.duration
            length = int(secs * self.audio_format.bytes_per_second)

        if length == 0:
            return None

        data = AudioData('a'*length, length, self.timestamp, secs, ())
        self.timestamp += secs
        self.duration -= secs
        self.seconds_buffered += secs
        self.bytes_buffered += length
        return data


class SilentAudioPlayerPacketConsumerTest(unittest.TestCase):
    def setUp(self):
        self.time_patcher = mock.patch('time.time')
        self.thread_patcher = mock.patch('pyglet.media.drivers.silent.MediaThread')
        self.mock_time = self.time_patcher.start()
        self.mock_thread = self.thread_patcher.start()

    def tearDown(self):
        self.time_patcher.stop()
        self.thread_patcher.stop()

    def set_time(self, t):
        self.mock_time.return_value = t

    def test_buffer_data_initial(self):
        mock_player = mock.MagicMock()
        mock_source_group = MockSourceGroup(1.)

        silent_player = SilentAudioPlayerPacketConsumer(mock_source_group.mock, mock_player)
        self.set_time(1000.)
        silent_player._buffer_data()
        self.assertAlmostEqual(.4, mock_source_group.seconds_buffered, delta=.01)

        self.assertAlmostEqual(0., silent_player.get_time(), delta=.01)

    def test_playing(self):
        mock_player = mock.MagicMock()
        mock_source_group = MockSourceGroup(1.)

        silent_player = SilentAudioPlayerPacketConsumer(mock_source_group.mock, mock_player)

        # Buffer initial data
        self.set_time(1000.)
        silent_player._buffer_data()
        self.assertAlmostEqual(.4, mock_source_group.seconds_buffered, delta=.01)

        # Start playing
        silent_player.play()
        self.assertAlmostEqual(0., silent_player.get_time(), delta=.01)

        # Check timestamp increases even when not consuming new data
        self.set_time(1000.2)
        self.assertAlmostEqual(.2, silent_player.get_time(), delta=.01)

        # Timestamp sill correct after consuming data
        silent_player._consume_data()
        self.assertAlmostEqual(.2, silent_player.get_time(), delta=.01)

        # Consuming data means we need to buffer more
        silent_player._buffer_data()
        self.assertAlmostEqual(.6, mock_source_group.seconds_buffered, delta=.01)
        self.assertAlmostEqual(.2, silent_player.get_time(), delta=.01)

    def test_not_started_yet(self):
        mock_player = mock.MagicMock()
        mock_source_group = MockSourceGroup(1.)
        silent_player = SilentAudioPlayerPacketConsumer(mock_source_group.mock, mock_player)

        # Do initial buffering even when not playing yet
        self.set_time(1000.)
        silent_player._buffer_data()
        self.assertAlmostEqual(.4, mock_source_group.seconds_buffered, delta=.01)
        self.assertAlmostEqual(0., silent_player.get_time(), delta=.01)

        # Increase of timestamp does not change anything
        self.set_time(1001.)
        self.assertAlmostEqual(0., silent_player.get_time(), delta=.01)

        # No data is consumed
        silent_player._consume_data()
        self.assertAlmostEqual(0., silent_player.get_time(), delta=.01)

        # No new data is buffered
        silent_player._buffer_data()
        self.assertAlmostEqual(.4, mock_source_group.seconds_buffered, delta=.01)
        self.assertAlmostEqual(0., silent_player.get_time(), delta=.01)

    def test_play_and_stop(self):
        mock_player = mock.MagicMock()
        mock_source_group = MockSourceGroup(1.)
        silent_player = SilentAudioPlayerPacketConsumer(mock_source_group.mock, mock_player)

        # Do initial buffering even when not playing yet
        self.set_time(1000.)
        silent_player._buffer_data()
        self.assertAlmostEqual(.4, mock_source_group.seconds_buffered, delta=.01)
        self.assertAlmostEqual(0., silent_player.get_time(), delta=.01)

        # Play a little bit
        silent_player.play()
        self.set_time(1000.2)
        silent_player._consume_data()
        silent_player._buffer_data()
        self.assertAlmostEqual(.6, mock_source_group.seconds_buffered, delta=.01)
        self.assertAlmostEqual(.2, silent_player.get_time(), delta=.01)

        # Now stop, this should consume data upto stopping moment
        self.set_time(1000.4)
        silent_player.stop()
        self.assertAlmostEqual(.4, silent_player.get_time(), delta=.01)

        # Buffering still happens
        silent_player._buffer_data()
        self.assertAlmostEqual(.8, mock_source_group.seconds_buffered, delta=.01)
        self.assertAlmostEqual(.4, silent_player.get_time(), delta=.01)

        # But now playback is really paused
        self.set_time(1001.)
        self.assertAlmostEqual(.4, silent_player.get_time(), delta=.01)

        # And no more buffering and consuming
        silent_player._consume_data()
        silent_player._buffer_data()
        self.assertAlmostEqual(.8, mock_source_group.seconds_buffered, delta=.01)
        self.assertAlmostEqual(.4, silent_player.get_time(), delta=.01)

